""" Mem0 配置数据类 用于管理 Mem0 长期记忆系统的配置参数 """ from dataclasses import dataclass from typing import TYPE_CHECKING, Optional # 避免循环导入 if TYPE_CHECKING: from langchain_core.language_models import BaseChatModel @dataclass class Mem0Config: """Mem0 长期记忆配置类""" # 功能开关 enabled: bool = False # 语义搜索配置 semantic_search_top_k: int = 20 memory_prompt_template: str = ( "\n\n=== 相关记忆 ===\n" "以下是从历史对话中检索到的相关信息,可以帮助你更好地回答用户问题:\n" "{memories}\n" "==================\n" ) # 多租户配置 user_id: Optional[str] = None # 用户标识 agent_id: Optional[str] = None # Bot 标识 session_id: Optional[str] = None # 会话标识 # LLM 实例(用于 Mem0 的记忆提取和增强) llm_instance: Optional["BaseChatModel"] = None # LangChain LLM 实例 def get_attribution_tuple(self) -> tuple[str, str]: """获取 attribution 所需的元组 (user_id, agent_id) Returns: (user_id, agent_id) 元组 """ if not self.user_id or not self.agent_id: raise ValueError("user_id and agent_id are required for attribution") return (self.user_id, self.agent_id) def is_enabled(self) -> bool: """检查 Mem0 功能是否启用 Returns: bool: 是否启用 """ return self.enabled def get_memory_prompt(self, memories: list[str]) -> str: """根据记忆列表生成注入提示词 Args: memories: 记忆内容列表 Returns: str: 格式化的记忆提示词 """ if not memories: return "" memory_text = "\n".join(f"- {m}" for m in memories) return self.memory_prompt_template.format(memories=memory_text) def with_session(self, session_id: str) -> "Mem0Config": """创建带有新 session_id 的配置副本 Args: session_id: 新的会话 ID Returns: 新的 Mem0Config 实例 """ new_config = Mem0Config(**self.__dict__) new_config.session_id = session_id return new_config