Agent Harness 工程：记忆与 Memory Store

上一篇我们讲了上下文压缩：messages 会随着 Agent Loop 不断膨胀，所以 Harness 需要在每次模型调用前整理上下文，把旧工具结果压缩成 artifact，把旧消息总结成摘要，同时保留当前 Todo、关键事实和最近几轮细节。

这解决的是“当前任务里，怎么在有限窗口内继续工作”的问题。

但还有另一个问题：

如果某些信息不仅当前任务有用，以后也会反复有用，应该放在哪里？

比如：

用户喜欢中文回答；
当前项目使用 pnpm，不使用 npm；
这个仓库的测试命令是 pnpm test；
某个目录是生成产物，不应该手动编辑；
用户明确说过“不要自动提交，除非我要求”；
某个长期任务已经确认过一组业务约束。

这些信息如果只留在当前 messages 里，任务结束后就丢了。下一次 Agent 又要重新问、重新读、重新踩坑。

所以这一篇继续把“记忆”单独展开：Memory Store。

记忆不是更长的上下文

先强调一个边界：记忆不是把所有历史都塞进上下文。

如果你把过去所有对话、所有工具结果、所有摘要都长期保存，并且每次调用模型都塞进去，那只是把上下文膨胀问题换了个名字。

记忆应该是一个外部状态库，而不是系统提示词的一部分。

它平时不在模型上下文里。只有当当前任务需要时，Harness 才从 Memory Store 检索相关条目，再把少量结果注入本次上下文。

这点很重要。否则 Memory Store 越写越多，最后又会变成另一种“大提示词”。记忆系统真正要解决的不是“让模型每次看见更多”，而是“让 Harness 能在合适的时候取出合适的信息”。

可以这样理解：

概念	作用
`messages`	当前任务的过程流水账
上下文摘要	当前任务里被压缩后的历史
Todo	当前任务的执行计划和进度
Memory Store	跨任务保存的稳定事实、偏好和约束

记忆解决的是跨任务延续性，不是当前窗口容量。

Memory Store 不直接等于上下文。Harness 先判断什么值得记，再在需要时检索少量相关记忆放回上下文。

什么值得记住

不是所有看起来重要的信息都应该进入长期记忆。

一个信息适合进入 Memory Store，通常要满足几个条件：

稳定。 它不是临时状态。比如“这次测试失败了”不一定值得长期记住；但“这个项目的测试命令是 pnpm test”更稳定。

可复用。 以后还会用到。比如用户偏好、项目约束、常用命令、长期业务背景。

已确认。 不是模型猜的。最好来自用户明确表达、工具验证结果，或者多次一致观察。

低风险。 不应该保存密码、token、隐私数据、临时文件内容、未经确认的敏感信息。

粒度合适。 一条记忆应该表达一个清楚事实，不要把一整段日志或一整篇文档塞进去。

可以先用一个简单判断表：

信息	是否适合记忆	原因
用户希望默认中文回答	适合	稳定偏好，可复用
本项目使用 `yarn build` 验证	适合	项目级事实，可复用
某次测试输出的完整错误栈	不适合	临时过程信息，太长
用户贴出来的 API token	不适合	敏感信息
模型猜测某模块可能有 bug	不适合	未确认
用户明确说某目录不要手动编辑	适合	明确约束，后续任务会用到

一句话：记忆应该保存长期有用的事实，而不是保存完整历史。

记忆也要分层

记忆最好不要只有一个大池子。

至少可以先分三层：

层级	例子	作用范围
`user`	用户偏好、输出语言、协作习惯	跨项目、跨任务
`project`	项目命令、目录约束、技术栈	当前仓库或当前项目
`task`	长期任务目标、阶段性决策	某个长任务或某条会话链路

不同层级的记忆，生命周期和风险不同。

用户偏好可能长期有效；项目命令随着仓库变化可能失效；任务记忆在任务完成后就应该归档或删除。

不同记忆有不同作用范围。不要把临时任务状态写进用户长期记忆，也不要把项目约束只藏在当前 messages 里。

定义 Memory 数据结构

先写一个最小的数据结构。

import hashlib
import json
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal


MemoryScope = Literal["user", "project", "task"]


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass
class MemoryRecord:
    id: str
    scope: MemoryScope
    content: str
    source: str
    tags: list[str] = field(default_factory=list)
    confidence: float = 1.0
    created_at: str = field(default_factory=now_iso)
    updated_at: str = field(default_factory=now_iso)

这里每个字段都有用：

字段	作用
`id`	稳定标识，方便更新和去重
`scope`	记忆作用范围
`content`	记忆正文，应该短而清楚
`source`	记忆来源，比如用户确认或工具结果
`tags`	检索和分类用
`confidence`	可信度
`created_at`	创建时间
`updated_at`	更新时间

source 很重要。

如果一条记忆没有来源，后面就很难判断它是不是模型幻觉。真实系统里还可以保存 run_id、message_id、tool_call_id，方便回溯。

实现一个简单 Memory Store

教学版先用 JSON 文件保存记忆。

真实项目里可以换成 SQLite、Postgres、向量数据库，或者带检索索引的对象存储。但最小实现里，先把流程跑通更重要。

class MemoryStore:
    def __init__(self, path: Path):
        self.path = path
        self.records: dict[str, MemoryRecord] = {}
        self.load()

    def load(self) -> None:
        if not self.path.exists():
            self.records = {}
            return

        raw_items = json.loads(self.path.read_text(encoding="utf-8"))
        self.records = {
            item["id"]: MemoryRecord(**item)
            for item in raw_items
        }

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        items = [asdict(record) for record in self.records.values()]
        self.path.write_text(
            json.dumps(items, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

    def make_id(self, scope: MemoryScope, content: str) -> str:
        digest = hashlib.sha256(f"{scope}:{content}".encode("utf-8")).hexdigest()
        return f"mem-{digest[:16]}"

    def upsert(
        self,
        scope: MemoryScope,
        content: str,
        source: str,
        tags: list[str] | None = None,
        confidence: float = 1.0,
    ) -> MemoryRecord:
        normalized_content = content.strip()
        memory_id = self.make_id(scope, normalized_content)

        existing = self.records.get(memory_id)
        if existing:
            existing.source = source
            existing.tags = tags or existing.tags
            existing.confidence = confidence
            existing.updated_at = now_iso()
            self.save()
            return existing

        record = MemoryRecord(
            id=memory_id,
            scope=scope,
            content=normalized_content,
            source=source,
            tags=tags or [],
            confidence=confidence,
        )
        self.records[memory_id] = record
        self.save()
        return record

这段代码先做了最简单的 upsert。

id 根据 scope + content 生成，所以同一层级里相同内容不会重复写很多次。真实系统里可以做更智能的相似度合并，比如“用户喜欢中文回答”和“用户希望默认用中文”其实应该是同一类记忆。

先做关键词检索

记忆写进去之后，还要能找出来。

教学版先用关键词匹配，不上向量数据库：

    def search(
        self,
        query: str,
        scopes: list[MemoryScope] | None = None,
        limit: int = 5,
    ) -> list[MemoryRecord]:
        query_terms = {
            term.lower()
            for term in query.replace("/", " ").replace("_", " ").split()
            if term.strip()
        }

        results: list[tuple[int, MemoryRecord]] = []
        for record in self.records.values():
            if scopes and record.scope not in scopes:
                continue

            haystack = " ".join(
                [record.content, record.scope, " ".join(record.tags)]
            ).lower()

            score = sum(1 for term in query_terms if term in haystack)
            if score > 0:
                results.append((score, record))

        results.sort(key=lambda item: item[0], reverse=True)
        return [record for _, record in results[:limit]]

这个检索很粗糙，但足够说明流程：

根据当前用户任务或模型意图生成 query；
在 Memory Store 里找相关记忆；
把少量结果渲染进上下文。

真实系统可以换成向量检索，也可以混合关键词、标签、时间、作用域、置信度。

把记忆注入上下文

下一步，把检索到的记忆放进模型输入。

注意，不是所有记忆都放进去。每次只放当前任务相关的少量记忆。

这里开始会出现一个新的工程问题：模型输入不再只是一个固定 SYSTEM_PROMPT + messages。

它开始由多块内容动态组装出来：

固定系统规则；
当前任务；
当前 Todo；
已加载 Skill；
上下文摘要；
检索出来的相关记忆；
最近几轮 messages。

这一篇先在 build_model_messages() 里手动拼起来，下一篇会把它进一步抽象成运行时 Prompt Assembler。

MEMORY_STORE = MemoryStore(WORKDIR / ".agent" / "memory.json")


def render_memories(records: list[MemoryRecord]) -> str:
    if not records:
        return ""

    lines = ["Relevant memories:"]
    for record in records:
        tags = ", ".join(record.tags) if record.tags else "none"
        lines.append(
            f"- [{record.scope}] {record.content} "
            f"(confidence={record.confidence}, tags={tags})"
        )

    return "\n".join(lines)


def build_model_messages(
    user_message: str,
    messages: list[dict[str, Any]],
    context_state: ContextState,
) -> list[dict[str, Any]]:
    relevant_memories = MEMORY_STORE.search(
        query=user_message,
        scopes=["user", "project", "task"],
        limit=5,
    )

    model_messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
    ]

    memory_text = render_memories(relevant_memories)
    if memory_text:
        model_messages.append({"role": "system", "content": memory_text})

    model_messages.extend(compact_messages_if_needed(messages, context_state))
    return model_messages

这里把记忆作为一条额外 system 消息注入。

也可以放到系统提示词某个固定段落里。关键是要让模型知道：这些是检索出来的历史记忆，不是用户本轮刚说的话。

让模型写记忆：remember 工具

Memory Store 不应该只靠 Harness 自动写。

有些时候，模型最知道“这条信息后面还会用”。所以可以暴露一个工具，让模型提出写记忆请求。

REMEMBER_TOOL = {
    "type": "function",
    "function": {
        "name": "remember",
        "description": (
            "Store a stable, reusable memory. "
            "Use only for confirmed user preferences, project facts, "
            "or long-lived task constraints. Never store secrets."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "scope": {
                    "type": "string",
                    "enum": ["user", "project", "task"],
                },
                "content": {
                    "type": "string",
                    "description": "A short confirmed fact or preference.",
                },
                "source": {
                    "type": "string",
                    "description": "Why this memory is trustworthy.",
                },
                "tags": {
                    "type": "array",
                    "items": {"type": "string"},
                },
                "confidence": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 1,
                },
            },
            "required": ["scope", "content", "source"],
        },
    },
}

然后实现真实函数：

def run_remember(
    scope: MemoryScope,
    content: str,
    source: str,
    tags: list[str] | None = None,
    confidence: float = 1.0,
) -> str:
    decision = check_memory_policy(
        scope=scope,
        content=content,
        source=source,
        confidence=confidence,
    )
    if decision != "allow":
        return f"Error: memory rejected - {decision}"

    record = MEMORY_STORE.upsert(
        scope=scope,
        content=content,
        source=source,
        tags=tags,
        confidence=confidence,
    )

    return f"Remembered {record.id}: {record.content}"

这里多了一个 check_memory_policy()。

记忆写入比普通读上下文更敏感，所以不能完全相信模型。

记忆写入策略

先写一个很保守的策略：

SENSITIVE_PATTERNS = [
    "password",
    "secret",
    "api_key",
    "token",
    "private key",
]


def check_memory_policy(
    scope: MemoryScope,
    content: str,
    source: str,
    confidence: float,
) -> str:
    text = content.lower()

    if not content.strip():
        return "empty memory"

    if len(content) > 500:
        return "memory is too long"

    if any(pattern in text for pattern in SENSITIVE_PATTERNS):
        return "memory may contain sensitive data"

    if confidence < 0.6:
        return "confidence is too low"

    if source.strip() == "":
        return "source is required"

    if scope not in {"user", "project", "task"}:
        return "invalid memory scope"

    return "allow"

真实系统里还可以做得更细：

写 user 级记忆前请求用户确认；
低置信度记忆只进入候选区，不直接保存；
敏感信息检测用更严格的规则；
保存前做去重和冲突检测；
每条记忆都有审计记录；
用户可以查看、修改、删除记忆。

记忆系统最怕的一件事是：把错误事实永久保存下来。

所以写入要比读取更谨慎。

加入工具注册表

和前几章一样，把工具加入 TOOLS 和 TOOL_HANDLERS：

TOOLS = [
    BASH_TOOL,
    LIST_DIR_TOOL,
    READ_FILE_TOOL,
    WRITE_FILE_TOOL,
    EDIT_FILE_TOOL,
    TODO_WRITE_TOOL,
    SUBAGENT_TOOL,
    LOAD_SKILL_TOOL,
    REMEMBER_TOOL,
]


TOOL_HANDLERS: dict[str, Callable[..., str]] = {
    "bash": bash,
    "list_dir": list_dir,
    "read_file": read_file,
    "write_file": write_file,
    "edit_file": edit_file,
    "run_todo_write": run_todo_write,
    "run_subagent": run_subagent,
    "load_skill": run_load_skill,
    "remember": run_remember,
}

主 Agent Loop 仍然不用大改。

模型如果发现一条信息值得长期保存，就调用 remember。Harness 做策略检查，通过后写入 Memory Store。

系统提示词怎么约束记忆

工具加好了，还要告诉模型什么时候该记、什么时候不该记。

这里先继续沿用一个固定的 SYSTEM_PROMPT。但你会看到，它已经开始变长了：工具规则、Todo 规则、Skill 规则、Subagent 规则、上下文压缩规则、记忆规则都想往里面塞。

这其实是下一篇要解决的问题：系统提示词不应该永远是一个固定大字符串，而应该在运行时按任务和状态组装。

SYSTEM_PROMPT = f"""
你是一个编程助手，工作在当前目录：{WORKDIR}。
你可以使用工具、Todo、Skill、Subagent 和 Memory Store 来完成任务。

记忆规则：
1. 只有稳定、可复用、已确认的信息才可以写入记忆。
2. 不要记住临时日志、一次性错误、完整文件内容或模型猜测。
3. 不要记住密码、token、密钥、隐私数据或敏感内容。
4. 写入 user 级记忆时要特别谨慎，最好来自用户明确表达。
5. 写入 project 级记忆时要来自工具验证或项目文件证据。
6. 如果信息只对当前任务有用，保留在 Todo 或当前上下文里，不要写入长期记忆。
"""

这里再次强调：提示词不是安全边界。真正的边界仍然是 check_memory_policy()。

记忆和上下文压缩怎么配合

上一章讲上下文压缩，这一章讲记忆。两者关系很紧。

上下文压缩处理的是当前任务历史：

旧消息摘要；
大工具结果 artifact；
最近几轮完整保留；
当前 Todo 不丢。

记忆处理的是跨任务信息：

用户偏好；
项目事实；
长期约束；
已确认的可复用经验。

一个简单原则是：

当前任务还在用的信息，放上下文；以后也会复用的稳定事实，才进入记忆。

比如一次修 bug 过程中，Agent 发现：

pytest 失败在 tests/test_auth.py::test_login_cookie；
失败原因是 cookie 字段名不一致；
项目测试命令是 pnpm test；
用户要求以后默认只总结关键修改。

前两条更像当前任务上下文；第三条适合项目记忆；第四条适合用户记忆。

还有一个细节：从记忆里检索出来的信息，也会占用上下文窗口。

所以记忆注入也要遵守上下文预算。不要因为“这是长期记忆”就无限塞。更合理的做法是先检索、再排序、再限制数量，最后交给 Prompt Assembler 和上下文压缩层一起决定本轮模型到底能看到什么。

记忆也需要更新和遗忘

记忆不是写进去就永远正确。

项目会变化，用户偏好会变化，旧约束会失效。

所以 Memory Store 还需要支持更新和遗忘。

最小可以先做三件事：

更新时间。 每次 upsert 时更新 updated_at，让旧记忆可以被识别出来。

置信度。 工具验证过的事实置信度高；模型推断或弱证据置信度低。

删除接口。 用户应该能明确要求“忘掉这条记忆”。

教学版可以再加一个删除函数：

def forget_memory(memory_id: str) -> str:
    if memory_id not in MEMORY_STORE.records:
        return f"Error: memory not found - {memory_id}"

    deleted = MEMORY_STORE.records.pop(memory_id)
    MEMORY_STORE.save()
    return f"Forgot {deleted.id}: {deleted.content}"

真实系统里删除也要做审计，尤其是团队或企业场景。

常见坑

第一，把所有历史都当记忆。 这会把 Memory Store 变成另一个膨胀的 messages。

第二，把模型猜测写成事实。 “可能是 X”不应该直接变成“X 是事实”。

第三，记住敏感信息。 密码、token、私钥、隐私数据都不应该进入普通记忆。

第四，不区分作用域。 用户偏好、项目事实、任务状态混在一起，后面检索会很乱。

第五，只写不删。 记忆如果不能更新和遗忘，迟早会变成过期事实的堆积。

第六，每次都注入所有记忆。 检索不相关记忆会污染上下文，让模型被旧信息误导。

小结

这一篇把记忆系统拆成了一个可落地的 Memory Store：

记忆不是更长的上下文，而是上下文外部的长期状态库；
只有稳定、可复用、已确认、低风险的信息才适合记住；
记忆至少要分 user、project、task 三种作用域；
用 MemoryRecord 保存内容、来源、标签、置信度和时间；
用 MemoryStore 做写入、去重、保存和检索；
用 remember 工具让模型提出写记忆请求；
用 check_memory_policy() 防止敏感信息、低置信度信息和临时信息进入长期记忆；
检索到的少量相关记忆再注入当前上下文。

到这里，Agent Harness 已经有了更完整的长期工作能力：上下文负责当前任务，记忆负责跨任务延续，二者通过检索和策略连接起来。

下一篇继续解决一个前面一直被我们简化处理的问题：提示词的运行时组装。之前我们一直使用固定的 SYSTEM_PROMPT，但现在 Harness 已经有了工具、权限、Hook、Todo、Subagent、Skill、上下文摘要和记忆。下一章会把这些输入拆成模块，在每轮调用模型前按任务状态动态组装，而不是把所有规则都塞进一个固定大提示词。