本文档描述 Agent 执行过程的结构化记录、计划管理和 Context 压缩机制。
StepType = Literal[
# 计划相关
"goal", # 目标/计划项(可以有子 steps)
# LLM 输出
"thought", # 思考/分析(中间过程)
"evaluation", # 评估总结(需要 summary)
"response", # 最终回复
# 工具相关
"action", # 工具调用(tool_call)
"result", # 工具结果(tool_result)
]
| 类型 | 来源 | 说明 |
|---|---|---|
goal |
LLM(通过 step 工具) | 设定目标/计划 |
thought |
LLM | 中间思考,不产生工具调用 |
evaluation |
LLM | 对一组操作的总结,需要 summary |
response |
LLM | 最终给用户的回复 |
action |
System | LLM 决定调用工具,系统记录 |
result |
System | 工具执行结果 |
Status = Literal[
"planned", # 计划中(未执行)
"in_progress", # 执行中
"completed", # 已完成
"failed", # 失败
"skipped", # 跳过
]
@dataclass
class Step:
step_id: str
trace_id: str
step_type: StepType
status: Status
sequence: int
# 树结构(单父节点)
parent_id: Optional[str] = None
# 内容
description: str # 所有节点都有
data: Dict[str, Any] = field(default_factory=dict)
# 仅 evaluation 类型需要
summary: Optional[str] = None
# 执行指标
duration_ms: Optional[int] = None
cost: Optional[float] = None
tokens: Optional[int] = None
# 时间
created_at: datetime = field(default_factory=datetime.now)
关键点:
parent_id 是单个值(树结构),不是列表(DAG)summary 仅在 evaluation 类型节点填充,不是每个节点都需要planned 状态的 step 相当于 TODO itemTrace
├── goal: "探索代码库" (completed)
│ ├── thought: "需要先了解项目结构"
│ ├── action: glob_files
│ ├── result: [15 files...]
│ ├── thought: "发现配置文件,需要查看内容"
│ ├── action: read_file
│ ├── result: [content...]
│ └── evaluation: "主配置在 /src/config.yaml" ← summary
│
├── goal: "修改配置" (in_progress)
│ ├── action: read_file
│ └── result: [content...]
│
└── goal: "运行测试" (planned)
| Step 类型 | parent 是谁 |
|---|---|
goal |
上一个 goal(或 None) |
thought |
当前 in_progress 的 goal |
action |
当前 in_progress 的 goal |
result |
对应的 action |
evaluation |
所属的 goal |
response |
当前 in_progress 的 goal(或 None) |
以下字段由系统自动填充,不需要 LLM 参与:
step_id: str # 自动生成
parent_id: str # 根据当前 focus 的 goal 自动设置
step_type: StepType # 根据 LLM 输出推断(见下)
sequence: int # 递增序号
tokens: int # API 返回
cost: float # 计算得出
duration_ms: int # 计时
created_at: datetime # 当前时间
系统根据 LLM 输出内容自动推断类型,不需要显式声明:
def infer_step_type(llm_response) -> StepType:
# 有工具调用 → action
if llm_response.tool_calls:
return "action"
# 调用了 step 工具且 complete=True → evaluation
if called_step_tool(llm_response, complete=True):
return "evaluation"
# 调用了 step 工具且 plan 不为空 → goal
if called_step_tool(llm_response, plan=True):
return "goal"
# 最终回复(无后续工具调用,对话结束)
if is_final_response(llm_response):
return "response"
# 默认:中间思考
return "thought"
description 字段由系统从 LLM 输出中提取:
| Step 类型 | description 来源 |
|---|---|
goal |
step 工具的 plan 参数 |
thought |
LLM 输出的第一句话(或截断) |
action |
工具名 + 关键参数 |
result |
工具返回的 title 或简要输出 |
evaluation |
step 工具的 summary 参数 |
response |
LLM 输出的第一句话(或截断) |
模型通过 step 工具管理执行进度:
@tool
def step(
plan: Optional[List[str]] = None, # 添加 planned goals
focus: Optional[str] = None, # 切换焦点到哪个 goal
complete: bool = False, # 完成当前 goal
summary: Optional[str] = None, # 评估总结(配合 complete)
):
"""管理执行步骤"""
# 1. 创建计划
step(plan=["探索代码库", "修改配置", "运行测试"])
# 2. 开始执行第一个
step(focus="探索代码库")
# 3. [执行各种 tool_call...]
# 4. 完成并切换到下一个
step(complete=True, summary="主配置在 /src/config.yaml", focus="修改配置")
# 5. 中途调整计划
step(plan=["备份配置"]) # 追加新的 goal
调用 step(plan=["A", "B", "C"]) 后:
├── goal: "A" (planned)
├── goal: "B" (planned)
└── goal: "C" (planned)
调用 step(focus="A") 后:
├── goal: "A" (in_progress) ← 当前焦点
├── goal: "B" (planned)
└── goal: "C" (planned)
调用 step(complete=True, summary="...", focus="B") 后:
├── goal: "A" (completed)
│ └── evaluation: "..." ← 自动创建
├── goal: "B" (in_progress) ← 新焦点
└── goal: "C" (planned)
不同用途需要不同的信息粒度:
| 用途 | 选择哪些节点 | 详略程度 |
|---|---|---|
| Todo 列表 | 仅 goal 类型 |
简略:描述 + 状态 |
| 历史压缩 | goal + result + evaluation |
详细:包含关键结果 |
def to_todo_string(tree: StepTree) -> str:
lines = []
for goal in tree.filter(step_type="goal"):
icon = {"completed": "✓", "in_progress": "→", "planned": " "}[goal.status]
lines.append(f"[{icon}] {goal.description}")
return "\n".join(lines)
输出:
[✓] 探索代码库
[→] 修改配置
[ ] 运行测试
def to_history_string(tree: StepTree) -> str:
lines = []
for goal in tree.filter(step_type="goal"):
status_label = {"completed": "完成", "in_progress": "进行中", "planned": "待做"}
lines.append(f"[{status_label[goal.status]}] {goal.description}")
if goal.status == "completed":
# 选择关键结果节点
for step in goal.children():
if step.step_type == "result":
lines.append(f" → {extract_brief(step.data)}")
elif step.step_type == "evaluation":
lines.append(f" 总结: {step.summary}")
return "\n".join(lines)
输出:
[完成] 探索代码库
→ glob_files: 找到 15 个文件
→ read_file(config.yaml): db_host=prod.db.com
总结: 主配置在 /src/config.yaml,包含数据库连接配置
[进行中] 修改配置
→ read_file(config.yaml): 已读取
[待做] 运行测试
def build_messages(messages: List, tree: StepTree) -> List:
# 正常情况:不压缩
if estimate_tokens(messages) < MAX_CONTEXT * 0.7:
return messages
# 超限时:用树摘要替代历史详情
history_summary = tree.to_history_string()
summary_msg = {"role": "assistant", "content": history_summary}
# 保留最近的详细消息
return [summary_msg] + recent_messages(messages)
模型可通过工具读取当前进度,而非每次都注入:
@tool
def read_progress() -> str:
"""读取当前执行进度"""
return tree.to_todo_string()
策略:
read_progress 按需读取(省 context)树结构天然支持可视化:
goal 节点 → 隐藏其子节点failed 或 skipped 状态的分支goal 下的多个 action(并行工具调用)可视化时,边(连接线)可展示:
Step.duration_msStep.costStep.description| 方面 | OpenCode | 本设计 |
|---|---|---|
| 计划存储 | Markdown 文件 + Todo 列表 | Step 树(planned 状态) |
| 计划与执行关联 | 无结构化关联 | 统一在树结构中 |
| 进度读取 | todoread 工具 |
read_progress 工具 |
| 进度更新 | todowrite 工具 |
step 工具 |
| Context 压缩 | 无 | 基于树结构自动压缩 |
参考:OpenCode 的实现见 src/tool/todo.ts、src/session/prompt.ts
开发调试时,可通过 dump_tree 将完整的 Step 树输出到文件:
from agent.debug import dump_tree
# 每次 step 变化后调用
dump_tree(trace, steps)
# 自定义路径
dump_tree(trace, steps, output_path=".debug/my_trace.txt")
# 方式1:终端实时刷新
watch -n 0.5 cat .trace/tree.txt
# 方式2:VS Code 打开(自动刷新)
code .trace/tree.txt
============================================================
Step Tree Debug
Generated: 2024-01-15 14:30:25
============================================================
## Trace
trace_id: abc123
task: 修改配置文件
status: running
total_steps: 5
total_tokens: 1234
total_cost: 0.0150
## Steps
├── [✓] goal: 探索代码库
│ id: a1b2c3d4...
│ duration: 1234ms
│ tokens: 500
│ cost: $0.0050
│ data:
│ description: 探索代码库
│ time: 14:30:10
│
│ ├── [✓] thought: 需要先了解项目结构
│ │ id: e5f6g7h8...
│ │ data:
│ │ content: 让我先看看项目的目录结构...
│ │ time: 14:30:11
│ │
│ ├── [✓] action: glob_files
│ │ id: i9j0k1l2...
│ │ duration: 50ms
│ │ data:
│ │ tool_name: glob_files
│ │ arguments: {"pattern": "**/*.py"}
│ │ time: 14:30:12
│ │
│ └── [✓] result: 找到 15 个文件
│ id: m3n4o5p6...
│ data:
│ output: ["src/main.py", "src/config.py", ...]
│ time: 14:30:12
│
└── [→] goal: 修改配置
id: q7r8s9t0...
time: 14:30:15
用于程序化分析:
from agent.debug import dump_json
dump_json(trace, steps) # 输出到 .trace/tree.json
实现:agent/debug/tree_dump.py
agent/models/trace.py:Step(待更新)agent/tools/builtin/step.py(待实现)agent/tools/builtin/step.py(待实现)agent/context/compressor.py(待实现)agent/debug/tree_dump.py(已实现)agent/skills/core.md(已实现)