跳转至

Agent 长期记忆架构与上下文窗口管理:四分类记忆 + 5 个生产故障案例

适用场景:生产环境跑 LLM Agent,对话一长就报 context length exceeded,或 Agent 反复召回已失效的旧事实。一句话结论:Agent 记忆要按"工作/情景/语义/程序"四分类落地,上下文窗口靠"预算硬上限 + 时效优先级驱逐"兜底,否则超限和幻觉是必然事故。

接手 Agent 生产运维后,我踩过最痛的坑不是模型不准,而是记忆失控。一个客服 Agent 跑 30 轮对话后上下文爆掉,单次请求烧掉 $4;一个诊断 Agent 因为向量库召回了一条三个月前的旧版本事实,给用户报了个早就不存在的配置项。这两类问题——记忆怎么存、上下文怎么管——正是本文要讲清楚的事。

本文延续本站 ai 栏目的实操视角,不重复 RAG 基础与混合检索(见混合检索与向量库监控),只聚焦 Agent 自身记忆生命周期 和 上下文窗口动态管理。

1. 为什么 Agent 需要记忆:一个 30 轮对话的教训

先看一个真实故障。某工单分流 Agent,单轮回答 300 token 上下,跑 30 轮后单次请求输入冲到 60k token——因为我把每一轮历史都原样塞回 prompt,没有做任何压缩。

round 1: prompt 300 tok   →  成本 $0.0004
round 10: prompt 3.1k tok →  成本 $0.004
round 30: prompt 9.3k tok →  成本 $0.012(还没爆)
round 45: prompt 13.6k tok →  context length exceeded 💥

线性增长只是表象,真正的问题是我把"记忆"和"上下文"当成了一回事。实际上,Agent 的记忆有四种本质不同的类型,该用什么载体、什么检索方式、什么过期策略,天差地别。把它们混在一个 prompt 里,是绝大多数 Agent 生产事故的总根源。

2. Agent 记忆四分类

这是我给生产 Agent 定的记忆模型,参考认知科学的记忆分类改造而来,运维场景下特别管用。

记忆类型 含义 典型内容 存储载体 检索方式 过期策略
工作记忆 当前会话轮次内 本轮 user/assistant 消息、工具调用结果 内存(进程内) 顺序拼接 会话结束即清
情景记忆 具体发生的事件 "上次 502 是 Nginx 配置改错" 向量库 相似度召回 按事件时效戳
语义记忆 事实与知识 "集群有 12 个节点""服务 A 依赖 B" Redis / SQLite 键值直接命中 TTL + 时效标记
程序记忆 技能与工具用法 诊断流程、命令参数、工具 schema 配置文件 / Prompt 编译期注入 版本发布才变

2.1 工作记忆:最该被截断的部分

工作记忆就是当前会话里还没"定论"的对话。它最占上下文,但价值衰减最快——第 30 轮时,第 1 轮的用户意图早就不重要了。

class WorkingMemory:
    """会话内工作记忆:只保留最近 N 轮,超限即截断"""

    def __init__(self, max_rounds=20):
        self.max_rounds = max_rounds
        self.rounds = []  # [(role, content, turn_id), ...]

    def append(self, role, content):
        self.rounds.append((role, content, len(self.rounds) + 1))
        self._trim()

    def _trim(self):
        """截断:丢最老的,保住最新的,中间可再压缩"""
        if len(self.rounds) > self.max_rounds:
            overflow = len(self.rounds) - self.max_rounds
            self.rounds = self.rounds[overflow:]

2.2 情景记忆:事件型,走向量召回

情景记忆是"具体发生过什么"。它天然适合向量检索——用"当前问题"做 query,去向量库找最相似的过往事件。但必须带时效戳,否则就会召回旧事实。

def recall_episodic(query, vector_store, top_k=3, max_age_days=90):
    """召回情景记忆,带 freshness 过滤,防旧事实"""
    hits = vector_store.similarity_search_with_score(query, k=top_k * 2)
    now = time.time()
    fresh = []
    for doc, score in hits:
        valid_until = doc.metadata.get("valid_until", 0)
        if valid_until and valid_until < now:
            continue  # 已过期,丢弃
        # recency 加权:越新的记忆,越值得采信
        age_days = (now - doc.metadata.get("created_at", now)) / 86400
        if age_days > max_age_days:
            continue
        fresh.append((doc, score))
    return fresh[:top_k]

2.3 语义记忆:事实型,走键值 + TTL

语义记忆是"这个世界长什么样"。它不该进向量库——事实是精确的,相似度检索会模糊化。我用 Redis 存键值,key = f"fact:{namespace}:{entity}",值带 valid_from 和 valid_until,命中后还要做时效校验。

2.4 程序记忆:技能型,固化不检索

程序记忆是 Agent 的"肌肉记忆"——怎么调工具、诊断流程长什么样。它不该被检索,检索会引入不确定性。正确做法是编译期注入:把工具 schema、系统指令、固定流程写进配置或 Prompt,发布新版本才更新。我在生产里见过把工具用法也塞向量库的,结果 Agent 检索到两条互相矛盾的工具说明,行为彻底乱套。

3. 上下文窗口管理:五种策略

上下文管理的本质是在有限窗口里,用最少 token 装最多有用信息。我按"从急到缓"排了五种,生产中通常是组合使用。

3.1 滑动窗口截断(兜底,最快)

丢最老的消息,保最新的。适合工作记忆,不适合语义/程序记忆(它们不该在窗口里)。

3.2 摘要压缩(信息损失可控)

把早段对话喂给 LLM 生成摘要,替换原文。信息有损,但压缩比高。

def compress_history(messages, budget=2000):
    """摘要压缩:超预算就把前半段压成一段摘要"""
    if count_tokens(messages) <= budget:
        return messages
    head = messages[: len(messages) // 2]
    tail = messages[len(messages) // 2 :]
    summary = llm.invoke(f"把以下对话压缩成要点,保留关键约束、决策、用户意图:\n{head}")
    return [SystemMessage(f"[早前对话摘要] {summary.content}")] + tail

3.3 RAG 外置(根本解法)

这是我最推荐的生产方案:语义记忆和情景记忆不进上下文,Agent 需要时用 query 去外部检索,只把命中的那几条塞回上下文。这样上下文长度与对话轮次解耦,只随"本次检索命中的信息量"波动,不再线性爆炸。

3.4 层级滚动摘要(长会话专用)

对话极长时,单层摘要会失真。我做三层:最老 → 摘要 L3 → 摘要 L2 → 摘要 L1 → 最近对话。每层触发阈值不同,最老的摘要落盘为情景记忆,不再占窗口。

3.5 上下文预算控制(硬上限,必配)

没有预算控制的 Agent 一定会超限。我给每次请求设硬预算,budget = 0.7 × 模型窗口,剩余 30% 留给模型输出和工具结果。预算控制器按"时效优先级"驱逐:工作记忆最老轮次 → 旧情景记忆 → 最老摘要。

class ContextBudget:
    """上下文预算控制器:token 计数 + 时效优先级驱逐"""

    def __init__(self, model_window=128000, reserve_ratio=0.7):
        self.hard_limit = int(model_window * reserve_ratio)
        self.current = 0

    def fit(self, messages):
        """把消息压缩进预算,返回可在窗口内发送的消息"""
        while count_tokens(messages) > self.hard_limit:
            if self._evictable(messages):
                messages = self._evict_oldest(messages)  # 驱逐最老
            else:
                # 实在压不进去,走摘要
                messages = self._compress_oldest(messages)
        return messages

    def _evictable(self, messages):
        """工作记忆轮次优先驱逐,程序记忆(system)永不驱逐"""
        return any(m.type != "system" for m in messages)

4. 记忆存储选型表

存储 适合记忆 优点 缺点 关键配置
内存 工作记忆 零延迟,会话结束即清 进程重启丢失 上限 max_rounds=20
Redis 语义记忆 毫秒命中,支持 TTL 需序列化,容量有限 EXPIRE fact:* 86400*7
向量库 情景记忆 跨会话相似检索 召回会模糊,需时效过滤 必须带 created_at/valid_until
SQLite 语义+审计 事务安全,可 SQL 查询 并发写需锁 WAL 模式,busy_timeout

经验:语义记忆用 Redis 做热缓存 + SQLite 做冷存储双写,命中快、可审计。别把四类全塞进一个 Redis,键前缀混乱后互相污染,这是我在生产里实际踩过的坑。

5. 五个真实故障案例

案例 1:记忆无限增长,上下文超限

现象:工单 Agent 跑到 45 轮,请求抛 context length exceeded,单次成本飙到 $0.3。 根因:全量历史回填,无预算控制器。 修复:加 ContextBudget 硬预算(70% 窗口),工作记忆 max_rounds=20,超出即截断。

案例 2:向量召回旧事实,Agent 幻觉

现象:诊断 Agent 报"Nginx 配置项 keepalive_timeout 已废弃",实际该配置仍在使用。 根因:情景记忆无 valid_until,召回 180 天前的旧记录。 修复:检索加 max_age_days=90 过滤 + recency 权重,系统指令声明"优先采信新时间戳事实"。

案例 3:会话恢复后记忆错乱

现象:用户刷新页面重开会话,Agent 引用了上一会话的私有上下文,答非所问。 根因:session_id 复用 + 多请求并发写同一记忆键。 修复:每次会话唯一 session_id,记忆键含 session_id,写入带版本号(乐观锁)拒绝乱序覆盖。

案例 4:记忆写入冲突

现象:两个分支并行写同一事实键,后写覆盖先写,丢了一条关键事实。 根因:无锁无版本。 修复:Redis 用 WATCH/MULTI 或 version 字段 CAS 写入,冲突时重试或合并。

案例 5:摘要压缩丢失关键约束

现象:Agent 压缩早段对话后,把用户明确的硬约束"只允许 GET 请求"压没了,随后发了 POST。 根因:摘要 prompt 没强调保留约束,压缩即丢。 修复:摘要 prompt 显式要求"必须原样保留所有约束性指令、安全边界、用户明确要求",且压缩后做约束校验(正则扫描关键词)。

6. 记忆评估方法

记忆系统上线前必须量化评估,否则"召回旧事实"这种事故根本发现不了。

6.1 命中率(Recall)

有多少该被记住的信息真的被检索到了。测试集:构造 100 个已知事件,Agent 检索,统计命中。目标 ≥ 0.8。

def recall_eval(queries, known_events, retrieve_fn):
    hit = sum(1 for q in queries if retrieve_fn(q) & known_events)
    return hit / len(queries)

6.2 污染率(Contamination)—— 最重要的指标

检索结果里有多少是已失效/无关的信息。污染率高说明时效过滤失效,是幻觉前兆。目标 ≤ 0.1。

def contamination_eval(queries, stale_events, retrieve_fn):
    stale_hits = sum(1 for q in queries if retrieve_fn(q) & stale_events)
    return stale_hits / len(queries)  # 理想 0

6.3 遗忘曲线(Forgetting Curve)

记忆该不该留、留多久,用遗忘曲线决定。艾宾浩斯曲线是"越旧越不值钱",但运维知识不完全适用——有些事实(集群规模、依赖关系)是长期有效的,有些(变更事件、告警)则快速过时。我按 valid_until 把记忆分成"短时效(7 天)""中时效(90 天)""长时效(永久)"三档,分别设置过期策略。

7. 生产级记忆管理器(完整示例)

下面是我在生产里用的记忆管理器骨架,四分类 + 预算 + 时效,一次说清。

import time, uuid, threading
from typing import Any

class AgentMemory:
    """四分类 Agent 记忆管理器"""

    def __init__(self, redis_client, vector_store, sqlite_conn):
        self.redis = redis_client          # 语义记忆
        self.vector = vector_store         # 情景记忆
        self.sqlite = sqlite_conn          # 冷存储/审计
        self.working = {}                  # 工作记忆:session_id -> list
        self._lock = threading.Lock()

    # ---- 工作记忆 ----
    def add_turn(self, session_id, role, content):
        with self._lock:
            self.working.setdefault(session_id, []).append(
                {"role": role, "content": content, "turn": len(self.working.get(session_id, [])) + 1}
            )
            # 截断
            if len(self.working[session_id]) > 20:
                self.working[session_id] = self.working[session_id][-20:]

    # ---- 语义记忆(Redis + TTL,带时效)----
    def set_fact(self, namespace, entity, value, ttl_days=7):
        key = f"fact:{namespace}:{entity}"
        payload = {"value": value, "valid_until": time.time() + ttl_days * 86400}
        self.redis.setex(key, ttl_days * 86400, json.dumps(payload))

    def get_fact(self, namespace, entity):
        raw = self.redis.get(f"fact:{namespace}:{entity}")
        if not raw:
            return None
        fact = json.loads(raw)
        if fact["valid_until"] < time.time():
            self.redis.delete(f"fact:{namespace}:{entity}")  # 过期即删
            return None
        return fact["value"]

    # ---- 情景记忆(向量库,带时效)----
    def remember_event(self, text, entity_id, ttl_days=90):
        doc = {
            "text": text,
            "metadata": {
                "entity_id": entity_id,
                "created_at": time.time(),
                "valid_until": time.time() + ttl_days * 86400,
            },
        }
        self.vector.add_documents([doc])

    def recall_events(self, query, top_k=3, max_age_days=90):
        return recall_episodic(query, self.vector, top_k, max_age_days)

    # ---- 程序记忆:固化注入,不检索 ----
    def build_system_prompt(self, tool_schemas, workflows):
        """程序记忆编译期注入,发布才变"""
        return json.dumps({"tools": tool_schemas, "workflows": workflows}, ensure_ascii=False)

    # ---- 上下文组装:预算控制 ----
    def assemble_context(self, session_id, budget):
        budget_ctl = ContextBudget(model_window=128000, reserve_ratio=budget)
        working = self.working.get(session_id, [])
        msgs = self._to_messages(working)
        return budget_ctl.fit(msgs)

8. 上线检查清单

写这篇时我复盘了所有记忆相关事故,沉淀出这份必查清单:

  1. 预算控制器是否在每次请求前执行?没有预算=必超限。
  2. 时效戳是否覆盖所有语义/情景记忆?没有时效=必召回旧事实。
  3. session_id 是否每次会话唯一?复用=必错乱。
  4. 写入锁/版本号 是否保护所有记忆写?并发无锁=必冲突。
  5. 摘要 prompt 是否显式要求保留约束?不保留=必丢关键信息。
  6. 污染率评估是否 < 0.1?超标=别上线。

9. FAQ

Q1: 上下文窗口超限了怎么办?

按时效优先级驱逐:① 先截断最老的工作记忆轮次 ② 再滚动摘要合并过期情景记忆 ③ 语义/程序记忆走 RAG 外置不占窗口。核心是设 token 硬预算(约 70% 窗口),由预算控制器主动压缩,而不是等 API 报错。

Q2: 向量召回旧事实造成幻觉怎么修?

三层:① 记忆带 valid_from/valid_until,检索过滤已过期 ② recency 权重,旧事实降权 ③ 系统指令声明"优先采信新时间戳事实"。生产上再补一层 max_age_days 过滤,超 90 天直接丢弃。

Q3: 会话恢复后记忆错乱是什么原因?

通常是 session_id 复用 + 并发写入。修复:唯一 session_id、记忆键含 session_id 维度、写入带版本号乐观锁,乱序到达的写入拒绝覆盖。

Q4: 记忆无限增长会不会拖垮成本?

会,每轮历史回填成本线性涨。修法:对话轮次上限(如 20 轮)→ 超限压成滚动摘要 → 摘要设 3 层上限,最老摘要落盘为情景记忆。目标:每轮上下文 token 稳定在预算内。

Q5: 记忆该存内存、Redis 还是向量库?

按类型分:工作记忆存内存;情景记忆存向量库;语义记忆存 Redis/SQLite(带 TTL);程序记忆固化配置。别全塞进一个 Redis,会互相污染。