Merge branch 'refs/heads/develop' into fix/memory_mcp2_1

api/app/cache/__init__.py

@@ -0,0 +1,11 @@
+"""
+Cache 缓存模块
+
+提供各种缓存功能的统一入口
+"""
+from .memory import EmotionMemoryCache, ImplicitMemoryCache
+
+__all__ = [
+    "EmotionMemoryCache",
+    "ImplicitMemoryCache",
+]

api/app/cache/memory/__init__.py

@@ -0,0 +1,12 @@
+"""
+Memory 缓存模块
+
+提供记忆系统相关的缓存功能
+"""
+from .emotion_memory import EmotionMemoryCache
+from .implicit_memory import ImplicitMemoryCache
+
+__all__ = [
+    "EmotionMemoryCache",
+    "ImplicitMemoryCache",
+]

api/app/cache/memory/emotion_memory.py

@@ -0,0 +1,134 @@
+"""
+Emotion Suggestions Cache
+
+情绪个性化建议缓存模块
+用于缓存用户的情绪个性化建议数据
+"""
+import json
+import logging
+from typing import Optional, Dict, Any
+from datetime import datetime
+
+from app.aioRedis import aio_redis
+
+logger = logging.getLogger(__name__)
+
+
+class EmotionMemoryCache:
+    """情绪建议缓存类"""
+    
+    # Key 前缀
+    PREFIX = "cache:memory:emotion_memory"
+    
+    @classmethod
+    def _get_key(cls, *parts: str) -> str:
+        """生成 Redis key
+        
+        Args:
+            *parts: key 的各个部分
+            
+        Returns:
+            完整的 Redis key
+        """
+        return ":".join([cls.PREFIX] + list(parts))
+    
+    @classmethod
+    async def set_emotion_suggestions(
+        cls,
+        user_id: str,
+        suggestions_data: Dict[str, Any],
+        expire: int = 86400
+    ) -> bool:
+        """设置用户情绪建议缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            suggestions_data: 建议数据字典，包含：
+                - health_summary: 健康状态摘要
+                - suggestions: 建议列表
+                - generated_at: 生成时间（可选）
+            expire: 过期时间（秒），默认24小时（86400秒）
+            
+        Returns:
+            是否设置成功
+        """
+        try:
+            key = cls._get_key("suggestions", user_id)
+            
+            # 添加生成时间戳
+            if "generated_at" not in suggestions_data:
+                suggestions_data["generated_at"] = datetime.now().isoformat()
+            
+            # 添加缓存标记
+            suggestions_data["cached"] = True
+            
+            value = json.dumps(suggestions_data, ensure_ascii=False)
+            await aio_redis.set(key, value, ex=expire)
+            logger.info(f"设置情绪建议缓存成功: {key}, 过期时间: {expire}秒")
+            return True
+        except Exception as e:
+            logger.error(f"设置情绪建议缓存失败: {e}", exc_info=True)
+            return False
+    
+    @classmethod
+    async def get_emotion_suggestions(cls, user_id: str) -> Optional[Dict[str, Any]]:
+        """获取用户情绪建议缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            建议数据字典，如果不存在或已过期返回 None
+        """
+        try:
+            key = cls._get_key("suggestions", user_id)
+            value = await aio_redis.get(key)
+            
+            if value:
+                data = json.loads(value)
+                logger.info(f"成功获取情绪建议缓存: {key}")
+                return data
+            
+            logger.info(f"情绪建议缓存不存在或已过期: {key}")
+            return None
+        except Exception as e:
+            logger.error(f"获取情绪建议缓存失败: {e}", exc_info=True)
+            return None
+    
+    @classmethod
+    async def delete_emotion_suggestions(cls, user_id: str) -> bool:
+        """删除用户情绪建议缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            是否删除成功
+        """
+        try:
+            key = cls._get_key("suggestions", user_id)
+            result = await aio_redis.delete(key)
+            logger.info(f"删除情绪建议缓存: {key}, 结果: {result}")
+            return result > 0
+        except Exception as e:
+            logger.error(f"删除情绪建议缓存失败: {e}", exc_info=True)
+            return False
+    
+    @classmethod
+    async def get_suggestions_ttl(cls, user_id: str) -> int:
+        """获取情绪建议缓存的剩余过期时间
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            剩余秒数，-1表示永不过期，-2表示key不存在
+        """
+        try:
+            key = cls._get_key("suggestions", user_id)
+            ttl = await aio_redis.ttl(key)
+            logger.debug(f"情绪建议缓存TTL: {key} = {ttl}秒")
+            return ttl
+        except Exception as e:
+            logger.error(f"获取情绪建议缓存TTL失败: {e}")
+            return -2

api/app/cache/memory/implicit_memory.py

@@ -0,0 +1,136 @@
+"""
+Implicit Memory Profile Cache
+
+隐式记忆用户画像缓存模块
+用于缓存用户的完整画像数据（偏好标签、四维画像、兴趣领域、行为习惯）
+"""
+import json
+import logging
+from typing import Optional, Dict, Any
+from datetime import datetime
+
+from app.aioRedis import aio_redis
+
+logger = logging.getLogger(__name__)
+
+
+class ImplicitMemoryCache:
+    """隐式记忆用户画像缓存类"""
+    
+    # Key 前缀
+    PREFIX = "cache:memory:implicit_memory"
+    
+    @classmethod
+    def _get_key(cls, *parts: str) -> str:
+        """生成 Redis key
+        
+        Args:
+            *parts: key 的各个部分
+            
+        Returns:
+            完整的 Redis key
+        """
+        return ":".join([cls.PREFIX] + list(parts))
+    
+    @classmethod
+    async def set_user_profile(
+        cls,
+        user_id: str,
+        profile_data: Dict[str, Any],
+        expire: int = 86400
+    ) -> bool:
+        """设置用户完整画像缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            profile_data: 画像数据字典，包含：
+                - preferences: 偏好标签列表
+                - portrait: 四维画像对象
+                - interest_areas: 兴趣领域分布对象
+                - habits: 行为习惯列表
+                - generated_at: 生成时间（可选）
+            expire: 过期时间（秒），默认24小时（86400秒）
+            
+        Returns:
+            是否设置成功
+        """
+        try:
+            key = cls._get_key("profile", user_id)
+            
+            # 添加生成时间戳
+            if "generated_at" not in profile_data:
+                profile_data["generated_at"] = datetime.now().isoformat()
+            
+            # 添加缓存标记
+            profile_data["cached"] = True
+            
+            value = json.dumps(profile_data, ensure_ascii=False)
+            await aio_redis.set(key, value, ex=expire)
+            logger.info(f"设置用户画像缓存成功: {key}, 过期时间: {expire}秒")
+            return True
+        except Exception as e:
+            logger.error(f"设置用户画像缓存失败: {e}", exc_info=True)
+            return False
+    
+    @classmethod
+    async def get_user_profile(cls, user_id: str) -> Optional[Dict[str, Any]]:
+        """获取用户完整画像缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            画像数据字典，如果不存在或已过期返回 None
+        """
+        try:
+            key = cls._get_key("profile", user_id)
+            value = await aio_redis.get(key)
+            
+            if value:
+                data = json.loads(value)
+                logger.info(f"成功获取用户画像缓存: {key}")
+                return data
+            
+            logger.info(f"用户画像缓存不存在或已过期: {key}")
+            return None
+        except Exception as e:
+            logger.error(f"获取用户画像缓存失败: {e}", exc_info=True)
+            return None
+    
+    @classmethod
+    async def delete_user_profile(cls, user_id: str) -> bool:
+        """删除用户完整画像缓存
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            是否删除成功
+        """
+        try:
+            key = cls._get_key("profile", user_id)
+            result = await aio_redis.delete(key)
+            logger.info(f"删除用户画像缓存: {key}, 结果: {result}")
+            return result > 0
+        except Exception as e:
+            logger.error(f"删除用户画像缓存失败: {e}", exc_info=True)
+            return False
+    
+    @classmethod
+    async def get_profile_ttl(cls, user_id: str) -> int:
+        """获取用户画像缓存的剩余过期时间
+        
+        Args:
+            user_id: 用户ID（end_user_id）
+            
+        Returns:
+            剩余秒数，-1表示永不过期，-2表示key不存在
+        """
+        try:
+            key = cls._get_key("profile", user_id)
+            ttl = await aio_redis.ttl(key)
+            logger.debug(f"用户画像缓存TTL: {key} = {ttl}秒")
+            return ttl
+        except Exception as e:
+            logger.error(f"获取用户画像缓存TTL失败: {e}")
+            return -2

api/app/controllers/emotion_controller.py

@@ -231,9 +231,9 @@ async def get_emotion_suggestions(
                 extra={"group_id": request.group_id}
             )
             return fail(
-                BizCode.RESOURCE_NOT_FOUND,
+                BizCode.NOT_FOUND,
                 "建议缓存不存在或已过期，请调用 /generate_suggestions 接口生成新建议",
-                None
+                ""
             )
 
         api_logger.info(
@@ -267,7 +267,7 @@ async def generate_emotion_suggestions(
     """生成个性化情绪建议（调用LLM并缓存）
 
     Args:
-        request: 包含 group_id、可选的 config_id 和 force_refresh
+        request: 包含 end_user_id
         db: 数据库会话
         current_user: 当前用户
 
@@ -275,47 +275,22 @@ async def generate_emotion_suggestions(
         新生成的个性化情绪建议响应
     """
     try:
-        # 验证 config_id（如果提供）
-        # 获取终端用户关联的配置
-        config_id = request.config_id
-        if config_id is None:
-            # 如果没有提供 config_id，尝试获取用户关联的配置
-            try:
-                from app.services.memory_agent_service import (
-                    get_end_user_connected_config,
-                )
-                connected_config = get_end_user_connected_config(request.group_id, db)
-                config_id = connected_config.get("memory_config_id")
-            except ValueError as e:
-                return fail(BizCode.INVALID_PARAMETER, "无法获取用户关联的配置", str(e))
-        else:
-            # 如果提供了 config_id，验证其有效性
-            from app.services.memory_config_service import MemoryConfigService
-            try:
-                config_service = MemoryConfigService(db)
-                config = config_service.get_config_by_id(config_id)
-                if not config:
-                    return fail(BizCode.INVALID_PARAMETER, "配置ID无效", f"配置 {config_id} 不存在")
-            except Exception as e:
-                return fail(BizCode.INVALID_PARAMETER, "配置ID验证失败", str(e))
-
         api_logger.info(
             f"用户 {current_user.username} 请求生成个性化情绪建议",
             extra={
-                "group_id": request.group_id,
-                "config_id": config_id
+                "end_user_id": request.end_user_id
             }
         )
 
         # 调用服务层生成建议
         data = await emotion_service.generate_emotion_suggestions(
-            end_user_id=request.group_id,
+            end_user_id=request.end_user_id,
             db=db
         )
 
         # 保存到缓存
         await emotion_service.save_suggestions_cache(
-            end_user_id=request.group_id,
+            end_user_id=request.end_user_id,
             suggestions_data=data,
             db=db,
             expires_hours=24
@@ -324,7 +299,7 @@ async def generate_emotion_suggestions(
         api_logger.info(
             "个性化建议生成成功",
             extra={
-                "group_id": request.group_id,
+                "end_user_id": request.end_user_id,
                 "suggestions_count": len(data.get("suggestions", []))
             }
         )
@@ -334,7 +309,7 @@ async def generate_emotion_suggestions(
     except Exception as e:
         api_logger.error(
             f"生成个性化建议失败: {str(e)}",
-            extra={"group_id": request.group_id},
+            extra={"end_user_id": request.end_user_id},
             exc_info=True
         )
         raise HTTPException(

api/app/controllers/implicit_memory_controller.py

@@ -161,9 +161,9 @@ async def get_preference_tags(
         if cached_profile is None:
             api_logger.info(f"用户 {user_id} 的画像缓存不存在或已过期")
             return fail(
-                BizCode.RESOURCE_NOT_FOUND,
+                BizCode.NOT_FOUND,
                 "画像缓存不存在或已过期，请调用 /generate_profile 接口生成新画像",
-                None
+                ""
             )
         
         # Extract preferences from cache
@@ -232,9 +232,9 @@ async def get_dimension_portrait(
         if cached_profile is None:
             api_logger.info(f"用户 {user_id} 的画像缓存不存在或已过期")
             return fail(
-                BizCode.RESOURCE_NOT_FOUND,
+                BizCode.NOT_FOUND,
                 "画像缓存不存在或已过期，请调用 /generate_profile 接口生成新画像",
-                None
+                ""
             )
         
         # Extract portrait from cache
@@ -280,9 +280,9 @@ async def get_interest_area_distribution(
         if cached_profile is None:
             api_logger.info(f"用户 {user_id} 的画像缓存不存在或已过期")
             return fail(
-                BizCode.RESOURCE_NOT_FOUND,
+                BizCode.NOT_FOUND,
                 "画像缓存不存在或已过期，请调用 /generate_profile 接口生成新画像",
-                None
+                ""
             )
         
         # Extract interest areas from cache
@@ -332,9 +332,9 @@ async def get_behavior_habits(
         if cached_profile is None:
             api_logger.info(f"用户 {user_id} 的画像缓存不存在或已过期")
             return fail(
-                BizCode.RESOURCE_NOT_FOUND,
+                BizCode.NOT_FOUND,
                 "画像缓存不存在或已过期，请调用 /generate_profile 接口生成新画像",
-                None
+                ""
             )
         
         # Extract habits from cache

api/app/controllers/public_share_controller.py

@@ -8,9 +8,10 @@ from sqlalchemy.orm import Session
 
 from app.core.logging_config import get_business_logger
 from app.core.response_utils import success
-from app.db import get_db
+from app.db import get_db, get_db_read
 from app.dependencies import get_share_user_id, ShareTokenData
 from app.repositories import knowledge_repository
+from app.repositories.workflow_repository import WorkflowConfigRepository
 from app.schemas import release_share_schema, conversation_schema
 from app.schemas.response_schema import PageData, PageMeta
 from app.services import workspace_service
@@ -19,7 +20,8 @@ from app.services.conversation_service import ConversationService
 from app.services.release_share_service import ReleaseShareService
 from app.services.shared_chat_service import SharedChatService
 from app.services.app_chat_service import AppChatService, get_app_chat_service
-from app.utils.app_config_utils import dict_to_multi_agent_config, workflow_config_4_app_release, agent_config_4_app_release, multi_agent_config_4_app_release
+from app.utils.app_config_utils import dict_to_multi_agent_config, workflow_config_4_app_release, \
+    agent_config_4_app_release, multi_agent_config_4_app_release
 
 router = APIRouter(prefix="/public/share", tags=["Public Share"])
 logger = get_business_logger()
@@ -65,10 +67,10 @@ def get_or_generate_user_id(payload_user_id: str, request: Request) -> str:
     summary="获取访问 token"
 )
 def get_access_token(
-    share_token: str,
-    payload: release_share_schema.TokenRequest,
-    request: Request,
-    db: Session = Depends(get_db),
+        share_token: str,
+        payload: release_share_schema.TokenRequest,
+        request: Request,
+        db: Session = Depends(get_db),
 ):
     """获取访问 token
 
@@ -113,9 +115,9 @@ def get_access_token(
     response_model=None
 )
 def get_shared_release(
-    password: str = Query(None, description="访问密码（如果需要）"),
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
+        password: str = Query(None, description="访问密码（如果需要）"),
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
 ):
     """获取公开分享的发布版本信息
 
@@ -137,9 +139,9 @@ def get_shared_release(
     summary="验证访问密码"
 )
 def verify_password(
-    payload: release_share_schema.PasswordVerifyRequest,
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
+        payload: release_share_schema.PasswordVerifyRequest,
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
 ):
     """验证分享的访问密码
 
@@ -159,11 +161,11 @@ def verify_password(
     summary="获取嵌入代码"
 )
 def get_embed_code(
-    width: str = Query("100%", description="iframe 宽度"),
-    height: str = Query("600px", description="iframe 高度"),
-    request: Request = None,
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
+        width: str = Query("100%", description="iframe 宽度"),
+        height: str = Query("600px", description="iframe 高度"),
+        request: Request = None,
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
 ):
     """获取嵌入代码
 
@@ -183,7 +185,6 @@ def get_embed_code(
     return success(data=embed_code)
 
 
-
 # ---------- 会话管理接口 ----------
 
 @router.get(
@@ -191,11 +192,11 @@ def get_embed_code(
     summary="获取会话列表"
 )
 def list_conversations(
-    password: str = Query(None, description="访问密码"),
-    page: int = Query(1, ge=1),
-    pagesize: int = Query(20, ge=1, le=100),
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
+        password: str = Query(None, description="访问密码"),
+        page: int = Query(1, ge=1),
+        pagesize: int = Query(20, ge=1, le=100),
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
 ):
     """获取分享应用的会话列表
 
@@ -209,9 +210,9 @@ def list_conversations(
     from app.repositories.end_user_repository import EndUserRepository
     end_user_repo = EndUserRepository(db)
     new_end_user = end_user_repo.get_or_create_end_user(
-            app_id=share.app_id,
-            other_id=other_id
-        )
+        app_id=share.app_id,
+        other_id=other_id
+    )
     logger.debug(new_end_user.id)
     service = SharedChatService(db)
     conversations, total = service.list_conversations(
@@ -233,10 +234,10 @@ def list_conversations(
     summary="获取会话详情（含消息）"
 )
 def get_conversation(
-    conversation_id: uuid.UUID,
-    password: str = Query(None, description="访问密码"),
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
+        conversation_id: uuid.UUID,
+        password: str = Query(None, description="访问密码"),
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
 ):
     """获取会话详情和消息历史"""
     chat_service = SharedChatService(db)
@@ -266,10 +267,10 @@ def get_conversation(
     summary="发送消息（支持流式和非流式）"
 )
 async def chat(
-    payload: conversation_schema.ChatRequest,
-    share_data: ShareTokenData = Depends(get_share_user_id),
-    db: Session = Depends(get_db),
-    app_chat_service: Annotated[AppChatService, Depends(get_app_chat_service)] = None,
+        payload: conversation_schema.ChatRequest,
+        share_data: ShareTokenData = Depends(get_share_user_id),
+        db: Session = Depends(get_db),
+        app_chat_service: Annotated[AppChatService, Depends(get_app_chat_service)] = None,
 ):
     """发送消息并获取回复
 
@@ -313,7 +314,7 @@ async def chat(
         )
         end_user_id = str(new_end_user.id)
 
-        appid=share.app_id
+        appid = share.app_id
         """获取存储类型和工作空间的ID"""
 
         # 直接通过 SQLAlchemy 查询 app
@@ -425,16 +426,16 @@ async def chat(
             # )
             async def event_generator():
                 async for event in app_chat_service.agnet_chat_stream(
-                    message=payload.message,
-                    conversation_id=conversation.id,  # 使用已创建的会话 ID
-                    user_id= str(new_end_user.id),  # 转换为字符串
-                    variables=payload.variables,
-                    web_search=payload.web_search,
-                    config=agent_config,
-                    memory=payload.memory,
-                    storage_type=storage_type,
-                    user_rag_memory_id=user_rag_memory_id,
-                    workspace_id=workspace_id
+                        message=payload.message,
+                        conversation_id=conversation.id,  # 使用已创建的会话 ID
+                        user_id=str(new_end_user.id),  # 转换为字符串
+                        variables=payload.variables,
+                        web_search=payload.web_search,
+                        config=agent_config,
+                        memory=payload.memory,
+                        storage_type=storage_type,
+                        user_rag_memory_id=user_rag_memory_id,
+                        workspace_id=workspace_id
                 ):
                     yield event
 
@@ -481,15 +482,15 @@ async def chat(
             async def event_generator():
                 async for event in app_chat_service.multi_agent_chat_stream(
 
-                    message=payload.message,
-                    conversation_id=conversation.id,  # 使用已创建的会话 ID
-                    user_id=str(new_end_user.id),  # 转换为字符串
-                    variables=payload.variables,
-                    config=config,
-                    web_search=payload.web_search,
-                    memory=payload.memory,
-                    storage_type=storage_type,
-                    user_rag_memory_id=user_rag_memory_id
+                        message=payload.message,
+                        conversation_id=conversation.id,  # 使用已创建的会话 ID
+                        user_id=str(new_end_user.id),  # 转换为字符串
+                        variables=payload.variables,
+                        config=config,
+                        web_search=payload.web_search,
+                        memory=payload.memory,
+                        storage_type=storage_type,
+                        user_rag_memory_id=user_rag_memory_id
                 ):
                     yield event
 
@@ -561,24 +562,27 @@ async def chat(
 
         # return success(data=conversation_schema.ChatResponse(**result))
     elif app_type == AppType.WORKFLOW:
-
         config = workflow_config_4_app_release(release)
+        if not config.id:
+            with get_db_read() as db:
+                source_config = WorkflowConfigRepository(db).get_by_app_id(release.app_id)
+                config.id = source_config.id
+        config.id = uuid.UUID(config.id)
         if payload.stream:
             async def event_generator():
-
                 async for event in app_chat_service.workflow_chat_stream(
-
-                    message=payload.message,
-                    conversation_id=conversation.id,  # 使用已创建的会话 ID
-                    user_id=end_user_id,  # 转换为字符串
-                    variables=payload.variables,
-                    config=config,
-                    web_search=payload.web_search,
-                    memory=payload.memory,
-                    storage_type=storage_type,
-                    user_rag_memory_id=user_rag_memory_id,
-                    app_id=release.app_id,
-                    workspace_id=workspace_id
+                        message=payload.message,
+                        conversation_id=conversation.id,  # 使用已创建的会话 ID
+                        user_id=end_user_id,  # 转换为字符串
+                        variables=payload.variables,
+                        config=config,
+                        web_search=payload.web_search,
+                        memory=payload.memory,
+                        storage_type=storage_type,
+                        user_rag_memory_id=user_rag_memory_id,
+                        app_id=release.app_id,
+                        workspace_id=workspace_id,
+                        release_id=release.id
                 ):
                     event_type = event.get("event", "message")
                     event_data = event.get("data", {})
@@ -610,7 +614,8 @@ async def chat(
             storage_type=storage_type,
             user_rag_memory_id=user_rag_memory_id,
             app_id=release.app_id,
-            workspace_id=workspace_id
+            workspace_id=workspace_id,
+            release_id=release.id
         )
         logger.debug(
             "工作流试运行返回结果",

api/app/controllers/service/app_api_controller.py

@@ -242,8 +242,9 @@ async def chat(
                     memory=payload.memory,
                     storage_type=storage_type,
                     user_rag_memory_id=user_rag_memory_id,
-                    app_id=app.app_id,
-                    workspace_id=workspace_id
+                    app_id=app.id,
+                    workspace_id=workspace_id,
+                    release_id=app.current_release.id,
                 ):
                     event_type = event.get("event", "message")
                     event_data = event.get("data", {})
@@ -274,8 +275,9 @@ async def chat(
             memory=payload.memory,
             storage_type=storage_type,
             user_rag_memory_id=user_rag_memory_id,
-            app_id=app.app_id,
-            workspace_id=workspace_id
+            app_id=app.id,
+            workspace_id=workspace_id,
+            release_id=app.current_release.id
         )
         logger.debug(
             "工作流试运行返回结果",

api/app/core/config.py

@@ -38,6 +38,7 @@ class Settings:
     REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
     REDIS_DB: int = int(os.getenv("REDIS_DB", "1"))
     REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD", "")
+    
 
     # ElasticSearch configuration
     ELASTICSEARCH_HOST: str = os.getenv("ELASTICSEARCH_HOST", "https://127.0.0.1")

api/app/core/rag/llm/cv_model.py

@@ -243,6 +243,33 @@ class QWenCV(GptV4):
             tmp_path = tmp.name
 
             video_path = f"file://{tmp_path}"
+            prompt_ch = """
+                    你是一名专业的视频转录助手，能够将视频文件的内容转写为文本，并**精确标记每句话或每个段落对应的时间戳**（开始时间-结束时间）。\n
+                    **任务要求**：
+                    1.输入是MP4等视频文件,解析带时间戳的文本。
+                    2.时间戳格式为 `[HH:MM:SS.mmm]`（毫秒可选），例如 `[00:01:23.456]`。
+                    3.时间戳需尽可能贴近实际视频的起止时间，误差不超过1秒。
+                    4.如果无法确定具体时间，请根据上下文合理估算。
+                    5.最后总结:这段视频的内容是什么?,并用恰当的句子总结这个视频。
+
+                    **示例输出**：
+                    [00:00:00.000] 今天天气真好，
+                    [00:00:02.500] 我们一起去公园散步吧。
+                    [00:00:05.800] 公园里的花开得非常漂亮。
+                    这段视频的内容是关于如何在CREAMS系统中进行楼宇管理集合的编辑或删除操作。视频演示了 ..."""
+            prompt_en = """
+                    You are a professional video transcription assistant, capable of transcribing the content of video files into text and **precisely marking the timestamp (start time-end time) corresponding to each sentence or paragraph**. 
+                    **Task requirements**: 
+                    1. Input is MP4 or other video files, and parse the text with timestamps.
+                    2. The timestamp format is `[HH:MM:SS.mmm]` (milliseconds are optional), for example, `[00:01:23.456]`.
+                    3. The timestamp should be as close as possible to the actual start and end time of the video, with an error not exceeding 1 second.
+                    4. If the specific time cannot be determined, please make a reasonable estimation based on the context.
+                    5. Final summary: What is the content of this video? Summarize this video in an appropriate sentence.
+                    
+                    **Example output**: 
+                    [00:00:00.000] The weather is really nice today, [00:00:02.500] let's go for a walk in the park together.
+                    [00:00:05.800] The flowers in the park are blooming beautifully.
+                    The content of this video is about how to edit or delete building management collections in the CREAMS system. The video demonstrates .."""
             messages = [
                 {
                     "role": "user",
@@ -252,7 +279,7 @@ class QWenCV(GptV4):
                             "fps": 2,
                         },
                         {
-                            "text": "视频的内容是什么?,并且，请用恰当的句子总结这个视频。" if self.lang.lower() == "chinese" else "What is the content of the video? And please summarize this video in proper sentences.",
+                            "text": prompt_ch if self.lang.lower() == "chinese" else prompt_en,
                         },
                     ],
                 }

api/app/core/rag/llm/sequence2txt_model.py

@@ -60,6 +60,34 @@ class QWenSeq2txt(Base):
         from dashscope import MultiModalConversation
 
         audio_path = f"file://{audio_path}"
+        prompt_ch = """
+        你是一名专业的音频转录助手，能够将MP3音频文件的内容转写为文本，并**精确标记每句话或每个段落对应的时间戳**（开始时间-结束时间）。\n
+        **任务要求**：
+        1.输入是MP3,解析带时间戳的文本。
+        2.时间戳格式为 `[HH:MM:SS.mmm]`（毫秒可选），例如 `[00:01:23.456]`。
+        3.时间戳需尽可能贴近实际语音的起止时间，误差不超过1秒。
+        4.如果无法确定具体时间，请根据上下文合理估算。
+        5.最后总结:这段音频在说什么?
+        
+        **示例输出**：
+        [00:00:00.000] 今天天气真好，
+        [00:00:02.500] 我们一起去公园散步吧。
+        [00:00:05.800] 公园里的花开得非常漂亮。
+        这段音频讲述的是一个关于**“吃水不忘挖井人”**的感人故事，主 ..."""
+        prompt_en = """
+        You are a professional audio transcription assistant, capable of transcribing the content of MP3 audio files into text and **precisely marking the timestamps (start time - end time) corresponding to each sentence or paragraph**. 
+        **Task requirements**: 
+        1. Input is MP3, parse text with timestamps.
+        2. The timestamp format is `[HH:MM:SS.mmm]` (milliseconds are optional), for example, `[00:01:23.456]`.
+        3. The timestamp should be as close as possible to the actual start and end time of the voice, with an error not exceeding 1 second.
+        4. If a specific time cannot be determined, please make a reasonable estimation based on the context.
+        5. Final summary: What is this audio talking about?
+        
+        **Example Output**: 
+        [00:00:00.000] The weather is really nice today, 
+        [00:00:02.500] let's go for a walk in the park together.
+        [00:00:05.800] The flowers in the park are blooming beautifully.
+        This audio tells a touching story about **"Remembering the one who dug the well when drinking water"** .."""
         messages = [
             {
                 "role": "user",
@@ -68,7 +96,7 @@ class QWenSeq2txt(Base):
                         "audio": audio_path
                     },
                     {
-                        "text": "这段音频在说什么?" if self.lang.lower() == "chinese" else "What is this audio saying?",
+                        "text": prompt_ch if self.lang.lower() == "chinese" else prompt_en,
                     },
                 ],
             }

api/app/core/workflow/executor.py

@@ -8,6 +8,7 @@ import logging
 import uuid
 from typing import Any
 
+from langchain_core.runnables import RunnableConfig
 from langgraph.graph.state import CompiledStateGraph
 
 from app.core.workflow.graph_builder import GraphBuilder
@@ -53,11 +54,11 @@ class WorkflowExecutor:
         self.edges = workflow_config.get("edges", [])
         self.execution_config = workflow_config.get("execution_config", {})
 
-        self.checkpoint_config = {
-            "configurable": {
+        self.checkpoint_config = RunnableConfig(
+            configurable={
                 "thread_id": uuid.uuid4(),
             }
-        }
+        )
 
     def _prepare_initial_state(self, input_data: dict[str, Any]) -> WorkflowState:
         """准备初始状态（注入系统变量和会话变量）
@@ -214,13 +215,13 @@ class WorkflowExecutor:
         return {
             "status": "completed",
             "output": final_output,
+            "variables": result.get("variables", {}),
             "node_outputs": node_outputs,
             "messages": result.get("messages", []),
             "conversation_id": conversation_id,
             "elapsed_time": elapsed_time,
             "token_usage": token_usage,
             "error": result.get("error"),
-            "variables": result.get("variables", {}),
         }
 
     def build_graph(self, stream=False) -> CompiledStateGraph:
@@ -326,11 +327,10 @@ class WorkflowExecutor:
         }
 
         # 1. 构建图
-        graph = self.build_graph(True)
+        graph = self.build_graph(stream=True)
 
         # 2. 初始化状态（自动注入系统变量）
         initial_state = self._prepare_initial_state(input_data)
-
         # 3. Execute workflow
         try:
             chunk_count = 0
@@ -346,14 +346,16 @@ class WorkflowExecutor:
                     mode, data = event
                 else:
                     # Unexpected format, log and skip
-                    logger.warning(f"[STREAM] Unexpected event format: {type(event)}, value: {event}")
+                    logger.warning(f"[STREAM] Unexpected event format: {type(event)}, value: {event}"
+                                   f"- execution_id: {self.execution_id}")
                     continue
 
                 if mode == "custom":
                     # Handle custom streaming events (chunks from nodes via stream writer)
                     chunk_count += 1
                     event_type = data.get("type", "node_chunk")  # "message" or "node_chunk"
-                    logger.info(f"[CUSTOM] ✅ 收到 {event_type} #{chunk_count} from {data.get('node_id')}")
+                    logger.info(f"[CUSTOM] ✅ 收到 {event_type} #{chunk_count} from {data.get('node_id')}"
+                                f"- execution_id: {self.execution_id}")
                     yield {
                         "event": event_type,  # "message" or "node_chunk"
                         "data": {
@@ -380,7 +382,8 @@ class WorkflowExecutor:
                         variables_sys = variables.get("sys", {})
                         conversation_id = input_data.get("conversation_id")
                         execution_id = variables_sys.get("execution_id")
-                        logger.info(f"[DEBUG] Node starts execution: {node_name}")
+                        logger.info(f"[NODE-START] Node starts execution: {node_name} "
+                                    f"- execution_id: {self.execution_id}")
 
                         yield {
                             "event": "node_start",
@@ -399,7 +402,8 @@ class WorkflowExecutor:
                         variables_sys = variables.get("sys", {})
                         conversation_id = input_data.get("conversation_id")
                         execution_id = variables_sys.get("execution_id")
-                        logger.info(f"[DEBUG] Node execution completed: {node_name}")
+                        logger.info(f"[NODE-END] Node execution completed: {node_name} "
+                                    f"- execution_id: {self.execution_id}")
 
                         yield {
                             "event": "node_end",
@@ -407,13 +411,15 @@ class WorkflowExecutor:
                                 "node_id": node_name,
                                 "conversation_id": conversation_id,
                                 "execution_id": execution_id,
-                                "timestamp": data.get("timestamp")
+                                "timestamp": data.get("timestamp"),
+                                "state": result.get("node_outputs", {}).get(node_name),
                             }
                         }
 
                 elif mode == "updates":
                     # Handle state updates - store final state
-                    logger.debug(f"[UPDATES] 收到 state 更新 from {list(data.keys())}")
+                    logger.debug(f"[UPDATES] 收到 state 更新 from {list(data.keys())} "
+                                 f"- execution_id: {self.execution_id}")
 
             # 计算耗时
             end_time = datetime.datetime.now()
@@ -421,7 +427,7 @@ class WorkflowExecutor:
             result = graph.get_state(self.checkpoint_config).values
             logger.info(
                 f"Workflow execution completed (streaming), "
-                f"total chunks: {chunk_count}, elapsed: {elapsed_time:.2f}s"
+                f"total chunks: {chunk_count}, elapsed: {elapsed_time:.2f}s, execution_id: {self.execution_id}"
             )
 
             # 发送 workflow_end 事件
@@ -449,7 +455,8 @@ class WorkflowExecutor:
                 }
             }
 
-    def _extract_final_output(self, node_outputs: dict[str, Any]) -> str | None:
+    @staticmethod
+    def _extract_final_output(node_outputs: dict[str, Any]) -> str | None:
         """从节点输出中提取最终输出
 
         优先级：
@@ -473,7 +480,8 @@ class WorkflowExecutor:
 
         return None
 
-    def _aggregate_token_usage(self, node_outputs: dict[str, Any]) -> dict[str, int] | None:
+    @staticmethod
+    def _aggregate_token_usage(node_outputs: dict[str, Any]) -> dict[str, int] | None:
         """聚合所有节点的 token 使用情况
 
         Args:

api/app/core/workflow/nodes/base_node.py

@@ -25,7 +25,7 @@ class WorkflowState(TypedDict):
     The state object passed between nodes in a workflow, containing messages, variables, node outputs, etc.
     """
     # List of messages (append mode)
-    messages: list[dict[str, str]]
+    messages: Annotated[list[dict[str, str]], lambda x, y: y]
 
     # Set of loop node IDs, used for assigning values in loop nodes
     cycle_nodes: list

api/app/core/workflow/nodes/cycle_graph/iteration.py

@@ -21,6 +21,7 @@ class IterationRuntime:
     optional parallel execution, flattening of output, and loop control via
     the workflow state.
     """
+
     def __init__(
             self,
             graph: CompiledStateGraph,
@@ -87,6 +88,7 @@ class IterationRuntime:
             self.result.append(output)
         if not result["looping"]:
             self.looping = False
+        return result
 
     def _create_iteration_tasks(self, array_obj, idx):
         """
@@ -124,7 +126,7 @@ class IterationRuntime:
         array_obj = VariablePool(self.state).get(input_expression)
         if not isinstance(array_obj, list):
             raise RuntimeError("Cannot iterate over a non-list variable")
-
+        child_state = []
         idx = 0
         if self.typed_config.parallel:
             # Execute iterations in parallel batches
@@ -132,15 +134,14 @@ class IterationRuntime:
                 tasks = self._create_iteration_tasks(array_obj, idx)
                 logger.info(f"Iteration node {self.node_id}: running, concurrency {len(tasks)}")
                 idx += self.typed_config.parallel_count
-                await asyncio.gather(*tasks)
-            logger.info(f"Iteration node {self.node_id}: execution completed")
-            return self.result
+                child_state.extend(await asyncio.gather(*tasks))
         else:
             # Execute iterations sequentially
             while idx < len(array_obj) and self.looping:
                 logger.info(f"Iteration node {self.node_id}: running")
                 item = array_obj[idx]
                 result = await self.graph.ainvoke(self._init_iteration_state(item, idx))
+                child_state.append(result)
                 output = VariablePool(result).get(self.output_value)
                 if isinstance(output, list) and self.typed_config.flatten:
                     self.result.extend(output)
@@ -150,5 +151,8 @@ class IterationRuntime:
                     self.looping = False
                 idx += 1
 
-            logger.info(f"Iteration node {self.node_id}: execution completed")
-            return self.result
+        logger.info(f"Iteration node {self.node_id}: execution completed")
+        return {
+            "output": self.result,
+            "__child_state": child_state
+        }

api/app/core/workflow/nodes/cycle_graph/loop.py

@@ -67,7 +67,9 @@ class LoopRuntime:
                 variables=pool.get_all_conversation_vars(),
                 node_outputs=pool.get_all_node_outputs(),
                 system_vars=pool.get_all_system_vars(),
-            ) if variable.input_type == ValueInputType.VARIABLE else TypeTransformer.transform(variable.value, variable.type)
+            )
+            if variable.input_type == ValueInputType.VARIABLE
+            else TypeTransformer.transform(variable.value, variable.type)
             for variable in self.typed_config.cycle_vars
         }
         self.state["node_outputs"][self.node_id] = {
@@ -76,7 +78,9 @@ class LoopRuntime:
                 variables=pool.get_all_conversation_vars(),
                 node_outputs=pool.get_all_node_outputs(),
                 system_vars=pool.get_all_system_vars(),
-            ) if variable.input_type == ValueInputType.VARIABLE else TypeTransformer.transform(variable.value, variable.type)
+            )
+            if variable.input_type == ValueInputType.VARIABLE
+            else TypeTransformer.transform(variable.value, variable.type)
             for variable in self.typed_config.cycle_vars
         }
         loopstate = WorkflowState(
@@ -171,10 +175,11 @@ class LoopRuntime:
         """
         loopstate = self._init_loop_state()
         loop_time = self.typed_config.max_loop
+        child_state = []
         while self.evaluate_conditional(loopstate) and loopstate["looping"] and loop_time > 0:
             logger.info(f"loop node {self.node_id}: running")
-            await self.graph.ainvoke(loopstate)
+            child_state.append(await self.graph.ainvoke(loopstate))
             loop_time -= 1
 
         logger.info(f"loop node {self.node_id}: execution completed")
-        return loopstate["runtime_vars"][self.node_id]
+        return loopstate["runtime_vars"][self.node_id] | {"__child_state": child_state}

api/app/core/workflow/nodes/knowledge/node.py

@@ -10,9 +10,8 @@ from app.core.workflow.nodes.base_node import BaseNode, WorkflowState
 from app.core.workflow.nodes.knowledge import KnowledgeRetrievalNodeConfig
 from app.db import get_db_read
 from app.models import knowledge_model, knowledgeshare_model, ModelType
-from app.repositories import knowledge_repository
+from app.repositories import knowledge_repository, knowledgeshare_repository
 from app.schemas.chunk_schema import RetrieveType
-from app.services import knowledge_service, knowledgeshare_service
 from app.services.model_service import ModelConfigService
 
 logger = logging.getLogger(__name__)
@@ -96,7 +95,7 @@ class KnowledgeRetrievalNode(BaseNode):
 
         filters = self._build_kb_filter(kb_ids, knowledge_model.PermissionType.Share)
 
-        share_ids = knowledge_service.knowledge_repository.get_chunked_knowledgeids(
+        share_ids = knowledge_repository.get_chunked_knowledgeids(
             db=db,
             filters=filters
         )
@@ -105,7 +104,7 @@ class KnowledgeRetrievalNode(BaseNode):
             filters = [
                 knowledgeshare_model.KnowledgeShare.target_kb_id.in_(kb_ids)
             ]
-            items = knowledgeshare_service.knowledgeshare_repository.get_source_kb_ids_by_target_kb_id(
+            items = knowledgeshare_repository.get_source_kb_ids_by_target_kb_id(
                 db=db,
                 filters=filters
             )

api/app/core/workflow/nodes/llm/config.py

@@ -66,7 +66,7 @@ class LLMNodeConfig(BaseNodeConfig):
     )
 
     memory: MemoryWindowSetting = Field(
-        ...,
+        default_factory=MemoryWindowSetting,
         description="对话上下文窗口"
     )
 

api/app/core/workflow/nodes/llm/node.py

@@ -85,6 +85,7 @@ class LLMNode(BaseNode):
         """
 
         # 1. 处理消息格式（优先使用 messages）
+        self.typed_config = LLMNodeConfig(**self.config)
         messages_config = self.typed_config.messages
 
         if messages_config:
@@ -167,7 +168,7 @@ class LLMNode(BaseNode):
         Returns:
             LLM 响应消息
         """
-        self.typed_config = LLMNodeConfig(**self.config)
+        # self.typed_config = LLMNodeConfig(**self.config)
         llm, prompt_or_messages = self._prepare_llm(state, True)
 
         logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（非流式）")
@@ -269,12 +270,16 @@ class LLMNode(BaseNode):
         chunk_count = 0
 
         # 调用 LLM（流式，支持字符串或消息列表）
-        async for chunk in llm.astream(prompt_or_messages):
+        last_meta_data = {}
+        async for chunk in llm.astream(prompt_or_messages, stream_usage=True):
             # 提取内容
             if hasattr(chunk, 'content'):
                 content = chunk.content
             else:
                 content = str(chunk)
+            if hasattr(chunk, 'response_metadata'):
+                if chunk.response_metadata:
+                    last_meta_data = chunk.response_metadata
 
             # 只有当内容不为空时才处理
             if content:
@@ -288,13 +293,10 @@ class LLMNode(BaseNode):
         logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(full_response)}, 总 chunks: {chunk_count}")
 
         # 构建完整的 AIMessage（包含元数据）
-        if isinstance(last_chunk, AIMessage):
-            final_message = AIMessage(
-                content=full_response,
-                response_metadata=last_chunk.response_metadata if hasattr(last_chunk, 'response_metadata') else {}
-            )
-        else:
-            final_message = AIMessage(content=full_response)
+        final_message = AIMessage(
+            content=full_response,
+            response_metadata=last_meta_data
+        )
 
         # yield 完成标记
         yield {"__final__": True, "result": final_message}

api/app/models/__init__.py

@@ -27,8 +27,6 @@ from .tool_model import (
     ToolExecution, ToolType, ToolStatus, AuthType, ExecutionStatus
 )
 from .memory_perceptual_model import MemoryPerceptualModel
-from .emotion_suggestions_cache_model import EmotionSuggestionsCache
-from .implicit_memory_cache_model import ImplicitMemoryCache
 
 __all__ = [
     "Tenants",
@@ -79,6 +77,4 @@ __all__ = [
     "AuthType",
     "ExecutionStatus",
     "MemoryPerceptualModel",
-    "EmotionSuggestionsCache",
-    "ImplicitMemoryCache"
 ]

api/app/models/emotion_suggestions_cache_model.py

@@ -1,24 +0,0 @@
-"""情绪建议缓存模型"""
-
-import uuid
-import datetime
-from sqlalchemy import Column, String, Text, Integer, DateTime, JSON
-from sqlalchemy.dialects.postgresql import UUID
-from app.db import Base
-
-
-class EmotionSuggestionsCache(Base):
-    """情绪建议缓存表
-    
-    用于缓存个性化情绪建议，减少 LLM 调用成本，提升响应速度。
-    """
-    __tablename__ = "emotion_suggestions_cache"
-
-    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
-    end_user_id = Column(String(255), nullable=False, unique=True, index=True, comment="终端用户ID（组ID）")
-    health_summary = Column(Text, nullable=False, comment="健康状态摘要")
-    suggestions = Column(JSON, nullable=False, comment="建议列表（JSON格式）")
-    generated_at = Column(DateTime, nullable=False, default=datetime.datetime.now, comment="生成时间")
-    expires_at = Column(DateTime, nullable=True, comment="过期时间")
-    created_at = Column(DateTime, default=datetime.datetime.now)
-    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

api/app/models/implicit_memory_cache_model.py

@@ -1,27 +0,0 @@
-"""隐性记忆缓存模型"""
-
-import uuid
-import datetime
-from sqlalchemy import Column, String, Integer, DateTime, JSON
-from sqlalchemy.dialects.postgresql import UUID
-from app.db import Base
-
-
-class ImplicitMemoryCache(Base):
-    """隐性记忆缓存表
-    
-    用于缓存用户的完整隐性记忆画像，包括偏好标签、四维画像、兴趣领域和行为习惯。
-    减少 LLM 调用成本，提升响应速度。
-    """
-    __tablename__ = "implicit_memory_cache"
-
-    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
-    end_user_id = Column(String(255), nullable=False, unique=True, index=True, comment="终端用户ID")
-    preferences = Column(JSON, nullable=False, comment="偏好标签列表（JSON格式）")
-    portrait = Column(JSON, nullable=False, comment="四维画像对象（JSON格式）")
-    interest_areas = Column(JSON, nullable=False, comment="兴趣领域分布对象（JSON格式）")
-    habits = Column(JSON, nullable=False, comment="行为习惯列表（JSON格式）")
-    generated_at = Column(DateTime, nullable=False, default=datetime.datetime.now, comment="生成时间")
-    expires_at = Column(DateTime, nullable=True, comment="过期时间")
-    created_at = Column(DateTime, default=datetime.datetime.now)
-    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

api/app/models/workflow_model.py

@@ -75,6 +75,14 @@ class WorkflowExecution(Base):
         nullable=False,
         index=True
     )
+
+    release_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey("app_releases.id", ondelete="CASCADE"),
+        nullable=True,
+        index=True
+    )
+
     app_id = Column(
         UUID(as_uuid=True),
         ForeignKey("apps.id", ondelete="CASCADE"),

api/app/repositories/emotion_suggestions_cache_repository.py

@@ -1,163 +0,0 @@
-"""情绪建议缓存仓储层"""
-
-from sqlalchemy.orm import Session
-from typing import Optional, Dict, Any
-import datetime
-
-from app.models.emotion_suggestions_cache_model import EmotionSuggestionsCache
-from app.core.logging_config import get_db_logger
-
-# 获取数据库专用日志器
-db_logger = get_db_logger()
-
-
-class EmotionSuggestionsCacheRepository:
-    """情绪建议缓存仓储类"""
-    
-    def __init__(self, db: Session):
-        self.db = db
-
-    def get_by_end_user_id(self, end_user_id: str) -> Optional[EmotionSuggestionsCache]:
-        """根据终端用户ID获取缓存
-        
-        Args:
-            end_user_id: 终端用户ID（组ID）
-            
-        Returns:
-            缓存记录，如果不存在返回 None
-        """
-        try:
-            cache = (
-                self.db.query(EmotionSuggestionsCache)
-                .filter(EmotionSuggestionsCache.end_user_id == end_user_id)
-                .first()
-            )
-            if cache:
-                db_logger.info(f"成功获取用户 {end_user_id} 的情绪建议缓存")
-            else:
-                db_logger.info(f"用户 {end_user_id} 的情绪建议缓存不存在")
-            return cache
-        except Exception as e:
-            db_logger.error(f"获取用户 {end_user_id} 的情绪建议缓存失败: {str(e)}")
-            raise
-
-    def create_or_update(
-        self,
-        end_user_id: str,
-        health_summary: str,
-        suggestions: list,
-        expires_hours: int = 24
-    ) -> EmotionSuggestionsCache:
-        """创建或更新缓存
-        
-        Args:
-            end_user_id: 终端用户ID（组ID）
-            health_summary: 健康状态摘要
-            suggestions: 建议列表
-            expires_hours: 过期时间（小时），默认24小时
-            
-        Returns:
-            缓存记录
-        """
-        try:
-            # 查找现有记录
-            cache = self.get_by_end_user_id(end_user_id)
-            
-            now = datetime.datetime.now()
-            expires_at = now + datetime.timedelta(hours=expires_hours)
-            
-            if cache:
-                # 更新现有记录
-                cache.health_summary = health_summary
-                cache.suggestions = suggestions
-                cache.generated_at = now
-                cache.expires_at = expires_at
-                cache.updated_at = now
-                db_logger.info(f"更新用户 {end_user_id} 的情绪建议缓存")
-            else:
-                # 创建新记录
-                cache = EmotionSuggestionsCache(
-                    end_user_id=end_user_id,
-                    health_summary=health_summary,
-                    suggestions=suggestions,
-                    generated_at=now,
-                    expires_at=expires_at,
-                    created_at=now,
-                    updated_at=now
-                )
-                self.db.add(cache)
-                db_logger.info(f"创建用户 {end_user_id} 的情绪建议缓存")
-            
-            self.db.commit()
-            self.db.refresh(cache)
-            return cache
-        except Exception as e:
-            self.db.rollback()
-            db_logger.error(f"创建或更新用户 {end_user_id} 的情绪建议缓存失败: {str(e)}")
-            raise
-
-    def delete_by_end_user_id(self, end_user_id: str) -> bool:
-        """删除缓存
-        
-        Args:
-            end_user_id: 终端用户ID（组ID）
-            
-        Returns:
-            是否删除成功
-        """
-        try:
-            cache = self.get_by_end_user_id(end_user_id)
-            if cache:
-                self.db.delete(cache)
-                self.db.commit()
-                db_logger.info(f"删除用户 {end_user_id} 的情绪建议缓存")
-                return True
-            return False
-        except Exception as e:
-            self.db.rollback()
-            db_logger.error(f"删除用户 {end_user_id} 的情绪建议缓存失败: {str(e)}")
-            raise
-
-    @staticmethod
-    def is_expired(cache: EmotionSuggestionsCache) -> bool:
-        """检查缓存是否过期
-        
-        Args:
-            cache: 缓存记录
-            
-        Returns:
-            是否过期
-        """
-        if cache.expires_at is None:
-            return False
-        return datetime.datetime.now() > cache.expires_at
-
-
-# 便捷函数
-def get_cache_by_end_user_id(db: Session, end_user_id: str) -> Optional[EmotionSuggestionsCache]:
-    """根据终端用户ID获取缓存"""
-    repo = EmotionSuggestionsCacheRepository(db)
-    return repo.get_by_end_user_id(end_user_id)
-
-
-def create_or_update_cache(
-    db: Session,
-    end_user_id: str,
-    health_summary: str,
-    suggestions: list,
-    expires_hours: int = 24
-) -> EmotionSuggestionsCache:
-    """创建或更新缓存"""
-    repo = EmotionSuggestionsCacheRepository(db)
-    return repo.create_or_update(end_user_id, health_summary, suggestions, expires_hours)
-
-
-def delete_cache_by_end_user_id(db: Session, end_user_id: str) -> bool:
-    """删除缓存"""
-    repo = EmotionSuggestionsCacheRepository(db)
-    return repo.delete_by_end_user_id(end_user_id)
-
-
-def is_cache_expired(cache: EmotionSuggestionsCache) -> bool:
-    """检查缓存是否过期"""
-    return EmotionSuggestionsCacheRepository.is_expired(cache)

api/app/repositories/implicit_memory_cache_repository.py

@@ -1,175 +0,0 @@
-"""隐性记忆缓存仓储层"""
-
-from sqlalchemy.orm import Session
-from typing import Optional, Dict, Any
-import datetime
-
-from app.models.implicit_memory_cache_model import ImplicitMemoryCache
-from app.core.logging_config import get_db_logger
-
-# 获取数据库专用日志器
-db_logger = get_db_logger()
-
-
-class ImplicitMemoryCacheRepository:
-    """隐性记忆缓存仓储类"""
-    
-    def __init__(self, db: Session):
-        self.db = db
-
-    def get_by_end_user_id(self, end_user_id: str) -> Optional[ImplicitMemoryCache]:
-        """根据终端用户ID获取缓存
-        
-        Args:
-            end_user_id: 终端用户ID
-            
-        Returns:
-            缓存记录，如果不存在返回 None
-        """
-        try:
-            cache = (
-                self.db.query(ImplicitMemoryCache)
-                .filter(ImplicitMemoryCache.end_user_id == end_user_id)
-                .first()
-            )
-            if cache:
-                db_logger.info(f"成功获取用户 {end_user_id} 的隐性记忆缓存")
-            else:
-                db_logger.info(f"用户 {end_user_id} 的隐性记忆缓存不存在")
-            return cache
-        except Exception as e:
-            db_logger.error(f"获取用户 {end_user_id} 的隐性记忆缓存失败: {str(e)}")
-            raise
-
-    def create_or_update(
-        self,
-        end_user_id: str,
-        preferences: list,
-        portrait: dict,
-        interest_areas: dict,
-        habits: list,
-        expires_hours: int = 168  # 默认7天
-    ) -> ImplicitMemoryCache:
-        """创建或更新缓存
-        
-        Args:
-            end_user_id: 终端用户ID
-            preferences: 偏好标签列表
-            portrait: 四维画像对象
-            interest_areas: 兴趣领域分布对象
-            habits: 行为习惯列表
-            expires_hours: 过期时间（小时），默认168小时（7天）
-            
-        Returns:
-            缓存记录
-        """
-        try:
-            # 查找现有记录
-            cache = self.get_by_end_user_id(end_user_id)
-            
-            now = datetime.datetime.now()
-            expires_at = now + datetime.timedelta(hours=expires_hours)
-            
-            if cache:
-                # 更新现有记录
-                cache.preferences = preferences
-                cache.portrait = portrait
-                cache.interest_areas = interest_areas
-                cache.habits = habits
-                cache.generated_at = now
-                cache.expires_at = expires_at
-                cache.updated_at = now
-                db_logger.info(f"更新用户 {end_user_id} 的隐性记忆缓存")
-            else:
-                # 创建新记录
-                cache = ImplicitMemoryCache(
-                    end_user_id=end_user_id,
-                    preferences=preferences,
-                    portrait=portrait,
-                    interest_areas=interest_areas,
-                    habits=habits,
-                    generated_at=now,
-                    expires_at=expires_at,
-                    created_at=now,
-                    updated_at=now
-                )
-                self.db.add(cache)
-                db_logger.info(f"创建用户 {end_user_id} 的隐性记忆缓存")
-            
-            self.db.commit()
-            self.db.refresh(cache)
-            return cache
-        except Exception as e:
-            self.db.rollback()
-            db_logger.error(f"创建或更新用户 {end_user_id} 的隐性记忆缓存失败: {str(e)}")
-            raise
-
-    def delete_by_end_user_id(self, end_user_id: str) -> bool:
-        """删除缓存
-        
-        Args:
-            end_user_id: 终端用户ID
-            
-        Returns:
-            是否删除成功
-        """
-        try:
-            cache = self.get_by_end_user_id(end_user_id)
-            if cache:
-                self.db.delete(cache)
-                self.db.commit()
-                db_logger.info(f"删除用户 {end_user_id} 的隐性记忆缓存")
-                return True
-            return False
-        except Exception as e:
-            self.db.rollback()
-            db_logger.error(f"删除用户 {end_user_id} 的隐性记忆缓存失败: {str(e)}")
-            raise
-
-    @staticmethod
-    def is_expired(cache: ImplicitMemoryCache) -> bool:
-        """检查缓存是否过期
-        
-        Args:
-            cache: 缓存记录
-            
-        Returns:
-            是否过期
-        """
-        if cache.expires_at is None:
-            return False
-        return datetime.datetime.now() > cache.expires_at
-
-
-# 便捷函数
-def get_cache_by_end_user_id(db: Session, end_user_id: str) -> Optional[ImplicitMemoryCache]:
-    """根据终端用户ID获取缓存"""
-    repo = ImplicitMemoryCacheRepository(db)
-    return repo.get_by_end_user_id(end_user_id)
-
-
-def create_or_update_cache(
-    db: Session,
-    end_user_id: str,
-    preferences: list,
-    portrait: dict,
-    interest_areas: dict,
-    habits: list,
-    expires_hours: int = 168
-) -> ImplicitMemoryCache:
-    """创建或更新缓存"""
-    repo = ImplicitMemoryCacheRepository(db)
-    return repo.create_or_update(
-        end_user_id, preferences, portrait, interest_areas, habits, expires_hours
-    )
-
-
-def delete_cache_by_end_user_id(db: Session, end_user_id: str) -> bool:
-    """删除缓存"""
-    repo = ImplicitMemoryCacheRepository(db)
-    return repo.delete_by_end_user_id(end_user_id)
-
-
-def is_cache_expired(cache: ImplicitMemoryCache) -> bool:
-    """检查缓存是否过期"""
-    return ImplicitMemoryCacheRepository.is_expired(cache)

api/app/schemas/emotion_schema.py

@@ -34,5 +34,4 @@ class EmotionSuggestionsRequest(BaseModel):
 
 class EmotionGenerateSuggestionsRequest(BaseModel):
     """生成个性化情绪建议请求"""
-    group_id: str = Field(..., description="组ID")
-    config_id: Optional[int] = Field(None, description="配置ID（用于指定LLM模型）")
+    end_user_id: str = Field(..., description="终端用户ID")

api/app/services/app_chat_service.py

@@ -527,6 +527,7 @@ class AppChatService:
             conversation_id: uuid.UUID,
             config: WorkflowConfig,
             app_id: uuid.UUID,
+            release_id: uuid.UUID,
             workspace_id: uuid.UUID,
             user_id: Optional[str] = None,
             variables: Optional[Dict[str, Any]] = None,
@@ -549,6 +550,7 @@ class AppChatService:
             payload=payload,
             config=config,
             workspace_id=workspace_id,
+            release_id=release_id,
         )
 
     async def workflow_chat_stream(
@@ -557,6 +559,7 @@ class AppChatService:
             conversation_id: uuid.UUID,
             config: WorkflowConfig,
             app_id: uuid.UUID,
+            release_id: uuid.UUID,
             workspace_id: uuid.UUID,
             user_id: str = None,
             variables: Optional[Dict[str, Any]] = None,
@@ -565,7 +568,7 @@ class AppChatService:
             storage_type: Optional[str] = None,
             user_rag_memory_id: Optional[str] = None,
 
-    ) -> AsyncGenerator[str, None]:
+    ) -> AsyncGenerator[dict, None]:
         """聊天（流式）"""
         workflow_service = WorkflowService(self.db)
         payload = DraftRunRequest(
@@ -580,6 +583,7 @@ class AppChatService:
                 payload=payload,
                 config=config,
                 workspace_id=workspace_id,
+                release_id=release_id
         ):
             yield event
 

api/app/services/app_service.py

@@ -129,7 +129,7 @@ class AppService:
         Raises:
             ResourceNotFoundException: 当应用不存在时
         """
-        app = get_apps_by_id(self.db,app_id)
+        app = get_apps_by_id(self.db, app_id)
         if not app:
             logger.warning("应用不存在", extra={"app_id": str(app_id)})
             raise ResourceNotFoundException("应用", str(app_id))
@@ -227,7 +227,6 @@ class AppService:
             if not model_api_key:
                 raise ResourceNotFoundException("模型配置", str(multi_agent_config.default_model_config_id))
 
-
         # 3. 检查子 Agent 配置
         if not multi_agent_config.sub_agents or len(multi_agent_config.sub_agents) == 0:
             raise BusinessException(
@@ -281,10 +280,10 @@ class AppService:
         )
 
     def _create_agent_config(
-        self,
-        app_id: uuid.UUID,
-        config_data: app_schema.AgentConfigCreate,
-        now: datetime.datetime
+            self,
+            app_id: uuid.UUID,
+            config_data: app_schema.AgentConfigCreate,
+            now: datetime.datetime
     ) -> None:
         """创建 Agent 配置（内部方法）
 
@@ -313,10 +312,10 @@ class AppService:
         logger.debug("Agent 配置已创建", extra={"app_id": str(app_id)})
 
     def _create_multi_agent_config(
-        self,
-        app_id: uuid.UUID,
-        config_data: Dict[str, Any],
-        now: datetime.datetime
+            self,
+            app_id: uuid.UUID,
+            config_data: Dict[str, Any],
+            now: datetime.datetime
     ) -> None:
         """创建多 Agent 配置（内部方法）
 
@@ -411,9 +410,9 @@ class AppService:
         return 1 if max_ver is None else int(max_ver) + 1
 
     def _convert_to_schema(
-        self,
-        app: App,
-        current_workspace_id: uuid.UUID
+            self,
+            app: App,
+            current_workspace_id: uuid.UUID
     ) -> app_schema.App:
         """将 App 模型转换为 Schema，并设置 is_shared 字段
 
@@ -447,9 +446,9 @@ class AppService:
     # ==================== 应用管理 ====================
 
     def get_app(
-        self,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> App:
         """获取应用详情
 
@@ -469,11 +468,11 @@ class AppService:
         return app
 
     def create_app(
-        self,
-        *,
-        user_id: uuid.UUID,
-        workspace_id: uuid.UUID,
-        data: app_schema.AppCreate
+            self,
+            *,
+            user_id: uuid.UUID,
+            workspace_id: uuid.UUID,
+            data: app_schema.AppCreate
     ) -> App:
         """创建应用
 
@@ -535,11 +534,11 @@ class AppService:
             raise BusinessException(f"应用创建失败: {str(e)}", BizCode.INTERNAL_ERROR, cause=e)
 
     def update_app(
-        self,
-        *,
-        app_id: uuid.UUID,
-        data: app_schema.AppUpdate,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            data: app_schema.AppUpdate,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> App:
         """更新应用基本信息
 
@@ -578,10 +577,10 @@ class AppService:
         return app
 
     def delete_app(
-        self,
-        *,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> None:
         """删除应用
 
@@ -612,12 +611,12 @@ class AppService:
         )
 
     def copy_app(
-        self,
-        *,
-        app_id: uuid.UUID,
-        user_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None,
-        new_name: Optional[str] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            user_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None,
+            new_name: Optional[str] = None
     ) -> App:
         """复制应用（包括基础信息和配置）
 
@@ -716,16 +715,16 @@ class AppService:
             raise BusinessException(f"应用复制失败: {str(e)}", BizCode.INTERNAL_ERROR, cause=e)
 
     def list_apps(
-        self,
-        *,
-        workspace_id: uuid.UUID,
-        type: Optional[str] = None,
-        visibility: Optional[str] = None,
-        status: Optional[str] = None,
-        search: Optional[str] = None,
-        include_shared: bool = True,
-        page: int = 1,
-        pagesize: int = 10,
+            self,
+            *,
+            workspace_id: uuid.UUID,
+            type: Optional[str] = None,
+            visibility: Optional[str] = None,
+            status: Optional[str] = None,
+            search: Optional[str] = None,
+            include_shared: bool = True,
+            page: int = 1,
+            pagesize: int = 10,
     ) -> Tuple[List[App], int]:
         """列出工作空间中的应用（分页）
 
@@ -759,8 +758,7 @@ class AppService:
         )
 
         # 构建查询条件
-        filters = []
-        filters.append(App.is_active == True)
+        filters = [App.is_active == True]
         if type:
             filters.append(App.type == type)
         if visibility:
@@ -813,9 +811,9 @@ class AppService:
         return items, int(total)
 
     def get_apps_by_ids(
-        self,
-        app_ids: List[str],
-        workspace_id: uuid.UUID
+            self,
+            app_ids: List[str],
+            workspace_id: uuid.UUID
     ) -> List[App]:
         """根据ID列表获取应用
 
@@ -846,11 +844,11 @@ class AppService:
     # ==================== Agent 配置管理 ====================
 
     def update_agent_config(
-        self,
-        *,
-        app_id: uuid.UUID,
-        data: app_schema.AgentConfigUpdate,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            data: app_schema.AgentConfigUpdate,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> AgentConfig:
         """更新 Agent 配置
 
@@ -875,7 +873,8 @@ class AppService:
 
         self._validate_workspace_access(app, workspace_id)
 
-        stmt = select(AgentConfig).where(AgentConfig.app_id == app_id, AgentConfig.is_active==True).order_by(AgentConfig.updated_at.desc())
+        stmt = select(AgentConfig).where(AgentConfig.app_id == app_id, AgentConfig.is_active == True).order_by(
+            AgentConfig.updated_at.desc())
         agent_cfg: Optional[AgentConfig] = self.db.scalars(stmt).first()
         now = datetime.datetime.now()
 
@@ -918,10 +917,10 @@ class AppService:
         return agent_cfg
 
     def get_agent_config(
-        self,
-        *,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> AgentConfig:
         """获取 Agent 配置
 
@@ -948,7 +947,12 @@ class AppService:
         # 只读操作，允许访问共享应用
         self._validate_app_accessible(app, workspace_id)
 
-        stmt = select(AgentConfig).where(AgentConfig.app_id == app_id, AgentConfig.is_active == True).order_by(AgentConfig.updated_at.desc())
+        stmt = select(AgentConfig).where(
+            AgentConfig.app_id == app_id,
+            AgentConfig.is_active.is_(True)
+        ).order_by(
+            AgentConfig.updated_at.desc()
+        )
         config = self.db.scalars(stmt).first()
 
         if config:
@@ -1166,13 +1170,13 @@ class AppService:
     # ==================== 应用发布管理 ====================
 
     def publish(
-        self,
-        *,
-        app_id: uuid.UUID,
-        publisher_id: uuid.UUID,
-        version_name: str,
-        workspace_id: Optional[uuid.UUID] = None,
-        release_notes: Optional[str] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            publisher_id: uuid.UUID,
+            version_name: str,
+            workspace_id: Optional[uuid.UUID] = None,
+            release_notes: Optional[str] = None
     ) -> AppRelease:
         """发布应用（创建不可变快照）
 
@@ -1200,7 +1204,8 @@ class AppService:
         default_model_config_id = None
 
         if app.type == AppType.AGENT:
-            stmt = select(AgentConfig).where(AgentConfig.app_id == app_id, AgentConfig.is_active == True).order_by(AgentConfig.updated_at.desc())
+            stmt = select(AgentConfig).where(AgentConfig.app_id == app_id, AgentConfig.is_active == True).order_by(
+                AgentConfig.updated_at.desc())
             agent_cfg = self.db.scalars(stmt).first()
             if not agent_cfg:
                 raise BusinessException("Agent 应用缺少配置，无法发布", BizCode.AGENT_CONFIG_MISSING)
@@ -1236,8 +1241,7 @@ class AppService:
             default_model_config_id = multi_agent_cfg.default_model_config_id
 
             # 4. 构建配置快照
-            
-            
+
             config = {
                 "model_parameters": model_parameters_to_dict(multi_agent_cfg.model_parameters),
                 "master_agent_id": str(multi_agent_cfg.master_agent_id),
@@ -1264,6 +1268,7 @@ class AppService:
                 raise BusinessException("应用缺少有效配置，无法发布", BizCode.CONFIG_MISSING)
 
             config = {
+                "id": str(workflow_cfg.id),
                 "nodes": workflow_cfg.nodes,
                 "edges": workflow_cfg.edges,
                 "variables": workflow_cfg.variables,
@@ -1285,7 +1290,7 @@ class AppService:
             id=uuid.uuid4(),
             app_id=app_id,
             version=version,
-            version_name = version_name,
+            version_name=version_name,
             release_notes=release_notes,
             name=app.name,
             description=app.description,
@@ -1319,10 +1324,10 @@ class AppService:
         return release
 
     def get_current_release(
-        self,
-        *,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> Optional[AppRelease]:
         """获取当前发布版本
 
@@ -1349,10 +1354,10 @@ class AppService:
         return self.db.get(AppRelease, app.current_release_id)
 
     def list_releases(
-        self,
-        *,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> List[AppRelease]:
         """列出应用的所有发布版本（倒序）
 
@@ -1381,11 +1386,11 @@ class AppService:
         return list(self.db.scalars(stmt).all())
 
     def rollback(
-        self,
-        *,
-        app_id: uuid.UUID,
-        version: int,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            version: int,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> AppRelease:
         """回滚到指定版本
 
@@ -1434,12 +1439,12 @@ class AppService:
     # ==================== 应用分享功能 ====================
 
     def share_app(
-        self,
-        *,
-        app_id: uuid.UUID,
-        target_workspace_ids: List[uuid.UUID],
-        user_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            target_workspace_ids: List[uuid.UUID],
+            user_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> AppShare:
         """分享应用到其他工作空间
 
@@ -1457,7 +1462,6 @@ class AppService:
             BusinessException: 当应用不在指定工作空间或目标工作空间无效时
         """
 
-
         logger.info(
             "分享应用",
             extra={
@@ -1536,11 +1540,11 @@ class AppService:
         return shares
 
     def unshare_app(
-        self,
-        *,
-        app_id: uuid.UUID,
-        target_workspace_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            target_workspace_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> None:
         """取消应用分享
 
@@ -1594,10 +1598,10 @@ class AppService:
         )
 
     def list_app_shares(
-        self,
-        *,
-        app_id: uuid.UUID,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> List[AppShare]:
         """列出应用的所有分享记录
 
@@ -1637,14 +1641,14 @@ class AppService:
     # ==================== 试运行功能 ====================
 
     async def draft_run(
-        self,
-        *,
-        app_id: uuid.UUID,
-        message: str,
-        conversation_id: Optional[str] = None,
-        user_id: Optional[str] = None,
-        variables: Optional[Dict[str, Any]] = None,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            message: str,
+            conversation_id: Optional[str] = None,
+            user_id: Optional[str] = None,
+            variables: Optional[Dict[str, Any]] = None,
+            workspace_id: Optional[uuid.UUID] = None
     ) -> Dict[str, Any]:
         """试运行 Agent（使用当前草稿配置）
 
@@ -1736,14 +1740,14 @@ class AppService:
         return result
 
     async def draft_run_stream(
-        self,
-        *,
-        app_id: uuid.UUID,
-        message: str,
-        conversation_id: Optional[str] = None,
-        user_id: Optional[str] = None,
-        variables: Optional[Dict[str, Any]] = None,
-        workspace_id: Optional[uuid.UUID] = None
+            self,
+            *,
+            app_id: uuid.UUID,
+            message: str,
+            conversation_id: Optional[str] = None,
+            user_id: Optional[str] = None,
+            variables: Optional[Dict[str, Any]] = None,
+            workspace_id: Optional[uuid.UUID] = None
     ):
         """试运行 Agent（流式返回）
 
@@ -1794,30 +1798,30 @@ class AppService:
         # 4. 调用流式试运行服务
         draft_service = DraftRunService(self.db)
         async for event in draft_service.run_stream(
-            agent_config=agent_cfg,
-            model_config=model_config,
-            message=message,
-            workspace_id=workspace_id,
-            conversation_id=conversation_id,
-            user_id=user_id,
-            variables=variables
+                agent_config=agent_cfg,
+                model_config=model_config,
+                message=message,
+                workspace_id=workspace_id,
+                conversation_id=conversation_id,
+                user_id=user_id,
+                variables=variables
         ):
             yield event
 
     # ==================== 多模型对比试运行 ====================
 
     async def draft_run_compare(
-        self,
-        *,
-        app_id: uuid.UUID,
-        message: str,
-        models: List[app_schema.ModelCompareItem],
-        conversation_id: Optional[str] = None,
-        user_id: Optional[str] = None,
-        variables: Optional[Dict[str, Any]] = None,
-        workspace_id: Optional[uuid.UUID] = None,
-        parallel: bool = True,
-        timeout: int = 60
+            self,
+            *,
+            app_id: uuid.UUID,
+            message: str,
+            models: List[app_schema.ModelCompareItem],
+            conversation_id: Optional[str] = None,
+            user_id: Optional[str] = None,
+            variables: Optional[Dict[str, Any]] = None,
+            workspace_id: Optional[uuid.UUID] = None,
+            parallel: bool = True,
+            timeout: int = 60
     ) -> Dict[str, Any]:
         """多模型对比试运行
 
@@ -1907,17 +1911,17 @@ class AppService:
         return result
 
     async def draft_run_compare_stream(
-        self,
-        *,
-        app_id: uuid.UUID,
-        message: str,
-        models: List[app_schema.ModelCompareItem],
-        conversation_id: Optional[str] = None,
-        user_id: Optional[str] = None,
-        variables: Optional[Dict[str, Any]] = None,
-        workspace_id: Optional[uuid.UUID] = None,
-        parallel: bool = True,
-        timeout: int = 60
+            self,
+            *,
+            app_id: uuid.UUID,
+            message: str,
+            models: List[app_schema.ModelCompareItem],
+            conversation_id: Optional[str] = None,
+            user_id: Optional[str] = None,
+            variables: Optional[Dict[str, Any]] = None,
+            workspace_id: Optional[uuid.UUID] = None,
+            parallel: bool = True,
+            timeout: int = 60
     ):
         """多模型对比试运行（流式返回）
 
@@ -1982,15 +1986,15 @@ class AppService:
         # 4. 调用 DraftRunService 的流式对比方法
         draft_service = DraftRunService(self.db)
         async for event in draft_service.run_compare_stream(
-            agent_config=agent_cfg,
-            models=model_configs,
-            message=message,
-            workspace_id=workspace_id,
-            conversation_id=conversation_id,
-            user_id=user_id,
-            variables=variables,
-            parallel=parallel,
-            timeout=timeout
+                agent_config=agent_cfg,
+                models=model_configs,
+                message=message,
+                workspace_id=workspace_id,
+                conversation_id=conversation_id,
+                user_id=user_id,
+                variables=variables,
+                parallel=parallel,
+                timeout=timeout
         ):
             yield event
 
@@ -2009,7 +2013,8 @@ def create_app(db: Session, *, user_id: uuid.UUID, workspace_id: uuid.UUID, data
     return service.create_app(user_id=user_id, workspace_id=workspace_id, data=data)
 
 
-def update_app(db: Session, *, app_id: uuid.UUID, data: app_schema.AppUpdate, workspace_id: uuid.UUID | None = None) -> App:
+def update_app(db: Session, *, app_id: uuid.UUID, data: app_schema.AppUpdate,
+               workspace_id: uuid.UUID | None = None) -> App:
     """更新应用（向后兼容接口）"""
     service = AppService(db)
     return service.update_app(app_id=app_id, data=data, workspace_id=workspace_id)
@@ -2021,12 +2026,15 @@ def delete_app(db: Session, *, app_id: uuid.UUID, workspace_id: uuid.UUID | None
     return service.delete_app(app_id=app_id, workspace_id=workspace_id)
 
 
-def update_agent_config(db: Session, *, app_id: uuid.UUID, data: app_schema.AgentConfigUpdate, workspace_id: uuid.UUID | None = None) -> AgentConfig:
+def update_agent_config(db: Session, *, app_id: uuid.UUID, data: app_schema.AgentConfigUpdate,
+                        workspace_id: uuid.UUID | None = None) -> AgentConfig:
     """更新 Agent 配置（向后兼容接口）"""
     service = AppService(db)
     return service.update_agent_config(app_id=app_id, data=data, workspace_id=workspace_id)
 
-def update_workflow_config(db: Session, *, app_id: uuid.UUID, data: WorkflowConfigUpdate, workspace_id: uuid.UUID | None = None) -> WorkflowConfig:
+
+def update_workflow_config(db: Session, *, app_id: uuid.UUID, data: WorkflowConfigUpdate,
+                           workspace_id: uuid.UUID | None = None) -> WorkflowConfig:
     """更新 Agent 配置（向后兼容接口）"""
     service = AppService(db)
     return service.update_workflow_config(app_id=app_id, data=data, workspace_id=workspace_id)
@@ -2040,6 +2048,7 @@ def get_agent_config(db: Session, *, app_id: uuid.UUID, workspace_id: uuid.UUID
     service = AppService(db)
     return service.get_agent_config(app_id=app_id, workspace_id=workspace_id)
 
+
 def get_workflow_config(db: Session, *, app_id: uuid.UUID, workspace_id: uuid.UUID | None = None) -> WorkflowConfig:
     """获取 Agent 配置（向后兼容接口）
 
@@ -2049,13 +2058,20 @@ def get_workflow_config(db: Session, *, app_id: uuid.UUID, workspace_id: uuid.UU
     return service.get_workflow_config(app_id=app_id, workspace_id=workspace_id)
 
 
-def publish(db: Session, *, app_id: uuid.UUID, publisher_id: uuid.UUID, workspace_id: uuid.UUID | None = None,version_name:str, release_notes: Optional[str] = None) -> AppRelease:
+def publish(db: Session, *, app_id: uuid.UUID, publisher_id: uuid.UUID, workspace_id: uuid.UUID | None = None,
+            version_name: str, release_notes: Optional[str] = None) -> AppRelease:
     """发布应用（向后兼容接口）"""
     service = AppService(db)
-    return service.publish(app_id=app_id, publisher_id=publisher_id,version_name = version_name, workspace_id=workspace_id, release_notes=release_notes)
+    return service.publish(app_id=app_id, publisher_id=publisher_id, version_name=version_name,
+                           workspace_id=workspace_id, release_notes=release_notes)
 
 
-def get_current_release(db: Session, *, app_id: uuid.UUID, workspace_id: uuid.UUID | None = None) -> Optional[AppRelease]:
+def get_current_release(
+        db: Session,
+        *,
+        app_id: uuid.UUID,
+        workspace_id: uuid.UUID | None = None
+) -> Optional[AppRelease]:
     """获取当前发布版本（向后兼容接口）"""
     service = AppService(db)
     return service.get_current_release(app_id=app_id, workspace_id=workspace_id)
@@ -2074,16 +2090,16 @@ def rollback(db: Session, *, app_id: uuid.UUID, version: int, workspace_id: uuid
 
 
 def list_apps(
-    db: Session,
-    *,
-    workspace_id: uuid.UUID,
-    type: Optional[str] = None,
-    visibility: Optional[str] = None,
-    status: Optional[str] = None,
-    search: Optional[str] = None,
-    include_shared: bool = True,
-    page: int = 1,
-    pagesize: int = 10,
+        db: Session,
+        *,
+        workspace_id: uuid.UUID,
+        type: Optional[str] = None,
+        visibility: Optional[str] = None,
+        status: Optional[str] = None,
+        search: Optional[str] = None,
+        include_shared: bool = True,
+        page: int = 1,
+        pagesize: int = 10,
 ) -> Tuple[List[App], int]:
     """列出应用（向后兼容接口）"""
     service = AppService(db)
@@ -2100,9 +2116,9 @@ def list_apps(
 
 
 def get_apps_by_ids(
-    db: Session,
-    app_ids: List[str],
-    workspace_id: uuid.UUID
+        db: Session,
+        app_ids: List[str],
+        workspace_id: uuid.UUID
 ) -> List[App]:
     """根据ID列表获取应用（向后兼容接口）"""
     service = AppService(db)
@@ -2112,14 +2128,14 @@ def get_apps_by_ids(
 # ==================== 向后兼容的函数接口 ====================
 
 async def draft_run(
-    db: Session,
-    *,
-    app_id: uuid.UUID,
-    message: str,
-    conversation_id: Optional[str] = None,
-    user_id: Optional[str] = None,
-    variables: Optional[Dict[str, Any]] = None,
-    workspace_id: Optional[uuid.UUID] = None
+        db: Session,
+        *,
+        app_id: uuid.UUID,
+        message: str,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        variables: Optional[Dict[str, Any]] = None,
+        workspace_id: Optional[uuid.UUID] = None
 ) -> Dict[str, Any]:
     """试运行 Agent（向后兼容接口）"""
     service = AppService(db)
@@ -2134,30 +2150,28 @@ async def draft_run(
 
 
 async def draft_run_stream(
-    db: Session,
-    *,
-    app_id: uuid.UUID,
-    message: str,
-    conversation_id: Optional[str] = None,
-    user_id: Optional[str] = None,
-    variables: Optional[Dict[str, Any]] = None,
-    workspace_id: Optional[uuid.UUID] = None
+        db: Session,
+        *,
+        app_id: uuid.UUID,
+        message: str,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        variables: Optional[Dict[str, Any]] = None,
+        workspace_id: Optional[uuid.UUID] = None
 ):
     """试运行 Agent 流式返回（向后兼容接口）"""
     service = AppService(db)
     async for event in service.draft_run_stream(
-        app_id=app_id,
-        message=message,
-        conversation_id=conversation_id,
-        user_id=user_id,
-        variables=variables,
-        workspace_id=workspace_id
+            app_id=app_id,
+            message=message,
+            conversation_id=conversation_id,
+            user_id=user_id,
+            variables=variables,
+            workspace_id=workspace_id
     ):
         yield event
 
 
-
-
 # ==================== 依赖注入函数 ====================
 
 def get_app_service(

api/app/services/emotion_analytics_service.py

@@ -711,45 +711,32 @@ class EmotionAnalyticsService:
         end_user_id: str,
         db: Session,
     ) -> Optional[Dict[str, Any]]:
-        """从缓存获取个性化情绪建议
+        """从 Redis 缓存获取个性化情绪建议
         
         Args:
             end_user_id: 宿主ID（用户组ID）
-            db: 数据库会话
+            db: 数据库会话（保留参数以保持接口兼容性）
             
         Returns:
             Dict: 缓存的建议数据，如果不存在或已过期返回 None
         """
         try:
-            from app.repositories.emotion_suggestions_cache_repository import (
-                EmotionSuggestionsCacheRepository,
-            )
+            from app.cache.memory.emotion_memory import EmotionMemoryCache
             
-            logger.info(f"尝试从缓存获取情绪建议: user={end_user_id}")
+            logger.info(f"尝试从 Redis 缓存获取情绪建议: user={end_user_id}")
             
-            cache_repo = EmotionSuggestionsCacheRepository(db)
-            cache = cache_repo.get_by_end_user_id(end_user_id)
+            # 从 Redis 获取缓存
+            cached_data = await EmotionMemoryCache.get_emotion_suggestions(end_user_id)
             
-            if cache is None:
-                logger.info(f"用户 {end_user_id} 的建议缓存不存在")
+            if cached_data is None:
+                logger.info(f"用户 {end_user_id} 的建议缓存不存在或已过期")
                 return None
             
-            # 检查是否过期
-            if cache_repo.is_expired(cache):
-                logger.info(f"用户 {end_user_id} 的建议缓存已过期")
-                return None
-            
-            logger.info(f"成功从缓存获取建议: user={end_user_id}")
-            
-            return {
-                "health_summary": cache.health_summary,
-                "suggestions": cache.suggestions,
-                "generated_at": cache.generated_at.isoformat(),
-                "cached": True
-            }
+            logger.info(f"成功从 Redis 缓存获取建议: user={end_user_id}")
+            return cached_data
             
         except Exception as e:
-            logger.error(f"从缓存获取建议失败: {str(e)}", exc_info=True)
+            logger.error(f"从 Redis 缓存获取建议失败: {str(e)}", exc_info=True)
             return None
     
     async def save_suggestions_cache(
@@ -759,30 +746,33 @@ class EmotionAnalyticsService:
         db: Session,
         expires_hours: int = 24
     ) -> None:
-        """保存建议到缓存
+        """保存建议到 Redis 缓存
         
         Args:
             end_user_id: 宿主ID（用户组ID）
             suggestions_data: 建议数据
-            db: 数据库会话
-            expires_hours: 过期时间（小时）
+            db: 数据库会话（保留参数以保持接口兼容性）
+            expires_hours: 过期时间（小时），默认24小时
         """
         try:
-            from app.repositories.emotion_suggestions_cache_repository import (
-                EmotionSuggestionsCacheRepository,
+            from app.cache.memory.emotion_memory import EmotionMemoryCache
+            
+            logger.info(f"保存建议到 Redis 缓存: user={end_user_id}, expires={expires_hours}小时")
+            
+            # 计算过期时间（秒）
+            expire_seconds = expires_hours * 3600
+            
+            # 保存到 Redis
+            success = await EmotionMemoryCache.set_emotion_suggestions(
+                user_id=end_user_id,
+                suggestions_data=suggestions_data,
+                expire=expire_seconds
             )
             
-            logger.info(f"保存建议到缓存: user={end_user_id}")
-            
-            cache_repo = EmotionSuggestionsCacheRepository(db)
-            cache_repo.create_or_update(
-                end_user_id=end_user_id,
-                health_summary=suggestions_data["health_summary"],
-                suggestions=suggestions_data["suggestions"],
-                expires_hours=expires_hours
-            )
-            
-            logger.info(f"建议缓存保存成功: user={end_user_id}")
+            if success:
+                logger.info(f"建议缓存保存成功: user={end_user_id}")
+            else:
+                logger.warning(f"建议缓存保存失败: user={end_user_id}")
             
         except Exception as e:
             logger.error(f"保存建议缓存失败: {str(e)}", exc_info=True)

api/app/services/implicit_memory_service.py

@@ -418,48 +418,32 @@ class ImplicitMemoryService:
         end_user_id: str,
         db: Session
     ) -> Optional[dict]:
-        """从缓存获取完整用户画像
+        """从 Redis 缓存获取完整用户画像
         
         Args:
             end_user_id: 终端用户ID
-            db: 数据库会话
+            db: 数据库会话（保留参数以保持接口兼容性）
             
         Returns:
             Dict: 缓存的画像数据，如果不存在或已过期返回 None
         """
         try:
-            from app.repositories.implicit_memory_cache_repository import (
-                ImplicitMemoryCacheRepository,
-            )
+            from app.cache.memory.implicit_memory import ImplicitMemoryCache
             
-            logger.info(f"尝试从缓存获取用户画像: user={end_user_id}")
+            logger.info(f"尝试从 Redis 缓存获取用户画像: user={end_user_id}")
             
-            cache_repo = ImplicitMemoryCacheRepository(db)
-            cache = cache_repo.get_by_end_user_id(end_user_id)
+            # 从 Redis 获取缓存
+            cached_data = await ImplicitMemoryCache.get_user_profile(end_user_id)
             
-            if cache is None:
-                logger.info(f"用户 {end_user_id} 的画像缓存不存在")
+            if cached_data is None:
+                logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
                 return None
             
-            # 检查是否过期
-            if cache_repo.is_expired(cache):
-                logger.info(f"用户 {end_user_id} 的画像缓存已过期")
-                return None
-            
-            logger.info(f"成功从缓存获取用户画像: user={end_user_id}")
-            
-            return {
-                "end_user_id": cache.end_user_id,
-                "preferences": cache.preferences,
-                "portrait": cache.portrait,
-                "interest_areas": cache.interest_areas,
-                "habits": cache.habits,
-                "generated_at": cache.generated_at.isoformat(),
-                "cached": True
-            }
+            logger.info(f"成功从 Redis 缓存获取用户画像: user={end_user_id}")
+            return cached_data
             
         except Exception as e:
-            logger.error(f"从缓存获取用户画像失败: {str(e)}", exc_info=True)
+            logger.error(f"从 Redis 缓存获取用户画像失败: {str(e)}", exc_info=True)
             return None
     
     async def save_profile_cache(
@@ -469,32 +453,33 @@ class ImplicitMemoryService:
         db: Session,
         expires_hours: int = 168  # 默认7天
     ) -> None:
-        """保存用户画像到缓存
+        """保存用户画像到 Redis 缓存
         
         Args:
             end_user_id: 终端用户ID
             profile_data: 画像数据
-            db: 数据库会话
+            db: 数据库会话（保留参数以保持接口兼容性）
             expires_hours: 过期时间（小时），默认168小时（7天）
         """
         try:
-            from app.repositories.implicit_memory_cache_repository import (
-                ImplicitMemoryCacheRepository,
+            from app.cache.memory.implicit_memory import ImplicitMemoryCache
+            
+            logger.info(f"保存用户画像到 Redis 缓存: user={end_user_id}, expires={expires_hours}小时")
+            
+            # 计算过期时间（秒）
+            expire_seconds = expires_hours * 3600
+            
+            # 保存到 Redis
+            success = await ImplicitMemoryCache.set_user_profile(
+                user_id=end_user_id,
+                profile_data=profile_data,
+                expire=expire_seconds
             )
             
-            logger.info(f"保存用户画像到缓存: user={end_user_id}")
-            
-            cache_repo = ImplicitMemoryCacheRepository(db)
-            cache_repo.create_or_update(
-                end_user_id=end_user_id,
-                preferences=profile_data["preferences"],
-                portrait=profile_data["portrait"],
-                interest_areas=profile_data["interest_areas"],
-                habits=profile_data["habits"],
-                expires_hours=expires_hours
-            )
-            
-            logger.info(f"用户画像缓存保存成功: user={end_user_id}")
+            if success:
+                logger.info(f"用户画像缓存保存成功: user={end_user_id}")
+            else:
+                logger.warning(f"用户画像缓存保存失败: user={end_user_id}")
             
         except Exception as e:
             logger.error(f"保存用户画像缓存失败: {str(e)}", exc_info=True)

api/app/services/workflow_service.py

@@ -4,7 +4,7 @@
 import datetime
 import logging
 import uuid
-from typing import Any, Annotated, AsyncGenerator
+from typing import Any, Annotated, AsyncGenerator, Optional
 
 from deprecated import deprecated
 from fastapi import Depends
@@ -14,15 +14,14 @@ from app.core.error_codes import BizCode
 from app.core.exceptions import BusinessException
 from app.core.workflow.validator import validate_workflow_config
 from app.db import get_db
-from app.models.conversation_model import Message
 from app.models.workflow_model import WorkflowConfig, WorkflowExecution
-from app.repositories.conversation_repository import MessageRepository
 from app.repositories.workflow_repository import (
     WorkflowConfigRepository,
     WorkflowExecutionRepository,
     WorkflowNodeExecutionRepository
 )
 from app.schemas import DraftRunRequest
+from app.services.conversation_service import ConversationService
 from app.services.multi_agent_service import convert_uuids_to_str
 
 logger = logging.getLogger(__name__)
@@ -36,7 +35,7 @@ class WorkflowService:
         self.config_repo = WorkflowConfigRepository(db)
         self.execution_repo = WorkflowExecutionRepository(db)
         self.node_execution_repo = WorkflowNodeExecutionRepository(db)
-        self.message_repo = MessageRepository(db)
+        self.conversation_service = ConversationService(db)
 
     # ==================== 配置管理 ====================
 
@@ -266,6 +265,7 @@ class WorkflowService:
             workflow_config_id: uuid.UUID,
             app_id: uuid.UUID,
             trigger_type: str,
+            release_id: uuid.UUID | None = None,
             triggered_by: uuid.UUID | None = None,
             conversation_id: uuid.UUID | None = None,
             input_data: dict[str, Any] | None = None
@@ -273,6 +273,7 @@ class WorkflowService:
         """创建工作流执行记录
 
         Args:
+            release_id: 应用发布 ID
             workflow_config_id: 工作流配置 ID
             app_id: 应用 ID
             trigger_type: 触发类型
@@ -289,6 +290,7 @@ class WorkflowService:
         execution = WorkflowExecution(
             workflow_config_id=workflow_config_id,
             app_id=app_id,
+            release_id=release_id,
             conversation_id=conversation_id,
             execution_id=execution_id,
             trigger_type=trigger_type,
@@ -337,6 +339,7 @@ class WorkflowService:
             self,
             execution_id: str,
             status: str,
+            token_usage: int | None = None,
             output_data: dict[str, Any] | None = None,
             error_message: str | None = None,
             error_node_id: str | None = None
@@ -346,6 +349,7 @@ class WorkflowService:
         Args:
             execution_id: 执行 ID
             status: 状态
+            token_usage: token消耗
             output_data: 输出数据
             error_message: 错误信息
             error_node_id: 出错节点 ID
@@ -364,6 +368,8 @@ class WorkflowService:
             )
 
         execution.status = status
+        if token_usage is not None:
+            execution.token_usage = token_usage
         if output_data is not None:
             execution.output_data = convert_uuids_to_str(output_data)
         if error_message is not None:
@@ -414,12 +420,14 @@ class WorkflowService:
             payload: DraftRunRequest,
             config: WorkflowConfig,
             workspace_id: uuid.UUID,
+            release_id: uuid.UUID | None = None,
     ):
         """运行工作流
 
         Args:
-            workspace_id:
-            config:
+            release_id: 发布 ID
+            workspace_id:工作空间 ID
+            config: 配置
             payload:
             app_id: 应用 ID
 
@@ -463,7 +471,8 @@ class WorkflowService:
             trigger_type="manual",
             triggered_by=None,
             conversation_id=conversation_id_uuid,
-            input_data=input_data
+            input_data=input_data,
+            release_id=release_id,
         )
 
         # 3. 构建工作流配置字典
@@ -507,20 +516,20 @@ class WorkflowService:
 
             # 更新执行结果
             if result.get("status") == "completed":
+                token_usage = result.get("token_usage", {}) or {}
                 self.update_execution_status(
                     execution.execution_id,
                     "completed",
-                    output_data=result
+                    output_data=result,
+                    token_usage=token_usage.get("total_tokens", None)
                 )
                 final_messages = result.get("messages", [])[init_message_length:]
                 for message in final_messages:
-                    message_obj = Message(
+                    self.conversation_service.add_message(
                         conversation_id=conversation_id_uuid,
                         role=message["role"],
-                        content=message["content"],
+                        content=message["content"]
                     )
-                    self.message_repo.add_message(message_obj)
-                self.db.commit()
                 logger.info(f"Workflow Run Success, "
                             f"execution_id: {execution.execution_id}, message count: {len(final_messages)}")
             else:
@@ -562,10 +571,12 @@ class WorkflowService:
             payload: DraftRunRequest,
             config: WorkflowConfig,
             workspace_id: uuid.UUID,
+            release_id: Optional[uuid.UUID] = None,
     ):
         """运行工作流（流式）
 
         Args:
+            release_id: 发布id
             workspace_id:
             app_id: 应用 ID
             payload: 请求对象（包含 message, variables, conversation_id 等）
@@ -611,7 +622,8 @@ class WorkflowService:
             trigger_type="manual",
             triggered_by=None,
             conversation_id=conversation_id_uuid,
-            input_data=input_data
+            input_data=input_data,
+            release_id=release_id,
         )
 
         # 3. 构建工作流配置字典
@@ -653,21 +665,21 @@ class WorkflowService:
                 if event.get("event") == "workflow_end":
 
                     status = event.get("data", {}).get("status")
+                    token_usage = event.get("data", {}).get("token_usage", {}) or {}
                     if status == "completed":
                         self.update_execution_status(
                             execution.execution_id,
                             "completed",
-                            output_data=event.get("data")
+                            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:
-                            message_obj = Message(
+                            self.conversation_service.add_message(
                                 conversation_id=conversation_id_uuid,
                                 role=message["role"],
-                                content=message["content"],
+                                content=message["content"]
                             )
-                            self.message_repo.add_message(message_obj)
-                        self.db.commit()
                         logger.info(f"Workflow Run Success, "
                                     f"execution_id: {execution.execution_id}, message count: {len(final_messages)}")
                     elif status == "failed":
@@ -784,10 +796,12 @@ class WorkflowService:
 
                 # 更新执行结果
                 if result.get("status") == "completed":
+                    token_usage = result.get("data").get("token_usage", {}) or {}
                     self.update_execution_status(
                         execution.execution_id,
                         "completed",
-                        output_data=result.get("node_outputs", {})
+                        output_data=result.get("node_outputs", {}),
+                        token_usage=token_usage.get("total_tokens", None)
                     )
                 else:
                     self.update_execution_status(
@@ -882,13 +896,14 @@ class WorkflowService:
             ):
                 # 直接转发事件（executor 已经返回正确格式）
                 if event.get("event") == "workflow_end":
-
+                    token_usage = event.get("data").get("token_usage", {}) or {}
                     status = event.get("data", {}).get("status")
                     if status == "completed":
                         self.update_execution_status(
                             execution_id,
                             "completed",
-                            output_data=event.get("data")
+                            output_data=event.get("data"),
+                            token_usage=token_usage.get("total_tokens", None)
                         )
                     elif status == "failed":
                         self.update_execution_status(

api/app/templates/workflows/simple_qa/template.yml

@@ -53,7 +53,7 @@ nodes:
     type: end
     name: 结束
     config:
-      output: "{{ llm_qa.output }}"
+      output: "{{llm_qa.output}}"
     position:
       x: 900
       y: 100

api/app/utils/app_config_utils.py

@@ -120,12 +120,9 @@ def multi_agent_config_4_app_release(release: AppRelease) -> MultiAgentConfig:
 
 def workflow_config_4_app_release(release: AppRelease) -> WorkflowConfig:
     config_dict = release.config
-    with get_db_read() as db:
-        source_config = WorkflowConfigRepository(db).get_by_app_id(release.app_id)
-        source_config_id = source_config.id
 
     config = WorkflowConfig(
-        id=source_config_id,
+        id=config_dict.get("id"),
         app_id=release.app_id,
         nodes=config_dict.get("nodes", []),
         edges=config_dict.get("edges", []),

api/migrations/versions/8cd790908f92_202601191615.py

@@ -0,0 +1,34 @@
+"""202601191615
+
+Revision ID: 8cd790908f92
+Revises: 1fd7d0e703b3
+Create Date: 2026-01-19 16:15:35.058649
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '8cd790908f92'
+down_revision: Union[str, None] = '1fd7d0e703b3'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('workflow_executions', sa.Column('release_id', sa.UUID(), nullable=True))
+    op.create_index(op.f('ix_workflow_executions_release_id'), 'workflow_executions', ['release_id'], unique=False)
+    op.create_foreign_key(None, 'workflow_executions', 'app_releases', ['release_id'], ['id'], ondelete='CASCADE')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'workflow_executions', type_='foreignkey')
+    op.drop_index(op.f('ix_workflow_executions_release_id'), table_name='workflow_executions')
+    op.drop_column('workflow_executions', 'release_id')
+    # ### end Alembic commands ###