Merge branch 'release/v0.2.6' into develop

* release/v0.2.6:
  fix(web): ontology class default tag bugfix
  fix(version): Version 0.2.6 Release Notes
  fix(web): chat file delete bugfix
  feat: support model load balancing and add message_id to API responses
  feat: support model load balancing and add message_id to API responses
  [changes] Work space isolation
  [add] Recently, memory activities have adopted Redis caching.
  [changes] Work space isolation
  [add] Recently, memory activities have adopted Redis caching.
  fix(web): upload add loading
  [changes] The enumeration check has been changed to a string.
  [changes] The enumeration check has been changed to a string.
  feat(web): http-request add headers variable
  fix(workflow): ensure file messages are written to messages in non-stream mode
  fix(workflow): fix Dify compatibility issues
  [changes] Memory write completion active failure interest cache
  feat(workflow): support multimodal context
  [changes] AI review and correction of code
  [add] Semantic pruning is unified with the ontology engineering scenario.
  feat(chat): add message_id field to chat API response

api/app/services/app_chat_service.py

@@ -144,7 +144,7 @@ class AppChatService:
         )
 
         # 保存消息
-        self.conversation_service.save_conversation_messages(
+        message_id = self.conversation_service.save_conversation_messages(
             conversation_id=conversation_id,
             user_message=message,
             assistant_message=result["content"],
@@ -163,6 +163,7 @@ class AppChatService:
 
         return {
             "conversation_id": conversation_id,
+            "message_id": str(message_id),
             "message": result["content"],
             "usage": result.get("usage", {
                 "prompt_tokens": 0,
@@ -191,7 +192,11 @@ class AppChatService:
         try:
             start_time = time.time()
             config_id = None
-            yield f"event: start\ndata: {json.dumps({'conversation_id': str(conversation_id)}, ensure_ascii=False)}\n\n"
+            message_id = uuid.uuid4()
+            yield f"event: start\ndata: {json.dumps({
+                'conversation_id': str(conversation_id), 
+                "message_id": str(message_id)
+            }, ensure_ascii=False)}\n\n"
 
             variables = self.agent_service.prepare_variables(variables, config.variables)
             # 获取模型配置ID
@@ -296,6 +301,7 @@ class AppChatService:
             )
 
             self.conversation_service.add_message(
+                message_id=message_id,
                 conversation_id=conversation_id,
                 role="assistant",
                 content=full_content,
@@ -373,7 +379,7 @@ class AppChatService:
             content=message
         )
 
-        self.conversation_service.add_message(
+        ai_message = self.conversation_service.add_message(
             conversation_id=conversation_id,
             role="assistant",
             content=result.get("message", ""),
@@ -391,6 +397,7 @@ class AppChatService:
         return {
             "conversation_id": conversation_id,
             "message": result.get("message", ""),
+            "message_id": str(ai_message.id),
             "usage": {
                 "prompt_tokens": 0,
                 "completion_tokens": 0,
@@ -419,9 +426,9 @@ class AppChatService:
             variables = {}
 
         try:
-
+            message_id = uuid.uuid4()
             # 发送开始事件
-            yield f"event: start\ndata: {json.dumps({'conversation_id': str(conversation_id)}, ensure_ascii=False)}\n\n"
+            yield f"event: start\ndata: {json.dumps({'conversation_id': str(conversation_id), "message_id": str(message_id)}, ensure_ascii=False)}\n\n"
 
             full_content = ""
             total_tokens = 0
@@ -429,6 +436,7 @@ class AppChatService:
             # 2. 创建编排器
             orchestrator = MultiAgentOrchestrator(self.db, config)
 
+
             # 3. 流式执行任务
             async for event in orchestrator.execute_stream(
                     message=message,
@@ -472,6 +480,7 @@ class AppChatService:
             )
 
             self.conversation_service.add_message(
+                message_id=message_id,
                 conversation_id=conversation_id,
                 role="assistant",
                 content=full_content,

api/app/services/conversation_service.py

@@ -178,7 +178,8 @@ class ConversationService:
             conversation_id: uuid.UUID,
             role: str,
             content: str,
-            meta_data: Optional[dict] = None
+            meta_data: Optional[dict] = None,
+            message_id: Optional[uuid.UUID] = None,
     ) -> Message:
         """
         Add a message to a conversation using UnitOfWork.
@@ -188,6 +189,7 @@ class ConversationService:
             role (str): Role of the message sender ('user' or 'assistant').
             content (str): Message content.
             meta_data (Optional[dict]): Optional metadata.
+            message_id (Optional[uuid.UUID]): Optional custom message UUID.
 
         Returns:
             Message: Newly created Message instance.
@@ -198,6 +200,7 @@ class ConversationService:
             )
 
             message = Message(
+                id=message_id if message_id else uuid.uuid4(),
                 conversation_id=conversation_id,
                 role=role,
                 content=content,
@@ -317,7 +320,7 @@ class ConversationService:
             content=user_message
         )
 
-        self.add_message(
+        ai_message = self.add_message(
             conversation_id=conversation_id,
             role="assistant",
             content=assistant_message,
@@ -332,6 +335,7 @@ class ConversationService:
                 "assistant_message_length": len(assistant_message)
             }
         )
+        return ai_message.id
 
     def delete_conversation(
             self,

api/app/services/memory_config_service.py

@@ -107,6 +107,40 @@ def _validate_config_id(config_id, db: Session = None):
     )
 
 
+# 专门场景的内置 key 集合，直接从 SceneConfigRegistry 派生，避免重复维护
+# 使用懒加载函数避免模块级循环导入
+def _get_builtin_pruning_scenes() -> set:
+    from app.core.memory.storage_services.extraction_engine.data_preprocessing.scene_config import SceneConfigRegistry
+    return set(SceneConfigRegistry.get_all_scenes())
+
+
+def _load_ontology_classes(db: Session, scene_id, pruning_scene: Optional[str]) -> Optional[list]:
+    """当 pruning_scene 不是内置场景时，从 ontology_class 表加载类型名称列表。
+
+    Args:
+        db: 数据库会话
+        scene_id: 本体场景 UUID
+        pruning_scene: 语义剪枝场景名称
+
+    Returns:
+        class_name 字符串列表，或 None（内置场景 / 无数据时）
+    """
+    if not scene_id:
+        return None
+    # 内置场景走 SceneConfigRegistry，不需要注入类型列表
+    if pruning_scene in _get_builtin_pruning_scenes():
+        return None
+    try:
+        from app.repositories.ontology_class_repository import OntologyClassRepository
+        repo = OntologyClassRepository(db)
+        classes = repo.get_classes_by_scene(scene_id)
+        names = [c.class_name for c in classes if c.class_name]
+        return names if names else None
+    except Exception as e:
+        logger.warning(f"Failed to load ontology classes for scene_id={scene_id}: {e}")
+        return None
+
+
 class MemoryConfigService:
     """
     Centralized service for memory configuration loading and validation.
@@ -359,6 +393,7 @@ class MemoryConfigService:
                 pruning_threshold=float(memory_config.pruning_threshold) if memory_config.pruning_threshold is not None else 0.5,
                 # Ontology scene association
                 scene_id=memory_config.scene_id,
+                ontology_classes=_load_ontology_classes(self.db, memory_config.scene_id, memory_config.pruning_scene),
             )
 
             elapsed_ms = (time.time() - start_time) * 1000

api/app/services/memory_storage_service.py

@@ -146,6 +146,10 @@ class DataConfigService: # 数据配置服务类（PostgreSQL）
         if not params.emotion_model_id:
             params.emotion_model_id = params.llm_id
 
+        # 根据关联的本体场景推导 pruning_scene（语义剪枝场景与本体工程场景保持一致）
+        if params.scene_id and not getattr(params, 'pruning_scene', None):
+            params.pruning_scene = self._resolve_pruning_scene_from_scene_id(params.scene_id)
+
         config = MemoryConfigRepository.create(self.db, params)
         self.db.commit()
         return {"affected": 1, "config_id": config.config_id}
@@ -161,6 +165,23 @@ class DataConfigService: # 数据配置服务类（PostgreSQL）
         finally:
             db_session.close()
 
+    def _resolve_pruning_scene_from_scene_id(self, scene_id) -> Optional[str]:
+        """根据本体场景ID获取对应的 scene_name，作为语义剪枝场景值
+
+        Args:
+            scene_id: 本体场景UUID
+
+        Returns:
+            scene_name 字符串，查询失败时返回 None
+        """
+        try:
+            from app.models.ontology_scene import OntologyScene
+            scene = self.db.query(OntologyScene).filter_by(scene_id=scene_id).first()
+            return scene.scene_name if scene else None
+        except Exception as e:
+            logger.warning(f"_resolve_pruning_scene_from_scene_id failed for scene_id={scene_id}: {e}", exc_info=True)
+            return None
+
     # --- Delete ---
     def delete(self, key: ConfigParamsDelete) -> Dict[str, Any]: # 删除配置参数（按配置ID）
         success = MemoryConfigRepository.delete(self.db, key.config_id)
@@ -196,6 +217,19 @@ class DataConfigService: # 数据配置服务类（PostgreSQL）
     def get_all(self, workspace_id = None) -> List[Dict[str, Any]]: # 获取所有配置参数
         results = MemoryConfigRepository.get_all(self.db, workspace_id)
 
+        # 检查并修正 pruning_scene 与 scene_name 不一致的记录
+        needs_commit = False
+        for config, scene_name in results:
+            if scene_name and config.pruning_scene != scene_name:
+                logger.info(
+                    f"修正 pruning_scene: config_id={config.config_id} "
+                    f"'{config.pruning_scene}' -> '{scene_name}'"
+                )
+                config.pruning_scene = scene_name
+                needs_commit = True
+        if needs_commit:
+            self.db.commit()
+
         # 将 ORM 对象转换为字典列表
         data_list = []
         for config, scene_name in results:
@@ -749,8 +783,37 @@ async def analytics_hot_memory_tags(
         await connector.close()
 
 
-async def analytics_recent_activity_stats() -> Dict[str, Any]:
-    stats, _msg = get_recent_activity_stats()
+async def analytics_recent_activity_stats(workspace_id: Optional[str] = None) -> Dict[str, Any]:
+    """获取最近记忆提取活动统计。
+
+    优先从 Redis 缓存读取（按 workspace_id），缓存不存在时降级到日志文件解析。
+
+    Args:
+        workspace_id: 工作空间ID，用于从 Redis 读取对应缓存
+
+    Returns:
+        包含 total、stats、latest_relative、source 的统计字典
+    """
+    stats = None
+    source = "log"
+
+    # 优先从 Redis 读取
+    if workspace_id:
+        try:
+            from app.cache.memory.activity_stats_cache import ActivityStatsCache
+            cached = await ActivityStatsCache.get_activity_stats(workspace_id)
+            if cached:
+                stats = cached.get("stats", {})
+                source = "redis"
+                logger.info(f"[ANALYTICS] 从 Redis 读取活动统计: workspace_id={workspace_id}")
+        except Exception as e:
+            logger.warning(f"[ANALYTICS] 读取 Redis 活动统计失败，降级到日志: {e}")
+
+    # 降级：从日志文件解析
+    if stats is None:
+        stats, _msg = get_recent_activity_stats()
+        source = "log"
+
     total = (
         stats.get("chunk_count", 0)
         + stats.get("statements_count", 0)
@@ -758,26 +821,29 @@ async def analytics_recent_activity_stats() -> Dict[str, Any]:
         + stats.get("triplet_relations_count", 0)
         + stats.get("temporal_count", 0)
     )
-    # 精简：仅提供“最新一次活动多久前”
-    latest_relative = None
-    try:
-        info = stats.get("log_path", "")
-        idx = info.rfind("最新：")
-        if idx != -1:
-            latest_path = info[idx + 3 :].strip()
-            if latest_path and os.path.exists(latest_path):
-                import time
-                diff = max(0.0, time.time() - os.path.getmtime(latest_path))
-                m = int(diff // 60)
-                if m < 1:
-                    latest_relative = "刚刚"
-                elif m < 60:
-                    latest_relative = "一会前"
-                else:
-                    latest_relative = "较早前"
-    except Exception:
-        pass
 
-    data = {"total": total, "stats": stats, "latest_relative": latest_relative}
+    # 计算"最新一次活动多久前"（仅日志来源时有效）
+    latest_relative = None
+    if source == "log":
+        try:
+            info = stats.get("log_path", "")
+            idx = info.rfind("最新：")
+            if idx != -1:
+                latest_path = info[idx + 3:].strip()
+                if latest_path and os.path.exists(latest_path):
+                    import time
+                    diff = max(0.0, time.time() - os.path.getmtime(latest_path))
+                    m = int(diff // 60)
+                    if m < 1:
+                        latest_relative = "刚刚"
+                    elif m < 60:
+                        latest_relative = "一会前"
+                    else:
+                        latest_relative = "较早前"
+        except Exception:
+            pass
+
+    data = {"total": total, "stats": stats, "latest_relative": latest_relative, "source": source}
     return data
 
+

api/app/services/pilot_run_service.py

@@ -326,6 +326,25 @@ async def run_pilot_extraction(
 
         logger.info("Pilot run completed: Skipping Neo4j save")
 
+        # 将提取统计写入 Redis，按 workspace_id 存储
+        try:
+            from app.cache.memory.activity_stats_cache import ActivityStatsCache
+
+            stats_to_cache = {
+                "chunk_count": len(chunk_nodes) if chunk_nodes else 0,
+                "statements_count": len(statement_nodes) if statement_nodes else 0,
+                "triplet_entities_count": len(entity_nodes) if entity_nodes else 0,
+                "triplet_relations_count": len(entity_edges) if entity_edges else 0,
+                "temporal_count": 0,  # temporal 数据在日志中，此处暂置0
+            }
+            await ActivityStatsCache.set_activity_stats(
+                workspace_id=str(memory_config.workspace_id),
+                stats=stats_to_cache,
+            )
+            logger.info(f"[PILOT_RUN] 活动统计已写入 Redis: workspace_id={memory_config.workspace_id}")
+        except Exception as cache_err:
+            logger.warning(f"[PILOT_RUN] 写入活动统计缓存失败（不影响主流程）: {cache_err}", exc_info=True)
+
     except Exception as e:
         logger.error(f"Pilot run failed: {e}", exc_info=True)
         raise

api/app/services/workflow_import_service.py

@@ -56,7 +56,7 @@ class WorkflowImportService:
                 success=False,
                 temp_id=None,
                 workflow_id=None,
-                errors=[InvalidConfiguration()]
+                errors=[InvalidConfiguration()] + adapter.errors
             )
 
         workflow_config = adapter.parse_workflow()

api/app/services/workflow_service.py

@@ -25,7 +25,7 @@ from app.repositories.workflow_repository import (
     WorkflowExecutionRepository,
     WorkflowNodeExecutionRepository
 )
-from app.schemas import DraftRunRequest, FileInput
+from app.schemas import DraftRunRequest, FileInput, FileType
 from app.services.conversation_service import ConversationService
 from app.services.multi_agent_service import convert_uuids_to_str
 from app.services.multimodal_service import MultimodalService
@@ -496,6 +496,7 @@ class WorkflowService:
                     "event": "start",
                     "data": {
                         "conversation_id": payload.get("conversation_id"),
+                        "message_id": payload.get("message_id")
                     }
                 }
             case "workflow_end":
@@ -600,6 +601,7 @@ class WorkflowService:
         try:
             files = await self._handle_file_input(payload.files)
             input_data["files"] = files
+            message_id = uuid.uuid4()
             # 更新状态为运行中
             self.update_execution_status(execution.execution_id, "running")
 
@@ -624,24 +626,45 @@ class WorkflowService:
                 workspace_id=str(workspace_id),
                 user_id=payload.user_id
             )
-
             # 更新执行结果
             if result.get("status") == "completed":
                 token_usage = result.get("token_usage", {}) or {}
+
+                final_messages = result.get("messages", [])[init_message_length:]
+                human_message = ""
+                assistant_message = ""
+                for message in final_messages:
+                    if message["role"] == "user":
+                        if isinstance(message["content"], str):
+                            human_message += message["content"]
+                        elif isinstance(message["content"], list):
+                            for file in message["content"]:
+                                if file.get("type") == FileType.IMAGE:
+                                    human_message += f"![image]({file.get('url', '')})"
+                                else:
+                                    human_message += f"[{file.get('type')}]({file.get('url', '')})"
+                    if message["role"] == "assistant":
+                        assistant_message = message["content"]
+                self.conversation_service.add_message(
+                    conversation_id=conversation_id_uuid,
+                    role="user",
+                    content=human_message,
+                    meta_data=None
+                )
+                self.conversation_service.add_message(
+                    message_id=message_id,
+                    conversation_id=conversation_id_uuid,
+                    role="assistant",
+                    content=assistant_message,
+                    meta_data={"usage": token_usage}
+                )
                 self.update_execution_status(
                     execution.execution_id,
                     "completed",
                     output_data=result,
                     token_usage=token_usage.get("total_tokens", None)
                 )
-                final_messages = result.get("messages", [])[init_message_length:]
-                for message in final_messages:
-                    self.conversation_service.add_message(
-                        conversation_id=conversation_id_uuid,
-                        role=message["role"],
-                        content=message["content"],
-                        meta_data=None if message["role"] == "user" else {"usage": token_usage}
-                    )
+
                 logger.info(f"Workflow Run Success, "
                             f"execution_id: {execution.execution_id}, message count: {len(final_messages)}")
             else:
@@ -650,6 +673,8 @@ class WorkflowService:
                     "failed",
                     error_message=result.get("error")
                 )
+                logger.error(f"Workflow Run Failed, execution_id: {execution.execution_id},"
+                             f" error: {result.get('error')}")
 
             # 返回增强的响应结构
             return {
@@ -659,6 +684,7 @@ class WorkflowService:
                 # "messages": result.get("messages"),
                 "output": result.get("output"),  # 最终输出（字符串）
                 "message": result.get("output"),  # 最终输出（字符串）
+                "message_id": str(message_id),
                 # "output_data": result.get("node_outputs", {}),  # 所有节点输出（详细数据）
                 "conversation_id": result.get("conversation_id"),  # 所有节点输出（详细数据）payload.,  # 会话 ID
                 "error_message": result.get("error"),
@@ -756,7 +782,7 @@ class WorkflowService:
                         input_data["conv_messages"] = last_state.get("messages") or []
                         break
             init_message_length = len(input_data.get("conv_messages", []))
-
+            message_id = uuid.uuid4()
             async for event in execute_workflow_stream(
                     workflow_config=workflow_config_dict,
                     input_data=input_data,
@@ -765,24 +791,43 @@ class WorkflowService:
                     user_id=payload.user_id,
             ):
                 if event.get("event") == "workflow_end":
-
                     status = event.get("data", {}).get("status")
                     token_usage = event.get("data", {}).get("token_usage", {}) or {}
                     if status == "completed":
+                        final_messages = event.get("data", {}).get("messages", [])[init_message_length:]
+                        human_message = ""
+                        assistant_message = ""
+                        for message in final_messages:
+                            if message["role"] == "user":
+                                if isinstance(message["content"], str):
+                                    human_message += message["content"]
+                                elif isinstance(message["content"], list):
+                                    for file in message["content"]:
+                                        if file.get("type") == FileType.IMAGE:
+                                            human_message += f"![image]({file.get('url', '')})"
+                                        else:
+                                            human_message += f"[{file.get('type')}]({file.get('url', '')})"
+                            if message["role"] == "assistant":
+                                assistant_message = message["content"]
+                        self.conversation_service.add_message(
+                            conversation_id=conversation_id_uuid,
+                            role="user",
+                            content=human_message,
+                            meta_data=None
+                        )
+                        self.conversation_service.add_message(
+                            message_id=message_id,
+                            conversation_id=conversation_id_uuid,
+                            role="assistant",
+                            content=assistant_message,
+                            meta_data={"usage": token_usage}
+                        )
                         self.update_execution_status(
                             execution.execution_id,
                             "completed",
                             output_data=event.get("data"),
                             token_usage=token_usage.get("total_tokens", None)
                         )
-                        final_messages = event.get("data", {}).get("messages", [])[init_message_length:]
-                        for message in final_messages:
-                            self.conversation_service.add_message(
-                                conversation_id=conversation_id_uuid,
-                                role=message["role"],
-                                content=message["content"],
-                                meta_data=None if message["role"] == "user" else {"usage": token_usage}
-                            )
                         logger.info(f"Workflow Run Success, "
                                     f"execution_id: {execution.execution_id}, message count: {len(final_messages)}")
                     elif status == "failed":
@@ -793,6 +838,8 @@ class WorkflowService:
                         )
                     else:
                         logger.error(f"unexpect workflow run status, status: {status}")
+                elif event.get("event") == "workflow_start":
+                    event["data"]["message_id"] = str(message_id)
                 event = self._emit(public, event)
                 if event:
                     yield event

api/app/services/workspace_service.py

@@ -152,6 +152,7 @@ def create_workspace(
 
         # Initialize default ontology scenes for the workspace (先创建本体场景)
         default_scene_id = None
+        default_scene_name = None
         try:
             initializer = DefaultOntologyInitializer(db)
             success, error_msg = initializer.initialize_default_scenes(
@@ -163,7 +164,7 @@ def create_workspace(
                     f"为工作空间 {db_workspace.id} 创建默认本体场景成功 (language={language})"
                 )
                 
-                # 获取默认场景ID，优先使用"在线教育"场景，如果不存在则使用"情感陪伴"场景
+        # 获取默认场景ID，优先使用"在线教育"场景，如果不存在则使用"情感陪伴"场景
                 from app.repositories.ontology_scene_repository import OntologySceneRepository
                 from app.config.default_ontology_config import (
                     ONLINE_EDUCATION_SCENE, 
@@ -179,6 +180,7 @@ def create_workspace(
                 
                 if education_scene:
                     default_scene_id = education_scene.scene_id
+                    default_scene_name = education_scene.scene_name
                     business_logger.info(
                         f"获取到教育场景ID用于默认记忆配置: {default_scene_id} (scene_name={education_scene_name})"
                     )
@@ -189,6 +191,7 @@ def create_workspace(
                     
                     if companion_scene:
                         default_scene_id = companion_scene.scene_id
+                        default_scene_name = companion_scene.scene_name
                         business_logger.info(
                             f"教育场景不存在，使用情感陪伴场景ID用于默认记忆配置: {default_scene_id} (scene_name={companion_scene_name})"
                         )
@@ -219,6 +222,7 @@ def create_workspace(
                     embedding_id=embedding,
                     rerank_id=rerank,
                     scene_id=default_scene_id,  # 传入默认场景ID（优先教育场景，其次情感陪伴场景）
+                    pruning_scene_name=default_scene_name,  # 传入场景名称作为语义剪枝场景值
                 )
                 business_logger.info(
                     f"为工作空间 {db_workspace.id} 创建默认记忆配置成功 (scene_id={default_scene_id})"
@@ -1159,6 +1163,7 @@ def _create_default_memory_config(
     embedding_id: Optional[uuid.UUID] = None,
     rerank_id: Optional[uuid.UUID] = None,
     scene_id: Optional[uuid.UUID] = None,
+    pruning_scene_name: Optional[str] = None,
 ) -> None:
     """Create a default memory config for a newly created workspace.
     
@@ -1170,6 +1175,7 @@ def _create_default_memory_config(
         embedding_id: Optional embedding model ID
         rerank_id: Optional rerank model ID
         scene_id: Optional ontology scene ID (默认关联教育场景)
+        pruning_scene_name: Optional pruning scene name，取自 ontology_scene.scene_name
     """
     from app.models.memory_config_model import MemoryConfig
     
@@ -1183,7 +1189,8 @@ def _create_default_memory_config(
         llm_id=str(llm_id) if llm_id else None,
         embedding_id=str(embedding_id) if embedding_id else None,
         rerank_id=str(rerank_id) if rerank_id else None,
-        scene_id=scene_id,  # 关联本体场景ID
+        scene_id=scene_id,  # 关联本体场景ID（默认为"在线教育"场景）
+        pruning_scene=pruning_scene_name,  # 语义剪枝场景直接使用 scene_name
         state=True,  # Active by default
         is_default=True,  # Mark as workspace default
     )