AutoGPT 子 Agent 派生架构重构：ExecutionContext、资源预算与 LATS 树搜索策略实现

本文基于 AutoGPT classic 仓库中的子 Agent 派生架构重构计划（SUB_AGENT_REFACTOR_PLAN.md），完整讲解如何为 Prompt 策略注入执行上下文（ExecutionContext），使其能够生成、调度并回收子 Agent，覆盖资源预算、AgentFactory 协议、策略接口扩展、Agent 集成以及 LATS 示例策略等核心设计，并逐一对照当前仓库中已落地的源码实现，帮助读者理解这套多 Agent 协调机制从设计到实现的完整脉络。

一、问题背景：策略为何无法生成子 Agent

重构计划的目标是让 prompt 策略（Prompt Strategy）能够派生（spawn）并协调子 Agent，从而支撑四类高级模式：

• LATS（Language Agent Tree Search）—— 并行探索多个推理分支；
• 多 Agent 辩论（Multi-agent debate）—— 通过 Agent 间交互达成共识；
• 分层分解（Hierarchical decomposition）—— 将子任务委派给专用 Agent；
• Agent 即工具（Agent-as-tool）—— 像调用函数一样调用 Agent。

重构前的瓶颈在于：策略（Strategy）只负责"构建 prompt → 调用 LLM → 解析响应"，完全没有以下任何资源的访问权：

• Agent 工厂（Agent factory）
• LLM 提供商（LLM provider）
• 文件存储（File storage）
• 执行上下文（Execution context）
• 其他 Agent

计划文档给出的重构前主循环示意如下：

┌─────────────────────────────────────────────────────────────────┐
│                         Main Loop                                │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │ while running:                                           │    │
│  │   prompt = strategy.build_prompt(messages, task, ...)   │    │
│  │   response = llm.call(prompt)                           │    │
│  │   proposal = strategy.parse_response(response)          │    │
│  │   result = agent.execute(proposal)  ← tools only        │    │
│  └─────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────┘

Strategy has NO access to:
- Agent factory
- LLM provider
- File storage
- Execution context
- Other agents

重构后的目标架构则引入一个贯穿整个 Agent 层级的执行上下文：

┌─────────────────────────────────────────────────────────────────┐
│                      Execution Context                           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Agent Factory│  │ LLM Provider │  │ File Storage │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│         │                 │                  │                  │
│         └─────────────────┼──────────────────┘                  │
│                           ▼                                      │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    Parent Agent                          │
│  │  ┌─────────────────────────────────────────────────┐    │    │
│  │  │              Prompt Strategy                     │    │    │
│  │  │  - Has access to ExecutionContext               │    │    │
│  │  │  - Can spawn sub-agents via context             │    │    │
│  │  │  - Can await sub-agent results                  │    │    │
│  │  └─────────────────────────────────────────────────┘    │    │
│  │           │                                               │    │
│  │           ▼             ▼             ▼                   │    │
│  │    ┌───────────┐ ┌───────────┐ ┌───────────┐             │    │
│  │    │ SubAgent1 │ │ SubAgent2 │ │ SubAgent3 │             │    │
│  │    │ (searcher)│ │ (analyzer)│ │ (coder)   │             │    │
│  │    └───────────┘ └───────────┘ └───────────┘             │    │
│  └─────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────┘

值得强调的是，这套设计并非停留在纸面：当前仓库中已能看到完整的落地实现，包括 execution_context.py、策略基类中的子 Agent 方法（prompt_strategies/base.py）、Agent 集成代码 以及 LATS 策略实现（下文以 classic/original_autogpt/autogpt/agents/prompt_strategies/lats.py 为准）。

二、Phase 1：核心基础设施

2.1 ExecutionContext 模型

计划将执行上下文放在 forge/agent/execution_context.py（即仓库中的 classic/forge/forge/agent/execution_context.py），由三个核心类型组成。

（1）ResourceBudget —— 层级化资源预算

class ResourceBudget(BaseModel):
    """Resource limits for an agent and its children."""

    max_tokens: Optional[int] = None
    max_cycles: Optional[int] = None
    max_sub_agents: int = 10
    max_depth: int = 3  # Nesting depth limit
    deadline: Optional[datetime] = None

    def remaining_time(self) -> Optional[timedelta]:
        if self.deadline:
            return self.deadline - datetime.now()
        return None

    def create_child_budget(self, fraction: float = 0.5) -> "ResourceBudget":
        """Create a budget for a child agent."""
        return ResourceBudget(
            max_tokens=int(self.max_tokens * fraction) if self.max_tokens else None,
            max_cycles=int(self.max_cycles * fraction) if self.max_cycles else None,
            max_sub_agents=max(1, self.max_sub_agents // 2),
            max_depth=self.max_depth - 1,
            deadline=self.deadline,
        )

预算的关键设计是 create_child_budget：每向下传递一层，token/周期预算按比例缩减、可派生数量减半、嵌套深度减一，从机制上杜绝了子 Agent 无限自我繁殖的"递归炸弹"。

（2）SubAgentHandle —— 子 Agent 句柄

class SubAgentHandle(BaseModel):
    """Reference to a spawned sub-agent."""

    agent_id: str
    task: str
    status: str = "pending"  # pending, running, completed, failed, cancelled
    result: Optional[Any] = None
    error: Optional[str] = None

    # Internal (excluded from serialization)
    _agent: Optional["BaseAgent"] = None
    _task: Optional[asyncio.Task] = None

句柄是父 Agent 观察子 Agent 的唯一窗口：状态在 pending → running → completed / failed / cancelled 间流转，内部字段（下划线前缀）保存真正的 Agent 实例与 asyncio.Task，不进入序列化输出。

（3）ExecutionContext —— 沿层级传递的上下文容器

@dataclass
class ExecutionContext:
    """Context passed down the agent hierarchy."""

    # Core dependencies
    llm_provider: "MultiProvider"
    file_storage: "FileStorage"

    # Agent factory function
    agent_factory: "AgentFactory"

    # Hierarchy tracking
    parent_agent_id: Optional[str] = None
    depth: int = 0

    # Resource management
    budget: ResourceBudget = field(default_factory=ResourceBudget)

    # Active sub-agents
    sub_agents: dict[str, SubAgentHandle] = field(default_factory=dict)

    # Cancellation
    cancelled: bool = False

    def can_spawn_sub_agent(self) -> bool:
        """Check if spawning another sub-agent is allowed."""
        if self.cancelled:
            return False
        if self.budget.max_depth <= 0:
            return False
        if len(self.sub_agents) >= self.budget.max_sub_agents:
            return False
        if self.budget.deadline and datetime.now() >= self.budget.deadline:
            return False
        return True

    def create_child_context(self, child_agent_id: str) -> "ExecutionContext":
        """Create a context for a child agent."""
        return ExecutionContext(
            llm_provider=self.llm_provider,
            file_storage=self.file_storage,
            agent_factory=self.agent_factory,
            parent_agent_id=child_agent_id,
            depth=self.depth + 1,
            budget=self.budget.create_child_budget(),
        )

    async def cancel_all_sub_agents(self):
        """Cancel all running sub-agents."""
        self.cancelled = True
        for handle in self.sub_agents.values():
            if handle._task and not handle._task.done():
                handle._task.cancel()
                handle.status = "cancelled"

三道准入门槛（已取消、深度耗尽、并发数超限、超时）集中在 can_spawn_sub_agent() 中判定；create_child_context() 则负责沿层级复制共享依赖并递减预算。

落地实现对照。 当前仓库中的 ExecutionContext 实现 与计划高度一致，但有几处可见的演进（从源码实现看）：

维度	计划稿	仓库当前实现
状态表示	字符串（"pending" 等）	独立枚举 SubAgentStatus（execution_context.py）
预算默认值	max_sub_agents=10, max_depth=3	max_depth=5, max_sub_agents=25, max_cycles_per_agent=50, max_tokens_total=0（0 表示不限），即"宽松默认"（ResourceBudget）
时间约束	预算内嵌 deadline	取消 deadline，改为运行时超时参数 sub_agent_timeout_seconds 在 run_sub_agent 层强制执行
文件隔离	子 Agent 直接复用父级 file_storage	子 Agent 通过 clone_with_subroot(".sub_agents/{agent_id}") 获得写受限存储（_create_child_storage）
权限继承	未涉及	预算中携带 inherited_deny_rules（继承父级拒绝规则，始终生效）与 explicit_allow_rules（显式允许规则逐层重置为空）（ResourceBudget.create_child_budget）

其中"子 Agent 可读取父级工作区、只能写入自己的子目录"正是计划文档 Open Questions 中"Shared State"一题的落地答案；_create_child_storage 的注释还坦承当前实现对读写的限制是"全部限制在子根目录"的过渡方案，说明文档与实现之间是持续迭代的关系。

2.2 AgentFactory 协议

工厂负责屏蔽"具体创建哪种 Agent"的细节，使策略无需知道 Agent 的构造细节即可派生：

class AgentFactory(Protocol):
    """Protocol for creating agents."""

    def create_agent(
        self,
        agent_id: str,
        task: str,
        context: ExecutionContext,
        ai_profile: Optional[AIProfile] = None,
        directives: Optional[AIDirectives] = None,
        strategy: Optional[str] = None,
    ) -> "BaseAgent":
        """Create a new agent instance."""
        ...


class DefaultAgentFactory:
    """Default implementation of AgentFactory."""

    def __init__(self, app_config: "AppConfig"):
        self.app_config = app_config

    def create_agent(
        self,
        agent_id: str,
        task: str,
        context: ExecutionContext,
        ai_profile: Optional[AIProfile] = None,
        directives: Optional[AIDirectives] = None,
        strategy: Optional[str] = None,
    ) -> "BaseAgent":
        from autogpt.agent_factory.configurators import create_agent_state
        from autogpt.agents.agent import Agent

        # Use provided or default profile/directives
        ai_profile = ai_profile or AIProfile(ai_name=f"SubAgent-{agent_id[:8]}")
        directives = directives or AIDirectives()

        # Create state
        state = create_agent_state(
            agent_id=agent_id,
            task=task,
            ai_profile=ai_profile,
            directives=directives,
            app_config=self.app_config,
        )

        # Override strategy if specified
        config = self.app_config.model_copy()
        if strategy:
            config.prompt_strategy = strategy

        return Agent(
            settings=state,
            llm_provider=context.llm_provider,
            file_storage=context.file_storage,
            app_config=config,
            execution_context=context,  # NEW: pass context
        )

两个设计要点：其一，ai_profile / directives / strategy 三个可选覆盖项让父 Agent 可以为不同职责的"专家"子 Agent 定制人格与策略（例如让"coder"子 Agent 用另一种 prompt 策略）；其二，最终构造 Agent 时把 execution_context 一路传下去——这正是子 Agent 继续向下派生（在预算允许时）的递归基础。

在仓库实现中，AgentFactory 被定义为 typing.Protocol（AgentFactory Protocol），与计划稿一致；而 ExecutionContext.agent_factory 字段设为可选（Optional[AgentFactory] = None），并加入了"工厂缺失则禁止派生"的防御性检查（can_spawn_sub_agent），比计划稿更稳妥。

三、Phase 2：策略接口更新

3.1 新增子 Agent 配置项

BasePromptStrategyConfiguration 增加三个 UserConfigurable 字段，使其可通过配置文件暴露给用户：

class BasePromptStrategyConfiguration(SystemConfiguration):
    # ... existing fields ...

    # Sub-agent configuration
    enable_sub_agents: bool = UserConfigurable(default=False)
    max_sub_agents: int = UserConfigurable(default=5)
    sub_agent_timeout_seconds: int = UserConfigurable(default=300)

对照仓库中的 实际配置定义，当前实现在保留上述三项的同时又补充了一个 sub_agent_max_cycles（默认 25，限制单个子 Agent 的"提案-执行"循环次数），且 enable_sub_agents 的默认值从计划稿的 False 调整为 True。计划稿的"默认关闭"是迁移期的保守选择，而实现阶段说明系统已被认为足够稳定，可直接默认开启。

3.2 核心方法：从派生到并行执行

BaseMultiStepPromptStrategy 被扩展出完整的子 Agent 生命周期方法族。核心调用链为：set_execution_context() 注入上下文 → can_spawn_sub_agent() 准入检查 → spawn_sub_agent() 创建句柄 → run_sub_agent() 驱动执行循环 → spawn_and_run() / run_parallel() 提供组合糖。

（1）上下文注入与准入检查

def set_execution_context(self, context: ExecutionContext) -> None:
    """Inject the execution context. Called by Agent after creation."""
    self._execution_context = context

def can_spawn_sub_agent(self) -> bool:
    """Check if this strategy can spawn sub-agents."""
    if not self.config.enable_sub_agents:
        return False
    if not self._execution_context:
        return False
    return self._execution_context.can_spawn_sub_agent()

注意 can_spawn_sub_agent 做了两层判断：策略配置层（enable_sub_agents）+ 执行上下文层（预算/深度/并发），前者是用户意图，后者是运行时资源约束。

（2）spawn_sub_agent —— 派生子 Agent

async def spawn_sub_agent(
    self,
    task: str,
    ai_profile: Optional[AIProfile] = None,
    directives: Optional[AIDirectives] = None,
    strategy: Optional[str] = None,
) -> SubAgentHandle:
    if not self.can_spawn_sub_agent():
        raise RuntimeError("Cannot spawn sub-agent: disabled or limit reached")

    ctx = self._execution_context
    agent_id = f"sub-{uuid4().hex[:8]}"

    # Create child context
    child_ctx = ctx.create_child_context(agent_id)

    # Create the sub-agent
    sub_agent = ctx.agent_factory.create_agent(
        agent_id=agent_id,
        task=task,
        context=child_ctx,
        ai_profile=ai_profile,
        directives=directives,
        strategy=strategy,
    )

    # Create handle
    handle = SubAgentHandle(agent_id=agent_id, task=task, status="pending")
    handle._agent = sub_agent

    # Track in context
    ctx.sub_agents[agent_id] = handle
    return handle

实现版（spawn_sub_agent）在此骨架上做了工程化加固：ID 改用层级化的 generate_sub_agent_id(parent_id)（形如 {parent_id}-sub-{8位uuid}，见 generate_sub_agent_id）；工厂调用包在 try/except 中，创建失败时将句柄标记为 FAILED 并记录 error，而不是直接抛出——派生失败是"可观测的状态"而非"不可见的异常"，这是异步多 Agent 系统中错误处理的关键细节。

（3）run_sub_agent —— 驱动子 Agent 执行循环

async def run_sub_agent(
    self,
    handle: SubAgentHandle,
    max_cycles: Optional[int] = None,
) -> Any:
    if handle._agent is None:
        raise RuntimeError("Sub-agent not initialized")

    agent = handle._agent
    handle.status = "running"
    cycles = 0
    max_cycles = max_cycles or self.config.max_sub_agent_cycles

    try:
        while cycles < max_cycles:
            # Check for cancellation
            if self._execution_context and self._execution_context.cancelled:
                handle.status = "cancelled"
                return None

            # Propose and execute
            proposal = await agent.propose_action()

            # Check for finish command
            if proposal.use_tool.name == "finish":
                handle.status = "completed"
                handle.result = proposal.use_tool.arguments.get("reason", "")
                return handle.result

            # Execute the action
            result = await agent.execute(proposal)
            cycles += 1

        # Max cycles reached
        handle.status = "completed"
        handle.result = "Max cycles reached"
        return handle.result

    except Exception as e:
        handle.status = "failed"
        handle.error = str(e)
        raise

这里的语义与 BaseAgent 的两个抽象方法（propose_action 与 execute）完全对齐：子 Agent 并不是特殊物种，就是被父策略"以协程方式驱动"的普通 Agent；以 finish 命令的 reason 参数作为返回值，使"Agent-as-tool"模式有了确定的输出契约。

实现版（run_sub_agent / _run_agent_loop）用 asyncio.wait_for(..., timeout=sub_agent_timeout_seconds) 包裹整个执行循环，把计划稿中"deadline 属于预算"的职责移到了运行层：超时返回 None 并将状态置为 FAILED，CancelledError 则单独归类为 CANCELLED——三种失败形态（超时/取消/异常）有了明确区分。

（4）spawn_and_run 与 run_parallel —— 组合模式

async def spawn_and_run(
    self,
    task: str,
    ai_profile: Optional[AIProfile] = None,
    directives: Optional[AIDirectives] = None,
    strategy: Optional[str] = None,
    max_cycles: Optional[int] = None,
) -> Any:
    """Convenience: spawn and immediately run a sub-agent."""
    handle = await self.spawn_sub_agent(task, ai_profile, directives, strategy)
    return await self.run_sub_agent(handle, max_cycles)


async def run_parallel(
    self,
    tasks: list[str],
    strategy: Optional[str] = None,
    max_cycles: Optional[int] = None,
) -> list[Any]:
    """Run multiple sub-agents in parallel."""
    handles = []
    for task in tasks:
        handle = await self.spawn_sub_agent(task, strategy=strategy)
        handles.append(handle)

    # Run all in parallel
    coros = [self.run_sub_agent(h, max_cycles) for h in handles]
    results = await asyncio.gather(*coros, return_exceptions=True)
    return results

run_parallel 是"多 Agent 辩论"与"LATS 并行分支"的直接支撑：先串行派生全部句柄（受 max_sub_agents 预算约束），再用 asyncio.gather 并发驱动。实现版还额外提供了 get_sub_agent_results()，只向父 Agent 汇总已完成子 Agent 的结果摘要——这回应了计划文档 Open Questions 中"History"一题的取舍：父级只见结果，不见子 Agent 的完整行动历史（这一点也写进了 ExecutionContext 的 docstring 设计决策第 4 条）。

四、Phase 3：Agent 集成

4.1 Agent 构造函数接入执行上下文

计划要求 Agent.__init__ 新增可选的 execution_context 参数，形成"子 Agent 复用父级上下文、根 Agent 自建上下文"的双分支：

class Agent(BaseAgent[AnyActionProposal], Configurable[AgentSettings]):

    def __init__(
        self,
        settings: AgentSettings,
        llm_provider: MultiProvider,
        file_storage: FileStorage,
        app_config: AppConfig,
        permission_manager: Optional[CommandPermissionManager] = None,
        execution_context: Optional[ExecutionContext] = None,  # NEW
    ):
        super().__init__(settings, permission_manager=permission_manager)

        self.llm_provider = llm_provider
        self.app_config = app_config

        # Create or use provided execution context
        if execution_context:
            self.execution_context = execution_context
        else:
            # Root agent - create new context
            from forge.agent.factory import DefaultAgentFactory
            self.execution_context = ExecutionContext(
                llm_provider=llm_provider,
                file_storage=file_storage,
                agent_factory=DefaultAgentFactory(app_config),
            )

        # Create strategy and inject context
        self.prompt_strategy = self._create_prompt_strategy(app_config)
        if hasattr(self.prompt_strategy, 'set_execution_context'):
            self.prompt_strategy.set_execution_context(self.execution_context)

仓库中的实际实现与计划稿逐行对应（Agent.init）：

• 传入 execution_context 则直接采用（子 Agent 场景，由工厂注入）；
• 未传入则走 _create_root_execution_context，内部实例化 DefaultAgentFactory(app_config) 后构造根级 ExecutionContext；
• 注入上下文时用 getattr(self.prompt_strategy, "set_execution_context", None) 做鸭子式探测再调用——因为 one_shot 这类单步策略根本没有该方法，这种"有则注入、无则跳过"的写法保证了现有策略零改动，与计划文档"Backward Compatibility"承诺完全一致。

4.2 工厂函数透传

计划同步更新了 agent_factory/configurators.py 中的 create_agent，在参数列表末尾追加 execution_context: Optional[ExecutionContext] = None 并在构造 Agent 时透传：

def create_agent(
    agent_id: str,
    task: str,
    app_config: AppConfig,
    file_storage: FileStorage,
    llm_provider: MultiProvider,
    ai_profile: Optional[AIProfile] = None,
    directives: Optional[AIDirectives] = None,
    permission_manager: Optional[CommandPermissionManager] = None,
    execution_context: Optional[ExecutionContext] = None,  # NEW
) -> Agent:
    # ... existing code ...

    return Agent(
        settings=agent_state,
        llm_provider=llm_provider,
        file_storage=file_storage,
        app_config=app_config,
        permission_manager=permission_manager,
        execution_context=execution_context,  # NEW
    )

至此，从 CLI/服务入口创建 Agent、到根 Agent 建上下文、到策略派生、到子 Agent 继承上下文的完整闭环成立。

五、Phase 4：示例策略 —— LATS

计划以 LATS（Language Agent Tree Search）作为首个完整示范，其核心是用蒙特卡洛树搜索（MCTS）驱动推理，并让子 Agent 充当"分支模拟器"。

5.1 搜索树节点与 UCB1 选择

class LATSPhase(str, Enum):
    SELECT = "select"
    EXPAND = "expand"
    SIMULATE = "simulate"
    BACKPROPAGATE = "backpropagate"


@dataclass
class LATSNode:
    """A node in the LATS search tree."""

    state: str  # Description of current state
    action: Optional[str] = None  # Action that led here
    parent: Optional["LATSNode"] = None
    children: list["LATSNode"] = field(default_factory=list)

    # MCTS statistics
    visits: int = 0
    value: float = 0.0

    # Simulation results
    simulated: bool = False
    simulation_result: Optional[str] = None

    @property
    def ucb1(self) -> float:
        """Upper Confidence Bound for tree policy."""
        if self.visits == 0:
            return float('inf')
        if self.parent is None:
            return self.value / self.visits

        exploration = math.sqrt(2 * math.log(self.parent.visits) / self.visits)
        return (self.value / self.visits) + exploration

    def best_child(self) -> Optional["LATSNode"]:
        """Select best child by UCB1."""
        if not self.children:
            return None
        return max(self.children, key=lambda n: n.ucb1)

UCB1 公式 value/visits + sqrt(2·ln(parent_visits)/visits) 是 MCTS 的经典权衡项：前一项利用（exploitation），后一项探索（exploration）——访问次数越少的子节点，探索加成越大；visits == 0 时返回 inf 强制优先扩展未访问节点。

5.2 策略配置

class LATSPromptConfiguration(BasePromptStrategyConfiguration):
    # MCTS parameters
    num_simulations: int = UserConfigurable(default=5)
    exploration_constant: float = UserConfigurable(default=1.414)
    max_depth: int = UserConfigurable(default=10)
    branching_factor: int = UserConfigurable(default=3)

    # Sub-agent configuration
    enable_sub_agents: bool = True  # Required for LATS
    simulation_strategy: str = UserConfigurable(default="one_shot")
    simulation_max_cycles: int = UserConfigurable(default=10)

注意 enable_sub_agents: bool = True 的注释"Required for LATS"：LATS 是这套子 Agent 机制的第一个"刚需消费者"——没有并行分支模拟，树搜索就退化成了普通 ReAct 循环。simulation_strategy 默认指向轻量的一发策略 one_shot（该策略文件位于 prompt_strategies/one_shot.py），配合 simulation_max_cycles=10 控制单次模拟成本——这正体现了预算思想在策略层的体现：搜索树每个节点的"模拟"都有明确的算力上限。

5.3 MCTS 主循环：子 Agent 在哪里登场

async def run_mcts(self, task: str, initial_state: str) -> LATSNode:
    """Run MCTS to find best action."""

    # Initialize root
    self.root = LATSNode(state=initial_state)

    for _ in range(self.config.num_simulations):
        # SELECT: traverse to promising leaf
        node = self._select(self.root)

        # EXPAND: generate candidate children
        if not node.children and node.visits > 0:
            await self._expand(node, task)

        # SIMULATE: run sub-agent to evaluate
        if node.children:
            child = node.children[0]  # Or random
            if not child.simulated:
                value = await self._simulate(child, task)
                child.simulated = True
                child.value = value

        # BACKPROPAGATE: update ancestor values
        self._backpropagate(node)

        self.simulations_completed += 1

    return self.root

run_mcts 每轮模拟执行 SELECT → EXPAND → SIMULATE → BACKPROPAGATE 四阶段；真正的创新点在 _simulate：

async def _simulate(self, node: LATSNode, task: str) -> float:
    """Simulate by running a sub-agent.

    This is where sub-agent spawning happens!
    """
    if not self.can_spawn_sub_agent():
        self.logger.warning("Cannot spawn sub-agent for simulation")
        return 0.5  # Neutral score

    # Build simulation task
    simulation_task = (
        f"Task: {task}\n\n"
        f"Current state: {node.state}\n"
        f"Action to take: {node.action}\n\n"
        f"Execute this action and report the outcome."
    )

    try:
        result = await self.spawn_and_run(
            task=simulation_task,
            strategy=self.config.simulation_strategy,
            max_cycles=self.config.simulation_max_cycles,
        )

        node.simulation_result = str(result)

        # Evaluate outcome (would call LLM)
        # For now, simple heuristic
        if "success" in str(result).lower():
            return 1.0
        elif "error" in str(result).lower():
            return 0.0
        else:
            return 0.5

    except Exception as e:
        self.logger.error(f"Simulation failed: {e}")
        return 0.0

可以看到设计中的三层防御：预算耗尽时回退到中性分 0.5（不惩罚也不鼓励该分支）；评估用关键词启发式占位（标注 "would call LLM"），保留升级为 LLM 评审的接口；子 Agent 异常返回 0 分。这正是"Agent 模拟价值函数"的工程化雏形。

仓库中的 LATS 实现。 lats.py 是计划稿 Phase 4 的完整落地：LATSPhase 扩展为五阶段（SELECTION / EXPANSION / EVALUATION / BACKPROPAGATION / EXECUTION，比计划稿多出"执行最优动作"阶段，使策略能闭环产出真实动作而非只返回树）；LATSNode 改为 Pydantic 模型并新增 reward、reflection（失败反思）字段，UCT 得分函数 uct_score 将探索系数参数化为 exploration_weight（默认 1.41 ≈ √2，与计划稿的 exploration_constant=1.414 呼应）；配置项 LATSPromptConfiguration 使用 num_candidates（默认 3）与 max_depth（默认 5）等参数。此外，仓库 prompt_strategies/ 目录下还有 multi_agent_debate.py 与 tree_of_thoughts.py 等策略，对应计划文档概述中列出的"多 Agent 辩论"等目标模式以及实施排期第 4 周"Additional example strategies"的交付项。

六、Phase 5：通信协议与资源追踪

6.1 子 Agent 通信协议

计划提出以显式消息模型描述父子通信（forge/agent/sub_agent_protocol.py），为未来跨进程/跨服务的 Agent 通信预留抽象：

class MessageType(str, Enum):
    REQUEST = "request"
    RESPONSE = "response"
    STATUS = "status"
    CANCEL = "cancel"


class SubAgentMessage(BaseModel):
    """Message passed between parent and sub-agent."""

    type: MessageType
    sender_id: str
    recipient_id: str
    content: Any
    correlation_id: Optional[str] = None  # 关联同一轮请求/响应


class SubAgentRequest(SubAgentMessage):
    """Request from parent to sub-agent."""
    type: MessageType = MessageType.REQUEST
    task: str
    context: dict[str, Any] = {}


class SubAgentResponse(SubAgentMessage):
    """Response from sub-agent to parent."""
    type: MessageType = MessageType.RESPONSE
    success: bool
    result: Any
    error: Optional[str] = None

correlation_id 是并发场景的关键：当多个子 Agent 并行返回时，父 Agent 靠它把响应匹配回当初的请求。当前仓库实现中，父级通过 SubAgentHandle（同进程内句柄）完成通信，correlation_id 的语义实际上由 agent_id 承担（ExecutionContext.sub_agents 字典以 ID 为键），消息模型属于前瞻性预留。

6.2 资源追踪组件

class ResourceTrackerComponent(AgentComponent, AfterExecute):
    """Tracks resource usage for budget enforcement."""

    def __init__(self, execution_context):
        self.context = execution_context
        self.tokens_used = 0
        self.cycles_completed = 0

    def after_execute(self, result: ActionResult) -> None:
        self.cycles_completed += 1

        # Check budget
        budget = self.context.budget
        if budget.max_cycles and self.cycles_completed >= budget.max_cycles:
            raise BudgetExceededError("Cycle budget exceeded")

该组件挂在 Agent 组件协议的 AfterExecute 钩子上（AutoGPT 的组件系统支持按协议在生命周期节点插入组件，参见 forge/components），把"预算强制执行"从策略代码中解耦出来——任何 Agent（无论何种策略）只要在组件链里挂了它，就会在每次执行后被审计一次周期计数。

七、实施路径、迁移兼容与遗留问题

7.1 四周实施排期

计划给出了明确的推进顺序，核心原则是先基础设施、再集成、后示例策略：

• 第 1 周（核心基础设施）：创建 ExecutionContext 模型 → 创建 AgentFactory 协议 → 为 BaseMultiStepPromptStrategy 添加子 Agent 方法 → 更新 Agent.__init__ 接收上下文；
• 第 2 周（集成）：更新 agent factory 函数 → 添加子 Agent 通信协议 → 创建资源追踪组件 → 补充子 Agent 派生测试；
• 第 3 周（示例策略）：实现 LATS 骨架 → 实现 MCTS 核心逻辑 → 集成子 Agent 模拟 → 端到端测试；
• 第 4 周（打磨）：错误处理与清理 → 文档 → 性能优化 → 更多示例策略。

对照仓库现状，前三周交付物均已在代码中可见（execution_context.py、base.py 方法族、agent.py 集成、lats.py），第 4 周的"更多示例策略"则对应 prompt_strategies/ 目录下的 debate、tree_of_thoughts、plan_execute、reflexion、rewoo 等文件。

7.2 迁移兼容性承诺

计划文档的 Migration Notes 明确了两点：

• 向后兼容：现有策略零改动继续工作；enable_sub_agents 计划默认 False（最终实现改为 True，见前文）；ExecutionContext 对根 Agent 是自动创建的可选参数；
• 无破坏性变更：既有代码无需修改；使用子 Agent 的新策略才要求更新后的 Agent 类。

测试策略按四层递进：ExecutionContext 单元测试 → 子 Agent 生命周期集成测试 → 用 mock 子 Agent 的 LATS 策略测试 → 真实 LLM 调用的端到端测试。其中"mock 子 Agent"一层很关键：子 Agent 机制的价值在于可被独立测试，工厂协议 + 句柄抽象正是这一点的实现基础。

7.3 开放问题

计划最后保留了五个开放问题，其中三个已有实现层面的答案（文件共享→写受限子目录；历史可见性→父级只见结果；取消→cancelled 标志 + asyncio.Task.cancel），权限继承与跨层级执行追踪两项在实现中也有对应痕迹（inherited_deny_rules 继承拒绝规则；层级化 agent_id 便于跨层级日志串联）。这些问题提示读者：该文档与源码应结合阅读，文档记录"设计意图"，源码记录"最终取舍"。

八、小结

这份子 Agent 重构计划展示了在既有单体 Agent 框架上叠加多 Agent 协调能力的完整方法论：ExecutionContext 提供共享资源与层级隔离，ResourceBudget 提供递归收敛保障，AgentFactory 协议解耦 Agent 构造，策略基类方法族（spawn/run/parallel）提供统一编程接口，LATS 则验证了"子 Agent 作为树搜索模拟器"这一具体范式。其工程细节——派生失败转句柄状态、超时与取消的三态区分、子 Agent 写目录隔离、拒绝规则逐层继承——都是多 Agent 系统从 Demo 走向可用的必经功课。读者可以沿 SUB_AGENT_REFACTOR_PLAN.md 的设计脉络，结合 execution_context.py、prompt_strategies/base.py、agent.py 与 lats.py 四个文件交叉阅读，获得"设计意图 ↔ 源码实现"的双向理解。