""" Memori 配置数据类 用于管理 Memori 长期记忆系统的配置参数 """ from dataclasses import dataclass from typing import Optional @dataclass class MemoriConfig: """Memori 长期记忆配置类""" # 功能开关 enabled: bool = False # API 配置 api_key: Optional[str] = None # 语义搜索配置 semantic_search_top_k: int = 5 semantic_search_threshold: float = 0.7 semantic_search_embeddings_limit: int = 1000 # 记忆注入配置 inject_memory_to_system_prompt: bool = True memory_prompt_template: str = ( "\n\n=== 相关记忆 ===\n" "以下是从历史对话中检索到的相关信息,可以帮助你更好地回答用户问题:\n" "{memories}\n" "==================\n" ) # 增强配置 augmentation_enabled: bool = True augmentation_wait_timeout: Optional[float] = None # None 表示后台异步执行 # 多租户配置 entity_id: Optional[str] = None # 用户标识 process_id: Optional[str] = None # Bot 标识 session_id: Optional[str] = None # 会话标识 def get_attribution_tuple(self) -> tuple[str, str]: """获取 attribution 所需的元组 (entity_id, process_id) Returns: (entity_id, process_id) 元组 """ if not self.entity_id or not self.process_id: raise ValueError("entity_id and process_id are required for attribution") return (self.entity_id, self.process_id) def is_enabled(self) -> bool: """检查 Memori 功能是否启用 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) -> "MemoriConfig": """创建带有新 session_id 的配置副本 Args: session_id: 新的会话 ID Returns: 新的 MemoriConfig 实例 """ new_config = MemoriConfig(**self.__dict__) new_config.session_id = session_id return new_config