Context 管理与执行计划

本文档描述 Agent 的 Context 管理、执行计划和 Sub-Agent 机制。

设计目标

自主长程执行：Agent 能独立执行复杂任务，无需人工频繁干预
有效的 Context 管理：长任务中保持关键信息，压缩次要细节
支持探索和回溯：能尝试多种方案，失败时能有效回溯
统一的 Agent 模型：主 Agent 和 Sub-Agent 使用相同的 Trace 结构
简单的工具接口：LLM 只需理解少量简单工具，复杂逻辑由系统处理

参考方案：OpenCode 的 Context 管理

核心架构

┌─────────────────┐
│   plan.md       │  ← 文本格式的计划（TODO 列表）
└─────────────────┘
         ↓
┌─────────────────┐
│  线性 Message   │  ← 对话历史
│     List        │
└─────────────────┘
         ↓
┌─────────────────┐
│  Prune + Full   │  ← 两阶段压缩
│   Compaction    │
└─────────────────┘
         ↓
┌─────────────────┐
│   Sub-Agent     │  ← 隔离大任务
└─────────────────┘

1. Message 管理

数据结构：

User Message: 用户输入，包含 TextPart, FilePart, CompactionPart, SubtaskPart 等
Assistant Message: LLM 输出，包含 TextPart, ToolPart, ReasoningPart 等
每个 Message 包含多个 Part，支持流式处理

存储：

Storage Key:
["message", sessionID, messageID] -> MessageV2.Info
["part", messageID, partID] -> MessageV2.Part

2. Context 压缩机制

两阶段压缩：

阶段 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 消息继续

3. Plan/Todo 机制

数据结构：

Todo.Info = {
  id: string
  content: string      // 任务描述
  status: string       // pending | in_progress | completed | cancelled
  priority: string     // high | medium | low
}

存储：文件系统（.opencode/plans/xxx.md）或 Storage

4. Sub-Agent 机制

Agent Mode：

primary: 主代理，执行工具
subagent: 子代理，独立 context，结果汇总回主会话

内置 Sub-Agents：

general: 通用代理，可并行执行多个任务
explore: 代码探索专用，仅允许查询工具
compaction: 上下文总结专用

Subtask 执行：

创建 SubtaskPart
子代理独立处理（独立 message list）
结果通过 "The following tool was executed by the user" 汇总

5. 优缺点分析

优点：

简单成熟，经过大量验证
Plan 和执行分离，用户可直接编辑 plan.md
Sub-agent 有效隔离大任务的 context

局限：

Plan 是纯文本，与执行记录无结构化关联
压缩是"事后"的，等满了再压缩
回溯能力有限，无法精确回到某个状态
不支持并行探索-合并的模式

我们的方案

核心思路

基于 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 作为折叠边显示        │
└─────────────────────────────────────────────┘

核心概念：统一的 Trace 模型

关键设计：每个 Agent（主 Agent 或 Sub-Agent）都是一个完整的 Trace。

# 主 Agent
Trace(trace_id="abc123", mode="agent", task="实现用户认证")

# Sub-Agent（并行探索）
Trace(trace_id="abc123.A", parent_trace_id="abc123", agent_type="explore", task="JWT 方案")
Trace(trace_id="abc123.B", parent_trace_id="abc123", agent_type="explore", task="Session 方案")

# Sub-Agent（单线委托）
Trace(trace_id="abc123.task1", parent_trace_id="abc123", agent_type="delegate", task="实现具体功能")

优点：

概念统一：不需要额外的 SubAgentContext/BranchContext 概念
ID 简洁：每个 Trace 内部独立编号（1, 2, 3），不需要前缀
完全隔离：每个 Trace 有独立的 GoalTree、Message List、LLM Context
自然分布式：每个 Trace 可以独立运行、迁移、存储
层级清晰：从 trace_id 可以直接解析出父子关系（abc123.A 的父是 abc123）

数据结构

后端存储两类数据，可视化的 DAG 是派生视图：

GoalTree：层级目标，注入 LLM
Messages：执行记录，通过 goal_id 关联 Goal

不存在独立的"边"数据结构，边在可视化时从 Messages 聚合生成。

Trace

@dataclass
class Trace:
    """
    执行轨迹 - 一次完整的 Agent 运行

    主 Agent 和 Sub-Agent 使用相同的 Trace 结构
    """
    trace_id: str                        # 层级化 ID："abc123", "abc123.A", "abc123.A.1"
    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：短随机 ID（如 "abc123"）
Sub-Trace：parent_id.suffix（如 "abc123.A", "abc123.task1"）
嵌套 Sub-Trace：继续追加（如 "abc123.A.1", "abc123.A.task2"）

从 trace_id 解析父子关系：

def parse_parent_trace_id(trace_id: str) -> Optional[str]:
    """从 trace_id 解析出 parent_trace_id"""
    parts = trace_id.rsplit('.', 1)
    return parts[0] if len(parts) > 1 else None

# 示例
parse_parent_trace_id("abc123.A")    # → "abc123"
parse_parent_trace_id("abc123.A.1")  # → "abc123.A"
parse_parent_trace_id("abc123")      # → None

Goal

@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"

    # 统计（后端维护，用于可视化边的数据）
    self_stats: GoalStats                # 自身统计（仅直接关联的 messages）
    cumulative_stats: GoalStats          # 累计统计（自身 + 所有后代）

实现：agent/goal/models.py:Goal

ID 设计：

内部 ID：每个 Trace 独立编号（"1", "2", "3"），简单自增
层级关系：通过 parent_id 字段维护
显示序号：to_prompt() 时动态生成连续有意义的编号（"1", "2", "2.1", "2.2"...）

统计更新逻辑：

每次添加 Message 时，更新对应 Goal 的 self_stats，并沿祖先链向上更新所有祖先的 cumulative_stats
可视化中，折叠边使用 target Goal 的 cumulative_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

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: str                         # 关联的 Goal 内部 ID（如 "1", "2"）
    tool_call_id: Optional[str]          # tool 消息关联对应的 tool_call
    content: Any                         # 消息内容（和 LLM API 格式一致）
    description: str                     # 消息描述（系统自动生成）

    # 元数据
    tokens: Optional[int] = None
    cost: Optional[float] = None
    created_at: datetime

description 字段（系统自动生成）：

assistant 消息：优先取 content 中的 text，若无 text 则生成 "tool call: XX, XX"
tool 消息：使用 tool name

实现：agent/execution/models.py:Message

Message 类型说明：

role="assistant"：模型的一次返回，可能同时包含文本和多个 tool_calls
role="tool"：一个工具的执行结果，通过 tool_call_id 关联对应的 tool_call

查询 Message：

# 查询主 Trace 的 Messages
GET /api/traces/abc123/messages?goal_id=2

# 查询 Sub-Trace 的 Messages
GET /api/traces/abc123.A/messages?goal_id=1

工具设计

goal 工具：计划管理

@tool
def goal(
    add: Optional[str] = None,       # 添加目标（逗号分隔多个）
    done: Optional[str] = None,      # 完成当前目标，值为 summary
    abandon: Optional[str] = None,   # 放弃当前目标，值为原因
    focus: Optional[str] = None,     # 切换焦点到指定显示序号
) -> str:
    """管理执行计划。"""

实现：agent/goal/tool.py:goal_tool

层级支持：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)
                        │              (级联：若所有兄弟都 completed，父 goal 自动 completed)
                        │
                     abandon
                        ↓
                   abandoned
                        ↓
                  (压缩 context)

explore 工具：并行探索

启动多个独立的 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. 汇总返回
    summary = format_explore_results(results)
    return summary

权限配置：

# 示例：系统配置中定义 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,
    }

delegate 工具：单线委托

将大任务委托给独立的 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：通常完整权限（执行完整任务）
但这些都不是固定的，可以根据具体场景调整

Context 管理

1. Plan 注入

每次 LLM 调用时，在 system prompt 末尾注入当前计划状态。注入时过滤掉 abandoned 目标，使用连续的显示序号：

## Current Plan

**Mission**: 实现用户认证功能
**Current**: 2.2 实现登录接口

**Progress**:
[✓] 1. 分析代码
    → 用户模型在 models/user.py，使用 bcrypt 加密
[→] 2. 实现功能
    [✓] 2.1 设计接口
    [→] 2.2 实现登录接口  ← current
    [ ] 2.3 实现注册接口
[ ] 3. 测试

实现：agent/goal/models.py:GoalTree.to_prompt

2. 完成时压缩

当调用 goal(done="...") 时：

找到该 goal 关联的所有 messages（通过 goal_id）
将详细 messages 替换为一条 summary message
更新 goal 状态为 completed

3. 回溯（Abandon）

两种模式：

模式 1：需要修改的计划还没有执行

直接修改计划并继续执行。Goal 状态为 pending 时，可以直接修改 description 或删除。

模式 2：需要修改的计划已经执行

将原 Goal 标记为 abandoned（保留在 GoalTree 数据中，但 to_prompt() 不展示）
将废弃分支关联的 messages 做 summary
将 summary 累积到新分支的第一条消息中（供 LLM 参考历史失败原因）
创建新的 Goal 继续执行

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

可视化

DAG 模型

可视化展示为 DAG（有向无环图），不是树。

核心概念：

节点 = Goal 完成后的结果/里程碑
边 = 从一个结果到下一个结果的执行过程（动作/策略）
每个节点对应一条入边，入边的数据从该 Goal 关联的 Messages 聚合

展开/折叠：对边操作，对应目标的层级展开。

折叠视图（只看顶层 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:测试]       ← 正常

API

后端提供 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 生成逻辑（前端实现）：

根据用户展开状态，确定可见 Goal 序列
相邻 Goal 之间形成边
边的统计数据从 target Goal 的 stats 获取（折叠用 cumulative_stats，展开用 self_stats）
边的详细内容通过 Messages API 查询

实现：见 frontend/API.md

WebSocket 实时推送

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/
├── abc123/                    # 主 Trace
│   ├── meta.json              # Trace 元数据
│   ├── goal.json              # GoalTree
│   ├── messages/              # Messages
│   │   ├── {message_id}.json
│   │   └── ...
│   └── events.jsonl           # 事件流
│
├── abc123.A/                  # Sub-Trace A（并行探索）
│   ├── meta.json              # parent_trace_id: "abc123"
│   ├── goal.json              # 独立的 GoalTree
│   ├── messages/
│   └── events.jsonl
│
├── abc123.B/                  # Sub-Trace B（并行探索）
│   └── ...
│
└── abc123.task1/              # Sub-Trace task1（单线委托）
    └── ...

关键点：

每个 Trace（主/子）都是完全独立的目录
从 trace_id 可以直接找到对应目录
通过 parent_trace_id 可以追溯父子关系

并行探索设计（explore 工具）

场景

主 Trace (abc123):
  [1] 分析问题
  [2] 并行探索认证方案 (type=agent_call, mode=explore)
      → 启动 Sub-Traces: abc123.A, abc123.B
  [3] 完善实现

Sub-Trace A (abc123.A):
  [1] JWT 设计
  [2] JWT 实现
  → 返回摘要："JWT 方案实现完成"

Sub-Trace B (abc123.B):
  [1] Session 设计
  [2] Session 实现
  [3] Session 测试
  → 返回摘要："Session 方案实现完成"

explore 工具返回:
  汇总两个分支的结果，主 Agent 继续决策

核心原则：

每个分支是独立的 Trace，有自己的 GoalTree 和 Message List
分支内的 Goal ID 简单编号（"1", "2", "3"），独立于主 Trace
主 Trace 的 Goal 通过 sub_trace_ids 字段关联分支
分支完全独立存储，可以并行运行
explore 工具自动汇总各分支 summary

数据结构

主 Trace 的 GoalTree：

# Trace: abc123
goals = [
    Goal(id="1", type="normal", description="分析问题"),
    Goal(
        id="2",
        type="agent_call",
        description="并行探索认证方案",
        agent_call_mode="explore",
        sub_trace_ids=["abc123.A", "abc123.B"],
    ),
    Goal(id="3", type="normal", description="完善实现"),
]

Sub-Trace A 的 GoalTree（独立编号）：

# Trace: abc123.A
goals = [
    Goal(id="1", type="normal", description="JWT 设计"),
    Goal(id="2", type="normal", description="JWT 实现"),
]

Sub-Trace B 的 GoalTree（独立编号）：

# Trace: abc123.B
goals = [
    Goal(id="1", type="normal", description="Session 设计"),
    Goal(id="2", type="normal", description="Session 实现"),
    Goal(id="3", type="normal", description="Session 测试"),
]

存储结构

.trace/
├── abc123/                # 主 Trace
│   ├── meta.json
│   ├── goal.json
│   └── messages/
│
├── abc123.A/              # Sub-Trace A
│   ├── meta.json          # parent_trace_id: "abc123", parent_goal_id: "2"
│   ├── goal.json          # 独立的 GoalTree
│   └── messages/
│
└── abc123.B/              # Sub-Trace B
    └── ...

DAG 可视化

折叠视图（Sub-Trace 作为单个节点）：

[1:分析] ──→ [2:并行探索] ──→ [3:完善]
                 │
          (启动2个Sub-Trace)

展开分支视图（显示并行路径）：

                  ┌──→ [A:JWT方案] ────┐
[1:分析] ──→ [2] ─┤                    ├──→ [3:完善]
                  └──→ [B:Session方案] ┘

继续展开分支 A 内部（加载 Sub-Trace abc123.A 的 GoalTree）：

                  ┌──→ [A.1:JWT设计] → [A.2:JWT实现] ──┐
[1:分析] ──→ [2] ─┤                                    ├──→ [3:完善]
                  └──→ [B:Session方案] ────────────────┘

注意：

[A:JWT方案] 是折叠视图，代表整个 Sub-Trace abc123.A
[A.1], [A.2] 是展开后显示 Sub-Trace 内部的 Goals
前端显示为 "A.1"，但后端查询是 GET /api/traces/abc123.A/messages?goal_id=1

前端 API

REST：返回主 Trace 的 GoalTree + Sub-Trace 列表（元数据）。

GET /api/traces/abc123

响应：

{
  "trace_id": "abc123",
  "status": "running",
  "goal_tree": {
    "goals": [
      {"id": "1", "type": "normal", "description": "分析问题"},
      {
        "id": "2",
        "type": "agent_call",
        "description": "并行探索认证方案",
        "agent_call_mode": "explore",
        "sub_trace_ids": ["abc123.A", "abc123.B"]
      },
      {"id": "3", "type": "normal", "description": "完善实现"}
    ]
  },
  "sub_traces": {
    "abc123.A": {
      "trace_id": "abc123.A",
      "parent_trace_id": "abc123",
      "parent_goal_id": "2",
      "agent_type": "explore",
      "task": "JWT 方案",
      "status": "completed",
      "total_messages": 8,
      "total_tokens": 4000,
      "total_cost": 0.05
    },
    "abc123.B": {
      "trace_id": "abc123.B",
      "agent_type": "explore",
      "task": "Session 方案",
      "status": "completed",
      ...
    }
  }
}

按需加载 Sub-Trace 详情：

GET /api/traces/abc123.A

响应：

{
  "trace_id": "abc123.A",
  "parent_trace_id": "abc123",
  "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

explore 工具执行流程

详细流程见前面"工具设计"部分，这里展示关键步骤：

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):
        suffix = chr(ord('A') + i)
        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={"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 (abc123.A): JWT 方案
实现完成。优点：无状态，易扩展。缺点：token 较大，无法主动失效。

### 方案 B (abc123.B): Session 方案
实现完成。优点：token 小，可主动失效。缺点：需要 Redis 存储。

---
两种方案都已实现，请选择一种继续。

Context 压缩

Sub-Trace 完成后的压缩策略：

Sub-Trace 完成时：Sub-Trace 的详细 context 压缩为 summary（存储在 Trace.summary）
explore 完成后：所有 Sub-Traces 的 summary 汇总到主 Trace 的 tool result
主 Trace context：explore 工具调用被压缩为一条包含汇总结果的 tool message

与 OpenCode 方案的对比

方面	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`	待调整（增加父子关系、context）
Goal 数据模型	`agent/goal/models.py`	待调整（简化，移除 branch_id）
goal 工具	`agent/goal/tool.py`	待调整
explore 工具	`agent/goal/explore.py`	待实现
delegate 工具	`agent/goal/delegate.py`	待实现
Trace ID 生成	`agent/execution/trace_id.py`	待实现
Context 压缩	`agent/goal/compaction.py`	待调整
TraceStore 协议	`agent/execution/protocols.py`	待调整（支持嵌套 Trace）
DAG 可视化 API	`agent/execution/api.py`	待调整（支持 Sub-Trace）
WebSocket 推送	`agent/execution/websocket.py`	待调整（统一事件格式）
Plan 注入	`agent/core/runner.py`	待调整

渐进式实现计划

Phase 1: 基础 goal 工具

GoalTree 数据结构（含 ID 映射）
goal 工具（add, done, focus）
Plan 注入到 system prompt（含显示序号生成）
Message 模型（替代 Step）

Phase 2: 回溯支持

abandon 操作（两种模式）
废弃分支 summary 累积到新分支
基于 goal 的 context 压缩

Phase 3: 可视化

DAG 视图 API
WebSocket goal/message 事件
展开/折叠逻辑

Phase 4: 并行探索

explore 工具
独立 message list 管理
结果汇总机制

context-management.md 38 KB Lịch sử Raw

Context 管理与执行计划

设计目标

参考方案：OpenCode 的 Context 管理

核心架构

1. Message 管理

2. Context 压缩机制

3. Plan/Todo 机制

4. Sub-Agent 机制

5. 优缺点分析

我们的方案

核心思路

架构

核心概念：统一的 Trace 模型

数据结构

Trace

Goal

Message

工具设计

goal 工具：计划管理

explore 工具：并行探索

delegate 工具：单线委托

Context 管理

1. Plan 注入

2. 完成时压缩

3. 回溯（Abandon）

可视化

DAG 模型

API

WebSocket 实时推送

存储结构

并行探索设计（explore 工具）

场景

数据结构

存储结构

DAG 可视化

前端 API

explore 工具执行流程

Context 压缩

与 OpenCode 方案的对比

实现位置

渐进式实现计划

Phase 1: 基础 goal 工具

Phase 2: 回溯支持

Phase 3: 可视化

Phase 4: 并行探索

context-management.md 38 KB

Lịch sử Raw