跳到主要内容

第 3 章 · 记忆与上下文管理

长任务最大的敌人是"忘事"和"爆 token"。本章讲 DB-GPT 用什么机制对抗这两件事:多层记忆负责"记住",任务进度追踪负责"不重复",四级压缩负责"不爆"。

3.1 它要解决的小问题​

Agent 跑十几步后,两个问题同时来:

  • 忘事:早期步骤的结果被挤出上下文窗口,模型不记得"第 2 步已经查到了销售额"。
  • 爆 token:把所有历史都塞进去,迟早超过模型的上下文上限,直接报错。

DB-GPT 用三套互补机制解决:分层记忆(决定记什么、怎么取)、任务进度追踪(一份永不丢的"已做清单")、渐进式压缩(token 紧张时分级瘦身)。

3.2 分层记忆:四种记忆各管一段​

记忆抽象在 dbgpt/agent/core/memory/base.py,按"人脑记忆模型"分了几类(都在 base.py 里定义为 Memory 的子类):

记忆类型类比干什么符号
感觉记忆一闪而过的感官最新输入的临时缓冲SensoryMemory
短期记忆工作记忆最近几条,默认 buffer 只有 5ShortTermMemory / EnhancedShortTermMemory
长期记忆长期存储带"时间权重 + 重要性"的检索LongTermMemory
混合记忆组合把上面几种组合起来用HybridMemory

对外统一入口是 AgentMemory(dbgpt/agent/core/memory/agent_memory.py:283-406),它包了一个具体 Memory 实现,默认就是 ShortTermMemory(buffer_size=5)(agent_memory.py:303-304)——只记最近 5 条。它还额外持有一个 gpts_memory,这个是多 Agent 之间共享的(存消息、存计划),structure_clone 时不克隆它(agent_memory.py:314-329)。

读 走 AgentMemory.read(observation)(agent_memory.py:371-390):按当前"观察"去检索相关记忆,带 alpha/beta/gamma 三个权重(重要性/时间/随机性)。写 走 write(agent_memory.py:352-360)。

记忆怎么被写进去?在第 1 章的循环里,每步结束都会调 Role.write_memories(dbgpt/agent/core/role.py:276-392),把"思考 + 动作 + 观察"按模板渲染成一条记忆片段:

# role.py:320-324  真实片段:一条记忆的骨架就是 思考/动作/观察
memory_map = {
    "thought": mem_thoughts,
    "action": action,
    "observation": observation,
}

3.3 巧妙之处一:任务进度追踪(永不丢的"已做清单")​

这是 DB-GPT 一个很聪明的设计。问题:记忆会被驱逐、上下文会被压缩,那"我已经做过哪几步"怎么保证不丢?

答案:用一个不进 pydantic 序列化的普通实例属性 _task_progress。看 Role 里的注释,意图写得很清楚:

# role.py:59-62  真实片段
# Task progress tracking: list of dicts with keys 'step', 'action', 'phase'
# This is NOT a pydantic field - managed as a plain instance attribute
# so it survives across retry rounds without being serialised into memory.
_task_progress: List[Dict] = []

为什么这么做: pydantic 字段会被序列化/反序列化,可能在记忆压缩、消息搬运中被改写或丢失。把它做成纯实例属性(用 object.__setattr__ 绕过 pydantic 写入,role.py:345-346),它就活在 Agent 对象的整个生命周期里,谁也动不了它。

每步成功就往里 append 一条(role.py:363-374),然后 task_progress_summary(role.py:210-238)把它渲染成一段人话,塞进每一次 LLM 调用的上下文(base_agent.py:1300-1304):

## Task Progress (do NOT repeat completed steps)
✅ Step 1: Action=run_sql | Intention: 查销售额 | ref: step_001_run_sql.json
✅ Step 2: Action=plot   | Intention: 画趋势图

这样无论记忆怎么被压缩,模型每次都能看到"已经干了啥、别重复"。

3.4 巧妙之处二:操作快照落盘​

压缩上下文会丢细节(精确的 SQL、文件路径、计算结果)。DB-GPT 的对策:每步成功操作都把完整细节写一个 JSON 快照到磁盘,_write_op_snapshot(role.py:394-456):

# role.py:435-448  真实片段(精简)
filename = f"step_{step:03d}_{safe_action}.json"
payload = {
    "step": step, "action": action, "thought": thought or "",
    "action_input": action_input or "",       # ← 完整入参,不截断
    "observation": observation or "",          # ← 完整观察,不截断
    "timestamp": datetime.utcnow().isoformat(), "conv_id": conv_id,
}

妙在: 上下文里只留摘要("Step 1 跑了 SQL"),精确值留在 step_001_run_sql.json 里。模型后续真需要那条 SQL 的原文,可以用 read_file 动作把它捞回来。这是"摘要省 token + 细节可回溯"的两全。

3.5 巧妙之处三:四级渐进式上下文压缩​

当上下文真的逼近上限,ContextManager.manage_context(dbgpt/agent/core/context/manager.py:93-167)按 token 状态分级压缩——能用便宜手段就不用贵的:

 token 用量上升 ──────────────────────────────►
 NORMAL          WARNING                ERROR        (LLM 报 too_long)
   │               │                      │                │
 啥都不做      Layer1 截旧观察 ──► Layer2 丢旧轮次 ──► Layer3 LLM摘要 ──► Layer4 紧急截断
             (无需LLM,便宜)    (无需LLM,靠      (要调LLM,贵)    (响应式,
                                task_progress                       最后一招)
                                当隐式摘要)
层触发状态手段要 LLM 吗符号
Layer 1WARNING截断旧的 Observation 消息否ObservationMicroCompact
Layer 2WARNING丢旧轮次,靠 task_progress 当隐式摘要否SessionMemoryCompact
Layer 3ERRORLLM 生成结构化摘要替换旧消息是FullContextCompression
Layer 4响应式LLM 真报 context_too_long 时紧急截断否ReactiveCompact

(分层定义见 context/compact.py:82-316;触发逻辑见 manager.py:139-167。)

两个细节体现工程成熟度:

  • 熔断器:manage_context 里若 circuit_breaker_tripped 就跳过压缩(manager.py:126-130),避免压缩本身反复失败把系统拖垮。
  • Layer 4 是响应式的:它不在主流程预防性触发,而是在第 1 章 think 那步真收到 context_too_long 报错时才补救(base_agent.py:572-596)——压完再 think 一次。这是"预防(Layer1-3)+ 兜底(Layer4)"的双保险。

注意:整套上下文管理是 opt-in 的——AgentContext.enable_context_management 默认 False(agent.py:220),只有显式打开才挂 ContextManager(base_agent.py:1376-1382 检查 _context_manager 是否存在)。

3.6 关键细节 / 坑​

  • 默认记忆很小:buffer 只有 5,不开任务进度追踪 + 压缩的话,长任务必然忘事。这套机制不是摆设,是长任务的刚需。
  • 进度只记成功步:write_memories 里只有 check_pass and action 才 append 进度(role.py:344),失败步不进"已做清单"——符合直觉(失败的不算"做完")。
  • 快照写盘可能静默失败:_write_op_snapshot 出异常只 log 不抛(role.py:452-456),返回 None,不影响主流程——细节回溯是"尽力而为"。

3.7 代码地图​

主题文件路径(相对 packages/dbgpt-core/src/)关键符号
记忆基类与分层dbgpt/agent/core/memory/base.pyMemory, SensoryMemory, ShortTermMemory, LongTermMemory
混合记忆dbgpt/agent/core/memory/hybrid.pyHybridMemory
对外记忆入口dbgpt/agent/core/memory/agent_memory.pyAgentMemory, AgentMemoryFragment
写记忆 + 进度 + 快照dbgpt/agent/core/role.pywrite_memories, task_progress_summary, _write_op_snapshot
上下文管理调度dbgpt/agent/core/context/manager.pyContextManager.manage_context, reactive_compact
四级压缩实现dbgpt/agent/core/context/compact.pyObservationMicroCompact, SessionMemoryCompact, FullContextCompression, ReactiveCompact