智能体记忆与上下文工程

上下文窗口不是记忆,智能体需要显式的记忆系统。本文系统讲解短期记忆、工作记忆与长期记忆的三层分层设计,滑动窗口与递归摘要等压缩策略,检索式记忆的写入、召回与遗忘工程,上下文预算的分配与降级裁剪,记忆冲突与时效性处理,多智能体共享记忆的边界划分,存储后端选型以及评估指标与常见失败模式排查。

把大模型塞进一个多轮任务里,最先崩的往往不是推理能力,而是记忆。窗口再大也会填满,而填满之后的表现不是优雅降级,而是灾难性的:关键指令被挤到注意力边缘,模型开始胡编,工具调用参数错位。上下文工程要解决的核心问题就是——在有限的 token 预算里,让该被记住的东西一直在场。

上下文窗口不是记忆。窗口是「此刻能看到的东西」,记忆是「跨时刻保持可用的东西」。把两者混为一谈,是绝大多数智能体在长任务里失忆的根因。

为什么智能体需要记忆

窗口扩大的三个幻觉

近两年上下文窗口从 4K 涨到 128K 甚至 1M,很多人据此认为记忆问题会自动消失。但三个经验事实否定了这一点:

假设现实
窗口够大就能记住「lost in the middle」:中段信息召回率显著低于首尾
塞进去就能用长上下文里无关内容会稀释注意力,指令遵循度下降
长上下文更便宜KV cache 显存与注意力开销随长度平方增长,成本不可忽视

也就是说,长上下文不是免费的记忆,而是昂贵的、带干扰的存储。真正的记忆系统要主动决定放什么、丢什么。

记忆让智能体具备的四类能力

  • 一致性:记住用户之前说过的偏好,不再重复询问。
  • 连续性:跨会话延续任务,而不是每次都从头开始。
  • 学习:把成功的解法沉淀下来,下次直接复用。
  • 个性化:累积用户画像,让回答贴合个体。

记忆分层

一个可落地的记忆架构通常分三层,对应不同的生命周期与访问模式。

三层模型

层级生命周期存储访问方式
短期记忆单次会话上下文窗口直接拼接
工作记忆单个子任务结构化变量显式读写
长期记忆跨会话向量库/数据库检索召回

短期记忆:对话历史

短期记忆就是当前的对话消息列表。它的问题不在「存」,而在「太长之后怎么办」。基本策略是滑动窗口 + 固定前缀:

def build_messages(system_prompt, history, max_turns=20, keep_pinned=3):
    """保留固定前缀 + 最近 max_turns 轮 + 所有被标记为 pinned 的消息"""
    pinned = [m for m in history if m.get("pinned")]
    recent = history[-max_turns * 2:]
    seen = set()
    merged = []
    for m in pinned + recent:
        key = id(m)
        if key not in seen:
            seen.add(key)
            merged.append(m)
    return [{"role": "system", "content": system_prompt}] + merged

pinned 机制很关键:用户明确说过的约束(「预算不超过 5000 元」「必须用中文回复」)应该被钉住,不随窗口滚动而丢失。

工作记忆:显式的任务状态

工作记忆不是消息,而是结构化状态。把任务进度、已收集的槽位、待办事项放进一个 JSON 对象,每轮注入:

from dataclasses import dataclass, field, asdict

@dataclass
class WorkingMemory:
    goal: str = ""
    slots: dict = field(default_factory=dict)
    todos: list = field(default_factory=list)
    done: list = field(default_factory=list)

    def render(self) -> str:
        return (
            f"<goal>{self.goal}</goal>\n"
            f"<slots>{self.slots}</slots>\n"
            f"<todo>{[t for t in self.todos if t not in self.done]}</todo>"
        )

结构化状态的好处是可校验、可回滚、可压缩。相比让模型从冗长对话里自己回忆进度,显式状态几乎消除了「忘了做到哪一步」这一类失败。

上下文预算管理

把上下文当成有限预算来分配,是上下文工程最重要的心法。

预算分配表

组件建议占比说明
系统提示10%角色、约束、输出格式
工具定义15%按需裁剪,不必全量注入
工作记忆10%结构化,密度高
检索记忆25%召回的长期记忆
对话历史30%最近若干轮
输出预留10%给生成留出空间

预算表的意义在于当总长超限时,你知道该砍哪一块。最常见的错误是先砍检索记忆——结果模型失去了关键事实,开始编造。

动态预算与降级

def fit_budget(components, total_budget, priorities):
    """按优先级从低到高裁剪,直到总长度进入预算"""
    lengths = {k: len(v) for k, v in components.items()}
    order = sorted(priorities, key=lambda k: priorities[k])
    for key in order:
        if sum(lengths.values()) <= total_budget:
            break
        overflow = sum(lengths.values()) - total_budget
        # 先尝试截断,再尝试整块丢弃
        if lengths[key] > overflow:
            lengths[key] -= overflow
        else:
            lengths[key] = 0
    return {k: v[:lengths[k]] for k, v in components.items()}

关键设计:降级要有层次。先截断冗余的历史轮次,再压缩成摘要,最后才丢弃。直接整块丢弃检索记忆,是导致「模型突然不认识用户」的典型原因。

压缩与摘要策略

三种压缩路线

策略压缩比信息损失成本
截断高高(丢尾部)零
递归摘要中中(丢细节)每次调用 LLM
结构化抽取高低(只留要点)一次抽取调用

递归摘要

递归摘要把旧对话压缩成一段不断演化的摘要,新对话保持原样:

def recursive_summarize(llm, old_summary, new_turns, max_tokens=400):
    prompt = (
        "把下面的对话历史压缩成一段不超过 400 字的摘要,"
        "保留:用户的明确约束、已完成的关键结论、尚未解决的问题、"
        "提到的具体数字与专有名词。不要保留寒暄。\n\n"
        f"已有摘要:\n{old_summary}\n\n新增对话:\n{format_turns(new_turns)}"
    )
    return llm.invoke(prompt).content

递归摘要最大的风险是「摘要漂移」:每一轮压缩都会丢一点信息,几轮之后关键约束就消失了。缓解办法有三条:

  1. 约束单独存放,不进摘要,用 pinned 机制常驻。
  2. 保留最近 N 轮原文,只压缩更早的部分。
  3. 摘要里保留具体数字与专有名词,明确禁止模型把「预算 5000 元」概括成「有预算限制」。

结构化抽取优于自由摘要

自由摘要是「让模型自己决定留什么」,结构化抽取是「按 schema 抽」。后者的信息保留率明显更高:

EXTRACT_SCHEMA = {
    "user_constraints": [],   # 用户明确提出的限制
    "decisions": [],          # 已达成的决定
    "open_questions": [],     # 待解决
    "entities": {},           # 人名/地名/订单号等
}

检索式记忆

长期记忆的核心是写入什么、如何召回、何时遗忘。

写入策略

不是所有对话都值得写入。常见做法是按事件切分并打分,只写入高分片段:

def should_write(turn, score):
    if turn["role"] != "user" and not turn.get("is_conclusion"):
        return False
    if score < 0.6:          # 信息量太低
        return False
    if is_greeting_or_smalltalk(turn["content"]):
        return False
    return True

写入时要做三件事:抽取结构化字段、生成 embedding、记录时间戳与来源。时间戳极其重要——没有它就无法实现时效性衰减。

召回:多路融合

单一向量检索容易漏掉精确匹配(订单号、人名)。实践中用向量 + 关键词 + 时间衰减的融合打分:

def recall_score(sim, bm25, age_days, w=(0.6, 0.3, 0.1), half_life=30.0):
    recency = 0.5 ** (age_days / half_life)
    return w[0] * sim + w[1] * bm25 + w[2] * recency

这种混合召回与 RAG 检索模式 里的多路融合思路完全一致,差别只在于记忆库的规模更小、时效权重更高。底层索引实现可以直接复用 向量数据库 的能力。

遗忘与淘汰

记忆库不能只增不减。三类淘汰策略并用:

  • 时间衰减:超过半衰期且长期未被召回的条目降权。
  • 容量上限:每个用户/会话设硬上限,超限时淘汰最低分条目。
  • 冲突合并:新记忆与旧记忆矛盾时,按时间戳覆盖并标记旧条为失效。
def consolidate(new_mem, existing, llm):
    conflicts = [m for m in existing if is_same_slot(m, new_mem)]
    if not conflicts:
        return [new_mem]
    if llm.confirm_contradiction(conflicts, new_mem):
        for c in conflicts:
            c["status"] = "superseded"      # 不删除,保留审计
        return conflicts + [new_mem]
    return conflicts + [new_mem]

冲突记忆不要物理删除,标记为 superseded 即可。生产环境里,用户改口(「我改主意了」)非常常见,保留历史能让你在出问题时回溯到底哪一轮改了主意。

存储后端选型

记忆系统不是一个数据库能包办的,不同层用不同的存储。

选型矩阵

层级推荐存储理由
短期记忆内存 / Redis生命周期短,需低延迟
工作记忆会话内变量 / Redis Hash结构化,随会话销毁
长期语义记忆向量库(Milvus/pgvector)语义召回为主
长期事实记忆关系库(PostgreSQL)需要精确查询与事务
事件日志追加写(Kafka/对象存储)审计与重放

双写:向量库 + 关系库

纯向量库难以支持「查某用户所有未失效的订单号」这类结构化查询,纯关系库又做不了语义召回。生产系统普遍双写:

def write_memory(mem, vec_store, sql_store):
    mem_id = sql_store.insert(mem)          # 事务写入,拿到自增 id
    vec_store.upsert(
        id=mem_id,
        vector=embed(mem["text"]),
        metadata={"user_id": mem["user_id"], "ts": mem["ts"],
                  "slot": mem.get("slot"), "status": "active"},
    )
    return mem_id

双写带来的一致性问题是真实的:向量库写入成功、关系库失败,就会出现「召回得到但查不到详情」的幽灵条目。工程上给向量库的 metadata 里带上 status 字段,召回后再回查关系库校验,比引入分布式事务简单得多。

隔离与命名空间

多租户场景下,必须在检索层强制过滤 user_id,而不是依赖检索后过滤。后者会污染 top-k,把别人的记忆挤进候选,再被过滤掉,等于白白浪费召回位。

results = vec_store.search(
    query_vector=q,
    filter={"user_id": user_id, "status": "active"},   # 检索时就过滤
    top_k=10,
)

多智能体共享记忆

当系统里不止一个智能体时,记忆的边界需要重新设计。

共享与私有的划分

  • 私有记忆:单个智能体的工作记忆与草稿,不共享。
  • 共享记忆:事实性结论、用户画像、任务进度,写入公共库。
  • 黑板模式:所有智能体读写同一块结构化状态,适合协作式任务。
class SharedBlackboard:
    def __init__(self, store):
        self.store = store
        self.version = 0

    def publish(self, agent_id, key, value):
        self.version += 1
        self.store.set(key, {"value": value, "by": agent_id,
                             "version": self.version})

    def read(self, key):
        return self.store.get(key)

版本与冲突

共享黑板最大的问题是并发写覆盖。加一个乐观锁:写入前检查版本号,不一致则重读重算:

def publish_cas(self, agent_id, key, value, expected_version):
    cur = self.store.get(key)
    if cur and cur["version"] != expected_version:
        raise ConflictError(cur["version"])
    self.publish(agent_id, key, value)

多智能体共享记忆的复杂度增长非常快。能不用就不用:大多数所谓「多智能体协作」,用单智能体加结构化工作记忆就能解决,而且调试成本低一个数量级。只有当子任务真正独立并行时才引入共享记忆。

记忆冲突与一致性

三类冲突

类型例子处理
时序冲突先说要 A,后说要 B时间戳新者胜
语义冲突「我喜欢辣」与「我不能吃辣」让 LLM 判断是否真矛盾
粒度冲突「预算 5000」与「预算 3000~8000」保留更具体的

一致性检查的时机

不要在每轮都做全量一致性检查——成本高且会引入噪声。合理的触发点:

  • 写入时:只与同一 slot 的条目比对,成本可控。
  • 召回首屏时:如果召回结果里有矛盾项,注入时显式标注「以下两条信息可能冲突,请以时间较新的为准」。
  • 会话结束时:做一次离线整理。

评估与排错

记忆系统的评估指标

指标定义目标
召回率该被记住的信息是否被召回> 0.9
精确率召回内容是否都相关> 0.7
一致性跨会话回答是否自洽人工抽检
预算利用率上下文是否被有效信息填满60%~85%

常见失败模式

  • 失忆:约束没被 pinned,滚动窗口把它挤掉。排查:打印每轮的最终 prompt,看约束还在不在。
  • 串味:多用户共用记忆库且未按用户隔离。排查:检查 namespace 或 user_id 过滤是否生效。
  • 摘要漂移:多轮压缩后细节丢失。排查:对比第 1 轮和第 10 轮的摘要,看数字是否还在。
  • 检索噪声:召回了一堆不相关的旧记忆,稀释了当前任务。排查:打印召回条目及其分数,看阈值是否过低。
  • 成本爆炸:每轮都注入全量记忆。排查:统计每轮 token 数与召回条数,设硬上限。

调试工具:把上下文打出来

最有效的调试手段是把每一轮实际送给模型的完整 prompt 落盘。绝大多数记忆 bug 都能在第一步定位:

import json

def log_context(step, messages, components):
    with open(f"ctx_trace_{step}.json", "w", encoding="utf-8") as f:
        json.dump({
            "messages": messages,
            "budget": {k: len(v) for k, v in components.items()},
            "total_chars": sum(len(v) for v in components.values()),
        }, f, ensure_ascii=False, indent=2)

小结

智能体记忆的本质是在有限预算下做取舍:什么常驻、什么压缩、什么检索、什么遗忘。三层分层(短期/工作/长期)给出了结构,预算分配给出了纪律,压缩与检索给出了手段,冲突处理与评估给出了闭环。把这几件事做扎实,智能体才能从「一问一答」进化成「记得住、接得上、学得会」的协作体。这与 智能体生产化部署 关注的可靠性与成本问题互为表里,也与 提示工程 中的上下文组织技巧直接衔接。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「ai」更多文章

  1. 排序学习与搜索召回排序系统
  2. 数据版本控制与血缘:DVC 与 LakeFS
  3. 模型可解释性:SHAP、LIME 与注意力归因