lixinyue11
@@ -40,5 +40,4 @@ async def short_term_configs(
|
||||
"long_term_number":len(long_result)
|
||||
}
|
||||
|
||||
return success(data=result, msg="短期记忆系统数据获取成功")
|
||||
|
||||
return success(data=result, msg="短期记忆系统数据获取成功")
|
||||
@@ -17,6 +17,7 @@ from app.services.user_memory_service import (
|
||||
analytics_memory_types,
|
||||
analytics_graph_data,
|
||||
)
|
||||
from app.services.memory_entity_relationship_service import MemoryEntityService,MemoryEmotion,MemoryInteraction
|
||||
from app.schemas.response_schema import ApiResponse
|
||||
from app.schemas.memory_storage_schema import GenerateCacheRequest
|
||||
from app.schemas.end_user_schema import (
|
||||
@@ -47,7 +48,7 @@ async def get_memory_insight_report_api(
|
||||
) -> dict:
|
||||
"""
|
||||
获取缓存的记忆洞察报告
|
||||
|
||||
|
||||
此接口仅查询数据库中已缓存的记忆洞察数据，不执行生成操作。
|
||||
如需生成新的洞察报告，请使用专门的生成接口。
|
||||
"""
|
||||
@@ -55,7 +56,7 @@ async def get_memory_insight_report_api(
|
||||
try:
|
||||
# 调用服务层获取缓存数据
|
||||
result = await user_memory_service.get_cached_memory_insight(db, end_user_id)
|
||||
|
||||
|
||||
if result["is_cached"]:
|
||||
api_logger.info(f"成功返回缓存的记忆洞察报告: end_user_id={end_user_id}")
|
||||
return success(data=result, msg="查询成功")
|
||||
@@ -75,7 +76,7 @@ async def get_user_summary_api(
|
||||
) -> dict:
|
||||
"""
|
||||
获取缓存的用户摘要
|
||||
|
||||
|
||||
此接口仅查询数据库中已缓存的用户摘要数据，不执行生成操作。
|
||||
如需生成新的用户摘要，请使用专门的生成接口。
|
||||
"""
|
||||
@@ -83,7 +84,7 @@ async def get_user_summary_api(
|
||||
try:
|
||||
# 调用服务层获取缓存数据
|
||||
result = await user_memory_service.get_cached_user_summary(db, end_user_id)
|
||||
|
||||
|
||||
if result["is_cached"]:
|
||||
api_logger.info(f"成功返回缓存的用户摘要: end_user_id={end_user_id}")
|
||||
return success(data=result, msg="查询成功")
|
||||
@@ -103,35 +104,35 @@ async def generate_cache_api(
|
||||
) -> dict:
|
||||
"""
|
||||
手动触发缓存生成
|
||||
|
||||
|
||||
- 如果提供 end_user_id，只为该用户生成
|
||||
- 如果不提供，为当前工作空间的所有用户生成
|
||||
"""
|
||||
workspace_id = current_user.current_workspace_id
|
||||
|
||||
|
||||
# 检查用户是否已选择工作空间
|
||||
if workspace_id is None:
|
||||
api_logger.warning(f"用户 {current_user.username} 尝试生成缓存但未选择工作空间")
|
||||
return fail(BizCode.INVALID_PARAMETER, "请先切换到一个工作空间", "current_workspace_id is None")
|
||||
|
||||
|
||||
group_id = request.end_user_id
|
||||
|
||||
|
||||
api_logger.info(
|
||||
f"缓存生成请求: user={current_user.username}, workspace={workspace_id}, "
|
||||
f"end_user_id={group_id if group_id else '全部用户'}"
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
if group_id:
|
||||
# 为单个用户生成
|
||||
api_logger.info(f"开始为单个用户生成缓存: end_user_id={group_id}")
|
||||
|
||||
|
||||
# 生成记忆洞察
|
||||
insight_result = await user_memory_service.generate_and_cache_insight(db, group_id, workspace_id)
|
||||
|
||||
|
||||
# 生成用户摘要
|
||||
summary_result = await user_memory_service.generate_and_cache_summary(db, group_id, workspace_id)
|
||||
|
||||
|
||||
# 构建响应
|
||||
result = {
|
||||
"end_user_id": group_id,
|
||||
@@ -139,7 +140,7 @@ async def generate_cache_api(
|
||||
"summary_success": summary_result["success"],
|
||||
"errors": []
|
||||
}
|
||||
|
||||
|
||||
# 收集错误信息
|
||||
if not insight_result["success"]:
|
||||
result["errors"].append({
|
||||
@@ -151,29 +152,29 @@ async def generate_cache_api(
|
||||
"type": "summary",
|
||||
"error": summary_result.get("error")
|
||||
})
|
||||
|
||||
|
||||
# 记录结果
|
||||
if result["insight_success"] and result["summary_success"]:
|
||||
api_logger.info(f"成功为用户 {group_id} 生成缓存")
|
||||
else:
|
||||
api_logger.warning(f"用户 {group_id} 的缓存生成部分失败: {result['errors']}")
|
||||
|
||||
|
||||
return success(data=result, msg="生成完成")
|
||||
|
||||
|
||||
else:
|
||||
# 为整个工作空间生成
|
||||
api_logger.info(f"开始为工作空间 {workspace_id} 批量生成缓存")
|
||||
|
||||
|
||||
result = await user_memory_service.generate_cache_for_workspace(db, workspace_id)
|
||||
|
||||
|
||||
# 记录统计信息
|
||||
api_logger.info(
|
||||
f"工作空间 {workspace_id} 批量生成完成: "
|
||||
f"总数={result['total_users']}, 成功={result['successful']}, 失败={result['failed']}"
|
||||
)
|
||||
|
||||
|
||||
return success(data=result, msg="批量生成完成")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(f"缓存生成失败: user={current_user.username}, error={str(e)}")
|
||||
return fail(BizCode.INTERNAL_ERROR, "缓存生成失败", str(e))
|
||||
@@ -186,18 +187,18 @@ async def get_node_statistics_api(
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
workspace_id = current_user.current_workspace_id
|
||||
|
||||
|
||||
# 检查用户是否已选择工作空间
|
||||
if workspace_id is None:
|
||||
api_logger.warning(f"用户 {current_user.username} 尝试查询节点统计但未选择工作空间")
|
||||
return fail(BizCode.INVALID_PARAMETER, "请先切换到一个工作空间", "current_workspace_id is None")
|
||||
|
||||
|
||||
api_logger.info(f"记忆类型统计请求: end_user_id={end_user_id}, user={current_user.username}, workspace={workspace_id}")
|
||||
|
||||
|
||||
try:
|
||||
# 调用新的记忆类型统计函数
|
||||
result = await analytics_memory_types(db, end_user_id)
|
||||
|
||||
|
||||
# 计算总数用于日志
|
||||
total_count = sum(item["count"] for item in result)
|
||||
api_logger.info(f"成功获取记忆类型统计: end_user_id={end_user_id}, 总记忆数={total_count}, 类型数={len(result)}")
|
||||
@@ -217,31 +218,31 @@ async def get_graph_data_api(
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
workspace_id = current_user.current_workspace_id
|
||||
|
||||
|
||||
# 检查用户是否已选择工作空间
|
||||
if workspace_id is None:
|
||||
api_logger.warning(f"用户 {current_user.username} 尝试查询图数据但未选择工作空间")
|
||||
return fail(BizCode.INVALID_PARAMETER, "请先切换到一个工作空间", "current_workspace_id is None")
|
||||
|
||||
|
||||
# 参数验证
|
||||
if limit > 1000:
|
||||
limit = 1000
|
||||
api_logger.warning("limit 参数超过最大值，已调整为 1000")
|
||||
|
||||
|
||||
if depth > 3:
|
||||
depth = 3
|
||||
api_logger.warning("depth 参数超过最大值，已调整为 3")
|
||||
|
||||
|
||||
# 解析 node_types 参数
|
||||
node_types_list = None
|
||||
if node_types:
|
||||
node_types_list = [t.strip() for t in node_types.split(",") if t.strip()]
|
||||
|
||||
|
||||
api_logger.info(
|
||||
f"图数据查询请求: end_user_id={end_user_id}, user={current_user.username}, "
|
||||
f"workspace={workspace_id}, node_types={node_types_list}, limit={limit}, depth={depth}"
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
result = await analytics_graph_data(
|
||||
db=db,
|
||||
@@ -251,19 +252,19 @@ async def get_graph_data_api(
|
||||
depth=depth,
|
||||
center_node_id=center_node_id
|
||||
)
|
||||
|
||||
|
||||
# 检查是否有错误消息
|
||||
if "message" in result and result["statistics"]["total_nodes"] == 0:
|
||||
api_logger.warning(f"图数据查询返回空结果: {result.get('message')}")
|
||||
return success(data=result, msg=result.get("message", "查询成功"))
|
||||
|
||||
|
||||
api_logger.info(
|
||||
f"成功获取图数据: end_user_id={end_user_id}, "
|
||||
f"nodes={result['statistics']['total_nodes']}, "
|
||||
f"edges={result['statistics']['total_edges']}"
|
||||
)
|
||||
return success(data=result, msg="查询成功")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(f"图数据查询失败: end_user_id={end_user_id}, error={str(e)}")
|
||||
return fail(BizCode.INTERNAL_ERROR, "图数据查询失败", str(e))
|
||||
@@ -276,25 +277,25 @@ async def get_end_user_profile(
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
workspace_id = current_user.current_workspace_id
|
||||
|
||||
|
||||
# 检查用户是否已选择工作空间
|
||||
if workspace_id is None:
|
||||
api_logger.warning(f"用户 {current_user.username} 尝试查询用户信息但未选择工作空间")
|
||||
return fail(BizCode.INVALID_PARAMETER, "请先切换到一个工作空间", "current_workspace_id is None")
|
||||
|
||||
|
||||
api_logger.info(
|
||||
f"用户信息查询请求: end_user_id={end_user_id}, user={current_user.username}, "
|
||||
f"workspace={workspace_id}"
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
# 查询终端用户
|
||||
end_user = db.query(EndUser).filter(EndUser.id == end_user_id).first()
|
||||
|
||||
|
||||
if not end_user:
|
||||
api_logger.warning(f"终端用户不存在: end_user_id={end_user_id}")
|
||||
return fail(BizCode.INVALID_PARAMETER, "终端用户不存在", f"end_user_id={end_user_id}")
|
||||
|
||||
|
||||
# 构建响应数据
|
||||
profile_data = EndUserProfileResponse(
|
||||
id=end_user.id,
|
||||
@@ -306,10 +307,10 @@ async def get_end_user_profile(
|
||||
hire_date=end_user.hire_date,
|
||||
updatetime_profile=end_user.updatetime_profile
|
||||
)
|
||||
|
||||
|
||||
api_logger.info(f"成功获取用户信息: end_user_id={end_user_id}")
|
||||
return success(data=UserMemoryService.convert_profile_to_dict_with_timestamp(profile_data), msg="查询成功")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(f"用户信息查询失败: end_user_id={end_user_id}, error={str(e)}")
|
||||
return fail(BizCode.INTERNAL_ERROR, "用户信息查询失败", str(e))
|
||||
@@ -323,56 +324,56 @@ async def update_end_user_profile(
|
||||
) -> dict:
|
||||
"""
|
||||
更新终端用户的基本信息
|
||||
|
||||
|
||||
该接口可以更新用户的姓名、职位、部门、联系方式、电话和入职日期等信息。
|
||||
所有字段都是可选的，只更新提供的字段。
|
||||
|
||||
|
||||
"""
|
||||
workspace_id = current_user.current_workspace_id
|
||||
end_user_id = profile_update.end_user_id
|
||||
|
||||
|
||||
# 检查用户是否已选择工作空间
|
||||
if workspace_id is None:
|
||||
api_logger.warning(f"用户 {current_user.username} 尝试更新用户信息但未选择工作空间")
|
||||
return fail(BizCode.INVALID_PARAMETER, "请先切换到一个工作空间", "current_workspace_id is None")
|
||||
|
||||
|
||||
api_logger.info(
|
||||
f"用户信息更新请求: end_user_id={end_user_id}, user={current_user.username}, "
|
||||
f"workspace={workspace_id}"
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
# 查询终端用户
|
||||
end_user = db.query(EndUser).filter(EndUser.id == end_user_id).first()
|
||||
|
||||
|
||||
if not end_user:
|
||||
api_logger.warning(f"终端用户不存在: end_user_id={end_user_id}")
|
||||
return fail(BizCode.INVALID_PARAMETER, "终端用户不存在", f"end_user_id={end_user_id}")
|
||||
|
||||
|
||||
# 更新字段（只更新提供的字段，排除 end_user_id）
|
||||
# 允许 None 值来重置字段（如 hire_date）
|
||||
update_data = profile_update.model_dump(exclude_unset=True, exclude={'end_user_id'})
|
||||
|
||||
|
||||
# 特殊处理 hire_date：如果提供了时间戳，转换为 DateTime
|
||||
if 'hire_date' in update_data:
|
||||
hire_date_timestamp = update_data['hire_date']
|
||||
if hire_date_timestamp is not None:
|
||||
update_data['hire_date'] = timestamp_to_datetime(hire_date_timestamp)
|
||||
# 如果是 None，保持 None（允许清空）
|
||||
|
||||
|
||||
for field, value in update_data.items():
|
||||
setattr(end_user, field, value)
|
||||
|
||||
|
||||
# 更新 updated_at 时间戳
|
||||
end_user.updated_at = datetime.datetime.now()
|
||||
|
||||
|
||||
# 更新 updatetime_profile 为当前时间
|
||||
end_user.updatetime_profile = datetime.datetime.now()
|
||||
|
||||
|
||||
# 提交更改
|
||||
db.commit()
|
||||
db.refresh(end_user)
|
||||
|
||||
|
||||
# 构建响应数据
|
||||
profile_data = EndUserProfileResponse(
|
||||
id=end_user.id,
|
||||
@@ -384,11 +385,50 @@ async def update_end_user_profile(
|
||||
hire_date=end_user.hire_date,
|
||||
updatetime_profile=end_user.updatetime_profile
|
||||
)
|
||||
|
||||
|
||||
api_logger.info(f"成功更新用户信息: end_user_id={end_user_id}, updated_fields={list(update_data.keys())}")
|
||||
return success(data=UserMemoryService.convert_profile_to_dict_with_timestamp(profile_data), msg="更新成功")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
api_logger.error(f"用户信息更新失败: end_user_id={end_user_id}, error={str(e)}")
|
||||
return fail(BizCode.INTERNAL_ERROR, "用户信息更新失败", str(e))
|
||||
@router.get("/memory_space/timeline_memories", response_model=ApiResponse)
|
||||
async def memory_space_timeline_of_shared_memories(id: str, label: str,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
MemoryEntity = MemoryEntityService(id, label)
|
||||
timeline_memories_result = await MemoryEntity.get_timeline_memories_server()
|
||||
return success(data=timeline_memories_result, msg="共同记忆时间线")
|
||||
@router.get("/memory_space/relationship_evolution", response_model=ApiResponse)
|
||||
async def memory_space_relationship_evolution(id: str, label: str,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
api_logger.info(f"关系演变查询请求: id={id}, table={label}, user={current_user.username}")
|
||||
|
||||
# 获取情绪数据
|
||||
emotion = MemoryEmotion(id, label)
|
||||
emotion_result = await emotion.get_emotion()
|
||||
|
||||
# 获取交互数据
|
||||
interaction = MemoryInteraction(id, label)
|
||||
interaction_result = await interaction.get_interaction_frequency()
|
||||
|
||||
# 关闭连接
|
||||
await emotion.close()
|
||||
await interaction.close()
|
||||
|
||||
result = {
|
||||
"emotion": emotion_result,
|
||||
"interaction": interaction_result
|
||||
}
|
||||
|
||||
api_logger.info(f"关系演变查询成功: id={id}, table={label}")
|
||||
return success(data=result, msg="关系演变")
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(f"关系演变查询失败: id={id}, table={label}, error={str(e)}", exc_info=True)
|
||||
return fail(BizCode.INTERNAL_ERROR, "关系演变查询失败", str(e))
|
||||
|
||||
@@ -862,3 +862,120 @@ neo4j_query_all = """
|
||||
"""
|
||||
|
||||
|
||||
'''针对当前节点下扩长的句子，实体和总结'''
|
||||
Memory_Timeline_ExtractedEntity="""
|
||||
MATCH (n)-[r1]-(e)-[r2]-(ms)
|
||||
WHERE elementId(n) =$id
|
||||
AND (ms:ExtractedEntity OR ms:MemorySummary)
|
||||
RETURN
|
||||
collect(DISTINCT coalesce(ms.name, n.name, e.name)) AS ExtractedEntity,
|
||||
collect(DISTINCT ms.content) AS MemorySummary,
|
||||
collect(DISTINCT e.statement) AS statement;
|
||||
"""
|
||||
Memory_Timeline_MemorySummary="""
|
||||
MATCH (n)-[r1]-(e)-[r2]-(ms)
|
||||
WHERE elementId(n) = $id
|
||||
AND (ms:MemorySummary OR ms:ExtractedEntity)
|
||||
RETURN
|
||||
collect(DISTINCT coalesce(ms.name, n.name, e.name)) AS ExtractedEntity,
|
||||
collect(DISTINCT ms.content) AS MemorySummary,
|
||||
collect(DISTINCT e.statement) AS statement;"""
|
||||
Memory_Timeline_Statement="""
|
||||
MATCH (n)
|
||||
WHERE elementId(n) = "4:f6039a9b-d553-4ba2-9b1c-d9a18917801f:77154"
|
||||
|
||||
CALL {
|
||||
WITH n
|
||||
MATCH (n)-[]-(m)
|
||||
WHERE m:ExtractedEntity
|
||||
AND NOT m:MemorySummary
|
||||
AND NOT m:Chunk
|
||||
RETURN collect(DISTINCT m.name) AS ExtractedEntity
|
||||
}
|
||||
|
||||
CALL {
|
||||
WITH n
|
||||
MATCH (n)-[]-(m)
|
||||
WHERE m:MemorySummary
|
||||
AND NOT m:Chunk
|
||||
RETURN collect(DISTINCT m.content) AS MemorySummary
|
||||
}
|
||||
|
||||
RETURN
|
||||
ExtractedEntity,
|
||||
MemorySummary,
|
||||
collect(DISTINCT n.statement) AS Statement;
|
||||
|
||||
"""
|
||||
|
||||
'''针对当前节点，主要获取更加完整的句子节点'''
|
||||
Memory_Space_Emotion_Statement="""
|
||||
MATCH (n)
|
||||
WHERE elementId(n) = $id
|
||||
RETURN
|
||||
n.emotion_intensity AS emotion_intensity,
|
||||
n.created_at AS created_at,
|
||||
n.emotion_type AS emotion_type,
|
||||
n.statement AS statement;
|
||||
|
||||
"""
|
||||
Memory_Space_Emotion_MemorySummary="""
|
||||
MATCH (n)-[]-(e)
|
||||
WHERE elementId(n) = "4:f6039a9b-d553-4ba2-9b1c-d9a18917801f:77019"
|
||||
AND EXISTS {
|
||||
MATCH (e)-[]-(ms)
|
||||
WHERE ms:MemorySummary OR ms:ExtractedEntity
|
||||
}
|
||||
RETURN DISTINCT
|
||||
e.emotion_intensity AS emotion_intensity,
|
||||
e.created_at AS created_at,
|
||||
e.emotion_type AS emotion_type,
|
||||
e.statement AS statement;
|
||||
"""
|
||||
Memory_Space_Emotion_ExtractedEntity="""
|
||||
MATCH (n)-[]-(e)
|
||||
WHERE elementId(n) = $id
|
||||
AND EXISTS {
|
||||
MATCH (e)-[]-(ms:ExtractedEntity)
|
||||
}
|
||||
RETURN DISTINCT
|
||||
e.emotion_intensity AS emotion_intensity,
|
||||
e.created_at AS created_at,
|
||||
e.emotion_type AS emotion_type,
|
||||
e.statement AS statement;
|
||||
"""
|
||||
|
||||
'''获取实体'''
|
||||
Memory_Space_Interaction_Statement="""
|
||||
MATCH (n)-[]-(m)
|
||||
WHERE elementId(n) = $id
|
||||
AND m.entity_type = "Person"
|
||||
RETURN
|
||||
m.name AS name,
|
||||
m.importance_score AS importance_score;
|
||||
|
||||
"""
|
||||
|
||||
Memory_Space_Interaction_ExtractedEntity="""
|
||||
MATCH (n)-[]-(e)
|
||||
WHERE elementId(n) = $id
|
||||
AND EXISTS {
|
||||
MATCH (e)-[]-(ms:ExtractedEntity)
|
||||
}
|
||||
RETURN DISTINCT
|
||||
e.name AS name,
|
||||
e.importance_score AS importance_score;
|
||||
|
||||
"""
|
||||
|
||||
Memory_Space_Interaction_Summary="""
|
||||
MATCH (n)-[]-(e)
|
||||
WHERE elementId(n) = $id
|
||||
AND EXISTS {
|
||||
MATCH (e)-[]-(ms:ExtractedEntity)
|
||||
}
|
||||
RETURN DISTINCT
|
||||
e.name AS name,
|
||||
e.importance_score AS importance_score;
|
||||
|
||||
"""
|
||||
464
api/app/services/memory_entity_relationship_service.py
Normal file
464
api/app/services/memory_entity_relationship_service.py
Normal file
@@ -0,0 +1,464 @@
|
||||
|
||||
from app.repositories.neo4j.cypher_queries import (
|
||||
Memory_Timeline_ExtractedEntity,
|
||||
Memory_Timeline_MemorySummary,
|
||||
Memory_Timeline_Statement,
|
||||
Memory_Space_Emotion_Statement,
|
||||
Memory_Space_Emotion_MemorySummary,
|
||||
Memory_Space_Emotion_ExtractedEntity,
|
||||
Memory_Space_Interaction_Statement,
|
||||
Memory_Space_Interaction_ExtractedEntity,
|
||||
Memory_Space_Interaction_Summary
|
||||
)
|
||||
from app.repositories.neo4j.neo4j_connector import Neo4jConnector
|
||||
from typing import Dict, List, Any, Optional
|
||||
import logging
|
||||
from neo4j.time import DateTime as Neo4jDateTime
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class MemoryEntityService:
|
||||
def __init__(self, id: str, table: str):
|
||||
self.id = id
|
||||
self.table = table
|
||||
self.connector = Neo4jConnector()
|
||||
|
||||
|
||||
|
||||
async def get_timeline_memories_server(self):
|
||||
"""
|
||||
获取时间线记忆数据
|
||||
|
||||
Args:
|
||||
id: 节点ID
|
||||
table: 节点类型/标签
|
||||
|
||||
Returns:
|
||||
Dict包含：
|
||||
- success: 是否成功
|
||||
- data: 时间线数据列表
|
||||
- total: 数据总数
|
||||
- error: 错误信息（如果有）
|
||||
|
||||
根据不同标签返回相应字段：
|
||||
- MemorySummary: content字段
|
||||
- Statement: statement字段
|
||||
- ExtractedEntity: name字段
|
||||
"""
|
||||
try:
|
||||
logger.info(f"获取时间线记忆数据 - ID: {self.id}, Table: {self.table}")
|
||||
|
||||
# 根据表类型选择查询
|
||||
if self.table == 'Statement':
|
||||
# Statement只需要输入ID，使用简化查询
|
||||
results = await self.connector.execute_query(Memory_Timeline_Statement, id=self.id)
|
||||
elif self.table == 'ExtractedEntity':
|
||||
# ExtractedEntity类型查询
|
||||
results = await self.connector.execute_query(Memory_Timeline_ExtractedEntity, id=self.id)
|
||||
else:
|
||||
# MemorySummary类型查询
|
||||
results = await self.connector.execute_query(Memory_Timeline_MemorySummary, id=self.id)
|
||||
|
||||
# 记录查询结果的类型和内容用于调试
|
||||
logger.info(f"时间线查询结果类型: {type(results)}, 长度: {len(results) if isinstance(results, list) else 'N/A'}")
|
||||
|
||||
# 处理查询结果
|
||||
timeline_data = self._process_timeline_results(results)
|
||||
|
||||
logger.info(f"成功获取时间线记忆数据: 总计 {len(timeline_data.get('timelines_memory', []))} 条")
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'data': timeline_data,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取时间线记忆数据失败: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e),
|
||||
'data': {
|
||||
"MemorySummary": [],
|
||||
"Statement": [],
|
||||
"ExtractedEntity": [],
|
||||
"timelines_memory": []
|
||||
},
|
||||
'total': 0
|
||||
}
|
||||
def _process_timeline_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""
|
||||
处理时间线查询结果
|
||||
|
||||
Args:
|
||||
results: Neo4j查询结果
|
||||
|
||||
Returns:
|
||||
处理后的时间线数据字典
|
||||
"""
|
||||
# 检查results是否为空或不是列表
|
||||
if not results or not isinstance(results, list):
|
||||
logger.warning(f"时间线查询结果为空或格式不正确: {type(results)}")
|
||||
return {
|
||||
"MemorySummary": [],
|
||||
"Statement": [],
|
||||
"ExtractedEntity": [],
|
||||
"timelines_memory": []
|
||||
}
|
||||
|
||||
memory_summary_list = []
|
||||
statement_list = []
|
||||
extracted_entity_list = []
|
||||
|
||||
for data in results:
|
||||
# 检查data是否为字典类型
|
||||
if not isinstance(data, dict):
|
||||
logger.warning(f"跳过非字典类型的记录: {type(data)} - {data}")
|
||||
continue
|
||||
|
||||
# 处理MemorySummary
|
||||
summary = data.get('MemorySummary')
|
||||
if summary is not None:
|
||||
processed_summary = self._process_field_value(summary, "MemorySummary")
|
||||
memory_summary_list.extend(processed_summary)
|
||||
|
||||
# 处理Statement
|
||||
statement = data.get('statement')
|
||||
if statement is not None:
|
||||
processed_statement = self._process_field_value(statement, "Statement")
|
||||
statement_list.extend(processed_statement)
|
||||
|
||||
# 处理ExtractedEntity
|
||||
extracted_entity = data.get('ExtractedEntity')
|
||||
if extracted_entity is not None:
|
||||
processed_entity = self._process_field_value(extracted_entity, "ExtractedEntity")
|
||||
extracted_entity_list.extend(processed_entity)
|
||||
|
||||
# 去重
|
||||
memory_summary_list = list(set(memory_summary_list))
|
||||
statement_list = list(set(statement_list))
|
||||
extracted_entity_list = list(set(extracted_entity_list))
|
||||
|
||||
# 合并所有数据
|
||||
all_timeline_data = memory_summary_list + statement_list + extracted_entity_list
|
||||
|
||||
result = {
|
||||
"MemorySummary": memory_summary_list,
|
||||
"Statement": statement_list,
|
||||
"ExtractedEntity": extracted_entity_list,
|
||||
"timelines_memory": all_timeline_data
|
||||
}
|
||||
|
||||
logger.info(f"时间线数据处理完成: MemorySummary={len(memory_summary_list)}, Statement={len(statement_list)}, ExtractedEntity={len(extracted_entity_list)}")
|
||||
|
||||
return result
|
||||
|
||||
def _process_field_value(self, value: Any, field_name: str) -> List[str]:
|
||||
"""
|
||||
处理字段值，支持字符串、列表等类型
|
||||
|
||||
Args:
|
||||
value: 字段值
|
||||
field_name: 字段名称（用于日志）
|
||||
|
||||
Returns:
|
||||
处理后的字符串列表
|
||||
"""
|
||||
processed_values = []
|
||||
|
||||
try:
|
||||
if isinstance(value, list):
|
||||
# 如果是列表，处理每个元素
|
||||
for item in value:
|
||||
if item is not None and str(item).strip() != '' and "MemorySummaryChunk" not in str(item):
|
||||
processed_values.append(str(item))
|
||||
elif isinstance(value, str):
|
||||
# 如果是字符串，直接处理
|
||||
if value.strip() != '' and "MemorySummaryChunk" not in value:
|
||||
processed_values.append(value)
|
||||
elif value is not None:
|
||||
# 其他类型转换为字符串
|
||||
str_value = str(value)
|
||||
if str_value.strip() != '' and "MemorySummaryChunk" not in str_value:
|
||||
processed_values.append(str_value)
|
||||
except Exception as e:
|
||||
logger.warning(f"处理字段 {field_name} 的值时出错: {e}, 值类型: {type(value)}, 值: {value}")
|
||||
|
||||
return processed_values
|
||||
|
||||
|
||||
|
||||
|
||||
async def close(self):
|
||||
"""关闭数据库连接"""
|
||||
await self.connector.close()
|
||||
|
||||
|
||||
|
||||
class MemoryEmotion:
|
||||
def __init__(self, id: str, table: str):
|
||||
self.id = id
|
||||
self.table = table
|
||||
self.connector = Neo4jConnector()
|
||||
|
||||
def _convert_neo4j_types(self, obj: Any) -> Any:
|
||||
"""
|
||||
递归转换Neo4j特殊类型为可序列化的Python类型
|
||||
"""
|
||||
if isinstance(obj, Neo4jDateTime):
|
||||
# 转换为用户友好的日期格式
|
||||
return self._format_datetime(obj.iso_format())
|
||||
elif hasattr(obj, '__class__') and 'neo4j' in str(obj.__class__):
|
||||
if hasattr(obj, 'iso_format'):
|
||||
return self._format_datetime(obj.iso_format())
|
||||
elif hasattr(obj, '__str__'):
|
||||
return str(obj)
|
||||
else:
|
||||
return repr(obj)
|
||||
elif isinstance(obj, dict):
|
||||
return {k: self._convert_neo4j_types(v) for k, v in obj.items()}
|
||||
elif isinstance(obj, list):
|
||||
return [self._convert_neo4j_types(item) for item in obj]
|
||||
elif isinstance(obj, tuple):
|
||||
return tuple(self._convert_neo4j_types(item) for item in obj)
|
||||
else:
|
||||
return obj
|
||||
|
||||
def _format_datetime(self, iso_string: str) -> str:
|
||||
"""
|
||||
将ISO格式的日期时间字符串转换为用户友好的格式
|
||||
|
||||
Args:
|
||||
iso_string: ISO格式的日期时间字符串，如 "2026-01-07T13:40:33.679530"
|
||||
|
||||
Returns:
|
||||
格式化后的日期时间字符串，如 "2026-01-07 13:40:33"
|
||||
"""
|
||||
try:
|
||||
# 解析ISO格式的日期时间
|
||||
dt = datetime.fromisoformat(iso_string.replace('Z', '+00:00'))
|
||||
# 返回用户友好的格式：YYYY-MM-DD HH:MM:SS
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, AttributeError):
|
||||
# 如果解析失败，返回原始字符串
|
||||
return iso_string
|
||||
|
||||
async def get_emotion(self) -> Dict[str, Any]:
|
||||
"""
|
||||
获取情绪随时间变化数据
|
||||
|
||||
Returns:
|
||||
包含情绪数据的字典
|
||||
"""
|
||||
try:
|
||||
logger.info(f"获取情绪数据 - ID: {self.id}, Table: {self.table}")
|
||||
|
||||
if self.table == 'Statement':
|
||||
results = await self.connector.execute_query(Memory_Space_Emotion_Statement, id=self.id)
|
||||
elif self.table == 'ExtractedEntity':
|
||||
results = await self.connector.execute_query(Memory_Space_Emotion_ExtractedEntity, id=self.id)
|
||||
else:
|
||||
# MemorySummary/Chunk类型查询
|
||||
results = await self.connector.execute_query(Memory_Space_Emotion_MemorySummary, id=self.id)
|
||||
|
||||
# 处理查询结果
|
||||
emotion_data = self._process_emotion_results(results)
|
||||
|
||||
# 转换Neo4j类型
|
||||
final_data = self._convert_neo4j_types(emotion_data)
|
||||
|
||||
logger.info(f"成功获取 {len(final_data)} 条情绪数据")
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'data': final_data,
|
||||
'total': len(final_data)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取情绪数据失败: {str(e)}")
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e),
|
||||
'data': [],
|
||||
'total': 0
|
||||
}
|
||||
|
||||
def _process_emotion_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
处理情绪查询结果
|
||||
|
||||
Args:
|
||||
results: Neo4j查询结果
|
||||
|
||||
Returns:
|
||||
处理后的情绪数据列表
|
||||
"""
|
||||
emotion_data = []
|
||||
|
||||
# 检查results是否为空或不是列表
|
||||
if not results or not isinstance(results, list):
|
||||
logger.warning(f"情绪查询结果为空或格式不正确: {type(results)}")
|
||||
return emotion_data
|
||||
|
||||
for record in results:
|
||||
# 检查record是否为字典类型
|
||||
if not isinstance(record, dict):
|
||||
logger.warning(f"跳过非字典类型的记录: {type(record)} - {record}")
|
||||
continue
|
||||
|
||||
# 获取创建时间并格式化
|
||||
created_at = record.get('created_at')
|
||||
formatted_created_at = created_at
|
||||
|
||||
# 如果created_at是字符串格式，尝试格式化
|
||||
if isinstance(created_at, str):
|
||||
formatted_created_at = self._format_datetime(created_at)
|
||||
|
||||
emotion_type = record.get('emotion_type')
|
||||
emotion_intensity = record.get('emotion_intensity')
|
||||
|
||||
if emotion_type is not None and emotion_intensity is not None:
|
||||
# 只保留情绪相关的字段
|
||||
emotion_record = {
|
||||
'emotion_intensity': emotion_intensity,
|
||||
'emotion_type': emotion_type,
|
||||
'created_at': formatted_created_at
|
||||
}
|
||||
emotion_data.append(emotion_record)
|
||||
|
||||
return emotion_data
|
||||
|
||||
async def close(self):
|
||||
"""关闭数据库连接"""
|
||||
await self.connector.close()
|
||||
|
||||
|
||||
class MemoryInteraction:
|
||||
def __init__(self, id: str, table: str):
|
||||
self.id = id
|
||||
self.table = table
|
||||
self.connector = Neo4jConnector()
|
||||
|
||||
def _convert_neo4j_types(self, obj: Any) -> Any:
|
||||
"""
|
||||
递归转换Neo4j特殊类型为可序列化的Python类型
|
||||
"""
|
||||
if isinstance(obj, Neo4jDateTime):
|
||||
# 转换为用户友好的日期格式
|
||||
return self._format_datetime(obj.iso_format())
|
||||
elif hasattr(obj, '__class__') and 'neo4j' in str(obj.__class__):
|
||||
if hasattr(obj, 'iso_format'):
|
||||
return self._format_datetime(obj.iso_format())
|
||||
elif hasattr(obj, '__str__'):
|
||||
return str(obj)
|
||||
else:
|
||||
return repr(obj)
|
||||
elif isinstance(obj, dict):
|
||||
return {k: self._convert_neo4j_types(v) for k, v in obj.items()}
|
||||
elif isinstance(obj, list):
|
||||
return [self._convert_neo4j_types(item) for item in obj]
|
||||
elif isinstance(obj, tuple):
|
||||
return tuple(self._convert_neo4j_types(item) for item in obj)
|
||||
else:
|
||||
return obj
|
||||
|
||||
def _format_datetime(self, iso_string: str) -> str:
|
||||
"""
|
||||
将ISO格式的日期时间字符串转换为用户友好的格式
|
||||
|
||||
Args:
|
||||
iso_string: ISO格式的日期时间字符串，如 "2026-01-07T13:40:33.679530"
|
||||
|
||||
Returns:
|
||||
格式化后的日期时间字符串，如 "2026-01-07 13:40:33"
|
||||
"""
|
||||
try:
|
||||
# 解析ISO格式的日期时间
|
||||
dt = datetime.fromisoformat(iso_string.replace('Z', '+00:00'))
|
||||
# 返回用户友好的格式：YYYY-MM-DD HH:MM:SS
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, AttributeError):
|
||||
# 如果解析失败，返回原始字符串
|
||||
return iso_string
|
||||
|
||||
async def get_interaction_frequency(self) -> Dict[str, Any]:
|
||||
"""
|
||||
获取交互频率数据
|
||||
|
||||
Returns:
|
||||
包含交互数据的字典
|
||||
"""
|
||||
try:
|
||||
logger.info(f"获取交互数据 - ID: {self.id}, Table: {self.table}")
|
||||
|
||||
if self.table == 'Statement':
|
||||
results = await self.connector.execute_query(Memory_Space_Interaction_Statement, id=self.id)
|
||||
elif self.table == 'ExtractedEntity':
|
||||
results = await self.connector.execute_query(Memory_Space_Interaction_ExtractedEntity, id=self.id)
|
||||
else:
|
||||
# MemorySummary/Chunk类型查询
|
||||
results = await self.connector.execute_query(Memory_Space_Interaction_Summary, id=self.id)
|
||||
|
||||
# 处理查询结果
|
||||
interaction_data = self._process_interaction_results(results)
|
||||
|
||||
# 转换Neo4j类型
|
||||
final_data = self._convert_neo4j_types(interaction_data)
|
||||
|
||||
logger.info(f"成功获取 {len(final_data)} 条交互数据")
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'data': final_data,
|
||||
'total': len(final_data)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取交互数据失败: {str(e)}")
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e),
|
||||
'data': [],
|
||||
'total': 0
|
||||
}
|
||||
|
||||
def _process_interaction_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
处理交互查询结果
|
||||
|
||||
Args:
|
||||
results: Neo4j查询结果
|
||||
|
||||
Returns:
|
||||
处理后的交互数据列表
|
||||
"""
|
||||
interaction_data = []
|
||||
|
||||
# 检查results是否为空或不是列表
|
||||
if not results or not isinstance(results, list):
|
||||
logger.warning(f"交互查询结果为空或格式不正确: {type(results)}")
|
||||
return interaction_data
|
||||
|
||||
for record in results:
|
||||
# 检查record是否为字典类型
|
||||
if not isinstance(record, dict):
|
||||
logger.warning(f"跳过非字典类型的记录: {type(record)} - {record}")
|
||||
continue
|
||||
|
||||
# 只保留交互相关的字段
|
||||
name = record.get('name')
|
||||
if name is not None:
|
||||
interaction_record = {
|
||||
'name': name,
|
||||
'importance_score': record.get('importance_score', 0.0),
|
||||
'interaction_count': record.get('interaction_count', 1) # 默认交互次数为1
|
||||
}
|
||||
interaction_data.append(interaction_record)
|
||||
|
||||
return interaction_data
|
||||
|
||||
async def close(self):
|
||||
"""关闭数据库连接"""
|
||||
await self.connector.close()
|
||||
@@ -27,7 +27,7 @@ class ShortService:
|
||||
for item in retrieved_content:
|
||||
if isinstance(item, dict):
|
||||
for key, values in item.items():
|
||||
retrieval_source.append({"query": key, "retrieval": values})
|
||||
retrieval_source.append({"query": key, "retrieval": values,"source":"上下文记忆"})
|
||||
|
||||
deep_expanded['retrieval'] = retrieval_source
|
||||
deep_expanded['message'] = messages # 修正拼写错误
|
||||
|
||||
Reference in New Issue
Block a user