手搓Claude Code-第九章 memory

写在前面

  这一章的改动量可以说更上了一层楼，本章只保留了基本工具、s08的压缩、子agent和新加入的memory策略，其他无关的功能一律删去了。不过不要慌，我们还是老样子，基于s08的内容来一步步的改造它，其中一些无关的修改我会详细的标注出来，然后再来一步步实现新功能的内容。
  在上一章，我们加入了对上下文的压缩功能。试想一下，如果任务一直跑下去，messages会不断面临一次又一次的压缩，此时必然丢失一些重要的信息，那我们可否采取一种持久化的策略，将这些重要的用户信息写入文件中，然后对这些文件进行管理，在必要时刻加入到我们的messages中，是不是就减轻了这些信息的折损。
  事实上，本章的memory就是一种持久化的策略，它主要解决的问题是：如何读记忆文件，如何将记忆文件加载到内存中，如何从用户的对话中抽取内容从而写成记忆文件等等这一系列组织memory的内容。那本章的完整代码依旧粘贴在下面：
https://github.com/shareAI-lab/learn-claude-code/blob/main/s09_memory/code.py
  我们的任务是：
  1， 先基于s08做一些修改，这部分与本章内容无直接关系。
  2， 一步步引入memory，感受整个机制如何在系统中发力的。
  3， 跑几个测试。

一、做一些无关的改动

  首先是一些冗余的内容，直接贴在下面，你只需要找到这些内容全部和部分注释掉即可。

# SKILL_REGISTRY: dict[str, dict] = {}

# def _scan_skills():
#     if not SKILLS_DIR.exists():
#         return
#     for d in sorted(SKILLS_DIR.iterdir()):
#         if not d.is_dir():
#             continue
#         manifest = d / "SKILL.md"
#         if manifest.exists():
#             raw = manifest.read_text()
#             meta, body = _parse_frontmatter(raw)
#             name = meta.get("name", d.name)
#             desc = meta.get("description", raw.split("\n")[0].lstrip("#").strip())
#             SKILL_REGISTRY[name] = {"name": name, "description": desc, "content": raw}

# _scan_skills()

# def list_skills() -> str:
#     if not SKILL_REGISTRY:
#         return "(no skills found)"
#     return "\n".join(f"- **{s['name']}**: {s['description']}" for s in SKILL_REGISTRY.values())

# def load_skill(name: str) -> str:
#     skill = SKILL_REGISTRY.get(name)
#     if not skill:
#         return f"Skill not found: {name}"
#     return skill["content"]

# s08: SYSTEM includes skill catalog (inherited from s07 build_system)
# def build_system() -> str:
#     # catalog = list_skills()
#     return (
#         f"You are a coding agent at {WORKDIR}. "
        
#     )

# SYSTEM = build_system()

# s08: subagent gets its own system prompt — no compact, no skill loading
# SUB_SYSTEM = (
#     f"You are a coding agent at {WORKDIR}. "
#     "Complete the task you were given, then return a concise summary. "
#     "Do not delegate further."
# )


# def _normalize_todos(todos):
#     if isinstance(todos, str):
#         try:
#             todos = json.loads(todos)
#         except json.JSONDecodeError:
#             try:
#                 todos = ast.literal_eval(todos)
#             except (SyntaxError, ValueError):
#                 return None, "Error: todos must be a list or JSON array string"
#     if not isinstance(todos, list):
#         return None, "Error: todos must be a list"
#     for i, t in enumerate(todos):
#         if not isinstance(t, dict):
#             return None, f"Error: todos[{i}] must be an object"
#         if "content" not in t or "status" not in t:
#             return None, f"Error: todos[{i}] missing 'content' or 'status'"
#         if t["status"] not in ("pending", "in_progress", "completed"):
#             return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
#     return todos, None

# def run_todo_write(todos: list) -> str:
#     global CURRENT_TODOS
#     todos, error = _normalize_todos(todos)
#     if error:
#         return error
#     CURRENT_TODOS = todos
#     lines = ["\n\033[33m## Current Tasks\033[0m"]
#     for t in CURRENT_TODOS:
#         icon = {"pending": " ", "in_progress": "\033[36m▸\033[0m", "completed": "\033[32m✓\033[0m"}[t["status"]]
#         lines.append(f"  [{icon}] {t['content']}")
#     print("\n".join(lines))
#     return f"Updated {len(CURRENT_TODOS)} tasks"

SUB_TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to a file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    # {"name": "edit_file", "description": "Replace exact text in a file once.",
    #  "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
    # {"name": "glob", "description": "Find files matching a glob pattern.",
    #  "input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}},
]
SUB_HANDLERS = {"bash": run_bash, "read_file": run_read, "write_file": run_write}

def spawn_subagent(description: str) -> str:
    print(f"\n\033[35m[Subagent spawned]\033[0m")
    messages = [{"role": "user", "content": description}]
    for _ in range(30):
        response = client.messages.create(model=MODEL, system=SUB_SYSTEM,
            messages=messages, tools=SUB_TOOLS, max_tokens=8000)
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            break
        results = []
        for block in response.content:
            if block.type == "tool_use":
                # blocked = trigger_hooks("PreToolUse", block)
                # if blocked:
                #     results.append({"type": "tool_result", "tool_use_id": block.id,
                #                     "content": str(blocked)})
                #     continue
                handler = SUB_HANDLERS.get(block.name)
                output = handler(**block.input) if handler else f"Unknown: {block.name}"
                # trigger_hooks("PostToolUse", block, output)
                print(f"  \033[90m[sub] {block.name}: {str(output)[:100]}\033[0m")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
    result = extract_text(messages[-1]["content"])
    if not result:
        for msg in reversed(messages):
            if msg["role"] == "assistant":
                result = extract_text(msg["content"])
                if result:
                    break
        if not result:
            result = "Subagent stopped after 30 turns without final answer."
    print(f"\033[35m[Subagent done]\033[0m")
    return result

def snip_compact(messages, max_messages=50):
    if len(messages) <= max_messages: return messages
    # keep_head, keep_tail = 3, max_messages - 3   其实无伤大雅
    head_end, tail_start = 3, len(messages) - (max_messages - 3)
    if head_end > 0 and _message_has_tool_use(messages[head_end - 1]):
        while head_end < len(messages) and _is_tool_result_message(messages[head_end]):
            head_end += 1
    if (tail_start > 0 and tail_start < len(messages)
            and _is_tool_result_message(messages[tail_start])
            and _message_has_tool_use(messages[tail_start - 1])):
        tail_start -= 1
    if head_end >= tail_start:
        return messages

    return messages[:head_end] + [{"role": "user", "content": f"[snipped {tail_start - head_end} messages]"}] + messages[tail_start:]


TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to a file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in a file once.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
    {"name": "glob", "description": "Find files matching a glob pattern.",
     "input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}},
    # {"name": "todo_write", "description": "Create and manage a task list for your current coding session.",
    #  "input_schema": {"type": "object", "properties": {"todos": {"type": "array", "items": {"type": "object", "properties": {"content": {"type": "string"}, "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]}}, "required": ["content", "status"]}}}, "required": ["todos"]}},
    {"name": "task", "description": "Launch a subagent to handle a complex subtask. Returns only the final conclusion.",
     "input_schema": {"type": "object", "properties": {"description": {"type": "string"}}, "required": ["description"]}},
    # {"name": "load_skill", "description": "Load the full content of a skill by name.",
    #  "input_schema": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}},
    # s08 change: new compact tool — triggers compact_history, not a no-op
    # {"name": "compact", "description": "Summarize earlier conversation to free context space.",
    #  "input_schema": {"type": "object", "properties": {"focus": {"type": "string"}}}},
]

TOOL_HANDLERS = {
    "bash": run_bash, "read_file": run_read, "write_file": run_write,
    "edit_file": run_edit, "glob": run_glob,  # "todo_write": run_todo_write,
    "task": spawn_subagent,  # "load_skill": load_skill,
}

# FROM s04 (unchanged): Hooks
# HOOKS = {"PreToolUse": [], "PostToolUse": []}
# def trigger_hooks(event, *args):
#     for cb in HOOKS[event]:
#         r = cb(*args)
#         if r is not None: return r
#     return None

# DENY_LIST = ["rm -rf /", "sudo", "shutdown"]
# def permission_hook(block):
#     if block.name == "bash":
#         for p in DENY_LIST:
#             if p in block.input.get("command", ""): return "Permission denied"
#     return None
# def log_hook(block):
#     print(f"\033[90m[HOOK] {block.name}\033[0m")
#     return None

# HOOKS["PreToolUse"].append(permission_hook)
# HOOKS["PreToolUse"].append(log_hook)



def spawn_subagent(description: str) -> str:
    print(f"\n\033[35m[Subagent spawned]\033[0m")
    messages = [{"role": "user", "content": description}]
    for _ in range(30):
        response = client.messages.create(model=MODEL, system=SUB_SYSTEM,
            messages=messages, tools=SUB_TOOLS, max_tokens=8000)
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            break
        results = []
        for block in response.content:
            if block.type == "tool_use":
                # blocked = trigger_hooks("PreToolUse", block)
                # if blocked:
                #     results.append({"type": "tool_result", "tool_use_id": block.id,
                #                     "content": str(blocked)})
                #     continue
                handler = SUB_HANDLERS.get(block.name)
                output = handler(**block.input) if handler else f"Unknown: {block.name}"
                # trigger_hooks("PostToolUse", block, output)
                print(f"  \033[90m[sub] {block.name}: {str(output)[:100]}\033[0m")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
    result = extract_text(messages[-1]["content"])
    if not result:
        for msg in reversed(messages):
            if msg["role"] == "assistant":
                result = extract_text(msg["content"])
                if result:
                    break
        if not result:
            result = "Subagent stopped after 30 turns without final answer."
    print(f"\033[35m[Subagent done]\033[0m")
    return result




# 下面这部分在主agent的循环中-------------------------
        for block in response.content:
            if block.type != "tool_use": continue
            print(f"\033[36m> {block.name}\033[0m")

            # s08: compact tool triggers compact_history, not a no-op string
            # 可能这里的压缩会放到memory中进行
            if block.name == "compact":
                messages[:] = compact_history(messages)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": "[Compacted. Conversation history has been summarized.]"})
                messages.append({"role": "user", "content": results})
                break  # end current turn, start fresh with compacted context

            # blocked = trigger_hooks("PreToolUse", block)
            # if blocked:
            #     results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)})
            #     continue
            handler = TOOL_HANDLERS.get(block.name)
            output = handler(**block.input) if handler else f"Unknown: {block.name}"
            # trigger_hooks("PostToolUse", block, output)
            print(str(output)[:200])
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
        else:
            messages.append({"role": "user", "content": results})
            continue
        # compact was called: results already appended above
        continue

  最后还有一些简单的修改。

# def summarize_history(messages):
#     conversation = json.dumps(messages, default=str)[:80000]
#     prompt = ("Summarize this coding-agent conversation so work can continue.\n"
#               "Preserve: 1. current goal, 2. key findings/decisions, 3. files read/changed, "
#               "4. remaining work, 5. user constraints.\nBe compact but concrete.\n\n" + conversation)
#     response = client.messages.create(model=MODEL, messages=[{"role": "user", "content": prompt}], max_tokens=2000)
#     return "\n".join(
#         getattr(block, "text", "")
#         for block in response.content
#         if getattr(block, "type", None) == "text").strip() or "(empty summary)"

# 这个变化也没啥，只是return的形式稍有差别，还有提示词稍微修改了一下
def summarize_history(msgs):
    conv = json.dumps(msgs, default=str)[:80000]
    r = client.messages.create(model=MODEL, messages=[{"role": "user", "content":
        "Summarize this coding-agent conversation so work can continue.\n"
        "Preserve: 1. current goal, 2. key findings, 3. files changed, 4. remaining work, 5. user constraints.\n\n" + conv}],
        max_tokens=2000)
    return extract_text(r.content).strip()

二、实现memory机制

  在开始之前，请先默许这个机制的几个事实，后面我会在代码的注释当中指出支持这段事实的逻辑。第一个是：有关所有的该机制的所有持久化的数据全部存储在.memory的文件夹中；第二，该文件夹下有两类记忆文件，一类名为*.md，这是从对话中提取出的所有的用户记忆，有两种命名方式，下面是示例；

# 一种是以简要信息命名
---
name: User Preference Tabs
description: 用户喜欢用空格缩进
type: user
---
用户明确表示...

# 一种是以时间戳命名
---
name: User Preference Tabs
description: 用户要求用python11版本
type: user
---
用户明确表示...

  还有一类名为MEMORY.md，这个文件中以简要的形式存放所有的记忆文件，相当于一个目录，其中空格会被短横杠替代，时间戳会被int化，下面是示例。

- [User prefers spaces for indentation](user-preference-tabs.md) — 用户要求使用空格缩进而不是Tab
- [Project uses FastAPI](project-framework.md) — 项目使用FastAPI + PostgreSQL
- [1778232323(1778232323.md) — 项目使用FastAPI + PostgreSQL
- ...

  接着，我们来写一些管理这些记忆文件的基本操作。

# 写记忆文件
def write_memory_file(name: str, mem_type: str, description: str, body: str):
    """写*.md文件的内容，在.memory下面，返回文件的路径"""
    slug = name.lower().replace(" ", "-").replace("/", "-") # 这个地方会将命名时的空格转化为-
    filename = f"{slug}.md"
    filepath = MEMORY_DIR / filename
    filepath.write_text(
        f"---\nname: {name}\ndescription: {description}\ntype: {mem_type}\n---\n\n{body}\n"
    )
    _rebuild_index()
    return filepath

# 自动生成MEMORY.md文件
def _rebuild_index():
    """扫描.memory下的所有*.md文件，生成一个MEMORY.md文件，这个文件中每行是一个lines的元素"""
    lines = []
    for f in sorted(MEMORY_DIR.glob("*.md")):
        if f.name == "MEMORY.md":
            continue
        raw = f.read_text()
        meta, body =_parse_frontmatter()
        name = meta.get("name", f.stem)
        desc = meta.get("description", body.split("\n")[0][:80])
        lines.append(f"- [{name}]({f.name}) — {desc}")
    MEMORY_INDEX.write_text("\n".join(lines) + "\n" if lines else "")

# 读MEMORY.md文件
def read_memory_index() -> str:
    """读这个.memory下面的MEMORY.md文件"""
    if not MEMORY_INDEX.exists():
        return ""
    text = MEMORY_INDEX.read_text().strip()
    return text if text else "" 

# 读其他的*.md文件
def read_memory_file(filename: str) -> str | None:
    """读这个.memory下面*.md文件"""
    path = MEMORY_DIR / filename
    if not path.exists():
        return None
    return path.read_text()

# 列出每个*.md文件的所有信息返回
def list_memory_files() -> list[dict]:
    """返回.memory下面，每个*.md文件的内容信息，具体是一个list[dict]"""
    result = []
    for f in sorted(MEMORY_DIR.glob("*.md")):
        if f.name == "MEMORY.md":
            continue
        meta, body = _parse_frontmatter(f.read_text())
        # 返回值的每个元素形如
        result.append({
            "filename": f.name,
            "name": meta.get("name", f.stem),
            "description": meta.get("description", ""),
            "type": meta.get("type", "user"),
            "body": body,
        })
    return result

  接着来完成整个机制最核心的两个功能，第一个是根据messages筛选出相关的memory文件。

def extract_memories(messages: list):
    """每轮对话后，自动根据最近的messages中提取相关的记忆。"""
    # 从最近的十个block中提取相关的，思考一下，我们需要找到什么样子的block？block["role"：user, "content"：{"type":"text", "text":"xxx"}]
    dialogue_parts = []
    for msg in messages[-10:]:
        role = msg.get("role", "?")
        content = msg.get("content", "")
        if isinstance(content, list):
            content = " ".join(
                str(getattr(b, "text", "")) for b in content
                if getattr(b, "type", None) == "text"
            )
        if isinstance(content, str) and content.strip():
            dialogue_parts.append(f"{role}: {content}")
    dialogue = "\n".join(dialogue_parts)

    # 如果不存在，那就退出
    if not dialogue.strip():
        return

    # 不需要让LLM全部拿到，因为太多了，只需要弄一个简要的摘要，由name和description组成
    existing = list_memory_files()
    existing_desc = "\n".join(f"- {m['name']}: {m['description']}" for m in existing) if existing else "(none)"

    prompt = (
        "Extract user preferences, constraints, or project facts from this dialogue.\n"
        "Return a JSON array. Each item: {name, type, description, body}.\n"
        "- name: short kebab-case identifier (e.g. 'user-preference-tabs')\n"
        "- type: one of 'user' (user preference), 'feedback' (guidance), "
        "'project' (project fact), 'reference' (external pointer)\n"
        "- description: one-line summary for index lookup\n"
        "- body: full detail in markdown\n"
        "If nothing new or already covered by existing memories, return [].\n\n"
        f"Existing memories:\n{existing_desc}\n\n"
        f"Dialogue:\n{dialogue[:4000]}"
    )

    try:
        response = client.messages.create(
            model=MODEL, messages=[{"role": "user", "content": prompt}], max_tokens=800
        )
        text = extract_text(response.content).strip()
        # 提示词中已经规定模型回答的格式了，但还是需要小心的提取一下。
        match = re.search(r'\[.*\]', text, re.DOTALL)
        if not match:
            return
        items = json.loads(match.group())
        if not items:
            return
        count = 0
        for mem in items:
            name = mem.get("name", f"memory_{int(time.time())}")
            mem_type = mem.get("type", "user")
            desc = mem.get("description", "")
            body = mem.get("body", "")
            if desc and body:
                write_memory_file(name, mem_type, desc, body)
                count += 1
        if count:
            print(f"\n\033[33m[Memory: extracted {count} new memories]\033[0m")
    except Exception:
        pass

  另外一个就是从memory文件中去找到与当前对话最相关的，然后加入的上下文中作为模型的提示词当中。这两个核心的功能会形成闭环。

def select_relevant_memories(messages: list, max_items: int = 5) -> list[str]:
    """根据最近的对话内容，从所有记忆文件中筛选出最相关的"""

    # 步骤1：拿到所有记忆文件----------------------------
    files = list_memory_files()
    if not files:
        return []

    # 步骤2:取最近的三条用户文本信息----------------------------
    recents_texts = [] # 存放最近的三条，messages中有text的block
    for msg in reversed(messages): # 只需要处理最近的用户消息
        if msg.get("role") == "user":
            content = msg.get("content", "")
            if isinstance(content, list):
                content = " ".join(
                    str(getattr(b, "text", "")) for b in content
                    if getattr(b, "type", None) == "text"
                )    
            if isinstance(content, str):
                recents_texts.append(content)
            if len(recents_texts) >= 3:
                break
    # 拿到了这些text，再取最近的[:2000]，防止过大
    recent = " ".join(reversed(recents_texts))[:2000] 

    # 检查下是不是空
    if not recent.strip():
        return []

    # 步骤3：构建一个记忆目录：相当于让模型方便“点菜”---------------------------
    catalog_lines = []
    for i, f in enumerate(files):
        catalog_lines.append(f"{i}: {f['name']} — {f['description']}")
    catalog = "\n".join(catalog_lines)

    # 可以看到这个其实还是让ai去选择的。
    prompt = (
        "Given the recent conversation and the memory catalog below, "
        "select the indices of memories that are clearly relevant. "
        "Return ONLY a JSON array of integers, e.g. [0, 3]. "
        "If none are relevant, return [].\n\n"
        f"Recent conversation:\n{recent}\n\n"
        f"Memory catalog:\n{catalog}"
    )

    try:
        response = client.messages.create(
            model=MODEL,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=200,
        )
        text = extract_text(response.content).strip()

        # 正则表达式：找到返回的json数组，更python数组没什么区别
        match = re.search(r'\[.*?\]', text, re.DOTALL)
        if match:
            indices = json.loads(match.group()) # 转化成python列表
            selected = []
            for idx in indices:
                if isinstance(idx, int):
                    selected.append(files[idx]["filename"]) # 下标变为name
                    if len(selected) >= max_items: # 找到5个最相关的就可以了
                        break
            return selected # 其实只是存放了相关的 [filename1, filename2, ...]
    except Exception:
        pass # 忽略了

    # 附加步骤：如果模型报错了：降级到关键词匹配-----------------------------
    # 这个相当于对前面三个最近的rencent block中取出相关的词，后面加判断，是防止短的介词的出现
    keywords = [w.lower() for w in recent.split() if len(w) > 3] 

    # 筛选出相关的文件名，就结束了。
    selected = []
    for f in files:
        text = (f["name"] + " " + f["description"]).lower()
        if any(kw in text for kw in keywords):
            selected.append(f["filename"])
            if len(selected) >= max_items:
                break
    return selected

  有细心的小伙伴可能会发现，其实这个逻辑跟我在第六章讲subagent的时候，最后改进spawn_subagent函数，让子智能体自己去总结摘要的做法是一模一样的（本质就是套娃，来回的把一些抽象的，无法用硬编码实现的任务交给LLM）。

  不过，这里面其实还是有些细节的。比如这里做了一些异常处理的情况，当LLM总结报错或者实效时，会降级为关键词的匹配，以此达到找到相关记忆文件的目的，而且在取的时候，还会通过判断当前关键词的长度来筛选出介词，宾语等等不可能相关的词语来降低复杂度。

  找到相关的memory后，把它加载到内存当中去。这里加载的原因和上面的list_memory_files()是类似的，区别就是一个加载全部的给模型筛选相关文件，一个加载相关的后续会在agent_loop中方便处理许多。

# 加载memory文件到内存中去
def load_memories(message: list) -> str:
    """加载一下，目前不知道何意味"""
    selected_files = select_relevant_memories(message)
    if not selected_files:
        return ""

    parts = ["<relevant_memories>"]
    for filename in selected_files:
        content = read_memory_file(filename)
        if content:
            parts.append(content)
    parts.append("</relevant_memories>")
    return "\n\n".join(parts)

CONSOLIDATE_THRESHOLD = 10

def consolidate_memories():
    """整理memroy文件，将相关的合并，删除等操作"""
    files = list_memory_files()
	# 小于这个数就不用整理
    if len(files) < CONSOLIDATE_THRESHOLD:
        return

	# 加载到目录中，就像一个菜单，加载到提示词当中
    catalog = "\n\n".join(
        f"## {f['filename']}\nname: {f['name']}\ndescription: {f['description']}\n{f['body']}"
        for f in files
    )

    prompt = (
        "Consolidate the following memory files. Rules:\n"
        "1. Merge duplicates into one\n"
        "2. Remove outdated/contradicted memories\n"
        "3. Keep the total under 30 memories\n"
        "4. Preserve important user preferences above all\n"
        "Return a JSON array. Each item: {name, type, description, body}.\n\n"
        f"{catalog[:16000]}"
    )

	# 下面处理和生成的时候类似
    try:
        response = client.messages.create(
            model=MODEL, messages=[{"role": "user", "content": prompt}], max_tokens=3000
        )
        text = extract_text(response.content).strip()
        match = re.search(r'\[.*\]', text, re.DOTALL)
        if not match:
            return
        items = json.loads(match.group())

        # 这部分可以看出，是先删除全部，再去一个个生成
        for f in MEMORY_DIR.glob("*.md"):
            if f.name != "MEMORY.md":
                f.unlink()

        for mem in items:
            name = mem.get("name", f"memory_{int(time.time())}")
            mem_type = mem.get("type", "user")
            desc = mem.get("description", "")
            body = mem.get("body", "")
            if desc and body:
                write_memory_file(name, mem_type, desc, body)

        print(f"\n\033[33m[Memory: consolidated {len(files)} → {len(items)} memories]\033[0m")
    except Exception:
        pass


def build_system():
    index = read_memory_index()
    memories_section = f"\n\nMemories available:\n{index}" if index else ""
    return (
        f"You are a coding agent at {WORKDIR}."
        f"{memories_section}\n"
        "Relevant memories are injected below. Respect user preferences from memory.\n"
        "When the user says 'remember' or expresses a clear preference, extract it as a memory."
    )

SUB_SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Complete the task you were given, then return a concise summary. "
    "Do not delegate further."
)

  最后，来修改agent_loop部分。动手之前一定要先问自己一个问题，而不是直接看代码：这个memory机制，如果是你，你会放在什么地方去触发？每当用户给出新的任务（提出新的问题），memory会记录以往执行任务的时候出现的用户偏好，所以放在while 之前是比较合适的。

MAX_REACTIVE_RETRIES = 1  # retry limit for reactive compact

def agent_loop(messages: list):
    reactive_retries = 0

    # 拿到string类型的所有相关的memory及其内容["start", "memory1", "memory2", ... ,"end"]
    memories_content = load_memories(messages)
    # 拿到当前的对话位置(下标)，这里有个问题：为什么在获取当前位置的时候要加上这个条件？因为此时相当于做了一个检查，检查上一段任务是不是已经做完了（最后的block中的content一定是string）。
    memory_turn = len(messages) - 1 if messages and isinstance(messages[-1].get("content"), str) else None
    system = build_system()

    while True:
    	# 专门提取压缩前的属于用户的对话，格式为{"role": "user", "content": "xxx"}。因为肯定是提取用户的偏好，不可能是模型或者工具调用的
        pre_compress = [m if isinstance(m, dict) 
                        else {"role": m.get("role",""), "content": str(m.get("content",""))} 
                        for m in messages]
        
        messages[:] = tool_result_budget(messages)    
        messages[:] = snip_compact(messages)          
        messages[:] = micro_compact(messages)         


        # s08的上下文        
        if estimate_size(messages) > CONTEXT_LIMIT:
            print("[auto compact]")
            messages[:] = compact_history(messages)

        try:
            request_messages = messages
            if memories_content and memory_turn is not None and memory_turn < len(messages):
                request_messages = messages.copy()
                request_messages[memory_turn] = {
                    **messages[memory_turn],
                    "content": memories_content + "\n\n" + messages[memory_turn]["content"],
                }

            response = client.messages.create(model=MODEL, system=system, messages=request_messages, tools=TOOLS, max_tokens=8000)
            reactive_retries = 0  # reset on successful API call
        except Exception as e:
            if ("prompt_too_long" in str(e).lower() or "too many tokens" in str(e).lower()) and reactive_retries < MAX_REACTIVE_RETRIES:
                print("[reactive compact]")
                messages[:] = reactive_compact(messages)
                reactive_retries += 1
                continue
            raise

        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use": 
            # s09: extract from pre-compression snapshot for full fidelity
            extract_memories(pre_compress)
            consolidate_memories()
            return

        results = []
        for block in response.content:
            if block.type != "tool_use": continue
            print(f"\033[36m> {block.name}\033[0m")

            # s08: compact tool triggers compact_history, not a no-op string
            # 可以看到，这里虽有压缩的组件，但其实还是没用到。
            # if block.name == "compact":
            #     messages[:] = compact_history(messages)
            #     results.append({"type": "tool_result", "tool_use_id": block.id,
            #                     "content": "[Compacted. Conversation history has been summarized.]"})
            #     messages.append({"role": "user", "content": results})
            #     break  

            # blocked = trigger_hooks("PreToolUse", block)
            # if blocked:
            #     results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)})
            #     continue
            handler = TOOL_HANDLERS.get(block.name)
            output = handler(**block.input) if handler else f"Unknown: {block.name}"
            # trigger_hooks("PostToolUse", block, output)
            print(str(output)[:200])
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
        
        # 这个地方的else去掉了，细细品这里面的逻辑。（区别在于触发break之后，这段代码都要执行）
        messages.append({"role": "user", "content": results})
        continue

三、跑几个测试

s08 >> I prefer using tabs for indentation, not spaces. Remember that
Got it! I'll remember your preference for using **tabs for indentation** instead of spaces. I'll apply this to any code I write or edit for you going forward.

  .memory文件下生成的内容。
在这里插入图片描述

s08 >> Create a Python file called test.py in /Users/bx/Documents/coding/learn_cladudecode/s09_memory
> write_file
Wrote 133 bytes to /Users/bx/Documents/coding/learn_cladudecode/s09_memory/test.py
Created `test.py` in the `s09_memory` directory! The file uses **tabs for indentation** as per your preference. It contains a simple `greet` function that takes a name and prints a greeting, with a main block to run it.

  新生成的test文件，观察是不是使用了刚刚写的偏好。
在这里插入图片描述
  再检查下。

s08 >> What did I tell you about my preferences?
You told me that you prefer using **tabs for indentation, not spaces**. I've remembered that and applied it when creating the `test.py` file — all indentation in that file uses tabs!

总结

  有没有发现一个现象：当每个章节涉及一些功能需要调整主agent和子agent的提示词时候，其实作者调整后的提示词都是有一些差别的。试想一个庞大的项目中，如果需要把涉及调整提示词的功能全部引入时，这个时候该拿提示词怎么办？原教程并没有给出答案，不过也许我们可以像consolidate_memories()和extract_memories()的操作那样继续套娃（结合提示词的动态注入，让LLM决定什么时候注入）

  此外，随着项目的深入，笔者还体会到，agent开发和传统后端在debug上有些差异。传统后端开发api工具的异常捕捉都很灵敏了，基本上一有报错就可以定位到。但agent开发貌似还没那么成熟，比如我在实现这个章节的时候，遇到了memory文件没有写入的情况，后面发现是consolidate_memories()的时机不对，但整个系统依旧流畅进行，LLM的上下文没有被压缩，用户偏好也没丢失，看起来风平浪静。

  最后总结下测试的整个流程。

  1. 提取记忆（第一轮）
    用户：明确说出偏好 “I prefer using tabs for indentation, not spaces. Remember that”
    AI：回复 “Got it! I’ll remember…”
    系统内部：在 agent_loop 的回合结束后（stop_reason != “tool_use”），调用 extract_memories(pre_compress)，利用 LLM 从对话中提取出新的记忆条目（例如 {“name”: “user-preference-tabs”, “type”: “user”, “description”: “用户要求使用Tab缩进”, “body”: “…”}），然后调用 write_memory_file 将其保存为 .md 文件（如 user-preference-tabs.md），并更新 MEMORY.md 索引。

  2. 存储与索引
    记忆文件被写入 .memory 目录，Front Matter 包含 name、description、type；body 存储完整细节。_rebuild_index() 更新 MEMORY.md，添加一行 - user-preference-tabs — 用户要求使用Tab缩进。

  3. 应用记忆（第二轮）
    用户：要求创建 test.py 文件。
    AI：调用 write_file 工具生成文件。此时，load_memories 在对话轮次开始时已检索到 user-preference-tabs 记忆，并将其内容注入当前用户消息之前，因此 AI 在生成文件内容时主动使用 Tab 缩进，遵循用户偏好。

  4. 检索与回答（第三轮）
    用户：问 “What did I tell you about my preferences?”
    AI：load_memories 再次根据对话关键词检索到相关记忆，注入上下文后，AI 准确回答 “You told me that you prefer using tabs for indentation, not spaces.”，验证了记忆的正确性。

  5. 关键机制
    提取时机：每轮对话结束后（无工具调用时）自动触发。
    检索时机：每轮对话开始时，load_memories 根据对话内容动态筛选相关记忆并注入。
    持久化：记忆以文件形式保存，重启程序后依然有效。

"""
┌─────────────────────────────────────────────────────┐
│ 1. 存储层：每个记忆是 .md 文件，头部有 YAML 元数据   │
├─────────────────────────────────────────────────────┤
│ 2. 索引层：MEMORY.md 自动生成所有记忆的摘要列表      │
├─────────────────────────────────────────────────────┤
│ 3. 检索层：根据对话内容，用 LLM 选择最相关的记忆     │
├─────────────────────────────────────────────────────┤
│ 4. 写入层：在每轮对话后自动提取新记忆并写入文件      │
├─────────────────────────────────────────────────────┤
│ 5. 维护层：当记忆数量 ≥ 10 时，自动合并去重          │
└─────────────────────────────────────────────────────┘


例如：
用户说："我讨厌用 Tab，请用空格"
         │
         ▼
extract_memories()  →  LLM 提取  →  写入 user-preference-tabs.md
         │
         ▼
_rebuild_index()  →  更新 MEMORY.md（加入新条目）
         │
         ▼
下次对话开始时，build_system() 把 MEMORY.md 注入 System Prompt
         │
         ▼
用户问："帮我格式化代码"
         │
         ▼
select_relevant_memories()  →  LLM 判断 "user-preference-tabs" 相关
         │
         ▼
load_memories()  →  把完整内容注入上下文
         │
         ▼
AI 回复："好的，我会使用空格缩进..."
"""

这里追加一个更正的bug，在跑完后发现.memory下没有生成MEMORY.md文件，我用trae帮我审察了之后才发现这个bug。下面是对话：
----------------------------分割线-------------------------------

用户问为什么运行代码后没有生成 MEMORY.md 文件。让我分析一下代码。

从代码看：

  1. write_memory_file 调用 _rebuild_index()（L96）
  2. _rebuild_index() 在 L107 调用 _parse_frontmatter() —— 但这里没有传参数！

让我看清楚：

def _rebuild_index():
    """扫描.memory下的所有*.md文件，生成一个MEMORY.md文件，这个文件中每行是一个lines的元素"""
    lines = []
    for f in sorted(MEMORY_DIR.glob("*.md")):
        if f.name == "MEMORY.md":
            continue
        raw = f.read_text()
        meta, body =_parse_frontmatter()   # ← 这里！没传 raw 参数
        name = meta.get("name", f.stem)
        desc = meta.get("description", body.split("\n")[0][:80])
        lines.append(f"- [{name}]({f.name}) — {desc}")
    MEMORY_INDEX.write_text("\n".join(lines) + "\n" if lines else "")

_parse_frontmatter 定义：

def _parse_frontmatter(text: str) -> tuple[dict, str]:
    if not text.startswith("---"):
        return {}, text
    ...

它需要一个 text 参数，但 _rebuild_index 调用时没传任何参数，会抛出 TypeError: _parse_frontmatter() missing 1 required positional argument: 'text'。

那么为什么 MEMORY.md 没生成？

让我追踪一下流程：

  1. 用户运行代码，输入对话
  2. agent_loop 执行
  3. 当 stop_reason != "tool_use" 时，调用 extract_memories(pre_compress) 和 consolidate_memories()
  4. extract_memories 中如果 LLM 返回了记忆项，会调用 write_memory_file()
  5. write_memory_file 调用 _rebuild_index()
  6. _rebuild_index 里有 bug，会抛 TypeError

但是 extract_memories 中有 try/except Exception: pass（L303-L304），所以即使 _rebuild_index 抛错也被吞掉了。

等等，让我再仔细看：

  • L280-L304 是 try 块
  • L293: for mem in items:
  • L299: write_memory_file(name, mem_type, desc, body) —— 这会调用 _rebuild_index()
  • 如果 _rebuild_index 抛错，会被 L303 的 except Exception: pass 捕获

所以即使有记忆被提取，因为 _rebuild_index 抛错，整个流程被吞掉了。

但是！等等 —— write_memory_file 的代码：

def write_memory_file(name: str, mem_type: str, description: str, body: str):
    slug = name.lower().replace(" ", "-").replace("/", "-")
    filename = f"{slug}.md"
    filepath = MEMORY_DIR / filename
    filepath.write_text(...)   # ← 先写文件
    _rebuild_index()           # ← 再调用 _rebuild_index
    return filepath

所以 .md 文件应该能写出来，但 _rebuild_index() 失败，所以 MEMORY.md 没生成。

但实际上，根据 .memory/ 目录的现有内容，只有两个 .md 文件：

  • project-working-directory.md
  • user-preference-tabs.md

这两个应该是用户手动创建的（或者从 s09 之前的章节带过来的）。如果用户运行 s09 后没有触发 extract_memories 提取出新记忆，那也不会调用到 write_memory_file。

让我重新想想。用户问：“为什么我运行了代码，在.memory文件夹下，没有生成MEMORY.md文件呢？”

两种可能：

  1. 用户运行了代码，但对话内容没有触发 extract_memories 提取出新记忆 → 那么 write_memory_file 没被调用 → _rebuild_index 也没被调用 → MEMORY.md 没生成
  2. 用户运行了代码，对话内容触发了 extract_memories，但因为 _rebuild_index 有 bug，抛错被吞，MEMORY.md 没生成

无论哪种情况，根因都是：MEMORY.md 只在 write_memory_file 被调用时才会通过 _rebuild_index 生成。代码没有在启动时主动扫描现有记忆生成 MEMORY.md。

而且即使触发了，_rebuild_index 还有 bug。

让我再确认下 _rebuild_index 的 bug：

def _rebuild_index():
    lines = []
    for f in sorted(MEMORY_DIR.glob("*.md")):
        if f.name == "MEMORY.md":
            continue
        raw = f.read_text()
        meta, body =_parse_frontmatter()   # ← bug: 没传 raw

应该是 _parse_frontmatter(raw)。这会导致 TypeError: _parse_frontmatter() missing 1 required positional argument: 'text'。

所以总结答案：

  1. MEMORY.md 只在 write_memory_file() 内部通过 _rebuild_index() 生成，启动时不会主动生成
  2. _rebuild_index 有 bug：L107 的 _parse_frontmatter() 缺少 raw 参数
  3. 即使触发了 extract_memories 提取出新记忆，write_memory_file 写出 .md 文件后调用 _rebuild_index 会抛 TypeError，被 extract_memories 的 except Exception: pass 吞掉，所以 MEMORY.md 永远不会生成

让我用代码引用回答用户。
问题出在 [_rebuild_index](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L100-L111) 的实现，有两个层面的原因。

直接原因：_rebuild_index 里的 bug

[L107](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L107) 调用 _parse_frontmatter() 时忘了传参数：

def _rebuild_index():
    lines = []
    for f in sorted(MEMORY_DIR.glob("*.md")):
        if f.name == "MEMORY.md":
            continue
        raw = f.read_text()
        meta, body =_parse_frontmatter()   # ← bug：应该是 _parse_frontmatter(raw)

而 [_parse_frontmatter](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L73-L85) 是这样定义的：

def _parse_frontmatter(text: str) -> tuple[dict, str]:
    if not text.startswith("---"):
        ...

text 是必填位置参数。所以一旦执行到这一行就会抛 TypeError: _parse_frontmatter() missing 1 required positional argument: 'text'。

为什么你看不到任何报错

调用链是这样的：

  1. agent_loop 退出前调用 [extract_memories](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L821)
  2. extract_memories 在 [L299](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L299) 调用 write_memory_file()
  3. write_memory_file 在 [L96](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L96) 调用 _rebuild_index() —— 这里抛 TypeError
  4. 但 extract_memories 用 [L303-L304](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L303-L304) 的 except Exception: pass 把异常吞掉了

所以即使提取出新记忆、即使 .md 文件成功写出（write_memory_file 先写文件再调 _rebuild_index），MEMORY.md 也永远生成不出来，且控制台没有任何错误提示。

还有一个更深的设计问题

即使修了 bug，代码启动时也不会主动生成 MEMORY.md。_rebuild_index 只在 write_memory_file 内部被调用，而 write_memory_file 只在 LLM 真的提取出新记忆时才被触发。

也就是说：

  • 你运行代码 → 直接退出（没有触发提取） → MEMORY.md 永远不生成
  • 你对话中说了 “remember xxx” → 触发提取 → 才会调 _rebuild_index → 才会生成 MEMORY.md

但 [build_system](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L361-L369) 在每次启动时就 read_memory_index() 想读 MEMORY.md：

def build_system():
    index = read_memory_index()   # ← 启动时就读，但此时文件不存在
    memories_section = f"\n\nMemories available:\n{index}" if index else ""

第一次运行必然读到空字符串，系统提示里就没有记忆目录。

修复方案

两处都要改：

  1. 修 bug：[L107](file:///Users/bx/Documents/coding/learn_cladudecode/s09_memory/code.py#L107) 改成 _parse_frontmatter(raw)
  2. 启动时主动建索引：在 if __name__ == "__main__": 块或模块加载阶段加一次 _rebuild_index() 调用，确保已有 .memory/*.md 文件能被纳入索引

需要我直接帮你改吗？
----------------------------分割线-------------------------------
AI真牛批！！再次测试，MEMORY文件成功生成。
在这里插入图片描述

# 人工智能 # python # 算法 # 语言模型 # 自然语言处理
Logo
亚马逊云科技技术品牌专区

更多推荐

Fuzzapi项目概览：为什么它是REST API渗透测试的首选工具？

Fuzzapi是一款专为REST API渗透测试设计的强大工具，它基于API_Fuzzer gem构建，为开发人员和安全测试人员提供了高效、自动化的API漏洞检测方案。无论是小型应用还是大型系统，Fuzzapi都能帮助您快速发现潜在的安全风险，确保API接口的安全性。## 🚀 Fuzzapi的核心优势作为一款专注于REST API渗透测试的工具，Fuzzapi具有以下显著优势：###

avatar 亚马逊云科技技术品牌专区

3步掌握无线信号安全测试：Flipper Zero本田固件完整实战指南

无线安全测试是当今物联网安全领域的重要研究方向，而Flipper Zero本田固件为安全研究人员提供了深入了解无线信号重放攻击防护的绝佳平台。这款定制固件专门支持本田汽车钥匙（FCC ID: KR5V2X）的信号分析功能，通过CVE-2022-27254漏洞的实战研究，帮助技术人员掌握无线安全测试的核心技术。## 技术原理简析：无线信号重放攻击机制无线信号重放攻击的核心在于捕获合法设备发送

avatar 亚马逊云科技技术品牌专区

无需Arduino IDE！PSLab ESP01 Firmware的命令行编译与烧录技巧

PSLab ESP01 Firmware是一款专为ESP8266 ESP-01模块设计的固件，本文将详细介绍如何通过命令行工具快速完成固件的编译与烧录，无需依赖Arduino IDE，让开发过程更加高效便捷。## 📋 准备工作：环境与工具安装在开始操作前，需要确保系统中已安装以下工具：- **Python环境**：用于运行esptool.py工具- **esptool.py**：E

avatar 亚马逊云科技技术品牌专区