adai/MemoryBear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into fix/memory-enduser-config

Browse Source
This commit is contained in:
Ke Sun
2026-02-03 19:38:21 +08:00
parent 48f95e0ea4 940c594066
commit 8c69199689
151 changed files with 11318 additions and 1208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
api/app/services/app_chat_service.py
View File

@@ -171,7 +171,14 @@ class AppChatService:
self.conversation_service.save_conversation_messages(
conversation_id=conversation_id,
user_message=message,
assistant_message=result["content"]
assistant_message=result["content"],
meta_data={
"usage": result.get("usage", {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
)
elapsed_time = time.time() - start_time
@@ -310,6 +317,7 @@ class AppChatService:
# 流式调用 Agent
full_content = ""
total_tokens = 0
async for chunk in agent.chat_stream(
message=message,
history=history,
@@ -320,9 +328,12 @@ class AppChatService:
config_id=config_id,
memory_flag=memory_flag
):
full_content += chunk
# 发送消息块事件
yield f"event: message\ndata: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
if isinstance(chunk, int):
total_tokens = chunk
else:
full_content += chunk
# 发送消息块事件
yield f"event: message\ndata: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
elapsed_time = time.time() - start_time
@@ -339,7 +350,7 @@ class AppChatService:
content=full_content,
meta_data={
"model": api_key_obj.model_name,
"usage": {}
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": total_tokens}
}
)
@@ -416,7 +427,11 @@ class AppChatService:
meta_data={
"mode": result.get("mode"),
"elapsed_time": result.get("elapsed_time"),
"sub_results": result.get("sub_results")
"usage": result.get("usage", {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
)
@@ -458,6 +473,7 @@ class AppChatService:
yield f"event: start\ndata: {json.dumps({'conversation_id': str(conversation_id)}, ensure_ascii=False)}\n\n"
full_content = ""
total_tokens = 0
# 2. 创建编排器
orchestrator = MultiAgentOrchestrator(self.db, config)
@@ -474,16 +490,26 @@ class AppChatService:
storage_type=storage_type,
user_rag_memory_id=user_rag_memory_id
):
yield event
# 尝试提取内容（用于保存）
if "data:" in event:
try:
data_line = event.split("data: ", 1)[1].strip()
data = json.loads(data_line)
if "content" in data:
full_content += data["content"]
except:
pass
if "sub_usage" in event:
if "data:" in event:
try:
data_line = event.split("data: ", 1)[1].strip()
data = json.loads(data_line)
if "total_tokens" in data:
total_tokens += data["total_tokens"]
except:
pass
else:
yield event
# 尝试提取内容（用于保存）
if "data:" in event:
try:
data_line = event.split("data: ", 1)[1].strip()
data = json.loads(data_line)
if "content" in data:
full_content += data["content"]
except:
pass
elapsed_time = time.time() - start_time
@@ -499,7 +525,12 @@ class AppChatService:
role="assistant",
content=full_content,
meta_data={
"elapsed_time": elapsed_time
"elapsed_time": elapsed_time,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": total_tokens
}
}
)

14
api/app/services/conversation_service.py
View File

@@ -1,4 +1,5 @@
"""会话服务"""
import os
import uuid
from datetime import datetime, timedelta
from typing import Annotated
@@ -298,7 +299,8 @@ class ConversationService:
self,
conversation_id: uuid.UUID,
user_message: str,
assistant_message: str
assistant_message: str,
meta_data: Optional[dict] = None
):
"""
Save a pair of user and assistant messages to the conversation.
@@ -307,6 +309,7 @@ class ConversationService:
conversation_id (uuid.UUID): Conversation UUID.
user_message (str): User's message content.
assistant_message (str): Assistant's response content.
meta_data (Optional[dict]): Optional metadata for the messages.
"""
self.add_message(
conversation_id=conversation_id,
@@ -317,7 +320,8 @@ class ConversationService:
self.add_message(
conversation_id=conversation_id,
role="assistant",
content=assistant_message
content=assistant_message,
meta_data=meta_data
)
logger.debug(
@@ -526,12 +530,12 @@ class ConversationService:
takeaways=[],
info_score=0,
)
with open('app/services/prompt/conversation_summary_system.jinja2', 'r', encoding='utf-8') as f:
prompt_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'prompt')
with open(os.path.join(prompt_path, 'conversation_summary_system.jinja2'), 'r', encoding='utf-8') as f:
system_prompt = f.read()
rendered_system_message = Template(system_prompt).render()
with open('app/services/prompt/conversation_summary_user.jinja2', 'r', encoding='utf-8') as f:
with open(os.path.join(prompt_path, 'conversation_summary_user.jinja2'), 'r', encoding='utf-8') as f:
user_prompt = f.read()
rendered_user_message = Template(user_prompt).render(
language=language,

38
api/app/services/draft_run_service.py
View File

@@ -442,7 +442,14 @@ class DraftRunService:
user_message=message,
assistant_message=result["content"],
app_id=agent_config.app_id,
user_id=user_id
user_id=user_id,
meta_data={
"usage": result.get("usage", {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
)
response = {
@@ -649,6 +656,7 @@ class DraftRunService:
# 9. 流式调用 Agent
full_content = ""
total_tokens = 0
async for chunk in agent.chat_stream(
message=message,
history=history,
@@ -659,14 +667,22 @@ class DraftRunService:
user_rag_memory_id=user_rag_memory_id,
memory_flag=memory_flag
):
full_content += chunk
# 发送消息块事件
yield self._format_sse_event("message", {
"content": chunk
})
if isinstance(chunk, int):
total_tokens = chunk
else:
full_content += chunk
# 发送消息块事件
yield self._format_sse_event("message", {
"content": chunk
})
elapsed_time = time.time() - start_time
if sub_agent:
yield self._format_sse_event("sub_usage", {
"total_tokens": total_tokens
})
# 10. 保存会话消息
if not sub_agent and agent_config.memory and agent_config.memory.get("enabled"):
await self._save_conversation_message(
@@ -674,7 +690,10 @@ class DraftRunService:
user_message=message,
assistant_message=full_content,
app_id=agent_config.app_id,
user_id=user_id
user_id=user_id,
meta_data={
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": total_tokens}
}
)
# 11. 发送结束事件
@@ -898,6 +917,7 @@ class DraftRunService:
conversation_id: str,
user_message: str,
assistant_message: str,
meta_data: dict,
app_id: Optional[uuid.UUID] = None,
user_id: Optional[str] = None
) -> None:
@@ -909,6 +929,7 @@ class DraftRunService:
assistant_message: AI 回复消息
app_id: 应用ID（未使用，保留用于兼容性）
user_id: 用户ID（未使用，保留用于兼容性）
meta_data: token消耗
"""
try:
from app.services.conversation_service import ConversationService
@@ -927,7 +948,8 @@ class DraftRunService:
conversation_service.add_message(
conversation_id=conv_uuid,
role="assistant",
content=assistant_message
content=assistant_message,
meta_data=meta_data
)
logger.debug(

329
api/app/services/emotion_analytics_service.py
View File

@@ -17,12 +17,15 @@ from app.repositories.neo4j.neo4j_connector import Neo4jConnector
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.utils.config_utils import resolve_config_id
logger = get_business_logger()
class EmotionSuggestion(BaseModel):
"""情绪建议模型"""
type: str = Field(..., description="建议类型：emotion_balance/activity_recommendation/social_connection/stress_management")
type: str = Field(...,
description="建议类型：emotion_balance/activity_recommendation/social_connection/stress_management")
title: str = Field(..., description="建议标题")
content: str = Field(..., description="建议内容")
priority: str = Field(..., description="优先级：high/medium/low")
@@ -37,33 +40,33 @@ class EmotionSuggestionsResponse(BaseModel):
class EmotionAnalyticsService:
"""情绪分析服务
提供情绪数据的分析和统计功能，包括：
- 情绪标签统计
- 情绪词云数据
- 情绪健康指数计算
- 个性化情绪建议生成
Attributes:
emotion_repo: 情绪数据仓储实例
"""
def __init__(self):
"""初始化情绪分析服务"""
connector = Neo4jConnector()
self.emotion_repo = EmotionRepository(connector)
logger.info("情绪分析服务初始化完成")
async def get_emotion_tags(
self,
end_user_id: str,
emotion_type: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
limit: int = 10
self,
end_user_id: str,
emotion_type: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
limit: int = 10
) -> Dict[str, Any]:
"""获取情绪标签统计
查询指定用户的情绪类型分布，包括计数、百分比和平均强度。
确保返回所有6个情绪维度（joy、sadness、anger、fear、surprise、neutral），
即使某些维度没有数据也会返回count=0的记录。
@@ -71,8 +74,8 @@ class EmotionAnalyticsService:
"""
try:
logger.info(f"获取情绪标签统计: user={end_user_id}, type={emotion_type}, "
f"start={start_date}, end={end_date}, limit={limit}")
f"start={start_date}, end={end_date}, limit={limit}")
# 调用仓储层查询
tags = await self.emotion_repo.get_emotion_tags(
end_user_id=end_user_id,
@@ -81,13 +84,13 @@ class EmotionAnalyticsService:
end_date=end_date,
limit=limit
)
# 定义所有6个情绪维度
all_emotion_types = ['joy', 'sadness', 'anger', 'fear', 'surprise', 'neutral']
# 将查询结果转换为字典，方便查找
tags_dict = {tag["emotion_type"]: tag for tag in tags}
# 补全缺失的情绪维度
complete_tags = []
for emotion in all_emotion_types:
@@ -101,52 +104,52 @@ class EmotionAnalyticsService:
"percentage": 0.0,
"avg_intensity": 0.0
})
# 计算总数
total_count = sum(tag["count"] for tag in complete_tags)
# 如果有数据，重新计算百分比（因为补全了0值项）
if total_count > 0:
for tag in complete_tags:
if tag["count"] > 0:
tag["percentage"] = round((tag["count"] / total_count) * 100, 2)
# 构建时间范围信息
time_range = {}
if start_date:
time_range["start_date"] = start_date
if end_date:
time_range["end_date"] = end_date
# 格式化响应
response = {
"tags": complete_tags,
"total_count": total_count,
"time_range": time_range if time_range else None
}
logger.info(f"情绪标签统计完成: total_count={total_count}, tags_count={len(complete_tags)}")
return response
except Exception as e:
logger.error(f"获取情绪标签统计失败: {str(e)}", exc_info=True)
raise
async def get_emotion_wordcloud(
self,
end_user_id: str,
emotion_type: Optional[str] = None,
limit: int = 50
self,
end_user_id: str,
emotion_type: Optional[str] = None,
limit: int = 50
) -> Dict[str, Any]:
"""获取情绪词云数据
查询情绪关键词及其频率，用于生成词云可视化。
Args:
end_user_id: 宿主ID（用户组ID）
emotion_type: 可选的情绪类型过滤
limit: 返回关键词的最大数量
Returns:
Dict: 包含情绪词云数据的响应：
- keywords: 关键词列表
@@ -154,39 +157,39 @@ class EmotionAnalyticsService:
"""
try:
logger.info(f"获取情绪词云数据: user={end_user_id}, type={emotion_type}, limit={limit}")
# 调用仓储层查询
keywords = await self.emotion_repo.get_emotion_wordcloud(
end_user_id=end_user_id,
emotion_type=emotion_type,
limit=limit
)
# 计算总关键词数量
total_keywords = len(keywords)
# 格式化响应
response = {
"keywords": keywords,
"total_keywords": total_keywords
}
logger.info(f"情绪词云数据获取完成: total_keywords={total_keywords}")
return response
except Exception as e:
logger.error(f"获取情绪词云数据失败: {str(e)}", exc_info=True)
raise
def _calculate_positivity_rate(self, emotions: List[Dict[str, Any]]) -> Dict[str, Any]:
"""计算积极率
根据情绪类型分类正面、负面和中性情绪，计算积极率。
公式：(正面数 / (正面数 + 负面数)) * 100
Args:
emotions: 情绪数据列表，每个包含 emotion_type 字段
Returns:
Dict: 包含积极率计算结果：
- score: 积极率分数（0-100）
@@ -197,38 +200,38 @@ class EmotionAnalyticsService:
# 定义情绪分类
positive_emotions = {'joy', 'surprise'}
negative_emotions = {'sadness', 'anger', 'fear'}
# 统计各类情绪数量
positive_count = sum(1 for e in emotions if e.get('emotion_type') in positive_emotions)
negative_count = sum(1 for e in emotions if e.get('emotion_type') in negative_emotions)
neutral_count = sum(1 for e in emotions if e.get('emotion_type') == 'neutral')
# 计算积极率
total_non_neutral = positive_count + negative_count
if total_non_neutral > 0:
score = (positive_count / total_non_neutral) * 100
else:
score = 50.0 # 如果没有非中性情绪，默认为50
logger.debug(f"积极率计算: positive={positive_count}, negative={negative_count}, "
f"neutral={neutral_count}, score={score:.2f}")
f"neutral={neutral_count}, score={score:.2f}")
return {
"score": round(score, 2),
"positive_count": positive_count,
"negative_count": negative_count,
"neutral_count": neutral_count
}
def _calculate_stability(self, emotions: List[Dict[str, Any]]) -> Dict[str, Any]:
"""计算稳定性
基于情绪强度的标准差计算情绪稳定性。
公式：(1 - min(std_deviation, 1.0)) * 100
Args:
emotions: 情绪数据列表，每个包含 emotion_intensity 字段
Returns:
Dict: 包含稳定性计算结果：
- score: 稳定性分数（0-100）
@@ -236,7 +239,7 @@ class EmotionAnalyticsService:
"""
# 提取所有情绪强度
intensities = [e.get('emotion_intensity', 0.0) for e in emotions if e.get('emotion_intensity') is not None]
# 计算标准差
if len(intensities) >= 2:
std_deviation = statistics.stdev(intensities)
@@ -244,29 +247,29 @@ class EmotionAnalyticsService:
std_deviation = 0.0 # 只有一个数据点，标准差为0
else:
std_deviation = 0.0 # 没有数据，标准差为0
# 计算稳定性分数
# 标准差越小，稳定性越高
score = (1 - min(std_deviation, 1.0)) * 100
logger.debug(f"稳定性计算: intensities_count={len(intensities)}, "
f"std_deviation={std_deviation:.3f}, score={score:.2f}")
f"std_deviation={std_deviation:.3f}, score={score:.2f}")
return {
"score": round(score, 2),
"std_deviation": round(std_deviation, 3)
}
def _calculate_resilience(self, emotions: List[Dict[str, Any]]) -> Dict[str, Any]:
"""计算恢复力
分析情绪转换模式，统计从负面情绪恢复到正面情绪的能力。
公式：(负面到正面转换次数 / 总负面情绪数) * 100
Args:
emotions: 情绪数据列表，每个包含 emotion_type 和 created_at 字段
应该按时间顺序排列
Returns:
Dict: 包含恢复力计算结果：
- score: 恢复力分数（0-100）
@@ -275,24 +278,24 @@ class EmotionAnalyticsService:
# 定义情绪分类
positive_emotions = {'joy', 'surprise'}
negative_emotions = {'sadness', 'anger', 'fear'}
# 统计负面到正面的转换次数
recovery_count = 0
negative_count = 0
for i in range(len(emotions)):
current_emotion = emotions[i].get('emotion_type')
# 统计负面情绪总数
if current_emotion in negative_emotions:
negative_count += 1
# 检查下一个情绪是否为正面
if i + 1 < len(emotions):
next_emotion = emotions[i + 1].get('emotion_type')
if next_emotion in positive_emotions:
recovery_count += 1
# 计算恢复力分数
if negative_count > 0:
recovery_rate = recovery_count / negative_count
@@ -301,28 +304,28 @@ class EmotionAnalyticsService:
# 如果没有负面情绪，恢复力设为100（最佳状态）
recovery_rate = 1.0
score = 100.0
logger.debug(f"恢复力计算: negative_count={negative_count}, "
f"recovery_count={recovery_count}, score={score:.2f}")
f"recovery_count={recovery_count}, score={score:.2f}")
return {
"score": round(score, 2),
"recovery_rate": round(recovery_rate, 3)
}
async def calculate_emotion_health_index(
self,
end_user_id: str,
time_range: str = "30d"
self,
end_user_id: str,
time_range: str = "30d"
) -> Dict[str, Any]:
"""计算情绪健康指数
综合积极率、稳定性和恢复力计算情绪健康指数。
Args:
end_user_id: 宿主ID（用户组ID）
time_range: 时间范围（7d/30d/90d）
Returns:
Dict: 包含情绪健康指数的完整响应：
- health_score: 综合健康分数（0-100）
@@ -336,13 +339,13 @@ class EmotionAnalyticsService:
"""
try:
logger.info(f"计算情绪健康指数: user={end_user_id}, time_range={time_range}")
# 获取时间范围内的情绪数据
emotions = await self.emotion_repo.get_emotions_in_range(
end_user_id=end_user_id,
time_range=time_range
)
# 如果没有数据，返回默认值
if not emotions:
logger.warning(f"用户 {end_user_id} 在时间范围 {time_range} 内没有情绪数据")
@@ -357,20 +360,20 @@ class EmotionAnalyticsService:
"emotion_distribution": {},
"time_range": time_range
}
# 计算各维度指标
positivity_rate = self._calculate_positivity_rate(emotions)
stability = self._calculate_stability(emotions)
resilience = self._calculate_resilience(emotions)
# 计算综合健康分数
# 公式：positivity_rate * 0.4 + stability * 0.3 + resilience * 0.3
health_score = (
positivity_rate["score"] * 0.4 +
stability["score"] * 0.3 +
resilience["score"] * 0.3
positivity_rate["score"] * 0.4 +
stability["score"] * 0.3 +
resilience["score"] * 0.3
)
# 确定健康等级
if health_score >= 80:
level = "优秀"
@@ -380,13 +383,13 @@ class EmotionAnalyticsService:
level = "一般"
else:
level = "较差"
# 统计情绪分布
emotion_distribution = {}
for emotion_type in ['joy', 'sadness', 'anger', 'fear', 'surprise', 'neutral']:
count = sum(1 for e in emotions if e.get('emotion_type') == emotion_type)
emotion_distribution[emotion_type] = count
# 格式化响应
response = {
"health_score": round(health_score, 2),
@@ -399,22 +402,22 @@ class EmotionAnalyticsService:
"emotion_distribution": emotion_distribution,
"time_range": time_range
}
logger.info(f"情绪健康指数计算完成: score={health_score:.2f}, level={level}")
return response
except Exception as e:
logger.error(f"计算情绪健康指数失败: {str(e)}", exc_info=True)
raise
def _analyze_emotion_patterns(self, emotions: List[Dict[str, Any]]) -> Dict[str, Any]:
"""分析情绪模式
识别主要负面情绪、情绪触发因素和波动时段。
Args:
emotions: 情绪数据列表，每个包含 emotion_type、emotion_intensity、created_at 字段
Returns:
Dict: 包含情绪模式分析结果：
- dominant_negative_emotion: 主要负面情绪类型
@@ -422,19 +425,19 @@ class EmotionAnalyticsService:
- emotion_volatility: 情绪波动性（高/中/低）
"""
negative_emotions = {'sadness', 'anger', 'fear'}
# 统计负面情绪分布
negative_emotion_counts = {}
for emotion in emotions:
emotion_type = emotion.get('emotion_type')
if emotion_type in negative_emotions:
negative_emotion_counts[emotion_type] = negative_emotion_counts.get(emotion_type, 0) + 1
# 识别主要负面情绪
dominant_negative_emotion = None
if negative_emotion_counts:
dominant_negative_emotion = max(negative_emotion_counts, key=negative_emotion_counts.get)
# 识别高强度情绪（强度 >= 0.7）
high_intensity_emotions = [
{
@@ -445,7 +448,7 @@ class EmotionAnalyticsService:
for e in emotions
if e.get('emotion_intensity', 0) >= 0.7
]
# 评估情绪波动性
intensities = [e.get('emotion_intensity', 0.0) for e in emotions if e.get('emotion_intensity') is not None]
if len(intensities) >= 2:
@@ -458,29 +461,29 @@ class EmotionAnalyticsService:
volatility = "低"
else:
volatility = "未知"
logger.debug(f"情绪模式分析: dominant_negative={dominant_negative_emotion}, "
f"high_intensity_count={len(high_intensity_emotions)}, volatility={volatility}")
f"high_intensity_count={len(high_intensity_emotions)}, volatility={volatility}")
return {
"dominant_negative_emotion": dominant_negative_emotion,
"high_intensity_emotions": high_intensity_emotions[:5], # 最多返回5个
"emotion_volatility": volatility
}
async def generate_emotion_suggestions(
self,
end_user_id: str,
db: Session,
self,
end_user_id: str,
db: Session,
) -> Dict[str, Any]:
"""生成个性化情绪建议
基于情绪健康数据和用户画像生成个性化建议。
Args:
end_user_id: 宿主ID（用户组ID）
db: 数据库会话
Returns:
Dict: 包含个性化建议的响应：
- health_summary: 健康状态摘要
@@ -488,17 +491,17 @@ class EmotionAnalyticsService:
"""
try:
logger.info(f"生成个性化情绪建议: user={end_user_id}")
# 1. 从 end_user_id 获取关联的 memory_config_id
llm_client = None
try:
from app.services.memory_agent_service import (
get_end_user_connected_config,
)
connected_config = get_end_user_connected_config(end_user_id, db)
config_id = connected_config.get("memory_config_id")
config_id = resolve_config_id(config_id, db)
if config_id is not None:
from app.services.memory_config_service import (
MemoryConfigService,
@@ -513,35 +516,35 @@ class EmotionAnalyticsService:
llm_client = factory.get_llm_client(str(memory_config.llm_model_id))
except Exception as e:
logger.warning(f"无法获取 end_user {end_user_id} 的配置，将使用默认配置: {e}")
# 2. 获取情绪健康数据
health_data = await self.calculate_emotion_health_index(end_user_id, time_range="30d")
# 3. 获取情绪数据用于模式分析
emotions = await self.emotion_repo.get_emotions_in_range(
end_user_id=end_user_id,
time_range="30d"
)
# 4. 分析情绪模式
patterns = self._analyze_emotion_patterns(emotions)
# 5. 获取用户画像数据（简化版，直接从Neo4j获取）
user_profile = await self._get_simple_user_profile(end_user_id)
# 6. 构建LLM prompt
prompt = await self._build_suggestion_prompt(health_data, patterns, user_profile)
# 7. 调用LLM生成建议（使用配置中的LLM）
if llm_client is None:
# 无法获取配置时，抛出错误而不是使用默认配置
raise ValueError("无法获取LLM配置，请确保end_user关联了有效的memory_config")
# 将 prompt 转换为 messages 格式
messages = [
{"role": "user", "content": prompt}
]
# 8. 使用结构化输出直接获取 Pydantic 模型
try:
suggestions_response = await llm_client.response_structured(
@@ -552,7 +555,7 @@ class EmotionAnalyticsService:
logger.error(f"LLM 结构化输出失败: {str(e)}")
# 返回默认建议
suggestions_response = self._get_default_suggestions(health_data)
# 8. 验证建议数量（3-5条）
if len(suggestions_response.suggestions) < 3:
logger.warning(f"建议数量不足: {len(suggestions_response.suggestions)}")
@@ -560,7 +563,7 @@ class EmotionAnalyticsService:
elif len(suggestions_response.suggestions) > 5:
logger.warning(f"建议数量过多: {len(suggestions_response.suggestions)}")
suggestions_response.suggestions = suggestions_response.suggestions[:5]
# 9. 格式化响应
response = {
"health_summary": suggestions_response.health_summary,
@@ -575,26 +578,26 @@ class EmotionAnalyticsService:
for s in suggestions_response.suggestions
]
}
logger.info(f"个性化建议生成完成: suggestions_count={len(response['suggestions'])}")
return response
except Exception as e:
logger.error(f"生成个性化建议失败: {str(e)}", exc_info=True)
raise
async def _get_simple_user_profile(self, end_user_id: str) -> Dict[str, Any]:
"""获取简化的用户画像数据
Args:
end_user_id: 用户ID
Returns:
Dict: 用户画像数据
"""
try:
connector = Neo4jConnector()
# 查询用户的实体和标签
query = """
MATCH (e:Entity)
@@ -603,59 +606,59 @@ class EmotionAnalyticsService:
ORDER BY e.created_at DESC
LIMIT 20
"""
entities = await connector.execute_query(query, end_user_id=end_user_id)
# 提取兴趣标签
interests = [e["name"] for e in entities if e.get("type") in ["INTEREST", "HOBBY"]][:5]
# 后期会引入用户的习惯。。
return {
"interests": interests if interests else ["未知"]
}
except Exception as e:
logger.error(f"获取用户画像失败: {str(e)}")
return {"interests": ["未知"]}
async def _build_suggestion_prompt(
self,
health_data: Dict[str, Any],
patterns: Dict[str, Any],
user_profile: Dict[str, Any]
self,
health_data: Dict[str, Any],
patterns: Dict[str, Any],
user_profile: Dict[str, Any]
) -> str:
"""构建情绪建议生成的prompt
Args:
health_data: 情绪健康数据
patterns: 情绪模式分析结果
user_profile: 用户画像数据
Returns:
str: LLM prompt
"""
from app.core.memory.utils.prompt.prompt_utils import (
render_emotion_suggestions_prompt,
)
prompt = await render_emotion_suggestions_prompt(
health_data=health_data,
patterns=patterns,
user_profile=user_profile
)
return prompt
def _get_default_suggestions(self, health_data: Dict[str, Any]) -> EmotionSuggestionsResponse:
"""获取默认建议（当LLM调用失败时使用）
Args:
health_data: 情绪健康数据
Returns:
EmotionSuggestionsResponse: 默认建议
"""
health_score = health_data.get('health_score', 0)
if health_score >= 80:
summary = "您的情绪健康状况优秀，请继续保持积极的生活态度。"
elif health_score >= 60:
@@ -664,7 +667,7 @@ class EmotionAnalyticsService:
summary = "您的情绪健康需要关注，建议采取一些改善措施。"
else:
summary = "您的情绪健康需要重点关注，建议寻求专业帮助。"
suggestions = [
EmotionSuggestion(
type="emotion_balance",
@@ -700,54 +703,54 @@ class EmotionAnalyticsService:
]
)
]
return EmotionSuggestionsResponse(
health_summary=summary,
suggestions=suggestions
)
async def get_cached_suggestions(
self,
end_user_id: str,
db: Session,
self,
end_user_id: str,
db: Session,
) -> Optional[Dict[str, Any]]:
"""从 Redis 缓存获取个性化情绪建议
Args:
end_user_id: 宿主ID（用户组ID）
db: 数据库会话（保留参数以保持接口兼容性）
Returns:
Dict: 缓存的建议数据，如果不存在或已过期返回 None
"""
try:
from app.cache.memory.emotion_memory import EmotionMemoryCache
logger.info(f"尝试从 Redis 缓存获取情绪建议: user={end_user_id}")
# 从 Redis 获取缓存
cached_data = await EmotionMemoryCache.get_emotion_suggestions(end_user_id)
if cached_data is None:
logger.info(f"用户 {end_user_id} 的建议缓存不存在或已过期")
return None
logger.info(f"成功从 Redis 缓存获取建议: user={end_user_id}")
return cached_data
except Exception as e:
logger.error(f"从 Redis 缓存获取建议失败: {str(e)}", exc_info=True)
return None
async def save_suggestions_cache(
self,
end_user_id: str,
suggestions_data: Dict[str, Any],
db: Session,
expires_hours: int = 24
self,
end_user_id: str,
suggestions_data: Dict[str, Any],
db: Session,
expires_hours: int = 24
) -> None:
"""保存建议到 Redis 缓存
Args:
end_user_id: 宿主ID（用户组ID）
suggestions_data: 建议数据
@@ -756,24 +759,24 @@ class EmotionAnalyticsService:
"""
try:
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
)
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)
# 不抛出异常，缓存失败不应影响主流程

18
api/app/services/handoffs_service.py
View File

@@ -4,7 +4,7 @@ import uuid
from typing import List, Dict, Any, Optional, AsyncGenerator, Annotated
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, AIMessageChunk
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langgraph.checkpoint.memory import MemorySaver
@@ -727,9 +727,12 @@ class HandoffsService:
# 提取响应
response_content = ""
total_tokens = 0
for msg in result.get("messages", []):
if isinstance(msg, AIMessage):
response_content = msg.content
response_meta = msg.response_metadata if hasattr(msg, 'response_metadata') else None
total_tokens = response_meta.get("token_usage", {}).get("total_tokens", 0) if response_meta else 0
break
return {
@@ -737,7 +740,12 @@ class HandoffsService:
"active_agent": result.get("active_agent"),
"response": response_content,
"message_count": len(result.get("messages", [])),
"handoff_count": result.get("handoff_count", 0)
"handoff_count": result.get("handoff_count", 0),
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": total_tokens
}
}
async def chat_stream(
@@ -830,6 +838,12 @@ class HandoffsService:
# 捕获 LLM 结束事件，输出收集到的工具调用
elif kind == "on_chat_model_end":
output_message = event.get("data", {}).get("output", {})
if isinstance(output_message, AIMessageChunk):
response_meta = output_message.response_metadata if hasattr(output_message, 'response_metadata') else None
total_tokens = response_meta.get("token_usage", {}).get("total_tokens",
0) if response_meta else 0
yield f"event: sub_usage\ndata: {json.dumps({"total_tokens": total_tokens}, ensure_ascii=False)}\n\n"
if collected_tool_calls:
# 找到参数最完整的 transfer 工具调用
best_tc = None

28
api/app/services/memory_reflection_service.py
View File

@@ -89,7 +89,6 @@ class WorkspaceAppService:
for release in app_releases:
memory_content = self._extract_memory_content(release.config)
memory_content=resolve_config_id(memory_content, self.db)
if memory_content and memory_content in processed_configs:
continue
@@ -122,16 +121,12 @@ class WorkspaceAppService:
def _get_memory_config(self, memory_content: str) -> Dict[str, Any]:
"""Retrieve memory_config information based on memory_content"""
try:
memory_config_result = MemoryConfigRepository.query_reflection_config_by_id(self.db, int(memory_content))
# memory_config_query, memory_config_params = MemoryConfigRepository.build_select_reflection(memory_content)
# memory_config_result = self.db.execute(text(memory_config_query), memory_config_params).fetchone()
# if memory_config_result is None:
# return None
memory_content = resolve_config_id(memory_content, self.db)
memory_config_result = MemoryConfigRepository.query_reflection_config_by_id(self.db, (memory_content))
if memory_config_result:
return {
"config_id": memory_config_result.config_id,
"config_id": memory_content,
"enable_self_reflexion": memory_config_result.enable_self_reflexion,
"iteration_period": memory_config_result.iteration_period,
"reflexion_range": memory_config_result.reflexion_range,
@@ -291,7 +286,7 @@ class MemoryReflectionService:
# 检查是否需要执行反思
should_execute = False
hours_diff = 0
if current_reflection_time is None:
# 首次执行反思
should_execute = True
@@ -303,11 +298,11 @@ class MemoryReflectionService:
reflection_time = datetime.fromisoformat(current_reflection_time)
else:
reflection_time = current_reflection_time
current_time = datetime.now()
time_diff = current_time - reflection_time
hours_diff = int(time_diff.total_seconds() / 3600)
# 检查是否达到反思周期
if hours_diff >= iteration_period:
should_execute = True
@@ -317,7 +312,7 @@ class MemoryReflectionService:
except (ValueError, TypeError) as e:
api_logger.warning(f"解析反思时间失败: {e}，将执行反思")
should_execute = True
if should_execute:
api_logger.info(f"与上次的反思时间间隔为: {hours_diff} 小时")
# 3. 执行反思引擎
@@ -350,7 +345,7 @@ class MemoryReflectionService:
"next_reflection_in_hours": iteration_period - hours_diff
}
except Exception as e:
config_id = config_data.get("config_id", "unknown")
api_logger.error(f"启动反思失败，config_id: {config_id}, end_user_id: {end_user_id}, 错误: {str(e)}")
@@ -361,7 +356,7 @@ class MemoryReflectionService:
"end_user_id": end_user_id,
"config_data": config_data
}
def _create_reflection_config_from_data(self, config_data: Dict[str, Any]) -> ReflectionConfig:
"""Create reflective configuration objects from configuration data"""
@@ -369,12 +364,12 @@ class MemoryReflectionService:
if reflexion_range_value is None or reflexion_range_value == "":
reflexion_range_value = "partial"
reflexion_range = ReflectionRange(reflexion_range_value)
baseline_value = config_data.get("baseline")
if baseline_value is None or baseline_value == "":
baseline_value = "TIME"
baseline = ReflectionBaseline(baseline_value)
# iteration_period =
iteration_period = config_data.get("iteration_period", 24)
if isinstance(iteration_period, str):
@@ -382,7 +377,6 @@ class MemoryReflectionService:
iteration_period = int(iteration_period)
except (ValueError, TypeError):
iteration_period = 24 # 默认24小时
return ReflectionConfig(
enabled=config_data.get("enable_self_reflexion", False),
iteration_period=str(iteration_period), # ReflectionConfig期望字符串

8
api/app/services/model_service.py
View File

@@ -508,10 +508,7 @@ class ModelApiKeyService:
)
if not validation_result["valid"]:
# 记录验证失败的模型，但不抛出异常
failed_models.append({
"model_name": model_name,
"error": validation_result["error"]
})
failed_models.append(model_name)
continue
# 创建API Key
@@ -692,6 +689,9 @@ class ModelBaseService:
@staticmethod
def create_model_base(db: Session, data: model_schema.ModelBaseCreate):
existing = ModelBaseRepository.get_by_name_and_provider(db, data.name, data.provider)
if existing:
raise BusinessException("模型已存在", BizCode.DUPLICATE_NAME)
model_base = ModelBaseRepository.create(db, data.model_dump())
db.commit()
db.refresh(model_base)

20
api/app/services/multi_agent_orchestrator.py
View File

@@ -280,14 +280,22 @@ class MultiAgentOrchestrator:
# 4. 提取子 Agent 的 conversation_id（用于多轮对话）
sub_conversation_id = None
total_tokens = 0
if isinstance(results, dict):
sub_conversation_id = results.get("conversation_id") or results.get("result", {}).get("conversation_id")
# 提取 token 信息
usage = results.get("usage", {}) or results.get("result", {}).get("usage", {})
total_tokens += usage.get("total_tokens", 0)
elif isinstance(results, list) and results:
for item in results:
if "result" in item:
sub_conversation_id = item["result"].get("conversation_id")
if sub_conversation_id:
break
# 累加每个子 Agent 的 token
usage = item.get("usage", {}) or item.get("result", {}).get("usage", {})
total_tokens += usage.get("total_tokens", 0)
logger.info(
"多 Agent 任务完成",
@@ -301,9 +309,15 @@ class MultiAgentOrchestrator:
return {
"message": final_result,
"conversation_id": sub_conversation_id,
"mode": OrchestrationMode.SUPERVISOR,
"elapsed_time": elapsed_time,
"strategy": routing_decision.get("collaboration_strategy", "single"),
"sub_results": results
"sub_results": results,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": total_tokens
}
}
except Exception as e:
@@ -1552,10 +1566,12 @@ class MultiAgentOrchestrator:
return {
"message": result.get("response", ""),
"conversation_id": result.get("conversation_id"),
"mode": OrchestrationMode.COLLABORATION,
"elapsed_time": elapsed_time,
"strategy": "collaboration",
"active_agent": result.get("active_agent"),
"sub_results": result
"sub_results": result,
"usage": result.get("usage")
}
except Exception as e:

106
api/app/services/multi_agent_service.py
View File

@@ -1,5 +1,6 @@
"""多 Agent 配置管理服务"""
import uuid
import json
from typing import Optional, List, Tuple, Any, Annotated
from fastapi import Depends
@@ -427,6 +428,23 @@ class MultiAgentService:
memory=getattr(request, 'memory', True) # 记忆功能参数
)
await self._save_conversation_message(
conversation_id=request.conversation_id,
user_message=request.message,
assistant_message=result.get("message", ""),
app_id=app_id,
user_id=request.user_id,
meta_data={
"mode": result.get("mode"),
"elapsed_time": result.get("elapsed_time"),
"usage": result.get("usage", {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
)
return result
async def run_stream(
@@ -451,11 +469,14 @@ class MultiAgentService:
raise ResourceNotFoundException("多 Agent 配置", str(app_id))
if not config.is_active:
raise BusinessException("多 Agent 配置已禁用", BizCode.RESOURCE_DISABLED)
raise BusinessException("多 Agent 配置已禁用", BizCode.NOT_FOUND)
# 2. 创建编排器
orchestrator = MultiAgentOrchestrator(self.db, config)
full_content = ""
total_tokens = 0
# 3. 流式执行任务
async for event in orchestrator.execute_stream(
message=request.message,
@@ -468,7 +489,88 @@ class MultiAgentService:
storage_type=storage_type,
user_rag_memory_id=user_rag_memory_id
):
yield event
if "sub_usage" in event:
if "data:" in event:
try:
data_line = event.split("data: ", 1)[1].strip()
data = json.loads(data_line)
if "total_tokens" in data:
total_tokens += data["total_tokens"]
except:
pass
else:
yield event
if "data:" in event:
try:
data_line = event.split("data: ", 1)[1].strip()
data = json.loads(data_line)
if "content" in data:
full_content += data["content"]
except:
pass
await self._save_conversation_message(
conversation_id=request.conversation_id,
user_message=request.message,
assistant_message=full_content,
app_id=app_id,
user_id=request.user_id,
meta_data={
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": total_tokens
}
}
)
async def _save_conversation_message(
self,
conversation_id: uuid.UUID,
user_message: str,
assistant_message: str,
meta_data: dict,
app_id: Optional[uuid.UUID] = None,
user_id: Optional[str] = None
) -> None:
"""保存会话消息
Args:
conversation_id: 会话ID
user_message: 用户消息
assistant_message: AI 回复消息
meta_data: 元数据（包括 token 消耗）
app_id: 应用ID
user_id: 用户ID
"""
try:
from app.services.conversation_service import ConversationService
conversation_service = ConversationService(self.db)
conversation_service.add_message(
conversation_id=conversation_id,
role="user",
content=user_message
)
conversation_service.add_message(
conversation_id=conversation_id,
role="assistant",
content=assistant_message,
meta_data=meta_data
)
logger.debug(
"保存多 Agent 会话消息",
extra={
"conversation_id": conversation_id,
"user_message_length": len(user_message),
"assistant_message_length": len(assistant_message)
}
)
except Exception as e:
logger.warning("保存会话消息失败", extra={"error": str(e)})
# def add_sub_agent(
# self,

1162
api/app/services/ontology_service.py Normal file
View File

File diff suppressed because it is too large Load Diff

178
api/app/services/prompt_optimizer_service.py
View File

@@ -1,3 +1,4 @@
import os
import re
import uuid
from typing import Any, AsyncGenerator
@@ -18,7 +19,8 @@ from app.models.prompt_optimizer_model import (
)
from app.repositories.model_repository import ModelConfigRepository, ModelApiKeyRepository
from app.repositories.prompt_optimizer_repository import (
PromptOptimizerSessionRepository
PromptOptimizerSessionRepository,
PromptReleaseRepository
)
from app.schemas.prompt_optimizer_schema import OptimizePromptResult
@@ -28,6 +30,8 @@ logger = get_business_logger()
class PromptOptimizerService:
def __init__(self, db: Session):
self.db = db
self.optim_repo = PromptOptimizerSessionRepository(self.db)
self.release_repo = PromptReleaseRepository(self.db)
def get_model_config(
self,
@@ -78,10 +82,12 @@ class PromptOptimizerService:
Returns:
PromptOptimzerSession: The newly created prompt optimization session.
"""
session = PromptOptimizerSessionRepository(self.db).create_session(
session = self.optim_repo.create_session(
tenant_id=tenant_id,
user_id=user_id
)
self.db.commit()
self.db.refresh(session)
return session
def get_session_message_history(
@@ -106,7 +112,7 @@ class PromptOptimizerService:
- role (str): The role of the message sender, e.g., 'system', 'user', or 'assistant'.
- content (str): The content of the message.
"""
history = PromptOptimizerSessionRepository(self.db).get_session_history(
history = self.optim_repo.get_session_history(
session_id=session_id,
user_id=user_id
)
@@ -177,11 +183,12 @@ class PromptOptimizerService:
base_url=api_config.api_base
), type=ModelType(model_config.type))
try:
with open('app/services/prompt/prompt_optimizer_system.jinja2', 'r', encoding='utf-8') as f:
prompt_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'prompt')
with open(os.path.join(prompt_path, 'prompt_optimizer_system.jinja2'), 'r', encoding='utf-8') as f:
opt_system_prompt = f.read()
rendered_system_message = Template(opt_system_prompt).render()
with open('app/services/prompt/prompt_optimizer_user.jinja2', 'r', encoding='utf-8') as f:
with open(os.path.join(prompt_path, 'prompt_optimizer_user.jinja2'), 'r', encoding='utf-8') as f:
opt_user_prompt = f.read()
except FileNotFoundError:
raise BusinessException(message="System prompt template not found", code=BizCode.NOT_FOUND)
@@ -296,4 +303,165 @@ class PromptOptimizerService:
role=role,
content=content
)
self.db.commit()
self.db.refresh(message)
return message
def save_prompt(
self,
tenant_id: uuid.UUID,
session_id: uuid.UUID,
title: str,
prompt: str
) -> dict:
"""
Create and save a new prompt release for a given session.
Args:
tenant_id (uuid.UUID): The ID of the tenant owning the prompt.
session_id (uuid.UUID): The ID of the session to associate with this prompt.
title (str): The title of the prompt release.
prompt (str): The content of the prompt.
Returns:
dict: A dictionary containing:
- id (UUID): The unique ID of the created prompt release.
- session_id (UUID): The session ID linked to the release.
- title (str): The title of the prompt.
- prompt (str): The prompt content.
- created_at (int): Timestamp (in milliseconds) of when the prompt was created.
Raises:
BusinessException: If a prompt release already exists for the given session.
"""
session = self.optim_repo.get_session_by_id(session_id)
if session is None or session.tenant_id != tenant_id:
raise BusinessException(
"Session does not exist or the current user has no access",
BizCode.BAD_REQUEST
)
if self.release_repo.get_prompt_by_session_id(session_id):
raise BusinessException(
"A release already exists for the current session",
BizCode.BAD_REQUEST
)
prompt_obj = self.release_repo.create_prompt_release(
tenant_id=tenant_id,
title=title,
session_id=session_id,
prompt=prompt
)
self.db.commit()
self.db.refresh(prompt_obj)
return {
"id": prompt_obj.id,
"session_id": prompt_obj.session_id,
"title": prompt_obj.title,
"prompt": prompt_obj.prompt,
"created_at": int(prompt_obj.created_at.timestamp() * 1000)
}
def delete_prompt(
self,
tenant_id: uuid.UUID,
prompt_id: uuid.UUID
) -> None:
"""
Soft delete a prompt release by prompt_id.
Args:
tenant_id (uuid.UUID): Tenant identifier.
prompt_id (uuid.UUID): Prompt identifier.
Raises:
BusinessException: If the prompt does not exist or already deleted.
"""
prompt_obj = self.release_repo.get_prompt_by_id(prompt_id)
if not prompt_obj or prompt_obj.is_delete:
raise BusinessException(
"Prompt does not exist or has already been deleted",
BizCode.NOT_FOUND
)
if prompt_obj.tenant_id != tenant_id:
raise BusinessException(
"No permission to delete this prompt",
BizCode.FORBIDDEN
)
self.release_repo.soft_delete_prompt(prompt_obj)
self.db.commit()
logger.info(f"Prompt soft deleted, prompt_id={prompt_id}, tenant_id={tenant_id}")
def get_release_list(
self,
tenant_id: uuid.UUID,
page: int,
page_size: int,
filter_keyword: str | None = None
) -> dict[str, int | list[Any]]:
"""
Get paginated list of prompt releases with optional filter.
Args:
tenant_id (uuid.UUID): Tenant identifier.
page (int): Page number (starting from 1).
page_size (int): Number of items per page.
filter_keyword (str | None): Optional keyword to filter by title.
Returns:
dict: Contains total count, pagination info, and list of releases.
"""
offset = (page - 1) * page_size
# Get total count and releases based on filter
if filter_keyword:
total = self.release_repo.count_prompts_by_keyword(tenant_id, filter_keyword)
releases = self.release_repo.search_prompts_paginated(
tenant_id=tenant_id,
keyword=filter_keyword,
offset=offset,
limit=page_size
)
else:
total = self.release_repo.count_prompts(tenant_id)
releases = self.release_repo.get_prompts_paginated(
tenant_id=tenant_id,
offset=offset,
limit=page_size
)
items = []
for release in releases:
# Get first user message from session
first_message = self.optim_repo.get_first_user_message(
session_id=release.session_id
)
items.append({
"id": release.id,
"title": release.title,
"prompt": release.prompt,
"created_at": int(release.created_at.timestamp() * 1000),
"first_message": first_message
})
log_msg = f"Retrieved {len(items)} prompt releases, page={page}, tenant_id={tenant_id}"
if filter_keyword:
log_msg += f", filter='{filter_keyword}'"
logger.info(log_msg)
result = {
"page": {
"total": total,
"page": page,
"page_size": page_size,
"hasnext": page * page_size < total
},
"keyword": filter_keyword,
"items": items
}
return result

21
api/app/services/shared_chat_service.py
View File

@@ -282,7 +282,14 @@ class SharedChatService:
self.conversation_service.save_conversation_messages(
conversation_id=conversation.id,
user_message=message,
assistant_message=result["content"]
assistant_message=result["content"],
meta_data={
"usage": result.get("usage", {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
})
}
)
# self.conversation_service.add_message(
# conversation_id=conversation.id,
@@ -469,6 +476,7 @@ class SharedChatService:
# 流式调用 Agent
full_content = ""
total_tokens = 0
async for chunk in agent.chat_stream(
message=message,
history=history,
@@ -479,9 +487,12 @@ class SharedChatService:
config_id=config_id,
memory_flag=memory_flag
):
full_content += chunk
# 发送消息块事件
yield f"event: message\ndata: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
if isinstance(chunk, int):
total_tokens = chunk
else:
full_content += chunk
# 发送消息块事件
yield f"event: message\ndata: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
elapsed_time = time.time() - start_time
@@ -498,7 +509,7 @@ class SharedChatService:
content=full_content,
meta_data={
"model": api_key_obj.model_name,
"usage": {}
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": total_tokens}
}
)