本文档描述 Agent 的 Context 管理、执行计划和 Sub-Agent 机制。
┌─────────────────┐
│ 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. 统一的 Trace 模型(主 Agent 和 Sub-Agent 都是 Trace)
3. 并行探索-合并(explore 工具,启动独立的 Sub-Traces)
4. 精确回溯(abandon + context 压缩)
┌─────────────────────────────────────────────┐
│ 每个 Agent = 一个 Trace │
│ 主 Trace 和 Sub-Trace 使用相同的数据结构 │
└─────────────────────────────────────────────┘
│
┌────────────┴────────────┐
↓ ↓
┌─────────────────┐ ┌─────────────────┐
│ GoalTree │ │ Messages │
│ (层级目标) │ │ (扁平列表) │
│ goal 工具维护 │ │ goal_id 关联 │
└─────────────────┘ └─────────────────┘
│ │
↓ ↓
┌─────────────────┐ ┌─────────────────┐
│ 工具调用 │ │ Sub-Traces │
│ explore/ │ │ 完全独立的 │
│ delegate │──────▶ Trace 实例 │
└─────────────────┘ └─────────────────┘
│
↓
┌─────────────────────────────────────────────┐
│ DAG 可视化(派生视图) │
│ 从 GoalTree + Messages 生成 │
│ 节点 = 结果/里程碑,边 = 动作/执行过程 │
│ 边可展开/折叠,Sub-Trace 作为折叠边显示 │
└─────────────────────────────────────────────┘
关键设计:每个 Agent(主 Agent 或 Sub-Agent)都是一个完整的 Trace。
# 主 Agent
Trace(trace_id="2f8d3a1c", mode="agent", task="实现用户认证")
# Sub-Agent(并行探索)
Trace(trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001", parent_trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", agent_type="explore", task="JWT 方案")
Trace(trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002", parent_trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", agent_type="explore", task="Session 方案")
# Sub-Agent(单线委托)
Trace(trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220030-001", parent_trace_id="2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", agent_type="delegate", task="实现具体功能")
优点:
@ 前是父 ID)后端存储两类数据,可视化的 DAG 是派生视图:
goal_id 关联 Goal不存在独立的"边"数据结构,边在可视化时从 Messages 聚合生成。
@dataclass
class Trace:
"""
执行轨迹 - 一次完整的 Agent 运行
主 Agent 和 Sub-Agent 使用相同的 Trace 结构
"""
trace_id: str # 主 Trace: UUID,Sub-Trace: parent@mode-timestamp-seq
mode: Literal["call", "agent"]
task: Optional[str] = None
# 父子关系
parent_trace_id: Optional[str] = None # 父 Trace ID
parent_goal_id: Optional[str] = None # 哪个 Goal 启动的
# 类型标记(语义)
agent_type: Optional[str] = None # "main", "explore", "delegate", "compaction"
# 权限和配置
context: Dict[str, Any] = field(default_factory=dict)
# 可以包含:
# - allowed_tools: ["read", "grep"] # 工具白名单
# - denied_tools: ["bash"] # 工具黑名单
# - max_turns: 10 # 最大对话轮数
# - timeout: 300 # 超时(秒)
status: Literal["running", "completed", "failed"] = "running"
# 统计
total_messages: int = 0
total_tokens: int = 0
total_cost: float = 0.0
created_at: datetime
completed_at: Optional[datetime] = None
实现:agent/execution/models.py:Trace
Trace ID 命名规则:
主 Trace(用户直接触发):
{uuid}
例如: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d
Sub-Trace(Agent 启动的子任务):
{parent_id}@{mode}-{timestamp}-{seq}
例如: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001
例如: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220015-002
格式说明:
parent_id:父 Trace 的完整 UUID(不截断,避免冲突)@:父子关系分隔符mode:运行模式(explore, delegate, compaction 等)timestamp:创建时间戳(YYYYMMDDHHmmss)seq:同一秒内的序号(001, 002, ...)优点:
@ 分隔符明确表示父子关系从 trace_id 解析父子关系:
def parse_parent_trace_id(trace_id: str) -> Optional[str]:
"""从 trace_id 解析出 parent_trace_id"""
if '@' in trace_id:
return trace_id.split('@')[0]
return None
# 示例
parse_parent_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001")
# → "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d"
parse_parent_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d")
# → None
生成 Sub-Trace ID:
from datetime import datetime
from threading import Lock
from typing import Dict
# 全局计数器(线程安全)
_seq_lock = Lock()
_seq_counter: Dict[str, int] = {} # key: "{parent_id}@{mode}-{timestamp}"
def generate_sub_trace_id(parent_id: str, mode: str) -> str:
"""
生成 Sub-Trace ID
格式: {parent_id}@{mode}-{timestamp}-{seq}
Args:
parent_id: 父 Trace ID(完整 UUID,不截断)
mode: 运行模式(explore, delegate, compaction)
Returns:
Sub-Trace ID(完整格式,无碰撞风险)
"""
# 直接使用完整 UUID,不截断
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
# 生成序号(同一秒内递增)
prefix = f"{parent_id}@{mode}-{timestamp}"
with _seq_lock:
seq = _seq_counter.get(prefix, 0) + 1
_seq_counter[prefix] = seq
return f"{prefix}-{seq:03d}"
# 示例
generate_sub_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", "explore")
# → "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001"
generate_sub_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", "delegate")
# → "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220030-001"
@dataclass
class GoalStats:
message_count: int = 0 # 消息数量
total_tokens: int = 0 # Token 总数
total_cost: float = 0.0 # 总成本
preview: Optional[str] = None # 工具调用摘要,如 "read_file → edit_file → bash"
GoalStatus = Literal["pending", "in_progress", "completed", "abandoned"]
GoalType = Literal["normal", "agent_call"] # agent_call: 启动了 Sub-Agent
@dataclass
class Goal:
id: str # 内部唯一 ID,纯自增("1", "2", "3"...)
parent_id: Optional[str] = None # 父 Goal ID(层级关系)
type: GoalType = "normal" # Goal 类型
description: str # 目标描述(做什么)
reason: str # 创建理由(为什么做)
status: GoalStatus # pending | in_progress | completed | abandoned
summary: Optional[str] = None # 完成/放弃时的总结
# agent_call 特有
sub_trace_ids: Optional[List[str]] = None # 启动的 Sub-Trace IDs
agent_call_mode: Optional[str] = None # "explore" | "delegate" | "sequential"
sub_trace_metadata: Optional[Dict[str, Dict[str, Any]]] = None # Sub-Trace 元数据
# 统计(后端维护,用于可视化边的数据)
self_stats: GoalStats # 自身统计(仅直接关联的 messages)
cumulative_stats: GoalStats # 累计统计(自身 + 所有后代)
实现:agent/goal/models.py:Goal
ID 设计:
parent_id 字段维护to_prompt() 时动态生成连续有意义的编号("1", "2", "2.1", "2.2"...)sub_trace_metadata 字段(agent_call 类型 Goal 专用):
存储各 Sub-Trace 的关键信息,用于辅助决策和可视化:
{
"sub_trace_id_1": {
"task": "JWT 方案", # Sub-Trace 任务描述
"status": "completed", # Sub-Trace 状态
"summary": "实现完成,使用 JWT token", # Sub-Trace 总结
"last_message": { # 最后一条 assistant 消息
"role": "assistant",
"description": "生成 JWT token 并返回",
"content": "...", # 截断至 500 字符
"created_at": "2026-02-05T10:30:00"
},
"stats": { # 统计信息
"message_count": 8,
"total_tokens": 4000,
"total_cost": 0.05
}
},
# ... 其他 Sub-Trace
}
用途:
统计更新逻辑:
self_stats,并沿祖先链向上更新所有祖先的 cumulative_statscumulative_stats,展开边使用 self_stats@dataclass
class GoalTree:
mission: str # 总任务描述
current_id: Optional[str] = None # 当前焦点(内部 ID)
goals: List[Goal] # 扁平列表,通过 parent_id 构建层级
_next_id: int = 1 # 内部 ID 计数器
实现:agent/goal/models.py:GoalTree
Message 对应 LLM API 的消息,加上元数据。每条 Message 通过 goal_id 关联所属 Goal。
@dataclass
class Message:
message_id: str
trace_id: str # 所属 Trace ID
role: Literal["assistant", "tool"] # 和 LLM API 一致
sequence: int # 当前 Trace 内的顺序
goal_id: Optional[str] = None # 关联的 Goal 内部 ID(None = 还没有创建 Goal)
tool_call_id: Optional[str] # tool 消息关联对应的 tool_call
content: Any # 消息内容(和 LLM API 格式一致)
description: str # 消息描述(系统自动生成)
# 元数据
tokens: Optional[int] = None
cost: Optional[float] = None
duration_ms: Optional[int] = None
created_at: datetime
description 字段(系统自动生成):
assistant 消息:优先取 content 中的 text,若无 text 则生成 "tool call: XX, XX"tool 消息:使用 tool namegoal_id 说明:
实现:agent/execution/models.py:Message
Message 类型说明:
role="assistant":模型的一次返回,可能同时包含文本和多个 tool_callsrole="tool":一个工具的执行结果,通过 tool_call_id 关联对应的 tool_call查询 Message:
# 查询主 Trace 的 Messages
GET /api/traces/2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d/messages?goal_id=2
# 查询 Sub-Trace 的 Messages
GET /api/traces/2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001/messages?goal_id=1
@tool
def goal(
add: Optional[str] = None, # 添加目标(逗号分隔多个)
reason: Optional[str] = None, # 创建理由(逗号分隔多个,与 add 一一对应)
after: Optional[str] = None, # 在指定目标后面添加(同层级)
under: Optional[str] = None, # 为指定目标添加子目标
done: Optional[str] = None, # 完成当前目标,值为 summary
abandon: Optional[str] = None, # 放弃当前目标,值为原因
focus: Optional[str] = None, # 切换焦点到指定 ID
) -> str:
"""管理执行计划。"""
实现:agent/goal/tool.py:goal_tool
位置控制:
after/under:添加到当前 focus 的 goal 下作为子目标(无 focus 时添加到顶层)after="X":在目标 X 后面添加兄弟节点(同层级)under="X":为目标 X 添加子目标(如已有子目标,追加到最后)after 和 under 互斥,不可同时指定设计原则:优先使用 after 明确指定位置,under 用于首次拆解或追加子任务。
示例:
# 1. 默认行为:添加顶层目标(无 focus)
goal(add="分析代码, 实现功能, 测试")
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 3. 测试
# 2. 拆解子任务
goal(add="设计接口, 实现代码", under="2")
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 2.1 设计接口
# [ ] 2.2 实现代码
# [ ] 3. 测试
# 3. 追加同层级任务
goal(add="编写文档", after="3")
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 2.1 设计接口
# [ ] 2.2 实现代码
# [ ] 3. 测试
# [ ] 4. 编写文档
# 4. 追加子任务(目标2已有子任务)
goal(add="编写单元测试", under="2")
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 2.1 设计接口
# [ ] 2.2 实现代码
# [ ] 2.3 编写单元测试
# [ ] 3. 测试
# [ ] 4. 编写文档
# 5. 使用 after 在子任务中间插入
goal(add="代码审查", after="2.2")
# [ ] 1. 分析代码
# [ ] 2. 实现功能
# [ ] 2.1 设计接口
# [ ] 2.2 实现代码
# [ ] 2.3 代码审查
# [ ] 2.4 编写单元测试
# [ ] 3. 测试
# [ ] 4. 编写文档
状态流转:
pending ──focus──→ in_progress ──done──→ completed
│ ↓
│ (压缩 context)
│ (级联:若所有兄弟都 completed,父 goal 自动 completed)
│
abandon
↓
abandoned
↓
(压缩 context)
启动多个独立的 Sub-Traces 并行执行。
@tool
def explore(
branches: List[str], # 探索方向列表
background: Optional[str] = None, # 背景概括(可选)
) -> str:
"""
并行探索多个方向,汇总结果。
每个方向会启动一个独立的 Sub-Trace(agent_type="explore")。
- background 有值:用它初始化各 Sub-Trace 的 context
- background 为空:继承主 Trace 的 message list
返回:各分支的汇总结果
"""
内部实现:
async def explore_tool(branches: List[str], background: Optional[str] = None) -> str:
current_trace_id = get_current_trace_id()
current_goal_id = get_current_goal_id()
# 1. 创建 agent_call Goal
goal = Goal(
id=next_id(),
type="agent_call",
description=f"并行探索 {len(branches)} 个方案",
agent_call_mode="explore",
sub_trace_ids=[],
)
# 2. 为每个分支创建 Sub-Trace
sub_traces = []
for i, desc in enumerate(branches):
suffix = chr(ord('A') + i) # A, B, C...
sub_trace = Trace.create(
trace_id=f"{current_trace_id}.{suffix}",
parent_trace_id=current_trace_id,
parent_goal_id=current_goal_id,
agent_type="explore",
task=desc,
context=get_explore_context(), # 从配置获取权限设置
)
sub_traces.append(sub_trace)
goal.sub_trace_ids.append(sub_trace.trace_id)
# 3. 并行执行
results = await asyncio.gather(
*[run_agent(st, background) for st in sub_traces]
)
# 4. 收集元数据并汇总
sub_trace_metadata = {}
summary_parts = []
for sub_trace, result in zip(sub_traces, results):
# 获取 Sub-Trace 最新状态
updated_trace = await store.get_trace(sub_trace.trace_id)
# 获取最后一条 assistant 消息
messages = await store.get_messages(sub_trace.trace_id)
last_message = None
for msg in reversed(messages):
if msg.role == "assistant":
last_message = msg
break
# 构建元数据
sub_trace_metadata[sub_trace.trace_id] = {
"task": sub_trace.task,
"status": updated_trace.status if updated_trace else "unknown",
"summary": result.get("summary", "") if isinstance(result, dict) else "",
"last_message": {
"role": last_message.role,
"description": last_message.description,
"content": last_message.content[:500] if last_message.content else None,
"created_at": last_message.created_at.isoformat()
} if last_message else None,
"stats": {
"message_count": updated_trace.total_messages,
"total_tokens": updated_trace.total_tokens,
"total_cost": updated_trace.total_cost
}
}
# 组装摘要文本
summary_parts.append(f"### {sub_trace.task}")
summary_parts.append(result.get("summary", "执行完成"))
# 5. 更新 Goal,包含元数据
await store.update_goal(
current_trace_id,
current_goal_id,
status="completed",
summary=f"探索了 {len(branches)} 个方案",
sub_trace_metadata=sub_trace_metadata # 保存元数据
)
# 6. 返回文本摘要(给 LLM)
return "\n".join(summary_parts)
权限配置:
# 示例:系统配置中定义 Sub-Trace 的默认权限
# 可以根据 agent_type 或具体场景调整
def get_explore_context() -> Dict[str, Any]:
"""获取 explore 模式的 context 配置"""
return {
# 工具权限:根据实际需求配置
# 选项 1:受限权限(只读,快速探索)
# "allowed_tools": ["read", "grep", "glob"],
# 选项 2:完整权限(需要实际实现来评估)
# "allowed_tools": None, # None = 所有工具
# 步数限制
"max_turns": 20,
}
将大任务委托给独立的 Sub-Trace 执行。
@tool
def delegate(task: str) -> str:
"""
将大任务委托给独立的 sub-agent。
创建一个独立的 Sub-Trace(agent_type="delegate"),
有独立的 context,权限由配置决定。
返回:任务执行结果摘要
"""
内部实现:
async def delegate_tool(task: str) -> str:
current_trace_id = get_current_trace_id()
current_goal_id = get_current_goal_id()
# 1. 创建 agent_call Goal
goal = Goal(
id=next_id(),
type="agent_call",
description=f"委托任务: {task}",
agent_call_mode="delegate",
sub_trace_ids=[],
)
# 2. 创建 Sub-Trace
suffix = f"task{next_task_num()}" # task1, task2...
sub_trace = Trace.create(
trace_id=f"{current_trace_id}.{suffix}",
parent_trace_id=current_trace_id,
parent_goal_id=current_goal_id,
agent_type="delegate",
task=task,
context=get_delegate_context(), # 从配置获取权限设置
)
goal.sub_trace_ids.append(sub_trace.trace_id)
# 3. 执行
result = await run_agent(sub_trace)
# 4. 返回摘要
return result.summary
权限配置:
def get_delegate_context() -> Dict[str, Any]:
"""获取 delegate 模式的 context 配置"""
return {
# 工具权限:根据实际需求配置
# 通常 delegate 用于完整任务,给完整权限
"allowed_tools": None, # None = 所有工具
# 或者也可以限制:
# "allowed_tools": ["read", "write", "edit", "bash"],
# "denied_tools": ["危险工具"],
# 步数限制
"max_turns": 50,
}
注意:
explore 和 delegate 的权限配置是独立的,可以根据需求调整Trace.context 配置explore:可能受限(快速探索)或完整(需要实现验证)delegate:通常完整权限(执行完整任务)每次 LLM 调用时,在 system prompt 末尾注入当前计划状态。注入时过滤掉 abandoned 目标,使用连续的显示序号。
展示策略:
## Current Plan
**Mission**: 实现用户认证功能
**Current**: 2.2 实现登录接口
**Progress**:
[✓] 1. 分析代码
→ 用户模型在 models/user.py,使用 bcrypt 加密
[→] 2. 实现功能
[✓] 2.1 设计接口
→ API 设计文档完成,使用 REST 风格
[→] 2.2 实现登录接口 ← current
[ ] 2.3 实现注册接口
[ ] 3. 测试
(3 subtasks)
实现:agent/goal/models.py:GoalTree.to_prompt
注意:当前 focus 目标的所有父目标及其子目标都会完整展示,确保 LLM 理解当前上下文。这样在使用 after 或 under 参数时,LLM 可以准确引用目标 ID。
当调用 goal(done="...") 时:
两种模式:
模式 1:需要修改的计划还没有执行
直接修改计划并继续执行。Goal 状态为 pending 时,可以直接修改 description 或删除。
模式 2:需要修改的计划已经执行
abandoned(保留在 GoalTree 数据中,但 to_prompt() 不展示)Before 回溯:
GoalTree 数据:
[✓] 1. 分析代码 (内部ID: 1)
[→] 2. 实现方案 A (内部ID: 2)
[ ] 3. 测试 (内部ID: 3)
Messages:
[分析代码的 20 条 message...]
[实现方案 A 的 30 条 message...]
[测试失败的 message...]
After 回溯:
GoalTree 数据(含废弃):
[✓] 1. 分析代码 (内部ID: 1)
[✗] 2. 实现方案 A (内部ID: 2, abandoned)
[→] 3. 实现方案 B (内部ID: 4, 新建)
[ ] 4. 测试 (内部ID: 3)
to_prompt() 输出(给 LLM,连续编号):
[✓] 1. 分析代码
[→] 2. 实现方案 B ← current
[ ] 3. 测试
Messages:
[分析代码的 20 条 message...]
[Summary: "尝试方案 A,因依赖问题失败"] ← 原 messages 压缩为 1 条
[方案 B 第一条消息,包含废弃分支的 summary] ← 供 LLM 参考
[方案 B 的后续 message...]
实现:agent/goal/compaction.py
用户可以通过编辑 Goal 内的 goal() 工具调用来修改执行计划。
核心思路:利用现有的 GoalTree 结构和 Message sequence,无需额外追踪字段。
编辑流程:
废弃判断:
UI 展示:
API:
GET /api/traces/{trace_id}/messages?goal_id=X
# 返回 Goal X 的所有 messages
PATCH /api/traces/{trace_id}/replay
{
"from_message_id": "msg_123",
"modified_tool_call": {
"tool_call_id": "call_abc",
"new_arguments": {"add": "新任务列表", "under": "2"}
}
}
实现:agent/goal/replay.py
注意:这是最通用、最简单的编辑方式,完全基于现有数据结构。
可视化展示为 DAG(有向无环图),不是树。
核心概念:
展开/折叠:对边操作,对应目标的层级展开。
折叠视图(只看顶层 Goals):
[START] ──→ [1:分析完成] ──→ [2:实现完成] ──→ [3:测试完成]
逻辑边
展开 [1]→[2] 的边(显示 Goal 2 的子目标):
[START] ──→ [1:分析完成] ──→ [2.1:设计完成] ──→ [2.2:代码完成] ──→ [3:测试完成]
执行边 执行边
展开时,父节点 [2] 被子节点 [2.1], [2.2] 替代。 折叠时,子节点合并回父节点 [2]。
嵌套展开:如果 2.1 也有子目标,可以继续展开 [1]→[2.1] 的边。
废弃分支:在可视化中以灰色样式展示废弃分支。
[1:分析完成] ──→ [2:方案A(废弃)] ──→ ... ← 灰色
──→ [4:方案B] ──→ [3:测试] ← 正常
后端提供 GoalTree 数据,前端负责生成 DAG 视图。
REST 端点:
GET /api/traces/{trace_id} # 获取 Trace + GoalTree
GET /api/traces/{trace_id}/messages?goal_id=2.1 # 获取 Messages(边详情)
响应(GoalTree 部分):
{
"goal_tree": {
"mission": "实现用户认证功能",
"current_id": "4",
"goals": [
{
"id": "1",
"parent_id": null,
"branch_id": null,
"type": "normal",
"description": "分析代码",
"reason": "了解现有结构",
"status": "completed",
"summary": "用户模型在 models/user.py",
"self_stats": {"message_count": 5, "total_tokens": 2300, "total_cost": 0.03, "preview": "glob → read × 2"},
"cumulative_stats": {"message_count": 5, "total_tokens": 2300, "total_cost": 0.03, "preview": "glob → read × 2"}
},
{
"id": "2",
"parent_id": null,
"branch_id": null,
"type": "normal",
"description": "实现功能",
"reason": "核心任务",
"status": "in_progress",
"self_stats": {"message_count": 0, ...},
"cumulative_stats": {"message_count": 11, "total_tokens": 5700, ...}
},
{
"id": "3",
"parent_id": "2",
"branch_id": null,
"type": "normal",
"description": "设计接口",
"status": "completed",
"self_stats": {...}
},
{
"id": "4",
"parent_id": "2",
"branch_id": null,
"type": "normal",
"description": "实现代码",
"status": "in_progress",
"self_stats": {...}
},
{
"id": "5",
"parent_id": null,
"branch_id": null,
"type": "normal",
"description": "测试",
"status": "pending",
...
}
]
}
}
DAG 生成逻辑(前端实现):
cumulative_stats,展开用 self_stats)实现:见 frontend/API.md
ws://localhost:8000/api/traces/{trace_id}/watch?since_event_id=0
事件类型:
| 事件 | 触发时机 | payload |
|---|---|---|
connected |
WebSocket 连接成功 | trace_id, current_event_id, goal_tree(完整 GoalTree) |
goal_added |
新增 Goal | goal 完整数据(含 self_stats, cumulative_stats) |
goal_updated |
Goal 状态变化(含级联完成) | goal_id, updates(含 cumulative_stats),affected_goals |
message_added |
新 Message | message 数据(含 goal_id),affected_goals |
trace_completed |
执行完成 | 统计信息 |
事件详情:
connected - 连接时推送完整 GoalTree,前端据此初始化 DAG:
{
"event": "connected",
"trace_id": "xxx",
"current_event_id": 42,
"goal_tree": { "mission": "...", "goals": [...] }
}
message_added - 新 Message 时,后端更新统计并推送受影响的 Goals:
{
"event": "message_added",
"message": { "message_id": "...", "role": "assistant", "goal_id": "2.1", "..." : "..." },
"affected_goals": [
{ "goal_id": "2.1", "self_stats": {"message_count": 4, "total_tokens": 2000, "total_cost": 0.03, "preview": "read → edit × 2"}, "cumulative_stats": {"message_count": 4, "total_tokens": 2000, "total_cost": 0.03, "preview": "read → edit × 2"} },
{ "goal_id": "2", "cumulative_stats": {"message_count": 12, "total_tokens": 6800, "total_cost": 0.08, "preview": "glob → read → edit × 5 → bash"} }
]
}
affected_goals 包含该 Message 直接关联的 Goal(更新 self_stats + cumulative_stats)以及所有祖先 Goal(仅更新 cumulative_stats)。前端根据当前展开状态选择使用哪个 stats 渲染边。
goal_updated - Goal 状态变化时推送,包含级联完成场景:
{
"event": "goal_updated",
"goal_id": "2.1",
"updates": { "status": "completed", "summary": "接口设计完成" },
"affected_goals": [
{ "goal_id": "2.1", "cumulative_stats": {"message_count": 4, "total_tokens": 2000, "total_cost": 0.03, "preview": "read → edit × 2"} },
{ "goal_id": "2", "status": "completed", "summary": "功能实现完成", "cumulative_stats": {"message_count": 12, "total_tokens": 6800, "total_cost": 0.08, "preview": "..."} }
]
}
当所有子 Goal 完成时,后端自动级联完成父 Goal,并在 affected_goals 中包含所有状态变更的祖先。前端收到后直接更新对应节点,无需自行计算。
goal_added - 新增 Goal,携带完整 Goal 数据:
{
"event": "goal_added",
"goal": { "id": "2.1", "description": "设计接口", "reason": "需要先定义 API", "status": "pending", "self_stats": {}, "cumulative_stats": {} },
"parent_id": "2"
}
实现:agent/execution/websocket.py
.trace/
├── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d/ # 主 Trace
│ ├── meta.json # Trace 元数据
│ ├── goal.json # GoalTree
│ ├── messages/ # Messages
│ │ ├── {message_id}.json
│ │ └── ...
│ └── events.jsonl # 事件流
│
├── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001/ # Sub-Trace A(并行探索)
│ ├── meta.json # parent_trace_id: "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d"
│ ├── goal.json # 独立的 GoalTree
│ ├── messages/
│ └── events.jsonl
│
├── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002/ # Sub-Trace B(并行探索)
│ └── ...
│
└── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220030-001/ # Sub-Trace task1(单线委托)
└── ...
关键点:
parent_trace_id 可以追溯父子关系主 Trace (2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d):
[1] 分析问题
[2] 并行探索认证方案 (type=agent_call, mode=explore)
→ 启动 Sub-Traces(完整 ID)
[3] 完善实现
Sub-Trace A (2f8d3a1c...@explore-20260204220012-001):
[1] JWT 设计
[2] JWT 实现
→ 返回摘要:"JWT 方案实现完成"
Sub-Trace B (2f8d3a1c...@explore-20260204220012-002):
[1] Session 设计
[2] Session 实现
[3] Session 测试
→ 返回摘要:"Session 方案实现完成"
explore 工具返回:
汇总两个分支的结果,主 Agent 继续决策
核心原则:
sub_trace_ids 字段关联分支主 Trace 的 GoalTree:
# Trace: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d
goals = [
Goal(id="1", type="normal", description="分析问题"),
Goal(
id="2",
type="agent_call",
description="并行探索认证方案",
agent_call_mode="explore",
sub_trace_ids=[
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001",
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002"
],
),
Goal(id="3", type="normal", description="完善实现"),
]
Sub-Trace A 的 GoalTree(独立编号):
# Trace: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001
goals = [
Goal(id="1", type="normal", description="JWT 设计"),
Goal(id="2", type="normal", description="JWT 实现"),
]
Sub-Trace B 的 GoalTree(独立编号):
# Trace: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002
goals = [
Goal(id="1", type="normal", description="Session 设计"),
Goal(id="2", type="normal", description="Session 实现"),
Goal(id="3", type="normal", description="Session 测试"),
]
.trace/
├── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d/ # 主 Trace
│ ├── meta.json
│ ├── goal.json
│ └── messages/
│
├── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001/ # Sub-Trace A
│ ├── meta.json # parent_trace_id: "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d"
│ ├── goal.json # 独立的 GoalTree
│ └── messages/
│
└── 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002/ # Sub-Trace B
└── ...
折叠视图(Sub-Trace 作为单个节点):
[1:分析] ──→ [2:并行探索] ──→ [3:完善]
│
(启动2个Sub-Trace)
展开分支视图(显示并行路径):
┌──→ [A:JWT方案] ────┐
[1:分析] ──→ [2] ─┤ ├──→ [3:完善]
└──→ [B:Session方案] ┘
继续展开分支 A 内部(加载 Sub-Trace 的 GoalTree):
┌──→ [A.1:JWT设计] → [A.2:JWT实现] ──┐
[1:分析] ──→ [2] ─┤ ├──→ [3:完善]
└──→ [B:Session方案] ────────────────┘
注意:
[A:JWT方案] 是折叠视图,代表整个 Sub-Trace[A.1], [A.2] 是展开后显示 Sub-Trace 内部的 GoalsGET /api/traces/2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001/messages?goal_id=1REST:返回主 Trace 的 GoalTree + Sub-Trace 列表(元数据)。
GET /api/traces/2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d
响应:
{
"trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d",
"status": "running",
"goal_tree": {
"goals": [
{"id": "1", "type": "normal", "description": "分析问题"},
{
"id": "2",
"type": "agent_call",
"description": "并行探索认证方案",
"agent_call_mode": "explore",
"sub_trace_ids": [
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001",
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002"
],
"sub_trace_metadata": {
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001": {
"task": "JWT 方案",
"status": "completed",
"summary": "实现完成,使用 JWT token",
"last_message": {
"role": "assistant",
"description": "生成 JWT token 并返回",
"content": "JWT 方案实现完成。使用 jsonwebtoken 库生成 token,包含 user_id 和过期时间...",
"created_at": "2026-02-05T10:30:00"
},
"stats": {
"message_count": 8,
"total_tokens": 4000,
"total_cost": 0.05
}
},
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002": {
"task": "Session 方案",
"status": "completed",
"summary": "实现完成,使用 Redis 存储 session",
"last_message": {
"role": "assistant",
"description": "配置 Redis 并实现 session 管理",
"content": "Session 方案实现完成。使用 Redis 存储 session,支持过期自动清理...",
"created_at": "2026-02-05T10:32:00"
},
"stats": {
"message_count": 12,
"total_tokens": 4000,
"total_cost": 0.05
}
}
}
},
{"id": "3", "type": "normal", "description": "完善实现"}
]
},
"sub_traces": {
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001": {
"trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001",
"parent_trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d",
"parent_goal_id": "2",
"agent_type": "explore",
"task": "JWT 方案",
"status": "completed",
"total_messages": 8,
"total_tokens": 4000,
"total_cost": 0.05
},
"2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002": {
"trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-002",
"agent_type": "explore",
"task": "Session 方案",
"status": "completed",
...
}
}
}
按需加载 Sub-Trace 详情:
GET /api/traces/2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001
响应:
{
"trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001",
"parent_trace_id": "2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d",
"parent_goal_id": "2",
"agent_type": "explore",
"task": "JWT 方案",
"status": "completed",
"goal_tree": {
"goals": [
{"id": "1", "description": "JWT 设计", ...},
{"id": "2", "description": "JWT 实现", ...}
]
},
"total_tokens": 4000,
"total_cost": 0.05
}
WebSocket 事件:
| 事件 | 触发时机 | payload |
|---|---|---|
sub_trace_started |
Sub-Trace 开始 | trace_id, parent_trace_id, parent_goal_id, agent_type |
goal_added |
任何 Trace 新增 Goal | trace_id, goal |
message_added |
任何 Trace 新 Message | trace_id, message, affected_goals |
sub_trace_completed |
Sub-Trace 完成 | trace_id, summary, stats |
trace_completed |
主 Trace 完成 | trace_id, stats |
详细流程见前面"工具设计"部分,这里展示关键步骤:
async def explore_tool(branches: List[str]) -> str:
current_trace_id = "abc123"
current_goal_id = "2"
# 1. 创建 agent_call Goal
goal = Goal(
id="2",
type="agent_call",
description=f"并行探索 {len(branches)} 个方案",
agent_call_mode="explore",
sub_trace_ids=[],
)
# 2. 创建多个 Sub-Traces
sub_traces = []
for i, desc in enumerate(branches):
sub_trace = Trace.create(
trace_id=generate_sub_trace_id(current_trace_id, "explore"),
parent_trace_id=current_trace_id,
parent_goal_id=current_goal_id,
agent_type="explore",
task=desc,
context={"allowed_tools": ["read", "grep", "glob"]},
)
sub_traces.append(sub_trace)
goal.sub_trace_ids.append(sub_trace.trace_id)
# 3. 并行执行
results = await asyncio.gather(
*[run_agent(st) for st in sub_traces]
)
# 4. 汇总返回
return format_results(results)
汇总结果示例:
## 探索结果
### 方案 A: JWT 方案
实现完成。优点:无状态,易扩展。缺点:token 较大,无法主动失效。
### 方案 B: Session 方案
实现完成。优点:token 小,可主动失效。缺点:需要 Redis 存储。
---
两种方案都已实现,请选择一种继续。
Sub-Trace 完成后的压缩策略:
| 方面 | OpenCode | 我们的方案 |
|---|---|---|
| Plan 格式 | 纯文本 (plan.md) | 结构化 (GoalTree JSON) |
| Plan 与执行关联 | 无 | 通过 goal_id 关联 |
| 执行记录 | Message List | Message List(加 goal_id 元数据) |
| Sub-Agent 模型 | SubagentPart(嵌入式) | 独立 Trace(统一模型) |
| Sub-Agent ID | 复杂(需要前缀) | 简单(每个 Trace 独立编号) |
| 压缩时机 | 事后(context 满时) | 增量(goal 完成/放弃时) |
| 并行探索 | Sub-agent(手动管理) | explore 工具(自动汇总) |
| 权限控制 | 在 Sub-Agent 类型上 | 在 Trace.context 上(更灵活) |
| 回溯能力 | 有限 | 精确(基于 goal 压缩 + 废弃分支 summary) |
| 可视化 | 无 | DAG(边可展开/折叠) |
| 分布式支持 | 困难 | 自然支持(每个 Trace 独立) |
| 功能 | 文件路径 | 状态 |
|---|---|---|
| Trace 数据模型 | agent/execution/models.py |
待调整(增加父子关系、移除 branch_id) |
| Goal 数据模型 | agent/goal/models.py |
待调整(移除 branch_id、BranchContext) |
| Trace ID 生成 | agent/execution/trace_id.py |
待实现(generate_sub_trace_id) |
| goal 工具 | agent/goal/tool.py |
待调整(新增 after/under 参数) |
| explore 工具 | agent/goal/explore.py |
待实现 |
| delegate 工具 | agent/goal/delegate.py |
待实现 |
| Context 压缩 | agent/goal/compaction.py |
待调整 |
| TraceStore 协议 | agent/execution/protocols.py |
待调整(移除 branch 相关接口) |
| FileSystem Store | agent/execution/fs_store.py |
待调整(移除 branches/ 目录) |
| DAG 可视化 API | agent/execution/api.py |
待调整(支持 Sub-Trace) |
| WebSocket 推送 | agent/execution/websocket.py |
待调整(统一事件格式) |
| Plan 注入 | agent/core/runner.py |
待调整 |