Merge branch 'refs/heads/release/v0.2.8' into fix/features_028

api/app/controllers/public_share_controller.py

@@ -663,6 +663,7 @@ async def config_query(
         content = {
             "app_type": release.app.type,
             "variables": workflow_service.get_start_node_variables(release.config),
+            "memory":  workflow_service.is_memory_enable(release.config),
             "features": release.config.get("features")
         }
     elif release.app.type == AppType.AGENT:

api/app/core/workflow/engine/stream_output_coordinator.py

@@ -5,7 +5,7 @@
 import re
 from typing import AsyncGenerator
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, PrivateAttr
 
 from app.core.logging_config import get_logger
 from app.core.workflow.engine.variable_pool import VariablePool
@@ -52,10 +52,11 @@ class OutputContent(BaseModel):
         )
     )
 
-    _SCOPE: str | None = None
+    _SCOPE: str | None = PrivateAttr(default=None)
 
-    def get_scope(self) -> str:
-        self._SCOPE = SCOPE_PATTERN.findall(self.literal)[0]
+    def get_scope(self) -> str | None:
+        matches = SCOPE_PATTERN.findall(self.literal)
+        self._SCOPE = matches[0] if matches else None
         return self._SCOPE
 
     def depends_on_scope(self, scope: str) -> bool:
@@ -68,6 +69,8 @@ class OutputContent(BaseModel):
         Returns:
             bool: True if this segment references the given scope.
         """
+        if not self.is_variable:
+            return False
         if self._SCOPE:
             return self._SCOPE == scope
         return self.get_scope() == scope
@@ -152,7 +155,7 @@ class StreamOutputConfig(BaseModel):
         """
 
         # Case 1: resolve control branch dependency
-        if scope in self.control_nodes.keys():
+        if scope in self.control_nodes:
             if status is None:
                 raise RuntimeError("[Stream Output] Control node activation status not provided")
             if status in self.control_nodes[scope]:

api/app/models/workflow_model.py

@@ -35,6 +35,7 @@ class WorkflowConfig(Base):
     
     # 执行配置
     execution_config = Column(JSONB, nullable=False, default=dict)
+    features = Column(JSONB, nullable=True, default=dict)
     
     # 触发器配置（可选）
     triggers = Column(JSONB, default=list)

api/app/schemas/workflow_schema.py

@@ -80,6 +80,7 @@ class WorkflowConfigCreate(BaseModel):
     variables: list[VariableDefinition] = Field(default_factory=list, description="变量列表")
     execution_config: ExecutionConfig = Field(default_factory=ExecutionConfig, description="执行配置")
     triggers: list[TriggerConfig] = Field(default_factory=list, description="触发器列表")
+    features: dict = Field(default_factory=dict, description="功能特性配置")
 
 
 class WorkflowConfigUpdate(BaseModel):
@@ -87,6 +88,7 @@ class WorkflowConfigUpdate(BaseModel):
     nodes: list[NodeDefinition] | None = None
     edges: list[EdgeDefinition] | None = None
     variables: list[VariableDefinition] | None = None
+    features: dict | None = None
     execution_config: ExecutionConfig | None = None
     triggers: list[TriggerConfig] | None = None
 
@@ -102,6 +104,7 @@ class WorkflowConfig(BaseModel):
     variables: list[dict[str, Any]]
     execution_config: dict[str, Any]
     triggers: list[dict[str, Any]]
+    features: dict | None
     is_active: bool
     created_at: datetime.datetime
     updated_at: datetime.datetime
@@ -114,6 +117,10 @@ class WorkflowConfig(BaseModel):
     def _serialize_updated_at(self, dt: datetime.datetime):
         return int(dt.timestamp() * 1000) if dt else None
 
+    @field_serializer("features", when_used="json")
+    def _serialize_features(self, features: dict | None):
+        return features or {}
+
 
 # ==================== 工作流执行 ====================
 

api/app/services/app_service.py

@@ -1609,6 +1609,7 @@ class AppService:
                 variables=[var.model_dump() for var in data.variables] if data.variables else [],
                 execution_config=data.execution_config.model_dump() if data.execution_config else {},
                 triggers=[trigger.model_dump() for trigger in data.triggers] if data.triggers else [],
+                features=data.features or {},
                 is_active=True,
                 created_at=now,
                 updated_at=now
@@ -1622,6 +1623,7 @@ class AppService:
             workflow_cfg.variables = [var.model_dump() for var in data.variables] if data.variables else []
             workflow_cfg.execution_config = data.execution_config.model_dump() if data.execution_config else {}
             workflow_cfg.triggers = [trigger.model_dump() for trigger in data.triggers] if data.triggers else []
+            workflow_cfg.features = data.features or {}
             workflow_cfg.updated_at = now
 
         self.db.commit()
@@ -1875,7 +1877,8 @@ class AppService:
                 "edges": workflow_cfg.edges,
                 "variables": workflow_cfg.variables,
                 "execution_config": workflow_cfg.execution_config,
-                "triggers": workflow_cfg.triggers
+                "triggers": workflow_cfg.triggers,
+                "features": workflow_cfg.features or {}
             }
 
             is_valid, errors = WorkflowValidator.validate_for_publish(config)

api/app/services/memory_perceptual_service.py

@@ -5,12 +5,14 @@ from urllib.parse import urlparse, unquote
 
 import json_repair
 from jinja2 import Template
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.core.error_codes import BizCode
 from app.core.exceptions import BusinessException
 from app.core.logging_config import get_business_logger
 from app.core.models import RedBearLLM, RedBearModelConfig
+from app.models import FileMetadata
 from app.models.memory_perceptual_model import PerceptualType, FileStorageService
 from app.models.prompt_optimizer_model import RoleType
 from app.repositories.memory_perceptual_repository import MemoryPerceptualRepository
@@ -245,6 +247,18 @@ class MemoryPerceptualService:
         filename = os.path.basename(path)
         filename = unquote(filename)
         file_ext = os.path.splitext(filename)[1]
+        try:
+            file_id = uuid.UUID(filename)
+            stmt = select(FileMetadata).where(
+                FileMetadata.id == file_id
+            )
+            file = self.db.execute(stmt).scalar_one_or_none()
+
+            if file:
+                filename = file.file_name
+                file_ext = file.file_ext
+        except ValueError:
+            business_logger.debug(f"Remote file, file_id={filename}")
         if not file_ext:
             if file_type == FileType.AUDIO:
                 file_ext = ".mp3"
@@ -262,17 +276,17 @@ class MemoryPerceptualService:
         }
         if file_type in [FileType.IMAGE, FileType.VIDEO]:
             file_modalities = {
-                "scene": content.get("scene")
+                "scene": content.get("scene", [])
             }
         elif file_type in [FileType.DOCUMENT]:
             file_modalities = {
-                "section_count": content.get("section_count"),
-                "title": content.get("title"),
-                "first_line": content.get("first_line")
+                "section_count": content.get("section_count", 0),
+                "title": content.get("title", ""),
+                "first_line": content.get("first_line", "")
             }
         else:
             file_modalities = {
-                "speaker_count": content.get("speaker_count")
+                "speaker_count": content.get("speaker_count", 0)
             }
         self.repository.create_perceptual_memory(
             end_user_id=uuid.UUID(end_user_id),
@@ -280,7 +294,7 @@ class MemoryPerceptualService:
             file_path=file_url,
             file_name=filename,
             file_ext=file_ext,
-            summary=content.get('summary'),
+            summary=content.get('summary', ""),
             meta_data={
                 "content": file_content,
                 "modalities": file_modalities

api/app/services/multimodal_service.py

@@ -59,22 +59,22 @@ class MultimodalFormatStrategy(ABC):
         self.file = file
 
     @abstractmethod
-    async def format_image(self, url: str, content: bytes | None = None) -> Dict[str, Any]:
+    async def format_image(self, url: str, content: bytes | None = None) -> tuple[bool, Dict[str, Any]]:
         """格式化图片"""
         pass
 
     @abstractmethod
-    async def format_document(self, file_name: str, text: str) -> Dict[str, Any]:
+    async def format_document(self, file_name: str, text: str) -> tuple[bool, Dict[str, Any]]:
         """格式化文档"""
         pass
 
     @abstractmethod
-    async def format_audio(self, file_type: str, url: str, content: bytes | None = None) -> Dict[str, Any]:
+    async def format_audio(self, file_type: str, url: str, content: bytes | None = None) -> tuple[bool, Dict[str, Any]]:
         """格式化音频"""
         pass
 
     @abstractmethod
-    async def format_video(self, url: str) -> Dict[str, Any]:
+    async def format_video(self, url: str) -> tuple[bool, Dict[str, Any]]:
         """格式化视频"""
         pass
 
@@ -82,16 +82,16 @@ class MultimodalFormatStrategy(ABC):
 class DashScopeFormatStrategy(MultimodalFormatStrategy):
     """通义千问策略"""
 
-    async def format_image(self, url: str, content: bytes | None = None) -> Dict[str, Any]:
+    async def format_image(self, url: str, content: bytes | None = None) -> tuple[bool, Dict[str, Any]]:
         """通义千问图片格式：{"type": "image", "image": "url"}"""
-        return {
+        return True, {
             "type": "image",
             "image": url
         }
 
-    async def format_document(self, file_name: str, text: str) -> Dict[str, Any]:
+    async def format_document(self, file_name: str, text: str) -> tuple[bool, Dict[str, Any]]:
         """通义千问文档格式"""
-        return {
+        return True, {
             "type": "text",
             "text": f"<document name=\"{file_name}\">\n{text}\n</document>"
         }
@@ -102,26 +102,26 @@ class DashScopeFormatStrategy(MultimodalFormatStrategy):
             url: str,
             content: bytes | None = None,
             transcription: Optional[str] = None
-    ) -> Dict[str, Any]:
+    ) -> tuple[bool, Dict[str, Any]]:
         """
         通义千问音频格式
         - 原生支持: qwen-audio 系列
         - 其他模型: 需要转录为文本
         """
         if transcription:
-            return {
+            return True, {
                 "type": "text",
                 "text": f"<audio url=\"{url}\">\ntext_transcription:{transcription}\n</audio>"
             }
         # 通义千问音频格式：{"type": "audio", "audio": "url"}
-        return {
+        return True, {
             "type": "audio",
             "audio": url
         }
 
-    async def format_video(self, url: str) -> Dict[str, Any]:
+    async def format_video(self, url: str) -> tuple[bool, Dict[str, Any]]:
         """通义千问视频格式（qwen-vl 系列原生支持）"""
-        return {
+        return True, {
             "type": "video",
             "video": url
         }
@@ -130,7 +130,7 @@ class DashScopeFormatStrategy(MultimodalFormatStrategy):
 class BedrockFormatStrategy(MultimodalFormatStrategy):
     """Bedrock/Anthropic 策略"""
 
-    async def format_image(self, url: str, content: bytes | None = None) -> Dict[str, Any]:
+    async def format_image(self, url: str, content: bytes | None = None) -> tuple[bool, Dict[str, Any]]:
         """
         Bedrock/Anthropic 格式: base64 编码
         {"type": "image", "source": {"type": "base64", "media_type": "...", "data": "..."}}
@@ -153,7 +153,7 @@ class BedrockFormatStrategy(MultimodalFormatStrategy):
 
         logger.info(f"图片编码完成: media_type={media_type}, size={len(base64_data)}")
 
-        return {
+        return True, {
             "type": "image",
             "source": {
                 "type": "base64",
@@ -162,13 +162,13 @@ class BedrockFormatStrategy(MultimodalFormatStrategy):
             }
         }
 
-    async def format_document(self, file_name: str, text: str) -> Dict[str, Any]:
+    async def format_document(self, file_name: str, text: str) -> tuple[bool, Dict[str, Any]]:
         """Bedrock/Anthropic 文档格式（需要 base64 编码）"""
         # Bedrock 文档需要 base64 编码
         text_bytes = text.encode('utf-8')
         base64_text = base64.b64encode(text_bytes).decode('utf-8')
 
-        return {
+        return True, {
             "type": "document",
             "source": {
                 "type": "base64",
@@ -182,24 +182,24 @@ class BedrockFormatStrategy(MultimodalFormatStrategy):
             url: str,
             content: bytes | None = None,
             transcription: Optional[str] = None
-    ) -> Dict[str, Any]:
+    ) -> tuple[bool, Dict[str, Any]]:
         """
         Bedrock/Anthropic 音频格式
         不支持原生音频，必须转录为文本
         """
         if transcription:
-            return {
+            return True, {
                 "type": "text",
                 "text": f"[音频转录]\n{transcription}"
             }
-        return {
+        return False, {
             "type": "text",
             "text": "[音频文件：Bedrock 不支持原生音频，请启用音频转文本功能]"
         }
 
-    async def format_video(self, url: str) -> Dict[str, Any]:
+    async def format_video(self, url: str) -> tuple[bool, Dict[str, Any]]:
         """Bedrock/Anthropic 视频格式"""
-        return {
+        return False, {
             "type": "text",
             "text": f"<video url=\"{url}\">\n[视频文件，当前 provider 暂不支持]\n</video>"
         }
@@ -208,18 +208,18 @@ class BedrockFormatStrategy(MultimodalFormatStrategy):
 class OpenAIFormatStrategy(MultimodalFormatStrategy):
     """OpenAI 策略"""
 
-    async def format_image(self, url: str, content: bytes | None = None) -> Dict[str, Any]:
+    async def format_image(self, url: str, content: bytes | None = None) -> tuple[bool, Dict[str, Any]]:
         """OpenAI 格式: {"type": "image_url", "image_url": {"url": "..."}}"""
-        return {
+        return True, {
             "type": "image_url",
             "image_url": {
                 "url": url
             }
         }
 
-    async def format_document(self, file_name: str, text: str) -> Dict[str, Any]:
+    async def format_document(self, file_name: str, text: str) -> tuple[bool, Dict[str, Any]]:
         """OpenAI 文档格式"""
-        return {
+        return True, {
             "type": "text",
             "text": f"<document name=\"{file_name}\">\n{text}\n</document>"
         }
@@ -230,14 +230,14 @@ class OpenAIFormatStrategy(MultimodalFormatStrategy):
             url: str,
             content: bytes | None = None,
             transcription: Optional[str] = None
-    ) -> Dict[str, Any]:
+    ) -> tuple[bool, Dict[str, Any]]:
         """
         OpenAI 音频格式
         - gpt-4o-audio 系列支持原生音频（需要 base64 编码）
         - 其他模型使用转录文本
         """
         if transcription:
-            return {
+            return True, {
                 "type": "text",
                 "text": f"<audio url=\"{url}\">\n{transcription}\n</audio>"
             }
@@ -266,7 +266,7 @@ class OpenAIFormatStrategy(MultimodalFormatStrategy):
             # supported_ext = {"wav", "mp3", "mp4", "ogg", "flac", "webm", "m4a", "wave", "x-m4a"}
             file_ext = "wav" if not file_ext else file_ext
 
-            return {
+            return True, {
                 "type": "input_audio",
                 "input_audio": {
                     "data": f"data:;base64,{base64_audio}",
@@ -275,14 +275,14 @@ class OpenAIFormatStrategy(MultimodalFormatStrategy):
             }
         except Exception as e:
             logger.error(f"下载音频失败: {e}")
-            return {
+            return False, {
                 "type": "text",
                 "text": f"[音频处理失败: {str(e)}]"
             }
 
-    async def format_video(self, url: str) -> Dict[str, Any]:
+    async def format_video(self, url: str) -> tuple[bool, Dict[str, Any]]:
         """OpenAI 视频格式"""
-        return {
+        return True, {
             "type": "video_url",
             "video_url": {
                 "url": url
@@ -377,21 +377,25 @@ class MultimodalService:
                 file.url = await self.get_file_url(file)
             try:
                 if file.type == FileType.IMAGE and "vision" in self.capability:
-                    content = await self._process_image(file, strategy)
+                    is_support, content = await self._process_image(file, strategy)
                     result.append(content)
-                    self.write_perceptual_memory(end_user_id, file.type, file.url, content)
+                    if is_support:
+                        self.write_perceptual_memory(end_user_id, file.type, file.url, content)
                 elif file.type == FileType.DOCUMENT:
-                    content = await self._process_document(file, strategy)
+                    is_support, content = await self._process_document(file, strategy)
                     result.append(content)
-                    self.write_perceptual_memory(end_user_id, file.type, file.url, content)
+                    if is_support:
+                        self.write_perceptual_memory(end_user_id, file.type, file.url, content)
                 elif file.type == FileType.AUDIO and "audio" in self.capability:
-                    content = await self._process_audio(file, strategy)
+                    is_support, content = await self._process_audio(file, strategy)
                     result.append(content)
-                    self.write_perceptual_memory(end_user_id, file.type, file.url, content)
+                    if is_support:
+                        self.write_perceptual_memory(end_user_id, file.type, file.url, content)
                 elif file.type == FileType.VIDEO and "video" in self.capability:
-                    content = await self._process_video(file, strategy)
+                    is_support, content = await self._process_video(file, strategy)
                     result.append(content)
-                    self.write_perceptual_memory(end_user_id, file.type, file.url, content)
+                    if is_support:
+                        self.write_perceptual_memory(end_user_id, file.type, file.url, content)
                 else:
                     logger.warning(f"不支持的文件类型: {file.type}")
             except Exception as e:
@@ -424,7 +428,7 @@ class MultimodalService:
         if end_user_id and self.api_config:
             write_perceptual_memory.delay(end_user_id, self.api_config.model_dump(), file_type, file_url, file_message)
 
-    async def _process_image(self, file: FileInput, strategy) -> Dict[str, Any]:
+    async def _process_image(self, file: FileInput, strategy) -> tuple[bool, Dict[str, Any]]:
         """
         处理图片文件
         
@@ -440,12 +444,12 @@ class MultimodalService:
             return await strategy.format_image(file.url, content=file.get_content())
         except Exception as e:
             logger.error(f"处理图片失败: {e}", exc_info=True)
-            return {
+            return False, {
                 "type": "text",
                 "text": f"[图片处理失败: {str(e)}]"
             }
 
-    async def _process_document(self, file: FileInput, strategy) -> Dict[str, Any]:
+    async def _process_document(self, file: FileInput, strategy) -> tuple[bool, Dict[str, Any]]:
         """
         处理文档文件（PDF、Word 等）
         
@@ -457,7 +461,7 @@ class MultimodalService:
             Dict: 根据 provider 返回不同格式的文档内容
         """
         if file.transfer_method == TransferMethod.REMOTE_URL:
-            return {
+            return True, {
                 "type": "text",
                 "text": f"<document url=\"{file.url}\">\n{await self._extract_document_text(file)}\n</document>"
             }
@@ -475,7 +479,7 @@ class MultimodalService:
             # 使用策略格式化文档
             return await strategy.format_document(file_name, text)
 
-    async def _process_audio(self, file: FileInput, strategy) -> Dict[str, Any]:
+    async def _process_audio(self, file: FileInput, strategy) -> tuple[bool, Dict[str, Any]]:
         """
         处理音频文件
         
@@ -503,12 +507,12 @@ class MultimodalService:
             return await strategy.format_audio(file.file_type, file.url, file.get_content(), transcription)
         except Exception as e:
             logger.error(f"处理音频失败: {e}", exc_info=True)
-            return {
+            return False, {
                 "type": "text",
                 "text": f"[音频处理失败: {str(e)}]"
             }
 
-    async def _process_video(self, file: FileInput, strategy) -> Dict[str, Any]:
+    async def _process_video(self, file: FileInput, strategy) -> tuple[bool, Dict[str, Any]]:
         """
         处理视频文件
         
@@ -524,7 +528,7 @@ class MultimodalService:
             return await strategy.format_video(file.url)
         except Exception as e:
             logger.error(f"处理视频失败: {e}", exc_info=True)
-            return {
+            return False, {
                 "type": "text",
                 "text": f"[视频处理失败: {str(e)}]"
             }

api/app/services/workflow_service.py

@@ -570,6 +570,9 @@ class WorkflowService:
                 message=f"工作流配置不存在: app_id={app_id}"
             )
 
+        feature_configs = config.features or {}
+        self._validate_file_upload(feature_configs, payload.files)
+
         input_data = {
             "message": payload.message, "variables": payload.variables,
             "conversation_id": payload.conversation_id,
@@ -737,6 +740,8 @@ class WorkflowService:
                 code=BizCode.CONFIG_MISSING,
                 message=f"工作流配置不存在: app_id={app_id}"
             )
+        feature_configs = config.features or {}
+        self._validate_file_upload(feature_configs, payload.files)
 
         input_data = {
             "message": payload.message, "variables": payload.variables,
@@ -845,7 +850,10 @@ class WorkflowService:
                     yield event
 
         except Exception as e:
-            logger.error(f"工作流流式执行失败: execution_id={execution.execution_id}, error={e}", exc_info=True)
+            logger.error(
+                f"Workflow streaming execution failed: execution_id={execution.execution_id}, error={e}",
+                exc_info=True
+            )
             self.update_execution_status(
                 execution.execution_id,
                 "failed",
@@ -868,6 +876,80 @@ class WorkflowService:
                 return node.get("config", {}).get("variables", [])
         raise BusinessException("workflow config error - start node not found")
 
+    @staticmethod
+    def is_memory_enable(config: dict) -> bool:
+        nodes = config.get("nodes", [])
+        for node in nodes:
+            if node.get("type") in [NodeType.MEMORY_READ, NodeType.MEMORY_WRITE]:
+                return True
+        return False
+
+    @staticmethod
+    def _validate_file_upload(
+            features_config: dict[str, Any],
+            files: Optional[list[FileInput]]
+    ) -> None:
+        """校验上传文件是否符合 file_upload 配置"""
+        if not files:
+            return
+        fu = features_config.get("file_upload")
+        if fu is None:
+            return
+        if not (isinstance(fu, dict) and fu.get("enabled")):
+            raise BusinessException(
+                "The application does not have file upload functionality enabled",
+                BizCode.BAD_REQUEST
+            )
+        max_count = fu.get("max_file_count", 5)
+        if len(files) > max_count:
+            raise BusinessException(
+                f"File count exceeds limit (maximum {max_count} files)",
+                BizCode.BAD_REQUEST
+            )
+
+        # 校验传输方式
+        allowed_methods = fu.get("allowed_transfer_methods", ["local_file", "remote_url"])
+        for f in files:
+            if f.transfer_method.value not in allowed_methods:
+                raise BusinessException(
+                    f"Unsupport file transfer method：{f.transfer_method.value},"
+                    f"allowed method:{', '.join(allowed_methods)}",
+                    BizCode.BAD_REQUEST
+                )
+
+        # 各类型对应的开关和大小限制配置键
+        type_cfg = {
+            "image": ("image_enabled", "image_max_size_mb", 20, "image"),
+            "audio": ("audio_enabled", "audio_max_size_mb", 50, "audio"),
+            "document": ("document_enabled", "document_max_size_mb", 100, "document"),
+            "video": ("video_enabled", "video_max_size_mb", 500, "video"),
+        }
+
+        for f in files:
+            ftype = str(f.type)  # 如 "image", "audio", "document", "video"
+            cfg = type_cfg.get(ftype)
+            if cfg is None:
+                continue
+            enabled_key, size_key, default_max_mb, label = cfg
+
+            # 校验类型开关
+            if not fu.get(enabled_key):
+                raise BusinessException(
+                    f"The application has not enabled {label} file upload",
+                    BizCode.BAD_REQUEST
+                )
+
+            # 校验文件大小（仅当内容已加载时）
+            content = f.get_content()
+            if content is not None:
+                max_mb = fu.get(size_key, default_max_mb)
+                size_mb = len(content) / (1024 * 1024)
+                if size_mb > max_mb:
+                    raise BusinessException(
+                        f"{label} File size exceeds the limit (maximum {max_mb} MB, current {size_mb:.1f} MB)",
+                        BizCode.BAD_REQUEST
+                    )
+
 
 # ==================== 依赖注入函数 ====================