本文档描述 Agent 的 Context 管理、执行计划和探索机制。
┌─────────────────┐
│ plan.md │ ← 文本格式的计划(TODO 列表)
└─────────────────┘
↓
┌─────────────────┐
│ 线性 Message │ ← 对话历史
│ List │
└─────────────────┘
↓
┌─────────────────┐
│ Prune + Full │ ← 两阶段压缩
│ Compaction │
└─────────────────┘
↓
┌─────────────────┐
│ Sub-Agent │ ← 隔离大任务
└─────────────────┘
数据结构:
存储:
Storage Key:
["message", sessionID, messageID] -> MessageV2.Info
["part", messageID, partID] -> MessageV2.Part
两阶段压缩:
阶段 1: Prune(清理旧工具输出)
参数:
- PRUNE_MINIMUM = 20,000 tokens(最少删除量)
- PRUNE_PROTECT = 40,000 tokens(保护阈值)
- PRUNE_PROTECTED_TOOLS = ["skill"](不删除的工具)
流程:
1. 从后向前遍历 messages
2. 跳过最后 2 轮 turns(保护最近交互)
3. 跳过已有 summary 标记的 assistant 消息
4. 收集已完成工具调用的输出
5. 当累计 > PRUNE_PROTECT 时,标记为已 compacted
6. 当删除量 > PRUNE_MINIMUM 时,执行删除
阶段 2: Full Compaction(上下文总结)
流程:
1. 创建新的 assistant 消息(summary=true)
2. 调用 "compaction" 专用 agent
3. 提示词: "Provide a detailed prompt for continuing our conversation..."
4. 返回 "continue" 时自动创建新的 user 消息继续
数据结构:
Todo.Info = {
id: string
content: string // 任务描述
status: string // pending | in_progress | completed | cancelled
priority: string // high | medium | low
}
存储:文件系统(.opencode/plans/xxx.md)或 Storage
Agent Mode:
primary: 主代理,执行工具subagent: 子代理,独立 context,结果汇总回主会话内置 Sub-Agents:
general: 通用代理,可并行执行多个任务explore: 代码探索专用,仅允许查询工具compaction: 上下文总结专用Subtask 执行:
优点:
局限:
基于 OpenCode 方案,增强三个能力:
1. 结构化 Plan(goal 工具)
2. 并行探索-合并(explore 工具)
3. 精确回溯(abandon + context 压缩)
┌─────────────────────────────────────────────┐
│ Plan (goal.json) │
│ 结构化的目标树,LLM 通过 goal 工具维护 │
└─────────────────────────────────────────────┘
│
┌────────────┴────────────┐
↓ ↓
┌─────────────────┐ ┌─────────────────┐
│ 线性执行 │ │ 并行探索 │
│ (主 message │ │ (explore 工具) │
│ list) │ │ 多个独立分支 │
└─────────────────┘ └─────────────────┘
│ │
↓ ↓
┌─────────────────┐ ┌─────────────────┐
│ 完成/回溯 │ │ 合并评估 │
│ done/abandon │ │ 返回主会话 │
│ 触发 context │ └─────────────────┘
│ 压缩 │
└─────────────────┘
@tool
def goal(
add: Optional[str] = None, # 添加目标(逗号分隔多个)
done: Optional[str] = None, # 完成当前目标,值为 summary
abandon: Optional[str] = None, # 放弃当前目标,值为原因
focus: Optional[str] = None, # 切换焦点到指定 id
) -> str:
"""管理执行计划。"""
层级支持:add 添加到当前 focus 的 goal 下作为子目标。
# 没有 focus 时,添加到顶层
goal(add="分析代码, 实现功能, 测试")
# 结果:
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 3. 测试
# focus 到某个 goal 后,add 添加为其子目标
goal(focus="2")
goal(add="设计接口, 实现代码")
# 结果:
# [ ] 1. 分析代码
# [→] 2. 实现功能
# [ ] 2.1 设计接口
# [ ] 2.2 实现代码
# [ ] 3. 测试
状态流转:
pending ──focus──→ in_progress ──done──→ completed
│ ↓
│ (压缩 context)
│
abandon
↓
abandoned
↓
(压缩 context)
基于 sub-agent 机制实现。
@tool
def explore(
branches: List[str], # 探索方向列表
background: Optional[str] = None, # 背景概括(可选)
) -> str:
"""
并行探索多个方向,汇总结果。
- background 有值:用它初始化各分支的 context
- background 为空:继承主 message list
"""
示例:
explore(
background="我们在实现用户认证。项目用 FastAPI,用户模型在 models/user.py。环境没有 Redis。",
branches=[
"调研 JWT 方案,考虑 token 刷新和撤销",
"调研 Session 方案,寻找 Redis 替代存储"
]
)
执行流程:
1. 为每个 branch 创建 sub-agent
- context = background(或继承主 msg list)
- prompt = branch 指令
2. 串行执行各 sub-agent
3. 收集结论,汇总返回主会话
分支 context 初始化:
background:LLM 概括的背景信息作为初始 contextbackground:继承全部主 message list(适用于 context 不长的情况)@dataclass
class Goal:
id: str # 自动生成: "1", "1.1", "2"
description: str # 目标描述
status: Status # pending | in_progress | completed | abandoned
summary: Optional[str] = None # 完成/放弃时的总结
children: List["Goal"] = field(default_factory=list)
Status = Literal["pending", "in_progress", "completed", "abandoned"]
@dataclass
class GoalTree:
mission: str # 总任务描述
current_id: Optional[str] = None # 当前焦点
goals: List[Goal] = field(default_factory=list)
# 每条 message 记录它属于哪个 goal
message = {
"role": "assistant",
"content": "...",
"goal_id": "2.1" # 关联到目标 2.1
}
每次 LLM 调用时,在 system prompt 末尾注入当前计划状态:
## Current Plan
**Mission**: 实现用户认证功能
**Current**: 2.1 实现登录接口
**Progress**:
[✓] 1. 分析代码
→ 用户模型在 models/user.py,使用 bcrypt 加密
[→] 2. 实现功能
[✓] 2.1 设计接口
[→] 2.2 实现登录接口 ← current
[ ] 2.3 实现注册接口
[ ] 3. 测试
当调用 goal(done="...") 时:
当调用 goal(abandon="...") 时:
Before 回溯:
Messages:
[分析代码的 20 条 message...]
[实现方案 A 的 30 条 message...] ← 这些要压缩
[测试失败的 message...]
Plan:
[✓] 1. 分析代码
[✓] 2. 实现方案 A
[→] 3. 测试
After 回溯:
Messages:
[分析代码的 20 条 message...]
[Summary: "尝试方案 A,因依赖问题失败"] ← 压缩为 1 条
[开始方案 B 的 message...]
Plan:
[✓] 1. 分析代码
[✗] 2. 实现方案 A (abandoned: 依赖问题)
[→] 2'. 实现方案 B
[ ] 3. 测试
.trace/{trace_id}/
├── goal.json # Goal Tree(LLM 通过工具维护)
├── messages.jsonl # 消息记录(系统自动,含 goal_id)
└── meta.json # Trace 元数据
Goal Tree + Messages 合并展示:
Mission: 实现用户认证功能
══════════════════════════════════════════
[✓] 1. 分析代码 (5 steps, 1.2s)
→ 用户模型在 models/user.py
┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄
├─ glob_files("**/user*.py")
├─ read_file("models/user.py")
└─ [详细步骤已折叠]
[✗] 2. 实现方案 A (abandoned)
→ 需要 Redis,环境没有
[→] 2'. 实现方案 B ← current
┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄
├─ read_file("requirements.txt")
└─ edit_file("app.py")
[ ] 3. 测试
──────────────────────────────────────────
Progress: 1/3 goals | Current: 2'
| 方面 | OpenCode | 我们的方案 |
|---|---|---|
| Plan 格式 | 纯文本 (plan.md) | 结构化 (goal.json) |
| Plan 与执行关联 | 无 | 通过 goal_id 关联 |
| 压缩时机 | 事后(context 满时) | 增量(goal 完成/放弃时) |
| 并行探索 | Sub-agent(手动管理) | explore 工具(自动汇总) |
| 回溯能力 | 有限 | 精确(基于 goal 压缩) |
| 工具复杂度 | todoread/todowrite | goal/explore(更简单) |
| 功能 | 文件路径 | 状态 |
|---|---|---|
| Goal 数据模型 | agent/goal/models.py |
待实现 |
| goal 工具 | agent/goal/tool.py |
待实现 |
| explore 工具 | agent/goal/explore.py |
待实现 |
| Context 压缩 | agent/goal/compaction.py |
待实现 |
| Plan 注入 | agent/core/runner.py |
待实现 |
| 可视化 | agent/goal/visualize.py |
待实现 |