扩展：加入显式 planner 或对标一个框架实现

把现有 loop 拆成 planner + executor

这一节你要在「上一版手写 loop」基础上，显式拆出 planner（计划） 和 executor（执行） 两个子组件。

在很多工业实践里，所谓的 agent loop 本质就是「感知 → 推理 → 执行动作 → 观察 → 再决策」的循环，你可以在 Oracle 这篇介绍里看到类似结构：What Is the AI Agent Loop?。拆 planner / executor，就是把其中的「推理决定做什么」和「实际把事做完」分开。

1. 先给 loop 起一个外壳

假设你上一版是一个类似这样的函数（伪代码）：

pythondef agent_loop(user_goal: str): state = {} for step in range(5): # 把当前目标和 state 丢给 LLM decision = call_llm(f"目标: {user_goal}\n当前状态: {state}\n你下一步该干嘛？") # 根据 decision 里的一些关键词，直接在这里 if/else 执行 if "search" in decision: state["search_result"] = web_search(...) elif "summarize" in decision: state["summary"] = summarize(state["search_result"]) # ... return state

这里决策和执行是耦合在一起的：LLM 输出被你立刻 if/else 消费。

现在我们要做的是：

• 抽一个 Planner 类：只负责决定「下一步做什么，以及需要哪些参数」
• 抽一个 Executor 类：只负责根据 planner 的指令执行工具 / 写状态

搭一个新文件，例如 planner_executor.py，至少准备这两个骨架：

pythonfrom dataclasses import dataclass from typing import Any, Dict, List, Optional @dataclass class PlanStep: action: str args: Dict[str, Any] reason: str class Planner: def __init__(self, llm_client): self.llm = llm_client def plan_next(self, goal: str, memory: Dict[str, Any]) -> PlanStep: """根据当前 goal + memory 产出下一步的计划""" raise NotImplementedError class Executor: def __init__(self, tools: Dict[str, Any]): self.tools = tools def execute(self, step: PlanStep, memory: Dict[str, Any]) -> Dict[str, Any]: """根据 PlanStep 调用对应工具，更新 memory""" raise NotImplementedError

2. 用 planner + executor 重写 loop

你可以写一个 AgentLoop 把两者串起来：

pythonclass AgentLoop: def __init__(self, planner: Planner, executor: Executor, max_steps: int = 5): self.planner = planner self.executor = executor self.max_steps = max_steps def run(self, goal: str, init_memory: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: memory: Dict[str, Any] = init_memory or {} trace: List[Dict[str, Any]] = [] for step_idx in range(self.max_steps): plan_step = self.planner.plan_next(goal, memory) memory = self.executor.execute(plan_step, memory) trace.append({ "step": step_idx, "plan": plan_step, "memory_snapshot": dict(memory), }) # 可选：添加终止条件 if memory.get("done"): break memory["_trace"] = trace return memory

这里的核心变化：

• AgentLoop.run 完全不 care「怎么做」，只管循环：
  • 问 planner：「下一步是什么？」
  • 让 executor 去做
  • 把每一步的 plan 和 memory 快照塞进 trace，后面做调优

这就是很多 agent 框架里所谓「可插拔 agent loop」的雏形，比如 verl 的 AgentLoopBase 也是只定义一个 run 接口，具体 loop 逻辑交给用户实现。