def _auto_recall(user_message: str) -> str:    results = memory_store.search_memory(user_message, top_k=3)    if not results:        return ""    return "\n".join(f"- [{r['path']}] {r['snippet']}" for r in results)# 在 agent 循环中, 每轮:memory_context = _auto_recall(user_input)system_prompt = build_system_prompt(    mode="full", bootstrap=bootstrap_data,    skills_block=skills_block, memory_context=memory_context,)

Frontend (AutoSave/Tools) → Memory trait → create_memory factory                                              │                                              ▼                                    Backend? (config.memory.backend)                                              │                    ┌─────────────────────────┼─────────────────────────┐                    │                         │                         │                    ▼                         ▼                         ▼                  sqlite                   markdown                    lucid                    │                         │                         │                    └─────────────────────────┼─────────────────────────┘                                              │                                              ▼                                    Memory Categories                                    (Conversation / Daily / Core)                                              │                                              ▼                                    Persistent Storage

Hardware RAG -.-> load_chunks -> Markdown (每日 .md 文件)

Session.messages → Filter & Format → [MemoryStore.consolidate()] → LLM Request → [_SAVE_MEMORY_TOOL] → Parse Result → Update Memory Files → Update Session State

    User Input        |        v    SessionStore.load_session()  --> rebuild messages[] from JSONL        |        v    ContextGuard.guard_api_call()        |        +-- Attempt 0: normal call        |       |        |   overflow? --no--> success        |       |yes        +-- Attempt 1: truncate oversized tool results        |       |        |   overflow? --no--> success        |       |yes        +-- Attempt 2: compact history via LLM summary        |       |        |   overflow? --yes--> raise        |    SessionStore.save_turn()  --> append to JSONL        |        v    Print response    File layout:    workspace/.sessions/agents/{agent_id}/sessions/{session_id}.jsonl    workspace/.sessions/agents/{agent_id}/sessions.json  (index)

def compact_history(self, messages, api_client, model):    keep_count = max(4, int(len(messages) * 0.2))    compress_count = max(2, int(len(messages) * 0.5))    compress_count = min(compress_count, len(messages) - keep_count)    old_text = _serialize_messages_for_summary(messages[:compress_count])    summary_resp = api_client.messages.create(        model=model, max_tokens=2048,        system="You are a conversation summarizer. Be concise and factual.",        messages=[{"role": "user", "content": summary_prompt}],    )    # 用摘要 + "Understood" 对替换旧消息    compacted = [        {"role": "user", "content": "[Previous conversation summary]\n" + summary},        {"role": "assistant", "content": [{"type": "text",         "text": "Understood, I have the context."}]},    ]    compacted.extend(messages[compress_count:])    return compacted

## When to Update MEMORY.mdWrite important facts immediately using `edit_file` or `write_file`:- User preferences ("I prefer dark mode")- Project context ("The API uses OAuth2")- Relationships ("Alice is the project lead")

用户输入："记住，我的名字是张三" → AI 分析请求意图 → 识别需要更新长期记忆 → 选择适当的文件操作工具（[EditFileTool.execute() → edit_file] 或 [WriteFileTool.execute() → write_file] → 构造工具调用参数 → 执行文件操作（更新 MEMORY.md） → 返回执行结果给 LLM → 更新 MEMORY.md 文件

# /new命令 pythonif cmd == "/new":    #档案全部消息    if notawait self._consolidate_memory(temp,archive_all=True):    #处理错误    session.clear）#清空会话

[ContextBuilder] 初始化 → 加载 MEMORY.md → 注入系统提示 → LLM 会话中持续访问记忆

用户输入："查找上次会议记录" → AI 识别搜索意图 → 构造 grep 命令 → exec 工具调用：{"command": "grep -i '会议' memory/HISTORY.md"} → [ExecTool.execute()] → 执行 grep 命令 → 返回搜索结果给 LLM → LLM 整理结果并回应用户

会话增长 → 达到阈值 → 触发合并 → LLM 处理 → [save_memory / _SAVE_MEMORY_TOOL] 工具调用 → 更新 MEMORY.md 和 HISTORY.md

ContextBuilder 初始化 → 加载 MemoryStore → 注册记忆相关工具

# [AgentLoop._process_message()]) → 检查 unconsolidated 消息数量 → 判断是否超过 [memory_window]阈值 → 触发记忆合并任务unconsolidated = len(session.messages) - session.last_consolidatedif (unconsolidated >= self.memory_window and session.key not in self._consolidating):

用户消息→_process_message(）→检查是否需要整合  →[AgentLoop._consolidate_memory()] → [MemoryStore.consolidate()] → 提取需要归档的消息（old_messages）→ 读取当前长期记忆（current_memory，作为合并过程的上下文）→ 构建 LLM 提示（prompt）

待整合消息 +当前记忆 → 调用 LLM with [_SAVE_MEMORY_TOOL] → 系统提示："You are a memory consolidation agent..." → 用户提示：包含当前记忆和待处理对话 → 工具定义：[_SAVE_MEMORY_TOOL]

        current_memory = self.read_long_term()        # 构建用户消息         prompt = f"""Process this conversation and call the save_memory tool with your consolidation.## Current Long-term Memory{current_memory or"(empty)"}## Conversation to Process{chr(10).join(lines)}"""        try:            response = await provider.chat(                messages=[ # 系统消息如下                    {"role": "system", "content": "You are a memory consolidation agent. Call the save_memory tool with your consolidation of the conversation."},                    {"role": "user", "content": prompt}, # 用户消息                ],                tools=_SAVE_MEMORY_TOOL,                model=model,            )

LLM 必须调用 save_memory 工具 → 参数：{history_entry, memory_update} → [MemoryStore.consolidate()] 解析工具调用结果 → 更新 HISTORY.md 和 MEMORY.md

# ===================== 核心类定义 =====================class MemoryStore:    """Two-layer memory: MEMORY.md (long-term facts) + HISTORY.md (grep-searchable log)."""    def __init__(self, workspace: Path):        # 初始化记忆目录（确保目录存在，不存在则创建）        self.memory_dir = ensure_dir(workspace / "memory")        # 定义长期记忆文件路径（MEMORY.md）        self.memory_file = self.memory_dir / "MEMORY.md"        # 定义历史日志文件路径（HISTORY.md）        self.history_file = self.memory_dir / "HISTORY.md"    def read_long_term(self) -> str:        """读取长期记忆文件内容"""        # 检查文件是否存在        if self.memory_file.exists():            # 读取文件内容（UTF-8编码保证中文等字符正常）            return self.memory_file.read_text(encoding="utf-8")        # 文件不存在时返回空字符串        return ""    def write_long_term(self, content: str) -> None:        """写入内容到长期记忆文件（覆盖写入）"""        self.memory_file.write_text(content, encoding="utf-8")    def append_history(self, entry: str) -> None:        """追加内容到历史日志文件（不会覆盖原有内容）"""        # 以追加模式打开文件（a），UTF-8编码        with open(self.history_file, "a", encoding="utf-8") as f:            # 去除entry末尾的空白符，添加两个换行（分隔不同条目）后写入            f.write(entry.rstrip() + "\n\n")    def get_memory_context(self) -> str:        """生成格式化的长期记忆上下文（供Agent对话使用）"""        # 读取当前长期记忆内容        long_term = self.read_long_term()        # 有内容时返回格式化字符串，无内容时返回空字符串        return f"## Long-term Memory\n{long_term}" if long_term else ""    async def consolidate(        self,        session: Session,        provider: LLMProvider,        model: str,        *,        archive_all: bool = False,        memory_window: int = 50,    ) -> bool:        """Consolidate old messages into MEMORY.md + HISTORY.md via LLM tool call.        Returns True on success (including no-op), False on failure.        """        # 分支1：全量归档模式（archive_all=True）        if archive_all:            # 取所有会话消息作为待整合的旧消息            old_messages = session.messages            # 全量归档后不保留旧消息（keep_count=0）            keep_count = 0            # 记录日志：全量归档的消息数量            logger.info("Memory consolidation (archive_all): {} messages", len(session.messages))        # 分支2：增量归档模式（默认）        else:            # 计算保留的消息数量（记忆窗口的一半，避免频繁整合）            keep_count = memory_window // 2            # 若会话消息总数 ≤ 保留数量，无需整合，直接返回成功            if len(session.messages) <= keep_count:                return True            # 若自上次整合后无新消息，无需整合，直接返回成功            if len(session.messages) - session.last_consolidated <= 0:                return True            # 提取待整合的旧消息：从上次整合位置到倒数keep_count条            old_messages = session.messages[session.last_consolidated:-keep_count]            # 若待整合消息为空，直接返回成功            if not old_messages:                return True            # 记录日志：待整合消息数量、保留消息数量            logger.info("Memory consolidation: {} to consolidate, {} keep", len(old_messages), keep_count)        # 构建待整合的对话文本（带时间戳、角色、工具使用信息）        lines = []        for m in old_messages:            # 跳过无内容的消息            if not m.get("content"):                continue            # 拼接工具使用信息（若有）            tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else ""            # 格式化每行内容：[时间戳] 角色(大写) [工具信息]: 消息内容            lines.append(f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}")        # 读取当前长期记忆内容        current_memory = self.read_long_term()        # 构建LLM提示词：告知LLM需要整合对话并调用save_memory工具        prompt = f"""Process this conversation and call the save_memory tool with your consolidation.## Current Long-term Memory{current_memory or"(empty)"}## Conversation to Process{chr(10).join(lines)}"""        try:            # 调用LLM接口，触发记忆整合            response = await provider.chat(                messages=[                    # 系统提示：定义LLM的角色为记忆整合代理                    {"role": "system", "content": "You are a memory consolidation agent. Call the save_memory tool with your consolidation of the conversation."},                    # 用户提示：传入待整合的对话和当前记忆                    {"role": "user", "content": prompt},                ],                tools=_SAVE_MEMORY_TOOL,  # 传入save_memory工具描述                model=model,             # 指定使用的LLM模型            )            # 检查LLM是否调用了save_memory工具            if not response.has_tool_calls:                # 未调用工具时记录警告日志，返回失败                logger.warning("Memory consolidation: LLM did not call save_memory, skipping")                return False            # 提取工具调用的参数（取第一个工具调用的参数）            args = response.tool_calls[0].arguments            # 兼容处理：部分LLM返回的参数是JSON字符串，需解析为字典            if isinstance(args, str):                args = json.loads(args)            # 校验参数类型：必须是字典，否则返回失败            if not isinstance(args, dict):                logger.warning("Memory consolidation: unexpected arguments type {}", type(args).__name__)                return False            # 处理历史条目：追加到HISTORY.md            if entry := args.get("history_entry"):                # 确保entry是字符串类型（非字符串则转为JSON字符串）                if not isinstance(entry, str):                    entry = json.dumps(entry, ensure_ascii=False)                # 追加到历史日志文件                self.append_history(entry)            # 处理记忆更新：写入MEMORY.md（仅当内容变化时）            if update := args.get("memory_update"):                # 确保update是字符串类型（非字符串则转为JSON字符串）                if not isinstance(update, str):                    update = json.dumps(update, ensure_ascii=False)                # 仅当内容与当前记忆不同时才写入（避免无意义的写入）                if update != current_memory:                    self.write_long_term(update)            # 更新会话的最后整合位置：            # - 全量归档：重置为0            # - 增量归档：设为当前消息总数 - 保留数量            session.last_consolidated = 0 if archive_all else len(session.messages) - keep_count            # 记录日志：记忆整合完成            logger.info("Memory consolidation done: {} messages, last_consolidated={}", len(session.messages), session.last_consolidated)            # 返回成功            return True        except Exception:            # 捕获所有异常，记录错误日志            logger.exception("Memory consolidation failed")            # 返回失败            return False

---name: memorydescription: Two-layer memory system with grep-based recall.always: true---# Memory## Structure- `memory/MEMORY.md` — Long-term facts (preferences, project context, relationships). Always loaded into your context.- `memory/HISTORY.md` — Append-only event log. NOT loaded into context. Search it with grep.## Search Past Events```bashgrep -i "keyword" memory/HISTORY.md```Use the `exec` tool to run grep. Combine patterns: `grep -iE "meeting|deadline" memory/HISTORY.md`## When to Update MEMORY.mdWrite important facts immediately using `edit_file` or `write_file`:- User preferences ("I prefer dark mode")- Project context ("The API uses OAuth2")- Relationships ("Alice is the project lead")## Auto-consolidationOld conversations are automatically summarized and appended to HISTORY.md when the session grows large. Long-term facts are extracted to MEMORY.md. You don't need to manage this.

class WriteFileTool(Tool):    """Tool to write content to a file."""    def __init__(self, workspace: Path | None = None, allowed_dir: Path | None = None):        self._workspace = workspace        self._allowed_dir = allowed_dir    @property    def name(self) -> str:        return "write_file"    @property    def description(self) -> str:        return "Write content to a file at the given path. Creates parent directories if needed."    @property    def parameters(self) -> dict[str, Any]:        return {            "type": "object",            "properties": {                "path": {                    "type": "string",                    "description": "The file path to write to"                },                "content": {                    "type": "string",                    "description": "The content to write"                }            },            "required": ["path", "content"]        }    async def execute(self, path: str, content: str, **kwargs: Any) -> str:        try:            file_path = _resolve_path(path, self._workspace, self._allowed_dir)            file_path.parent.mkdir(parents=True, exist_ok=True)            file_path.write_text(content, encoding="utf-8")            return f"Successfully wrote {len(content)} bytes to {file_path}"        except PermissionError as e:            return f"Error: {e}"        except Exception as e:            return f"Error writing file: {str(e)}"class EditFileTool(Tool):    """Tool to edit a file by replacing text."""    def __init__(self, workspace: Path | None = None, allowed_dir: Path | None = None):        self._workspace = workspace        self._allowed_dir = allowed_dir    @property    def name(self) -> str:        return "edit_file"    @property    def description(self) -> str:        return "Edit a file by replacing old_text with new_text. The old_text must exist exactly in the file."    @property    def parameters(self) -> dict[str, Any]:        return {            "type": "object",            "properties": {                "path": {                    "type": "string",                    "description": "The file path to edit"                },                "old_text": {                    "type": "string",                    "description": "The exact text to find and replace"                },                "new_text": {                    "type": "string",                    "description": "The text to replace with"                }            },            "required": ["path", "old_text", "new_text"]        }    async def execute(self, path: str, old_text: str, new_text: str, **kwargs: Any) -> str:        try:            file_path = _resolve_path(path, self._workspace, self._allowed_dir)            if not file_path.exists():                return f"Error: File not found: {path}"            content = file_path.read_text(encoding="utf-8")            if old_text not in content:                return self._not_found_message(old_text, content, path)            # Count occurrences            count = content.count(old_text)            if count > 1:                return f"Warning: old_text appears {count} times. Please provide more context to make it unique."            new_content = content.replace(old_text, new_text, 1)            file_path.write_text(new_content, encoding="utf-8")            return f"Successfully edited {file_path}"        except PermissionError as e:            return f"Error: {e}"        except Exception as e:            return f"Error editing file: {str(e)}"    @staticmethod    def _not_found_message(old_text: str, content: str, path: str) -> str:        """Build a helpful error when old_text is not found."""        lines = content.splitlines(keepends=True)        old_lines = old_text.splitlines(keepends=True)        window = len(old_lines)        best_ratio, best_start = 0.0, 0        for i in range(max(1, len(lines) - window + 1)):            ratio = difflib.SequenceMatcher(None, old_lines, lines[i : i + window]).ratio()            if ratio > best_ratio:                best_ratio, best_start = ratio, i        if best_ratio > 0.5:            diff = "\n".join(difflib.unified_diff(                old_lines, lines[best_start : best_start + window],                fromfile="old_text (provided)", tofile=f"{path} (actual, line {best_start + 1})",                lineterm="",            ))            return f"Error: old_text not found in {path}.\nBest match ({best_ratio:.0%} similar) at line {best_start + 1}:\n{diff}"        return f"Error: old_text not found in {path}. No similar text found. Verify the file content."

# ===================== 核心常量定义 =====================# 定义save_memory工具的元数据（供LLM调用的工具描述）_SAVE_MEMORY_TOOL = [    {        "type": "function",  # 工具类型（固定为function）        "function": {            "name": "save_memory",  # 工具名称（唯一标识）            # 工具描述：告知LLM该工具的作用是将记忆整合结果保存到持久化存储            "description": "Save the memory consolidation result to persistent storage.",            # 工具参数定义（JSON Schema格式）            "parameters": {                "type": "object",  # 参数整体为对象类型                "properties": {                    # 第一个参数：历史条目（会话摘要）                    "history_entry": {                        "type": "string",  # 参数类型为字符串                        "description": "A paragraph (2-5 sentences) summarizing key events/decisions/topics. "                        "Start with [YYYY-MM-DD HH:MM]. Include detail useful for grep search.",                    },                    # 第二个参数：记忆更新（完整的长期记忆内容）                    "memory_update": {                        "type": "string",  # 参数类型为字符串                        "description": "Full updated long-term memory as markdown. Include all existing "                        "facts plus new ones. Return unchanged if nothing new.",                    },                },                "required": ["history_entry", "memory_update"],  # 必填参数列表            },        },    }]