把大模型塞进一个多轮任务里,最先崩的往往不是推理能力,而是记忆。窗口再大也会填满,而填满之后的表现不是优雅降级,而是灾难性的:关键指令被挤到注意力边缘,模型开始胡编,工具调用参数错位。上下文工程要解决的核心问题就是——在有限的 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
递归摘要最大的风险是「摘要漂移」:每一轮压缩都会丢一点信息,几轮之后关键约束就消失了。缓解办法有三条:
- 约束单独存放,不进摘要,用 pinned 机制常驻。
- 保留最近 N 轮原文,只压缩更早的部分。
- 摘要里保留具体数字与专有名词,明确禁止模型把「预算 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)
小结
智能体记忆的本质是在有限预算下做取舍:什么常驻、什么压缩、什么检索、什么遗忘。三层分层(短期/工作/长期)给出了结构,预算分配给出了纪律,压缩与检索给出了手段,冲突处理与评估给出了闭环。把这几件事做扎实,智能体才能从「一问一答」进化成「记得住、接得上、学得会」的协作体。这与 智能体生产化部署 关注的可靠性与成本问题互为表里,也与 提示工程 中的上下文组织技巧直接衔接。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。