数据截至 2026-08-17(上游 commit db580e952e54)

第 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 只有 5	`ShortTermMemory` / `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:278-406),把"思考 + 动作 + 观察"按模板渲染成一条记忆片段:

# role.py:332-336  真实片段:一条记忆的骨架就是 思考/动作/观察
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:359-360),它就活在 Agent 对象的整个生命周期里,谁也动不了它。

每步成功就往里 append 一条(role.py:377-388),然后 task_progress_summary(role.py:212-240)把它渲染成一段人话,塞进每一次 LLM 调用的上下文(base_agent.py:1359-1361):

## 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:408-470):

# role.py:449-462  真实片段(精简)
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-170)按 token 状态分级压缩——能用便宜手段就不用贵的:

 token 用量上升 ──────────────────────────────►
 NORMAL          WARNING                ERROR        (LLM 报 too_long)
   │               │                      │                │
 啥都不做      Layer1 截旧观察 ──► Layer2 丢旧轮次 ──► Layer3 LLM摘要 ──► Layer4 紧急截断
             (无需LLM,便宜)    (无需LLM,靠      (要调LLM,贵)    (响应式,
                                task_progress                       最后一招)
                                当隐式摘要)

层	触发状态	手段	要 LLM 吗	符号
Layer 1	WARNING	截断旧的 Observation 消息	否	`ObservationMicroCompact`
Layer 2	WARNING	丢旧轮次,靠 task_progress 当隐式摘要	否	`SessionMemoryCompact`
Layer 3	ERROR	LLM 生成结构化摘要替换旧消息	是	`FullContextCompression`
Layer 4	响应式	LLM 真报 context_too_long 时紧急截断	否	`ReactiveCompact`

(分层定义见 context/compact.py:83-339;触发逻辑见 manager.py:139-170。)

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

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

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

3.6 关键细节 / 坑

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

3.7 代码地图

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

3.1 它要解决的小问题​

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

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

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

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

3.6 关键细节 / 坑​

3.7 代码地图​