Agent 上下文工程与记忆管理实战：Token 爆炸、上下文丢失、长期记忆

做 Agent 的人都踩过这几个坑：跑着跑着上下文爆了、模型忘了前面说过的话、多轮对话越答越傻、长任务到后面完全跑偏。根因不在模型，而在**上下文工程（Context Engineering）**没做好。这是 2025 年以来 CSDN/AI 技术圈最热的 Agent 落地话题。本文不讲空泛概念，全部围绕真实问题：现象 → 根因 → 可运行方案（代码 + 配置），覆盖 Token 预算、上下文裁剪、压缩摘要、结构化记忆、长期记忆（向量/文件）、多 Agent 共享上下文、常见坑，全程带可运行 Python 代码。

---

一、先看三个真实问题

1.1 问题一：跑到第 15 轮，上下文爆了

现象：
- 单次对话 20 轮后，请求报错：
  400 Bad Request: context_length_exceeded
  (maximum context length is 128000 tokens, your messages resulted in 131072)
- 或者费用暴涨（每轮都把历史全带上）

根因：
- 每轮把完整历史 + 所有工具返回结果无脑塞进 prompt
- 工具结果（网页抓取/日志/文件内容）动辄几万 token

1.2 问题二：模型"失忆"

现象：
- 用户第 3 轮说"用 Redis"，第 10 轮 Agent 又选了 MySQL
- 前面已经确认过的约束（不能删数据）后面被无视
- 多 Agent 之间互相不知道对方干了什么

根因：
- 关键约束没被"固化"，只是混在历史消息里
- 历史被截断时，重要的早期信息被裁掉

1.3 问题三：越答越傻（Lost in the Middle）

现象：
- 上下文越长，模型对中间内容的注意力越差
- 塞了 50 篇文档，模型只记得开头和结尾
- 回答开始"编"（幻觉），因为有效信息被稀释

根因：
- 长上下文 ≠ 好上下文
- 无关信息越多，信噪比越低

一句话总结：
上下文不是"越多越好"，而是"越准越好、越省越好"。
Context Engineering = 在有限 token 预算里，放最该放的信息。

---

二、上下文工程的核心框架

2.1 四层结构

Agent 上下文 = 系统层 + 任务层 + 记忆层 + 工具层

1. 系统层（System Prompt）
   - 角色、规则、输出格式、红线
   - 稳定不变，放最前（模型最关注开头）

2. 任务层（Task State）
   - 当前目标、已完成步骤、待办、约束
   - 结构化！不要用自然语言散着放

3. 记忆层（Memory）
   - 短期：最近 N 轮对话
   - 长期：摘要/事实/用户偏好（存外部）
   - 按需检索，不是全塞

4. 工具层（Tool Results）
   - 工具返回结果（最占 token）
   - 必须裁剪/摘要后再入上下文

2.2 Token 预算分配（关键）

总预算 = 模型上下文窗口 × 安全系数（如 0.7）

示例（128k 窗口）：
- 系统提示：≤ 2k
- 任务状态：≤ 4k
- 短期对话：≤ 8k（保留最近 N 轮）
- 检索/记忆：≤ 6k
- 工具结果：≤ 10k
- 输出预留：≤ 8k
→ 合计 ~38k，远低于 128k，留足余量

原则：
- 永远不要贴满窗口（留 30% 余量给输出和突发）
- 每层有硬上限，超了就裁剪

---

三、方案一：Token 预算与裁剪（解决"爆上下文"）

3.1 计数与监控

# 依赖：pip install tiktoken
import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

class TokenBudget:
    def __init__(self, limit: int, reserve_output: int = 8000):
        self.limit = limit
        self.reserve = reserve_output
        self.used = 0

    @property
    def available(self) -> int:
        return self.limit - self.reserve - self.used

    def add(self, text: str, model: str = "gpt-4o") -> bool:
        t = count_tokens(text, model)
        if t > self.available:
            return False        # 超预算，拒绝加入
        self.used += t
        return True

# 使用：每次拼上下文前先算预算
budget = TokenBudget(limit=96000, reserve_output=12000)
budget.add(SYSTEM_PROMPT)
budget.add(task_state_text)
for msg in recent_messages:
    if not budget.add(msg["content"]):
        break        # 超了就不再加更早的

3.2 滑动窗口（保留最近 N 轮）

def sliding_window(messages: list, max_turns: int = 12) -> list:
    """保留 system + 最近 N 轮"""
    system = [m for m in messages if m["role"] == "system"]
    rest = [m for m in messages if m["role"] != "system"]
    return system + rest[-max_turns:]

要点：
- 系统提示永远保留（不参与滑动）
- 按"轮"而不是"条"滑（一轮含 user+assistant+tool）
- 关键约束不放在会被滑掉的位置 → 放系统层或任务状态

3.3 工具结果裁剪（最大 token 杀手）

def trim_tool_result(content: str, max_chars: int = 3000) -> str:
    """工具结果太长 → 头尾保留 + 中间摘要"""
    if len(content) <= max_chars:
        return content
    head = content[:max_chars // 3]
    tail = content[-(max_chars // 3):]
    mid = f"\n...[已截断 {len(content) - len(head) - len(tail)} 字符]...\n"
    return head + mid + tail

# 更优：让模型先摘要工具结果，再入上下文
def summarize_tool_result(content: str, llm, max_tokens: int = 500) -> str:
    prompt = f"""把下面工具结果压缩成不超过 {max_tokens} token 的要点，
只保留与任务相关的事实，丢弃样板内容：
{content}
"""
    return llm.invoke(prompt).content

适用场景：
- 网页抓取的 HTML → 只留正文
- 日志文件 → 只留关键错误行（grep error）
- 大 JSON → 只留需要的字段
- SQL 结果 → 只留前 N 行

---

四、方案二：压缩与摘要（解决"越聊越忘"）

4.1 滚动摘要

def rolling_summary(messages: list, llm, keep_recent: int = 6) -> list:
    """
    历史太长时，把较早的对话压缩成一段摘要，
    保留最近几轮原文。
    """
    if len(messages) <= keep_recent:
        return messages

    old = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    summary_prompt = f"""把下面的对话历史压缩成摘要，要求：
1. 保留：已确认的需求/约束/决策/关键事实
2. 丢弃：寒暄、重复、无结论的讨论
3. 输出格式：条列式，每条一句话
历史：
{format_messages(old)}
"""
    summary = llm.invoke(summary_prompt).content

    return [
        {"role": "system", "content": f"【历史摘要】\n{summary}"}
    ] + recent

效果：
- 20 轮 → 1 段摘要 + 最近 6 轮
- token 从 30k 降到 5k
- 关键决策不丢（在摘要里）

4.2 关键信息固化（防遗忘）

把"必须记住的事"从对话历史里提出来，单独存：
- 约束清单：不删数据、不改表结构、只允许 Redis
- 决策清单：选了方案 B、接口路径 /api/v2
- 事实清单：用户是 VIP、订单号 10086

这些放进"任务状态"，每轮都注入（不参与滑动/裁剪）

class TaskState:
    """结构化任务状态：每轮都完整注入"""

    def __init__(self):
        self.goal = ""
        self.constraints = []      # 约束
        self.decisions = []        # 已做决策
        self.facts = []            # 关键事实
        self.done = []             # 已完成步骤
        self.todo = []             # 待办

    def to_prompt(self) -> str:
        return f"""【任务状态】
目标：{self.goal}
约束：{'; '.join(self.constraints) or '无'}
已决策：{'; '.join(self.decisions) or '无'}
关键事实：{'; '.join(self.facts) or '无'}
已完成：{'; '.join(self.done) or '无'}
待办：{'; '.join(self.todo) or '无'}
"""

# 每轮都放在 system 尾部（最稳定位置）
messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "system", "content": task_state.to_prompt()},
    *recent_messages,
]

4.3 结构化 > 自然语言

对比：
❌ 散在对话里：
"上次说了要用 Redis，还有那个不能删数据的要求，
  对了接口是 /api/orders..."

✅ 结构化清单：
constraints: ["不删数据", "只读写接口"]
decisions: ["缓存用 Redis"]
facts: ["订单接口 /api/orders"]

原因：
- 结构化的信息密度高（token 少、歧义少）
- 每轮固定注入，不会被裁/被稀释
- 模型照着清单执行，可靠性高

---

五、方案三：长期记忆（跨会话不丢）

5.1 记忆分类

会话内记忆（短期）：
- 最近对话 + 任务状态
- 存在内存里

跨会话记忆（长期）：
1. 事实记忆：用户偏好、业务规则（键值/JSON）
2. 情景记忆：历史任务、做过什么（向量检索）
3. 程序记忆：成功的执行方式（模板/SOP）

5.2 事实记忆（结构化存储）

import json, os

class FactMemory:
    """简单键值记忆：存用户偏好/业务事实"""

    def __init__(self, path="memory.json"):
        self.path = path
        self.data = {}
        if os.path.exists(path):
            with open(path, encoding="utf-8") as f:
                self.data = json.load(f)

    def save(self):
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)

    def remember(self, key: str, value):
        self.data[key] = value
        self.save()

    def recall(self, key: str, default=None):
        return self.data.get(key, default)

memory = FactMemory()
memory.remember("user.preferred_db", "PostgreSQL")
memory.remember("user.timezone", "Asia/Shanghai")

# 每轮注入
prefs = f"用户偏好：DB={memory.recall('user.preferred_db')}"

5.3 情景记忆（向量检索）

# 依赖：pip install chromadb
import chromadb

class EpisodicMemory:
    """向量库：存历史任务，按相似度检索"""

    def __init__(self, path="./chroma"):
        self.client = chromadb.PersistentClient(path=path)
        self.col = self.client.get_or_create_collection("episodes")

    def add(self, episode_id: str, content: str, metadata: dict = None):
        self.col.add(
            ids=[episode_id],
            documents=[content],
            metadatas=[metadata or {}],
        )

    def search(self, query: str, k: int = 3) -> list:
        results = self.col.query(query_texts=[query], n_results=k)
        return [doc for doc in results["documents"][0]]

# 使用：任务开始前检索相似历史
episodes = mem.search("怎么处理 MySQL 死锁")
context = "\n".join(episodes[:3])
# 把相关历史注入上下文（只注入相关的，不是全部）

关键：
- 只注入"检索命中的相关记忆"，不是全量塞
- 检索分数低的不注入（信噪比）
- 记忆要定期清理/更新（过期事实别用）

5.4 记忆写入的时机

该写：
- 用户明确说"记住..."、"以后都用..."
- 确认了重要决策（方案、约束）
- 任务完成后的复盘（成功经验/踩坑）

不该写：
- 一次性临时信息
- 未经确认的猜测
- 敏感数据（密钥/密码，不落盘）

---

六、方案四：多 Agent 共享上下文

6.1 问题

多 Agent 协作（Planner + Executor + Reviewer）：
- Executor 不知道 Planner 的约束
- Reviewer 不知道 Executor 改了哪些文件
- 每个 Agent 各自维护一份上下文 → 不一致

6.2 共享黑板的模式

import json

class SharedBoard:
    """多 Agent 共享状态：单一真相源"""

    def __init__(self):
        self.goal = ""
        self.constraints = []
        self.artifacts = {}      # 产物（文件路径/结果）
        self.log = []            # 事件流
        self.errors = []

    def snapshot_for(self, agent_role: str) -> str:
        """按角色裁剪：只给它需要的"""
        base = f"""目标：{self.goal}
约束：{'; '.join(self.constraints)}
产物：{json.dumps(self.artifacts, ensure_ascii=False)}
"""
        if agent_role == "reviewer":
            base += f"\n近期改动：{self.log[-5:]}"
            base += f"\n错误：{self.errors[-3:]}"
        return base

要点：
- 全局状态只存一份（避免各自漂移）
- 每个 Agent 拿到"裁剪过的快照"（不是全量）
- 事件流按时间追加，便于追溯

6.3 消息传递而非共享内存

class AgentMessage:
    def __init__(self, sender, receiver, intent, payload):
        self.sender = sender
        self.receiver = receiver
        self.intent = intent      # 如 "handoff" / "review_request"
        self.payload = payload    # 结构化数据

# Planner → Executor
msg = AgentMessage(
    sender="planner",
    receiver="executor",
    intent="执行任务",
    payload={
        "task": "给登录接口加限流",
        "constraints": ["不改表结构", "用 Redis"],
        "acceptance": ["单测通过", "接口返回 429"],
        "relevant_files": ["AuthService.java"],
    },
)

好处：
- 上下文明确、可审计
- 不依赖隐式共享状态
- 出问题能定位"谁传错了什么"

---

七、方案五：检索增强（只放该放的）

7.1 RAG 上下文注入的正确姿势

def build_context(query: str, retriever, max_docs: int = 3,
                  max_chars: int = 6000) -> str:
    """检索 → 过滤 → 裁剪 → 注入"""
    docs = retriever.search(query, k=10)      # 先多召回

    # 1. 分数过滤（低相关丢弃）
    docs = [d for d in docs if d.score > 0.75]

    # 2. 去重（相似内容只留一条）
    seen, unique = set(), []
    for d in docs:
        sig = d.text[:80]
        if sig not in seen:
            seen.add(sig)
            unique.append(d)

    # 3. 按分数取 top-N
    top = unique[:max_docs]

    # 4. 总长裁剪
    out, total = [], 0
    for d in top:
        if total + len(d.text) > max_chars:
            break
        out.append(f"[来源:{d.source}]\n{d.text}")
        total += len(d.text)

    return "\n\n".join(out)

7.2 常见错误

❌ 检索 20 篇全塞进去 → 噪音大、token 爆、模型混乱
❌ 不做分数过滤 → 不相关内容污染
❌ 不去重 → 同一内容多版本互相矛盾
❌ 不标来源 → 模型编造引用

✅ 少而准（top 3-5）+ 分数阈值 + 标明来源 + 总长上限

---

八、完整实战：一个会"记事儿"的 Agent

8.1 组装

class MemoryAgent:
    def __init__(self, llm, fact_mem, episodic_mem):
        self.llm = llm
        self.facts = fact_mem
        self.episodes = episodic_mem
        self.state = TaskState()
        self.history = []

    def build_messages(self, user_input: str) -> list:
        # 1. 检索相关历史
        related = self.episodes.search(user_input, k=3)
        related_text = "\n".join(related) if related else "无"

        # 2. 事实记忆
        facts_text = json.dumps(self.facts.data, ensure_ascii=False)

        # 3. 组装（分层，带预算）
        system = f"""你是执行型 Agent。
【长期事实】{facts_text}
【相关历史】{related_text}
"""
        messages = [
            {"role": "system", "content": system},
            {"role": "system", "content": self.state.to_prompt()},
        ]
        messages += rolling_summary(self.history, self.llm, keep_recent=8)
        messages.append({"role": "user", "content": user_input})
        return messages

    def run(self, user_input: str) -> str:
        messages = self.build_messages(user_input)
        resp = self.llm.invoke(messages)
        self.history.append({"role": "user", "content": user_input})
        self.history.append({"role": "assistant", "content": resp.content})
        return resp.content

8.2 效果对比

优化前：
- 20 轮后 token 3.5 万、费用高、偶发超限
- 关键约束常被忽略

优化后：
- token 稳定在 6000-9000
- 约束在任务状态里，每轮都在
- 历史记忆按需检索，不丢关键事实

---

九、常见坑速查

现象	根因	解决
上下文超限	历史/工具结果全塞	token 预算 + 滑动窗口 + 裁剪
越聊越忘	长上下文注意力衰减	结构化任务状态 + 滚动摘要
越答越编	噪音多、信噪比低	检索过滤 + top-N + 去重
工具结果撑爆	HTML/日志原样塞	摘要/截断/只留关键行
关键约束丢失	混在历史里被裁	提成清单，每轮固定注入
多 Agent 不一致	各自维护上下文	共享黑板 + 消息传递
成本失控	全量历史每轮重发	预算封顶 + 轻量模型分流

---

本章小结

Agent 上下文工程的主线：分层（系统/任务/记忆/工具）→ 预算（每层硬上限 + 留 30% 余量）→ 裁剪（滑动窗口 + 工具结果摘要）→ 压缩（滚动摘要 + 关键信息固化）→ 长期记忆（事实键值 + 情景向量）→ 多 Agent 共享（黑板 + 消息）→ 检索（少而准）。核心心法：上下文不是越多越好——系统层放规则、任务层放约束清单（每轮必现）、记忆层按需检索、工具层先摘要再入；token 永远留余量，关键信息永远不放进会被裁掉的位置。做好上下文工程，Agent 才能从"聊两句就崩"变成"长任务稳定跑"。

第二篇讲 Agent 工具调用不稳定问题与解决方案。

（全文完）