diff --git a/api/app/cache/__init__.py b/api/app/cache/__init__.py
new file mode 100644
index 00000000..a79d4cb2
--- /dev/null
+++ b/api/app/cache/__init__.py
@@ -0,0 +1,11 @@
+"""
+Cache 缓存模块
+
+提供各种缓存功能的统一入口
+"""
+from .memory import EmotionMemoryCache, ImplicitMemoryCache
+
+__all__ = [
+ "EmotionMemoryCache",
+ "ImplicitMemoryCache",
+]
diff --git a/api/app/cache/memory/__init__.py b/api/app/cache/memory/__init__.py
new file mode 100644
index 00000000..4ada3153
--- /dev/null
+++ b/api/app/cache/memory/__init__.py
@@ -0,0 +1,12 @@
+"""
+Memory 缓存模块
+
+提供记忆系统相关的缓存功能
+"""
+from .emotion_memory import EmotionMemoryCache
+from .implicit_memory import ImplicitMemoryCache
+
+__all__ = [
+ "EmotionMemoryCache",
+ "ImplicitMemoryCache",
+]
diff --git a/api/app/cache/memory/emotion_memory.py b/api/app/cache/memory/emotion_memory.py
new file mode 100644
index 00000000..45ea90de
--- /dev/null
+++ b/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
diff --git a/api/app/cache/memory/implicit_memory.py b/api/app/cache/memory/implicit_memory.py
new file mode 100644
index 00000000..21f08e9a
--- /dev/null
+++ b/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
diff --git a/api/app/controllers/emotion_controller.py b/api/app/controllers/emotion_controller.py
index 24bdc434..b5cd7250 100644
--- a/api/app/controllers/emotion_controller.py
+++ b/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(
diff --git a/api/app/controllers/implicit_memory_controller.py b/api/app/controllers/implicit_memory_controller.py
index eb7037ff..62d1e428 100644
--- a/api/app/controllers/implicit_memory_controller.py
+++ b/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
diff --git a/api/app/controllers/public_share_controller.py b/api/app/controllers/public_share_controller.py
index 04da05df..17ad70a7 100644
--- a/api/app/controllers/public_share_controller.py
+++ b/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(
"工作流试运行返回结果",
diff --git a/api/app/controllers/service/app_api_controller.py b/api/app/controllers/service/app_api_controller.py
index 583b4700..677e1623 100644
--- a/api/app/controllers/service/app_api_controller.py
+++ b/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(
"工作流试运行返回结果",
diff --git a/api/app/core/config.py b/api/app/core/config.py
index 774f4a0f..a31caab4 100644
--- a/api/app/core/config.py
+++ b/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")
diff --git a/api/app/core/rag/llm/cv_model.py b/api/app/core/rag/llm/cv_model.py
index 24d4a35b..4207304a 100644
--- a/api/app/core/rag/llm/cv_model.py
+++ b/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,
},
],
}
diff --git a/api/app/core/rag/llm/sequence2txt_model.py b/api/app/core/rag/llm/sequence2txt_model.py
index be4d3649..468dda55 100644
--- a/api/app/core/rag/llm/sequence2txt_model.py
+++ b/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,
},
],
}
diff --git a/api/app/core/workflow/executor.py b/api/app/core/workflow/executor.py
index c048f447..ad03fec1 100644
--- a/api/app/core/workflow/executor.py
+++ b/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:
diff --git a/api/app/core/workflow/nodes/base_node.py b/api/app/core/workflow/nodes/base_node.py
index 72fd0bb5..b31213d8 100644
--- a/api/app/core/workflow/nodes/base_node.py
+++ b/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
diff --git a/api/app/core/workflow/nodes/cycle_graph/iteration.py b/api/app/core/workflow/nodes/cycle_graph/iteration.py
index 4ae8e118..66c3a700 100644
--- a/api/app/core/workflow/nodes/cycle_graph/iteration.py
+++ b/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
+ }
diff --git a/api/app/core/workflow/nodes/cycle_graph/loop.py b/api/app/core/workflow/nodes/cycle_graph/loop.py
index 2e2ab4fb..38d4b21c 100644
--- a/api/app/core/workflow/nodes/cycle_graph/loop.py
+++ b/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}
diff --git a/api/app/core/workflow/nodes/knowledge/node.py b/api/app/core/workflow/nodes/knowledge/node.py
index 221ca079..997135f3 100644
--- a/api/app/core/workflow/nodes/knowledge/node.py
+++ b/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
)
diff --git a/api/app/core/workflow/nodes/llm/config.py b/api/app/core/workflow/nodes/llm/config.py
index f65d5879..265724f3 100644
--- a/api/app/core/workflow/nodes/llm/config.py
+++ b/api/app/core/workflow/nodes/llm/config.py
@@ -66,7 +66,7 @@ class LLMNodeConfig(BaseNodeConfig):
)
memory: MemoryWindowSetting = Field(
- ...,
+ default_factory=MemoryWindowSetting,
description="对话上下文窗口"
)
diff --git a/api/app/core/workflow/nodes/llm/node.py b/api/app/core/workflow/nodes/llm/node.py
index e25bd35d..a74e0b60 100644
--- a/api/app/core/workflow/nodes/llm/node.py
+++ b/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}
diff --git a/api/app/models/__init__.py b/api/app/models/__init__.py
index f45991cd..81cc6ead 100644
--- a/api/app/models/__init__.py
+++ b/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"
]
diff --git a/api/app/models/emotion_suggestions_cache_model.py b/api/app/models/emotion_suggestions_cache_model.py
deleted file mode 100644
index 9b32f424..00000000
--- a/api/app/models/emotion_suggestions_cache_model.py
+++ /dev/null
@@ -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)
diff --git a/api/app/models/implicit_memory_cache_model.py b/api/app/models/implicit_memory_cache_model.py
deleted file mode 100644
index 32defbab..00000000
--- a/api/app/models/implicit_memory_cache_model.py
+++ /dev/null
@@ -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)
diff --git a/api/app/models/workflow_model.py b/api/app/models/workflow_model.py
index d599f717..4f9ffe68 100644
--- a/api/app/models/workflow_model.py
+++ b/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"),
diff --git a/api/app/repositories/emotion_suggestions_cache_repository.py b/api/app/repositories/emotion_suggestions_cache_repository.py
deleted file mode 100644
index 1c0430d5..00000000
--- a/api/app/repositories/emotion_suggestions_cache_repository.py
+++ /dev/null
@@ -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)
diff --git a/api/app/repositories/implicit_memory_cache_repository.py b/api/app/repositories/implicit_memory_cache_repository.py
deleted file mode 100644
index 65356980..00000000
--- a/api/app/repositories/implicit_memory_cache_repository.py
+++ /dev/null
@@ -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)
diff --git a/api/app/schemas/emotion_schema.py b/api/app/schemas/emotion_schema.py
index 37e9a2e3..5175fed1 100644
--- a/api/app/schemas/emotion_schema.py
+++ b/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")
diff --git a/api/app/services/app_chat_service.py b/api/app/services/app_chat_service.py
index bc2d6ca3..c0a66e03 100644
--- a/api/app/services/app_chat_service.py
+++ b/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
diff --git a/api/app/services/app_service.py b/api/app/services/app_service.py
index 6d5204f8..2ac9ac05 100644
--- a/api/app/services/app_service.py
+++ b/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(
diff --git a/api/app/services/emotion_analytics_service.py b/api/app/services/emotion_analytics_service.py
index 50773b91..601d2921 100644
--- a/api/app/services/emotion_analytics_service.py
+++ b/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)
diff --git a/api/app/services/implicit_memory_service.py b/api/app/services/implicit_memory_service.py
index c98f14bc..106fa808 100644
--- a/api/app/services/implicit_memory_service.py
+++ b/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)
diff --git a/api/app/services/workflow_service.py b/api/app/services/workflow_service.py
index 974d5418..b7d5df02 100644
--- a/api/app/services/workflow_service.py
+++ b/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(
diff --git a/api/app/templates/workflows/simple_qa/template.yml b/api/app/templates/workflows/simple_qa/template.yml
index 14de4a73..2cf0f9b1 100644
--- a/api/app/templates/workflows/simple_qa/template.yml
+++ b/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
diff --git a/api/app/utils/app_config_utils.py b/api/app/utils/app_config_utils.py
index 4a35a4cc..514e4565 100644
--- a/api/app/utils/app_config_utils.py
+++ b/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", []),
diff --git a/api/migrations/versions/8cd790908f92_202601191615.py b/api/migrations/versions/8cd790908f92_202601191615.py
new file mode 100644
index 00000000..8e4624ee
--- /dev/null
+++ b/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 ###
diff --git a/web/src/api/memory.ts b/web/src/api/memory.ts
index 0ac14451..bbd9f6b0 100644
--- a/web/src/api/memory.ts
+++ b/web/src/api/memory.ts
@@ -18,174 +18,180 @@ import type { TestParams } from '@/views/MemoryConversation'
import type { EndUser } from '@/views/UserMemoryDetail/types'
import { handleSSE, type SSEMessage } from '@/utils/stream'
-// 记忆对话
+// Memory conversation
export const readService = (query: TestParams) => {
return request.post('/memory/read_service', query)
}
-/****************** 记忆看板 相关接口 *******************************/
-// 记忆看板-记忆总量
+/****************** Memory Dashboard APIs *******************************/
+// Memory Dashboard - Total memory count
export const getTotalMemoryCount = () => {
return request.get(`/dashboard/total_memory_count`)
}
-// 记忆看板-知识库类型分布
+// Memory Dashboard - Knowledge base type distribution
export const getKbTypes = () => {
return request.get(`/memory/stats/types`)
}
-// 记忆看板-热门记忆标签
+// Memory Dashboard - Hot memory tags
export const getHotMemoryTags = () => {
return request.get(`/memory-storage/analytics/hot_memory_tags`)
}
-// 记忆看板-最近活动统计
+// Memory Dashboard - Recent activity statistics
export const getRecentActivityStats = () => {
return request.get(`/memory-storage/analytics/recent_activity_stats`)
}
-// 记忆看板-记忆增长趋势
+// Memory Dashboard - Memory growth trend
export const getMemoryIncrement = (limit: number) => {
return request.get(`/dashboard/memory_increment`, { limit })
}
-// 记忆看板-API调用趋势
+// Memory Dashboard - API call trend
export const getApiTrend = () => {
return request.get(`/dashboard/api_increment`)
}
-// 记忆看板-总数据
+// Memory Dashboard - Total data
export const getDashboardData = () => {
return request.get(`/dashboard/dashboard_data`)
}
-/*************** end 记忆看板 相关接口 ******************************/
+/*************** end Memory Dashboard APIs ******************************/
-/****************** 用户记忆 相关接口 *******************************/
+/****************** User Memory APIs *******************************/
export const userMemoryListUrl = '/dashboard/end_users'
export const getUserMemoryList = () => {
return request.get(userMemoryListUrl)
}
-// 用户记忆-用户记忆总量
+// User Memory - Total end users
export const getTotalEndUsers = () => {
return request.get(`/dashboard/total_end_users`)
}
-// 用户记忆-用户详情
+// User Memory - User profile
export const getUserProfile = (end_user_id: string) => {
return request.get(`/memory/analytics/user_profile`, { end_user_id })
}
-// 用户记忆-记忆洞察
+// User Memory - Memory insight
export const getMemoryInsightReport = (end_user_id: string) => {
return request.get(`/memory-storage/analytics/memory_insight/report`, { end_user_id })
}
-// 用户记忆-用户摘要
+// User Memory - User summary
export const getUserSummary = (end_user_id: string) => {
return request.get(`/memory-storage/analytics/user_summary`, { end_user_id })
}
-// 记忆分类
+// Memory classification
export const getNodeStatistics = (end_user_id: string) => {
return request.get(`/memory-storage/analytics/node_statistics`, { end_user_id })
}
-// 基本信息
+// Basic information
export const getEndUserProfile = (end_user_id: string) => {
return request.get(`/memory-storage/read_end_user/profile`, { end_user_id })
}
export const updatedEndUserProfile = (values: EndUser) => {
return request.post(`/memory-storage/updated_end_user/profile`, values)
}
-// 用户记忆-关系网络
+// User Memory - Relationship network
export const getMemorySearchEdges = (end_user_id: string) => {
return request.get(`/memory-storage/analytics/graph_data`, { end_user_id })
}
-// 用户记忆-用户兴趣分布
+// User Memory - User interest distribution
export const getHotMemoryTagsByUser = (end_user_id: string) => {
return request.get(`/memory/analytics/hot_memory_tags/by_user`, { end_user_id })
}
-// 用户记忆-记忆总量
+// User Memory - Total memory count
export const getTotalMemoryCountByUser = (end_user_id: string) => {
return request.get(`/memory-storage/search`, { end_user_id })
}
-// RAG 用户记忆-记忆总量
+// RAG User Memory - Total memory count
export const getTotalRagMemoryCountByUser = (end_user_id: string) => {
return request.get(`/dashboard/current_user_rag_total_num`, { end_user_id })
}
-// RAG 用户记忆-用户摘要
+// RAG User Memory - User summary
export const getChunkSummaryTag = (end_user_id: string) => {
return request.get(`/dashboard/chunk_summary_tag`, { end_user_id })
}
-// RAG 用户记忆-记忆洞察
+// RAG User Memory - Memory insight
export const getChunkInsight = (end_user_id: string) => {
return request.get(`/dashboard/chunk_insight`, { end_user_id })
}
-// RAG 用户记忆-存储内容
+// RAG User Memory - Storage content
export const getRagContent = (end_user_id: string) => {
return request.get(`/dashboard/rag_content`, { end_user_id, limit: 20 })
}
-// 情感分布分析
+// Emotion distribution analysis
export const getWordCloud = (group_id: string) => {
return request.post(`/memory/emotion-memory/wordcloud`, { group_id, limit: 20 })
}
-// 高频情绪关键词
+// High-frequency emotion keywords
export const getEmotionTags = (group_id: string) => {
return request.post(`/memory/emotion-memory/tags`, { group_id, limit: 20 })
}
-// 情绪健康指数
+// Emotion health index
export const getEmotionHealth = (group_id: string) => {
return request.post(`/memory/emotion-memory/health`, { group_id, limit: 20 })
}
-// 个性化建议
+// Personalized suggestions
export const getEmotionSuggestions = (group_id: string) => {
return request.post(`/memory/emotion-memory/suggestions`, { group_id, limit: 20 })
}
+export const generateSuggestions = (end_user_id: string) => {
+ return request.post(`/memory/emotion-memory/generate_suggestions`, { end_user_id })
+}
export const analyticsRefresh = (end_user_id: string) => {
return request.post('/memory-storage/analytics/generate_cache', { end_user_id })
}
-// 遗忘
+// Forgetting stats
export const getForgetStats = (group_id: string) => {
return request.get(`/memory/forget-memory/stats`, { group_id })
}
-// 隐性记忆-偏好
+// Implicit Memory - Preferences
export const getImplicitPreferences = (end_user_id: string) => {
return request.get(`/memory/implicit-memory/preferences/${end_user_id}`)
}
-// 隐性记忆-核心特质
+// Implicit Memory - Core traits
export const getImplicitPortrait = (end_user_id: string) => {
return request.get(`/memory/implicit-memory/portrait/${end_user_id}`)
}
-// 隐性记忆-兴趣领域分布
+// Implicit Memory - Interest areas distribution
export const getImplicitInterestAreas = (end_user_id: string) => {
return request.get(`/memory/implicit-memory/interest-areas/${end_user_id}`)
}
-// 隐性记忆-用户习惯分析
+// Implicit Memory - User habits analysis
export const getImplicitHabits = (end_user_id: string) => {
return request.get(`/memory/implicit-memory/habits/${end_user_id}`)
}
-// 短期记忆
+export const generateProfile = (end_user_id: string) => {
+ return request.post(`/memory/implicit-memory/generate_profile`, { end_user_id })
+}
+// Short-term memory
export const getShortTerm = (end_user_id: string) => {
return request.get(`/memory/short/short_term`, { end_user_id })
}
-// 感知记忆-视觉记忆
+// Perceptual Memory - Visual memory
export const getPerceptualLastVisual = (end_user: string) => {
return request.get(`/memory/perceptual/${end_user}/last_visual`)
}
-// 感知记忆-音频记忆
+// Perceptual Memory - Audio memory
export const getPerceptualLastListen = (end_user: string) => {
return request.get(`/memory/perceptual/${end_user}/last_listen`)
}
-// 感知记忆-文本记忆
+// Perceptual Memory - Text memory
export const getPerceptualLastText = (end_user: string) => {
return request.get(`/memory/perceptual/${end_user}/last_text`)
}
-// 感知记忆-感知记忆时间线
+// Perceptual Memory - Perceptual memory timeline
export const getPerceptualTimeline = (end_user: string) => {
return request.get(`/memory/perceptual/${end_user}/timeline`)
}
-// 情景记忆-总览
+// Episodic Memory - Overview
export const getEpisodicOverview = (data: { end_user_id: string; time_range: string; episodic_type: string; } ) => {
return request.post(`/memory/episodic-memory/overview`, data)
}
export const getEpisodicDetail = (data: { end_user_id: string; summary_id: string; } ) => {
return request.post(`/memory/episodic-memory/details`, data)
}
-// 关系演化
+// Relationship evolution
export const getRelationshipEvolution = (data: { id: string; label: string; } ) => {
return request.get(`/memory-storage/memory_space/relationship_evolution`, data)
}
-// 共同记忆时间线
+// Shared memory timeline
export const getTimelineMemories = (data: { id: string; label: string; }) => {
return request.get(`/memory-storage/memory_space/timeline_memories`, data)
}
@@ -207,72 +213,72 @@ export const getConversationDetail = (end_user: string, conversation_id: string)
export const forgetTrigger = (data: { max_merge_batch_size: number; min_days_since_access: number; end_user_id: string;}) => {
return request.post(`/memory/forget-memory/trigger`, data)
}
-/*************** end 用户记忆 相关接口 ******************************/
+/*************** end User Memory APIs ******************************/
-/****************** 记忆管理 相关接口 *******************************/
-// 记忆管理-获取所有配置
+/****************** Memory Management APIs *******************************/
+// Memory Management - Get all configurations
export const memoryConfigListUrl = '/memory-storage/read_all_config'
export const getMemoryConfigList = () => {
return request.get(memoryConfigListUrl)
}
-// 记忆管理-创建配置
+// Memory Management - Create configuration
export const createMemoryConfig = (values: MemoryFormData) => {
return request.post('/memory-storage/create_config', values)
}
-// 记忆管理-更新配置
+// Memory Management - Update configuration
export const updateMemoryConfig = (values: MemoryFormData) => {
return request.post('/memory-storage/update_config', values)
}
-// 记忆管理-删除配置
+// Memory Management - Delete configuration
export const deleteMemoryConfig = (config_id: number) => {
return request.delete(`/memory-storage/delete_config?config_id=${config_id}`)
}
-// 遗忘引擎-获取配置
+// Forgetting Engine - Get configuration
export const getMemoryForgetConfig = (config_id: number | string) => {
return request.get('/memory/forget-memory/read_config', { config_id })
}
-// 遗忘引擎-更新配置
+// Forgetting Engine - Update configuration
export const updateMemoryForgetConfig = (values: ForgetConfigForm) => {
return request.post('/memory/forget-memory/update_config', values)
}
-// 记忆萃取引擎-获取配置
+// Memory Extraction Engine - Get configuration
export const getMemoryExtractionConfig = (config_id: number | string) => {
return request.get('/memory-storage/read_config_extracted', { config_id: config_id })
}
-// 记忆萃取引擎-更新配置
+// Memory Extraction Engine - Update configuration
export const updateMemoryExtractionConfig = (values: ExtractionConfigForm) => {
return request.post('/memory-storage/update_config_extracted', values)
}
-// 记忆萃取引擎-试运行
+// Memory Extraction Engine - Pilot run
export const pilotRunMemoryExtractionConfig = (values: { config_id: number | string; dialogue_text: string; }, onMessage?: (data: SSEMessage[]) => void) => {
return handleSSE('/memory-storage/pilot_run', values, onMessage)
}
-// 情绪引擎-获取配置
+// Emotion Engine - Get configuration
export const getMemoryEmotionConfig = (config_id: number | string) => {
return request.get('/memory/emotion/read_config', { config_id: config_id })
}
-// 情绪引擎-更新配置
+// Emotion Engine - Update configuration
export const updateMemoryEmotionConfig = (values: EmotionConfig) => {
return request.post('/memory/emotion/updated_config', values)
}
-// 反思引擎-获取配置
+// Reflection Engine - Get configuration
export const getMemoryReflectionConfig = (config_id: number | string) => {
return request.get('/memory/reflection/configs', { config_id: config_id })
}
-// 反思引擎-更新配置
+// Reflection Engine - Update configuration
export const updateMemoryReflectionConfig = (values: SelfReflectionEngineConfig) => {
return request.post('/memory/reflection/save', values)
}
-// 反思引擎-试运行
+// Reflection Engine - Pilot run
export const pilotRunMemoryReflectionConfig = (values: { config_id: number | string; language_type: string; }) => {
return request.get('/memory/reflection/run', values)
}
-/*************** end 记忆管理 相关接口 ******************************/
+/*************** end Memory Management APIs ******************************/
-/****************** API参数 相关接口 *******************************/
+/****************** API Parameters APIs *******************************/
export const getMemoryApi = () => {
return request.get('/memory/docs/api')
}
-/*************** end API参数 相关接口 ******************************/
\ No newline at end of file
+/*************** end API Parameters APIs ******************************/
\ No newline at end of file
diff --git a/web/src/assets/images/workflow/deleteBg.svg b/web/src/assets/images/workflow/deleteBg.svg
new file mode 100644
index 00000000..f3827fef
--- /dev/null
+++ b/web/src/assets/images/workflow/deleteBg.svg
@@ -0,0 +1,21 @@
+
+
\ No newline at end of file
diff --git a/web/src/assets/images/workflow/deleteBg_hover.svg b/web/src/assets/images/workflow/deleteBg_hover.svg
new file mode 100644
index 00000000..9e92cf75
--- /dev/null
+++ b/web/src/assets/images/workflow/deleteBg_hover.svg
@@ -0,0 +1,22 @@
+
+
\ No newline at end of file
diff --git a/web/src/assets/images/workflow/delete_cycle.svg b/web/src/assets/images/workflow/delete_cycle.svg
new file mode 100644
index 00000000..0d85650d
--- /dev/null
+++ b/web/src/assets/images/workflow/delete_cycle.svg
@@ -0,0 +1,18 @@
+
+
\ No newline at end of file
diff --git a/web/src/assets/images/workflow/recall.svg b/web/src/assets/images/workflow/recall.svg
new file mode 100644
index 00000000..756f9060
--- /dev/null
+++ b/web/src/assets/images/workflow/recall.svg
@@ -0,0 +1,18 @@
+
+
\ No newline at end of file
diff --git a/web/src/assets/images/workflow/recall_hover.svg b/web/src/assets/images/workflow/recall_hover.svg
new file mode 100644
index 00000000..a2e949a0
--- /dev/null
+++ b/web/src/assets/images/workflow/recall_hover.svg
@@ -0,0 +1,18 @@
+
+
\ No newline at end of file
diff --git a/web/src/components/FormItem/DescWrapper.tsx b/web/src/components/FormItem/DescWrapper.tsx
new file mode 100644
index 00000000..300fc2b6
--- /dev/null
+++ b/web/src/components/FormItem/DescWrapper.tsx
@@ -0,0 +1,12 @@
+import clsx from "clsx";
+import type { FC, ReactNode } from "react";
+
+const DescWrapper: FC<{desc: string | ReactNode, className?: string}> = ({desc, className}) => {
+ return (
+
+ {desc}
+
+ )
+}
+
+export default DescWrapper
\ No newline at end of file
diff --git a/web/src/components/FormItem/LabelWrapper.tsx b/web/src/components/FormItem/LabelWrapper.tsx
new file mode 100644
index 00000000..461250d8
--- /dev/null
+++ b/web/src/components/FormItem/LabelWrapper.tsx
@@ -0,0 +1,13 @@
+import clsx from "clsx";
+import type { FC, ReactNode } from "react";
+
+const LabelWrapper: FC<{ title: string | ReactNode, className?: string; children?: ReactNode}> = ({title, className, children}) => {
+ return (
+
+ )
+}
+
+export default LabelWrapper
\ No newline at end of file
diff --git a/web/src/components/FormItem/SwitchFormItem.tsx b/web/src/components/FormItem/SwitchFormItem.tsx
new file mode 100644
index 00000000..e17a8728
--- /dev/null
+++ b/web/src/components/FormItem/SwitchFormItem.tsx
@@ -0,0 +1,45 @@
+import { Switch, Form, ConfigProvider } from "antd";
+import useSize from 'antd/lib/config-provider/hooks/useSize'
+import type { FC, ReactNode } from "react";
+import { useContext } from "react";
+
+import LabelWrapper from './LabelWrapper'
+import DescWrapper from './DescWrapper'
+
+interface SwitchFormItemProps {
+ title: string | ReactNode;
+ desc?: string | ReactNode;
+ name: string | string[];
+ size?: 'small' | 'default'
+ className?: string;
+ disabled?: boolean;
+}
+
+const SwitchFormItem: FC = ({
+ title,
+ desc,
+ name,
+ size = 'default',
+ className,
+ disabled
+}) => {
+ const componentSize = useSize()
+ console.log('componentSize', componentSize)
+
+ return (
+
+
+ {desc && }
+
+
+
+
+
+ )
+}
+
+export default SwitchFormItem
\ No newline at end of file
diff --git a/web/src/components/Header/index.tsx b/web/src/components/Header/index.tsx
index 9aeeab6b..fac432f5 100644
--- a/web/src/components/Header/index.tsx
+++ b/web/src/components/Header/index.tsx
@@ -54,7 +54,7 @@ const AppHeader: FC<{source?: 'space' | 'manage';}> = ({source = 'manage'}) => {
key: '1',
label: (<>
{user.username}
- {user.email}
+ {user.email}
),
},
{
diff --git a/web/src/components/Markdown/index.tsx b/web/src/components/Markdown/index.tsx
index d16b72e4..58650207 100644
--- a/web/src/components/Markdown/index.tsx
+++ b/web/src/components/Markdown/index.tsx
@@ -150,9 +150,19 @@ const RbMarkdown: FC = ({
)
}
+ // 处理键盘快捷键
+ const handleKeyDown = (e: React.KeyboardEvent) => {
+ if ((e.ctrlKey || e.metaKey) && e.key === 'c') {
+ const selection = window.getSelection()
+ if (selection && selection.toString()) {
+ navigator.clipboard.writeText(selection.toString())
+ }
+ }
+ }
+
// 预览模式
return (
-