From bcb3d587a1dd98b3c775ffe410fe013dad0ca497 Mon Sep 17 00:00:00 2001 From: lixinyue11 <94037597+lixinyue11@users.noreply.github.com> Date: Wed, 7 Jan 2026 16:36:11 +0800 Subject: [PATCH] =?UTF-8?q?dev=E6=96=B0=E5=A2=9E=E7=9F=AD=E6=9C=9F?= =?UTF-8?q?=E8=AE=B0=E5=BF=86=E5=8A=9F=E8=83=BD=20(#47)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * dev新增短期记忆功能 * dev新增短期记忆功能 * dev新增短期记忆功能 * dev新增短期记忆功能 * dev新增短期记忆功能 * dev新增短期记忆功能 * dev新增短期记忆功能 --- api/app/controllers/__init__.py | 2 + .../memory_short_term_controller.py | 44 ++ api/app/core/agent/langchain_agent.py | 66 ++- api/app/models/__init__.py | 3 + api/app/models/memory_short_model.py | 60 +++ .../repositories/memory_short_repository.py | 503 ++++++++++++++++++ api/app/services/memory_agent_service.py | 72 ++- api/app/services/memory_short_service.py | 56 ++ api/app/services/user_memory_service.py | 4 +- 9 files changed, 765 insertions(+), 45 deletions(-) create mode 100644 api/app/controllers/memory_short_term_controller.py create mode 100644 api/app/models/memory_short_model.py create mode 100644 api/app/repositories/memory_short_repository.py create mode 100644 api/app/services/memory_short_service.py diff --git a/api/app/controllers/__init__.py b/api/app/controllers/__init__.py index 0b07d0c9..9d4b9248 100644 --- a/api/app/controllers/__init__.py +++ b/api/app/controllers/__init__.py @@ -24,6 +24,7 @@ from . import ( memory_storage_controller, memory_dashboard_controller, memory_reflection_controller, + memory_short_term_controller, api_key_controller, release_share_controller, public_share_controller, @@ -71,6 +72,7 @@ manager_router.include_router(emotion_controller.router) manager_router.include_router(emotion_config_controller.router) manager_router.include_router(prompt_optimizer_controller.router) manager_router.include_router(memory_reflection_controller.router) +manager_router.include_router(memory_short_term_controller.router) manager_router.include_router(tool_controller.router) manager_router.include_router(memory_forget_controller.router) manager_router.include_router(home_page_controller.router) diff --git a/api/app/controllers/memory_short_term_controller.py b/api/app/controllers/memory_short_term_controller.py new file mode 100644 index 00000000..f21a00b6 --- /dev/null +++ b/api/app/controllers/memory_short_term_controller.py @@ -0,0 +1,44 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from app.core.logging_config import get_api_logger +from app.core.response_utils import success +from app.db import get_db +from app.dependencies import get_current_user +from app.models.user_model import User + +from app.services.memory_storage_service import search_entity +from app.services.memory_short_service import ShortService,LongService +from dotenv import load_dotenv +from sqlalchemy.orm import Session +from typing import Optional +load_dotenv() +api_logger = get_api_logger() + +router = APIRouter( + prefix="/memory/short", + tags=["Memory"], +) +@router.get("/short_term") +async def short_term_configs( + end_user_id: str, + current_user: User = Depends(get_current_user), + db: Session = Depends(get_db), +): + # 获取短期记忆数据 + short_term=ShortService(end_user_id) + short_result=short_term.get_short_databasets() + short_count=short_term.get_short_count() + + long_term=LongService(end_user_id) + long_result=long_term.get_long_databasets() + + entity_result = await search_entity(end_user_id) + result = { + 'short_term': short_result, + 'long_term': long_result, + 'entity': entity_result.get('num', 0), + "retrieval_number":short_count, + "long_term_number":len(long_result) + } + + return success(data=result, msg="短期记忆系统数据获取成功") + diff --git a/api/app/core/agent/langchain_agent.py b/api/app/core/agent/langchain_agent.py index ef9a489f..91445b12 100644 --- a/api/app/core/agent/langchain_agent.py +++ b/api/app/core/agent/langchain_agent.py @@ -7,13 +7,20 @@ LangChain Agent 封装 - 支持流式输出 - 使用 RedBearLLM 支持多提供商 """ +import os import time from typing import Any, AsyncGenerator, Dict, List, Optional, Sequence + +from app.db import get_db from app.core.logging_config import get_business_logger from app.core.memory.agent.utils.redis_tool import store from app.core.models import RedBearLLM, RedBearModelConfig from app.models.models_model import ModelType +from app.repositories.memory_short_repository import LongTermMemoryRepository +from app.services.memory_agent_service import ( + get_end_user_connected_config, +) from app.services.memory_konwledges_server import write_rag from app.services.task_service import get_task_memory_write_result from app.tasks import write_message_task @@ -96,7 +103,8 @@ class LangChainAgent: "temperature": temperature, "streaming": streaming, "tool_count": len(self.tools), - "tool_names": [tool.name for tool in self.tools] if self.tools else [] + "tool_names": [tool.name for tool in self.tools] if self.tools else [], + "tool_count": len(self.tools) } ) @@ -137,11 +145,8 @@ class LangChainAgent: messages.append(HumanMessage(content=user_content)) return messages - async def term_memory_save(self,messages,end_user_end,aimessages): - """ - 短长期存储redis，为不影响正常使用6句一段话，存储用户名加一个前缀，当数据存够6条返回给neo4j - """ + '''短长期存储redis，为不影响正常使用6句一段话，存储用户名加一个前缀，当数据存够6条返回给neo4j''' end_user_end=f"Term_{end_user_end}" print(messages) print(aimessages) @@ -155,17 +160,18 @@ class LangChainAgent: store.delete_duplicate_sessions() # logger.info(f'Redis_Agent:{end_user_end};{session_id}') return session_id - async def term_memory_redis_read(self,end_user_end): end_user_end = f"Term_{end_user_end}" history = store.find_user_apply_group(end_user_end, end_user_end, end_user_end) # logger.info(f'Redis_Agent:{end_user_end};{history}') messagss_list=[] + retrieved_content=[] for messages in history: query = messages.get("Query") aimessages = messages.get("Answer") messagss_list.append(f'用户:{query}。AI回复:{aimessages}') - return messagss_list + retrieved_content.append({query: aimessages}) + return messagss_list,retrieved_content async def write(self,storage_type,end_user_id,message,user_rag_memory_id,actual_end_user_id,content,actual_config_id): @@ -205,7 +211,6 @@ class LangChainAgent: # If config_id is None, try to get from end_user's connected config if actual_config_id is None and end_user_id: try: - from app.db import get_db from app.services.memory_agent_service import ( get_end_user_connected_config, ) @@ -223,11 +228,26 @@ class LangChainAgent: logger.info(f'写入类型{storage_type,str(end_user_id), message, str(user_rag_memory_id)}') print(f'写入类型{storage_type,str(end_user_id), message, str(user_rag_memory_id)}') - history_term_memory=await self.term_memory_redis_read(end_user_id) + history_term_memory_result = await self.term_memory_redis_read(end_user_id) + history_term_memory = history_term_memory_result[0] + db_for_memory = next(get_db()) if memory_flag: if len(history_term_memory)>=4 and storage_type != "rag": - history_term_memory=';'.join(history_term_memory) - logger.info(f'写入短长期：{storage_type, str(end_user_id), history_term_memory, str(user_rag_memory_id)}') + history_term_memory = ';'.join(history_term_memory) + retrieved_content = history_term_memory_result[1] + print(retrieved_content) + # 为长期记忆操作获取新的数据库连接 + try: + repo = LongTermMemoryRepository(db_for_memory) + repo.upsert(end_user_id, retrieved_content) + logger.info( + f'写入短长期：{storage_type, str(end_user_id), history_term_memory, str(user_rag_memory_id)}') + except Exception as e: + logger.error(f"Failed to write to LongTermMemory: {e}") + raise + finally: + db_for_memory.close() + await self.write(storage_type,end_user_id,history_term_memory,user_rag_memory_id,actual_end_user_id,history_term_memory,actual_config_id) await self.write(storage_type,end_user_id,message,user_rag_memory_id,actual_end_user_id,message,actual_config_id) try: @@ -316,10 +336,6 @@ class LangChainAgent: # If config_id is None, try to get from end_user's connected config if actual_config_id is None and end_user_id: try: - from app.db import get_db - from app.services.memory_agent_service import ( - get_end_user_connected_config, - ) db = next(get_db()) try: connected_config = get_end_user_connected_config(end_user_id, db) @@ -331,14 +347,24 @@ class LangChainAgent: except Exception as e: logger.warning(f"Failed to get db session: {e}") - history_term_memory = await self.term_memory_redis_read(end_user_id) + history_term_memory_result = await self.term_memory_redis_read(end_user_id) + history_term_memory = history_term_memory_result[0] if memory_flag: if len(history_term_memory) >= 4 and storage_type != "rag": history_term_memory = ';'.join(history_term_memory) - logger.info( - f'写入短长期：{storage_type, str(end_user_id), history_term_memory, str(user_rag_memory_id)}') - await self.write(storage_type, end_user_id, history_term_memory, user_rag_memory_id, end_user_id, - history_term_memory, actual_config_id) + retrieved_content = history_term_memory_result[1] + db_for_memory = next(get_db()) + try: + repo = LongTermMemoryRepository(db_for_memory) + repo.upsert(end_user_id, retrieved_content) + logger.info( + f'写入短长期：{storage_type, str(end_user_id), history_term_memory, str(user_rag_memory_id)}') + await self.write(storage_type, end_user_id, history_term_memory, user_rag_memory_id, end_user_id, + history_term_memory, actual_config_id) + except Exception as e: + logger.error(f"Failed to write to long term memory: {e}") + finally: + db_for_memory.close() await self.write(storage_type, end_user_id, message, user_rag_memory_id, end_user_id, message, actual_config_id) try: diff --git a/api/app/models/__init__.py b/api/app/models/__init__.py index 01dad24e..158e607e 100644 --- a/api/app/models/__init__.py +++ b/api/app/models/__init__.py @@ -6,6 +6,7 @@ from .document_model import Document from .file_model import File from .generic_file_model import GenericFile from .models_model import ModelConfig, ModelProvider, ModelType, ModelApiKey +from .memory_short_model import ShortTermMemory, LongTermMemory from .knowledgeshare_model import KnowledgeShare from .app_model import App from .agent_app_config_model import AgentConfig @@ -67,6 +68,8 @@ __all__ = [ "BuiltinToolConfig", "CustomToolConfig", "MCPToolConfig", + "ShortTermMemory", + "LongTermMemory", "ToolExecution", "ToolType", "ToolStatus", diff --git a/api/app/models/memory_short_model.py b/api/app/models/memory_short_model.py new file mode 100644 index 00000000..6c3b1920 --- /dev/null +++ b/api/app/models/memory_short_model.py @@ -0,0 +1,60 @@ +""" +记忆模型 - 短期记忆和长期记忆表 +""" +import uuid +import datetime +from sqlalchemy import Column, String, DateTime, Text, JSON +from sqlalchemy.dialects.postgresql import UUID + +from app.db import Base + + +class ShortTermMemory(Base): + """短期记忆表 + + 用于存储临时的对话记忆，通常保存较短时间 + """ + __tablename__ = "memory_short_term" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True, comment="记忆ID") + + # 用户信息 + end_user_id = Column(String(255), nullable=False, index=True, comment="终端用户ID") + + # 对话内容 + messages = Column(Text, nullable=False, comment="用户消息内容") + aimessages = Column(Text, nullable=True, comment="AI回复消息内容") + + # 搜索开关 + search_switch = Column(String(50), nullable=True, comment="搜索开关状态") + + # 检索内容 - 存储为JSON格式的列表，包含字典 [{}, {}] + retrieved_content = Column(JSON, nullable=True, default=list, comment="检索到的相关内容，格式为[{}, {}]") + + # 时间戳 + created_at = Column(DateTime, default=datetime.datetime.now, nullable=False, index=True, comment="创建时间") + + def __repr__(self): + return f"" + + +class LongTermMemory(Base): + """长期记忆表 + + 用于存储重要的对话记忆，长期保存 + """ + __tablename__ = "memory_long_term" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True, comment="记忆ID") + + # 用户信息 + end_user_id = Column(String(255), nullable=False, index=True, comment="终端用户ID") + + # 检索内容 - 存储为JSON格式的列表，包含字典 [{}, {}] + retrieved_content = Column(JSON, nullable=True, default=list, comment="检索到的相关内容，格式为[{}, {}]") + + # 时间戳 + created_at = Column(DateTime, default=datetime.datetime.now, nullable=False, index=True, comment="创建时间") + + def __repr__(self): + return f"" \ No newline at end of file diff --git a/api/app/repositories/memory_short_repository.py b/api/app/repositories/memory_short_repository.py new file mode 100644 index 00000000..9a6e39c6 --- /dev/null +++ b/api/app/repositories/memory_short_repository.py @@ -0,0 +1,503 @@ +""" +记忆仓储模块 - 短期记忆和长期记忆的数据访问层 +""" +from sqlalchemy.orm import Session +from typing import List, Optional, Dict, Any +import uuid +import datetime + +from app.models.memory_short_model import ShortTermMemory, LongTermMemory +from app.core.logging_config import get_db_logger + +# 获取数据库专用日志器 +db_logger = get_db_logger() + + +class ShortTermMemoryRepository: + """短期记忆仓储类""" + + def __init__(self, db: Session): + self.db = db + + def create(self, end_user_id: str, messages: str, aimessages: str = None, search_switch: str = None, retrieved_content: List[Dict] = None) -> ShortTermMemory: + """创建短期记忆记录 + + Args: + end_user_id: 终端用户ID + messages: 用户消息内容 + aimessages: AI回复消息内容 + search_switch: 搜索开关状态 + retrieved_content: 检索到的相关内容，格式为[{}, {}] + + Returns: + ShortTermMemory: 创建的短期记忆对象 + """ + try: + memory = ShortTermMemory( + end_user_id=end_user_id, + messages=messages, + aimessages=aimessages, + search_switch=search_switch, + retrieved_content=retrieved_content or [] + ) + + self.db.add(memory) + self.db.commit() + self.db.refresh(memory) + + db_logger.info(f"成功创建短期记忆记录: {memory.id} for user {end_user_id}") + return memory + + except Exception as e: + self.db.rollback() + db_logger.error(f"创建短期记忆记录时出错: {str(e)}") + raise + + def count_by_user_id(self,end_user_id: str) -> int: + """根据ID获取短期记忆记录 + + Args: + memory_id: 记忆ID + + Returns: + Optional[ShortTermMemory]: 记忆对象，如果不存在则返回None + """ + try: + count = ( + self.db.query(ShortTermMemory) + .filter(ShortTermMemory.end_user_id == end_user_id) + .count() + ) + db_logger.debug(f"成功统计用户 {end_user_id} 的短期记忆数量: {count}") + + return count + + except Exception as e: + self.db.rollback() + db_logger.error(f"查询短期记忆记录 {count} 时出错: {str(e)}") + raise + + + def get_latest_by_user_id(self, end_user_id: str, limit: int = 5) -> List[ShortTermMemory]: + """获取用户最新的短期记忆记录 + + Args: + end_user_id: 终端用户ID + limit: 返回记录数限制，默认5条 + + Returns: + List[ShortTermMemory]: 最新的记忆记录列表，按创建时间倒序 + """ + try: + # 使用复合索引 ix_memory_short_term_user_time 优化查询 + memories = ( + self.db.query(ShortTermMemory) + .filter(ShortTermMemory.end_user_id == end_user_id) + .order_by(ShortTermMemory.created_at.desc()) + .limit(limit) + .all() + ) + + db_logger.info(f"成功查询用户 {end_user_id} 的最新 {len(memories)} 条短期记忆记录") + return memories + + except Exception as e: + self.db.rollback() + db_logger.error(f"查询用户 {end_user_id} 的最新短期记忆记录时出错: {str(e)}") + raise + + def get_recent_by_user_id(self, end_user_id: str, hours: int = 24) -> List[ShortTermMemory]: + """获取用户最近指定小时内的短期记忆记录 + + Args: + end_user_id: 终端用户ID + hours: 时间范围（小时），默认24小时 + + Returns: + List[ShortTermMemory]: 记忆记录列表，按创建时间倒序 + """ + try: + cutoff_time = datetime.datetime.now() - datetime.timedelta(hours=hours) + + # 使用复合索引 ix_memory_short_term_user_time 优化查询 + memories = ( + self.db.query(ShortTermMemory) + .filter( + ShortTermMemory.end_user_id == end_user_id, + ShortTermMemory.created_at >= cutoff_time + ) + .order_by(ShortTermMemory.created_at.desc()) + .all() + ) + + db_logger.info(f"成功查询用户 {end_user_id} 最近 {hours} 小时的 {len(memories)} 条短期记忆记录") + return memories + + except Exception as e: + self.db.rollback() + db_logger.error(f"查询用户 {end_user_id} 最近 {hours} 小时的短期记忆记录时出错: {str(e)}") + raise + + def delete_by_id(self, memory_id: uuid.UUID) -> bool: + """删除指定ID的短期记忆记录 + + Args: + memory_id: 记忆ID + + Returns: + bool: 删除成功返回True，否则返回False + """ + try: + deleted_count = ( + self.db.query(ShortTermMemory) + .filter(ShortTermMemory.id == memory_id) + .delete(synchronize_session=False) + ) + + self.db.commit() + + if deleted_count > 0: + db_logger.info(f"成功删除短期记忆记录 {memory_id}") + return True + else: + db_logger.warning(f"未找到短期记忆记录 {memory_id}，无法删除") + return False + + except Exception as e: + self.db.rollback() + db_logger.error(f"删除短期记忆记录 {memory_id} 时出错: {str(e)}") + raise + + def delete_old_memories(self, days: int = 7) -> int: + """删除指定天数之前的短期记忆记录 + + Args: + days: 保留天数，默认7天 + + Returns: + int: 删除的记录数 + """ + try: + cutoff_time = datetime.datetime.now() - datetime.timedelta(days=days) + + deleted_count = ( + self.db.query(ShortTermMemory) + .filter(ShortTermMemory.created_at < cutoff_time) + .delete(synchronize_session=False) + ) + + self.db.commit() + + db_logger.info(f"成功删除 {days} 天前的 {deleted_count} 条短期记忆记录") + return deleted_count + + except Exception as e: + self.db.rollback() + db_logger.error(f"删除 {days} 天前的短期记忆记录时出错: {str(e)}") + raise + + def upsert(self, end_user_id: str, messages: str, aimessages: str = None, search_switch: str = None, retrieved_content: List[Dict] = None) -> ShortTermMemory: + """创建或更新短期记忆记录 + + 根据 end_user_id、messages 和 aimessages 查找现有记录： + - 如果找到匹配的记录，则更新 messages、aimessages、search_switch 和 retrieved_content + - 如果没有找到匹配的记录，则创建新记录 + + Args: + end_user_id: 终端用户ID + messages: 用户消息内容 + aimessages: AI回复消息内容 + search_switch: 搜索开关状态 + retrieved_content: 检索到的相关内容，格式为[{}, {}] + + Returns: + ShortTermMemory: 创建或更新的短期记忆对象 + """ + try: + # 构建查询条件，使用复合索引 ix_memory_short_term_user_messages 优化查询 + query_filters = [ + ShortTermMemory.end_user_id == end_user_id, + ShortTermMemory.messages == messages + ] + + # 如果 aimessages 不为空，则加入查询条件 + if aimessages is not None: + query_filters.append(ShortTermMemory.aimessages == aimessages) + else: + # 如果 aimessages 为 None，则查找 aimessages 为 NULL 的记录 + query_filters.append(ShortTermMemory.aimessages.is_(None)) + + # 查找现有记录 + existing_memory = ( + self.db.query(ShortTermMemory) + .filter(*query_filters) + .first() + ) + + if existing_memory: + # 更新现有记录 + existing_memory.messages = messages + existing_memory.aimessages = aimessages + existing_memory.search_switch = search_switch + existing_memory.retrieved_content = retrieved_content or [] + + self.db.commit() + self.db.refresh(existing_memory) + + db_logger.info(f"成功更新短期记忆记录: {existing_memory.id} for user {end_user_id}") + return existing_memory + else: + # 创建新记录 + new_memory = ShortTermMemory( + end_user_id=end_user_id, + messages=messages, + aimessages=aimessages, + search_switch=search_switch, + retrieved_content=retrieved_content or [] + ) + + self.db.add(new_memory) + self.db.commit() + self.db.refresh(new_memory) + + db_logger.info(f"成功创建新的短期记忆记录: {new_memory.id} for user {end_user_id}") + return new_memory + + except Exception as e: + self.db.rollback() + db_logger.error(f"创建或更新短期记忆记录时出错: {str(e)}") + raise + + +class LongTermMemoryRepository: + """长期记忆仓储类""" + + def __init__(self, db: Session): + self.db = db + + def create(self, end_user_id: str, retrieved_content: List[Dict] = None) -> LongTermMemory: + """创建长期记忆记录 + + Args: + end_user_id: 终端用户ID + retrieved_content: 检索到的相关内容，格式为[{}, {}] + + Returns: + LongTermMemory: 创建的长期记忆对象 + """ + try: + memory = LongTermMemory( + end_user_id=end_user_id, + retrieved_content=retrieved_content or [] + ) + + self.db.add(memory) + self.db.commit() + self.db.refresh(memory) + + db_logger.info(f"成功创建长期记忆记录: {memory.id} for user {end_user_id}") + return memory + + except Exception as e: + self.db.rollback() + db_logger.error(f"创建长期记忆记录时出错: {str(e)}") + raise + + def get_by_id(self, memory_id: uuid.UUID) -> Optional[LongTermMemory]: + """根据ID获取长期记忆记录 + + Args: + memory_id: 记忆ID + + Returns: + Optional[LongTermMemory]: 记忆对象，如果不存在则返回None + """ + try: + memory = ( + self.db.query(LongTermMemory) + .filter(LongTermMemory.id == memory_id) + .first() + ) + + if memory: + db_logger.debug(f"成功查询到长期记忆记录 {memory_id}") + else: + db_logger.debug(f"未找到长期记忆记录 {memory_id}") + + return memory + + except Exception as e: + self.db.rollback() + db_logger.error(f"查询长期记忆记录 {memory_id} 时出错: {str(e)}") + raise + + def get_by_user_id(self, end_user_id: str, limit: int = 100, offset: int = 0) -> List[LongTermMemory]: + """根据用户ID获取长期记忆记录列表 + + Args: + end_user_id: 终端用户ID + limit: 返回记录数限制，默认100 + offset: 偏移量，默认0 + + Returns: + List[LongTermMemory]: 记忆记录列表，按创建时间倒序 + """ + try: + # 使用复合索引 ix_memory_long_term_user_time 优化查询 + memories = ( + self.db.query(LongTermMemory) + .filter(LongTermMemory.end_user_id == end_user_id) + .order_by(LongTermMemory.created_at.desc()) + .limit(limit) + .offset(offset) + .all() + ) + + db_logger.info(f"成功查询用户 {end_user_id} 的 {len(memories)} 条长期记忆记录") + return memories + + except Exception as e: + self.db.rollback() + db_logger.error(f"查询用户 {end_user_id} 的长期记忆记录时出错: {str(e)}") + raise + + def search_by_content(self, end_user_id: str, keyword: str, limit: int = 50) -> List[LongTermMemory]: + """根据内容关键词搜索长期记忆记录 + + Args: + end_user_id: 终端用户ID + keyword: 搜索关键词 + limit: 返回记录数限制，默认50 + + Returns: + List[LongTermMemory]: 匹配的记忆记录列表，按创建时间倒序 + """ + try: + # 使用 GIN 索引 ix_memory_long_term_retrieved_content_gin 优化 JSON 搜索 + # 同时使用复合索引 ix_memory_long_term_user_time 优化用户过滤 + memories = ( + self.db.query(LongTermMemory) + .filter( + LongTermMemory.end_user_id == end_user_id, + LongTermMemory.retrieved_content.astext.contains(keyword) + ) + .order_by(LongTermMemory.created_at.desc()) + .limit(limit) + .all() + ) + + db_logger.info(f"成功搜索用户 {end_user_id} 包含关键词 '{keyword}' 的 {len(memories)} 条长期记忆记录") + return memories + + except Exception as e: + self.db.rollback() + db_logger.error(f"搜索用户 {end_user_id} 包含关键词 '{keyword}' 的长期记忆记录时出错: {str(e)}") + raise + + def delete_by_id(self, memory_id: uuid.UUID) -> bool: + """删除指定ID的长期记忆记录 + + Args: + memory_id: 记忆ID + + Returns: + bool: 删除成功返回True，否则返回False + """ + try: + deleted_count = ( + self.db.query(LongTermMemory) + .filter(LongTermMemory.id == memory_id) + .delete(synchronize_session=False) + ) + + self.db.commit() + + if deleted_count > 0: + db_logger.info(f"成功删除长期记忆记录 {memory_id}") + return True + else: + db_logger.warning(f"未找到长期记忆记录 {memory_id}，无法删除") + return False + + except Exception as e: + self.db.rollback() + db_logger.error(f"删除长期记忆记录 {memory_id} 时出错: {str(e)}") + raise + + def count_by_user_id(self, end_user_id: str) -> int: + """统计用户的长期记忆记录数量 + + Args: + end_user_id: 终端用户ID + + Returns: + int: 记录数量 + """ + try: + count = ( + self.db.query(LongTermMemory) + .filter(LongTermMemory.end_user_id == end_user_id) + .count() + ) + + db_logger.debug(f"用户 {end_user_id} 共有 {count} 条长期记忆记录") + return count + + except Exception as e: + self.db.rollback() + db_logger.error(f"统计用户 {end_user_id} 的长期记忆记录数量时出错: {str(e)}") + raise + + def upsert(self, end_user_id: str, retrieved_content: List[Dict] = None) -> Optional[LongTermMemory]: + """创建或更新长期记忆记录 + + 根据 end_user_id 和 retrieved_content 判断是否需要写入： + - 如果找到相同的 end_user_id 和 retrieved_content，则不写入，返回 None + - 如果没有找到相同的记录，则创建新记录 + + Args: + end_user_id: 终端用户ID + retrieved_content: 检索到的相关内容，格式为[{}, {}] + + Returns: + Optional[LongTermMemory]: 创建的长期记忆对象，如果不需要写入则返回 None + """ + try: + retrieved_content = retrieved_content or [] + + # 优化查询：使用复合索引 ix_memory_long_term_user_time 先过滤用户 + # 然后在应用层比较 JSON 内容，避免复杂的数据库 JSON 比较 + existing_memories = ( + self.db.query(LongTermMemory) + .filter(LongTermMemory.end_user_id == end_user_id) + .order_by(LongTermMemory.created_at.desc()) + .limit(100) # 限制查询数量，避免加载过多数据 + .all() + ) + + # 在 Python 中比较 retrieved_content + for memory in existing_memories: + if memory.retrieved_content == retrieved_content: + # 如果找到相同的记录，不写入 + db_logger.info(f"长期记忆记录已存在，跳过写入: user {end_user_id}") + return None + + # 如果没有找到相同的记录，创建新记录 + new_memory = LongTermMemory( + end_user_id=end_user_id, + retrieved_content=retrieved_content + ) + + self.db.add(new_memory) + self.db.commit() + self.db.refresh(new_memory) + + db_logger.info(f"成功创建新的长期记忆记录: {new_memory.id} for user {end_user_id}") + return new_memory + + except Exception as e: + self.db.rollback() + db_logger.error(f"创建或更新长期记忆记录时出错: {str(e)}") + raise + + diff --git a/api/app/services/memory_agent_service.py b/api/app/services/memory_agent_service.py index 8193da8a..d44408fe 100644 --- a/api/app/services/memory_agent_service.py +++ b/api/app/services/memory_agent_service.py @@ -4,6 +4,7 @@ Memory Agent Service Handles business logic for memory agent operations including read/write services, health checks, and message type classification. """ +import datetime import json import os import re @@ -24,6 +25,7 @@ from app.core.memory.analytics.hot_memory_tags import get_hot_memory_tags from app.core.memory.utils.llm.llm_utils import MemoryClientFactory from app.db import get_db_context from app.models.knowledge_model import Knowledge, KnowledgeType +from app.repositories.memory_short_repository import ShortTermMemoryRepository from app.repositories.neo4j.neo4j_connector import Neo4jConnector from app.schemas.memory_config_schema import ConfigurationError, MemoryConfig from app.services.memory_config_service import MemoryConfigService @@ -393,7 +395,7 @@ class MemoryAgentService: import time start_time = time.time() - + ori_message=message # Resolve config_id if None using end_user's connected config if config_id is None: try: @@ -406,15 +408,15 @@ class MemoryAgentService: raise # Re-raise our specific error logger.error(f"Failed to get connected config for end_user {group_id}: {e}") raise ValueError(f"Unable to determine memory configuration for end_user {group_id}: {e}") - + logger.info(f"Read operation for group {group_id} with config_id {config_id}") - + # 导入审计日志记录器 try: from app.core.memory.utils.log.audit_logger import audit_logger except ImportError: audit_logger = None - + # Get group lock to prevent concurrent processing group_lock = self.get_group_lock(group_id) @@ -430,7 +432,7 @@ class MemoryAgentService: except ConfigurationError as e: error_msg = f"Failed to load configuration for config_id: {config_id}: {e}" logger.error(error_msg) - + # Log failed operation if audit_logger: duration = time.time() - start_time @@ -442,9 +444,9 @@ class MemoryAgentService: duration=duration, error=error_msg ) - + raise ValueError(error_msg) - + # Step 2: Prepare history history.append({"role": "user", "content": message}) logger.debug(f"Group ID:{group_id}, Message:{message}, History:{history}, Config ID:{config_id}") @@ -452,7 +454,7 @@ class MemoryAgentService: # Step 3: Initialize MCP client and execute read workflow mcp_config = get_mcp_server_config() client = MultiServerMCPClient(mcp_config) - + async with client.session('data_flow') as session: logger.debug("Connected to MCP Server: data_flow") tools = await load_mcp_tools(session) @@ -475,7 +477,7 @@ class MemoryAgentService: # Capture any errors from the state if event.get('errors'): workflow_errors.extend(event.get('errors', [])) - + for msg in messages: msg_content = msg.content msg_role = msg.__class__.__name__.lower().replace("message", "") @@ -483,7 +485,7 @@ class MemoryAgentService: "role": msg_role, "content": msg_content }) - + # Extract intermediate outputs if hasattr(msg, 'content'): try: @@ -496,7 +498,7 @@ class MemoryAgentService: break else: continue # No text block found - + # Try to parse content as JSON if isinstance(content_to_parse, str): try: @@ -506,16 +508,16 @@ class MemoryAgentService: if '_intermediate' in parsed: intermediate_data = parsed['_intermediate'] output_key = self._create_intermediate_key(intermediate_data) - + if output_key not in seen_intermediates: seen_intermediates.add(output_key) intermediate_outputs.append(self._format_intermediate_output(intermediate_data)) - + # Check for multiple intermediate outputs (from Retrieve) if '_intermediates' in parsed: for intermediate_data in parsed['_intermediates']: output_key = self._create_intermediate_key(intermediate_data) - + if output_key not in seen_intermediates: seen_intermediates.add(output_key) intermediate_outputs.append(self._format_intermediate_output(intermediate_data)) @@ -523,7 +525,7 @@ class MemoryAgentService: pass except Exception as e: logger.debug(f"Failed to extract intermediate output: {e}") - + workflow_duration = time.time() - start logger.info(f"Read graph workflow completed in {workflow_duration}s") @@ -532,7 +534,7 @@ class MemoryAgentService: for messages in outputs: if messages['role'] == 'tool': message = messages['content'] - + # Handle MCP content format: [{'type': 'text', 'text': '...'}] if isinstance(message, list): # Extract text from MCP content blocks @@ -542,7 +544,7 @@ class MemoryAgentService: break else: continue # No text block found - + try: parsed = json.loads(message) if isinstance(message, str) else message if isinstance(parsed, dict): @@ -552,15 +554,15 @@ class MemoryAgentService: final_answer = summary_result except (json.JSONDecodeError, ValueError): pass - + # 记录成功的操作 total_duration = time.time() - start_time - + # Check for workflow errors if workflow_errors: error_details = "; ".join([f"{e['tool']}: {e['error']}" for e in workflow_errors]) logger.warning(f"Read workflow completed with errors: {error_details}") - + if audit_logger: audit_logger.log_operation( operation="READ", @@ -577,11 +579,11 @@ class MemoryAgentService: "errors": workflow_errors } ) - + # Raise error if no answer was produced if not final_answer: raise ValueError(f"Read workflow failed: {error_details}") - + if audit_logger and not workflow_errors: audit_logger.log_operation( operation="READ", @@ -596,7 +598,31 @@ class MemoryAgentService: "has_answer": bool(final_answer) } ) - + retrieved_content=[] + repo = ShortTermMemoryRepository(db) + if str(search_switch)!="2": + for intermediate in intermediate_outputs: + intermediate_type=intermediate['type'] + if intermediate_type=="search_result": + query=intermediate['query'] + raw_results=intermediate['raw_results'] + reranked_results=raw_results.get('reranked_results',[]) + statements=[statement['statement'] for statement in reranked_results.get('statements', [])] + statements=list(set(statements)) + retrieved_content.append({query:statements}) + if '信息不足，无法回答' in str(final_answer) or retrieved_content!=[]: + # 使用 upsert 方法 + repo.upsert( + end_user_id=group_id, # 确保这个变量在作用域内 + messages=ori_message, + aimessages=final_answer, + retrieved_content=retrieved_content, + search_switch=str(search_switch) + ) + print("写入成功") + + + return { "answer": final_answer, "intermediate_outputs": intermediate_outputs diff --git a/api/app/services/memory_short_service.py b/api/app/services/memory_short_service.py new file mode 100644 index 00000000..ac9f86e0 --- /dev/null +++ b/api/app/services/memory_short_service.py @@ -0,0 +1,56 @@ + +from app.core.logging_config import get_api_logger +from app.db import get_db +from app.repositories.memory_short_repository import LongTermMemoryRepository +from app.repositories.memory_short_repository import ShortTermMemoryRepository + + +api_logger = get_api_logger() +db=next(get_db()) +class ShortService: + def __init__(self, end_user_id): + self.short_repo = ShortTermMemoryRepository(db) + self.end_user_id = end_user_id + + def get_short_databasets(self): + short_memories = self.short_repo.get_latest_by_user_id(self.end_user_id, 3) + short_result = [] + for memory in short_memories: + deep_expanded = {} # Create a new dictionary for each memory + messages = memory.messages + aimessages = memory.aimessages + retrieved_content = memory.retrieved_content or [] + + api_logger.debug(f"Retrieved content: {retrieved_content}") + + retrieval_source = [] + for item in retrieved_content: + if isinstance(item, dict): + for key, values in item.items(): + retrieval_source.append({"query": key, "retrieval": values}) + + deep_expanded['retrieval'] = retrieval_source + deep_expanded['message'] = messages # 修正拼写错误 + deep_expanded['answer'] = aimessages + short_result.append(deep_expanded) + return short_result + def get_short_count(self): + short_count = self.short_repo.count_by_user_id(self.end_user_id) + return short_count + +class LongService: + def __init__(self, end_user_id): + self.long_repo = LongTermMemoryRepository(db) + self.end_user_id = end_user_id + def get_long_databasets(self): + # 获取长期记忆数据 + long_memories = self.long_repo.get_by_user_id(self.end_user_id, 1) + + long_result = [] + for long_memory in long_memories: + if long_memory.retrieved_content: + for memory_item in long_memory.retrieved_content: + if isinstance(memory_item, dict): + for key, values in memory_item.items(): + long_result.append({"query": key, "retrieval": values}) + return long_result diff --git a/api/app/services/user_memory_service.py b/api/app/services/user_memory_service.py index 40851835..25577cbf 100644 --- a/api/app/services/user_memory_service.py +++ b/api/app/services/user_memory_service.py @@ -1496,8 +1496,8 @@ def _extract_node_properties(label: str, properties: Dict[str, Any]) -> Dict[str field_whitelist = { "Dialogue": ["content", "created_at"], "Chunk": ["content", "created_at"], - "Statement": ["temporal_info", "stmt_type", "statement", "valid_at", "created_at", "caption"], - "ExtractedEntity": ["description", "name", "entity_type", "created_at", "caption"], + "Statement": ["temporal_info", "stmt_type", "statement", "valid_at", "created_at", "caption","emotion_keywords","emotion_type","emotion_subject"], + "ExtractedEntity": ["description", "name", "entity_type", "created_at", "caption","aliases","connect_strength"], "MemorySummary": ["summary", "content", "created_at", "caption"] # 添加 content 字段 }