[!info] related notes

• 所属 MOC: AI Agent Application MOC
• 相关: Agent Engine, Plan-and-Execute Pattern

Planner

一句话定义

Planner 是 Agent 中负责将复杂任务分解为可执行步骤的组件。用户说”帮我分析最近的销售趋势并生成报告”，Planner 把它拆成：查询数据 → 分析趋势 → 生成图表 → 撰写报告。

它解决什么问题

直接让 LLM 一次性完成复杂任务：

• 容易遗漏步骤
• 中间结果无法检查
• 失败后需要从头开始
• 无法并行执行独立步骤

Planner 让任务”可分解、可追踪、可恢复”。

核心原理

Plan 数据结构

@dataclass
class PlanStep:
    id: str
    description: str
    tool: str  # 需要调用的工具
    args_template: dict  # 参数模板（可以引用前序步骤的结果）
    status: str = "pending"  # pending, running, completed, failed
    result: any = None

@dataclass
class Plan:
    id: str
    goal: str
    steps: list[PlanStep]
    current_step: int = 0
    status: str = "pending"

LLM 生成 Plan

class Planner:
    def __init__(self, llm, tools: list[dict]):
        self.llm = llm
        self.tools = tools

    async def create_plan(self, goal: str, context: dict = None) -> Plan:
        prompt = f"""
你是一个任务规划器。给定一个目标，将其分解为可执行的步骤。

目标: {goal}

可用工具:
{json.dumps(self.tools, indent=2)}

请输出执行计划（JSON 格式）:
{{
  "steps": [
    {{"description": "步骤描述", "tool": "工具名", "args": {{参数}}}}
  ]
}}
"""
        result = await self.llm.chat(prompt)
        plan_data = json.loads(result)

        return Plan(
            id=generate_id(),
            goal=goal,
            steps=[PlanStep(**step) for step in plan_data["steps"]],
        )

执行 Plan

class PlanExecutor:
    def __init__(self, tool_runtime: ToolRuntime):
        self.tool_runtime = tool_runtime

    async def execute(self, plan: Plan, context: ExecutionContext) -> Plan:
        for step in plan.steps:
            step.status = "running"

            # 解析参数（可以引用前序步骤的结果）
            args = self.resolve_args(step.args_template, plan)

            # 执行工具
            result = await self.tool_runtime.execute(
                ToolCall(name=step.tool, arguments=args),
                context,
            )

            if result.error:
                step.status = "failed"
                step.result = result.error
                break

            step.status = "completed"
            step.result = result.data
            plan.current_step += 1

        plan.status = "completed" if all(s.status == "completed" for s in plan.steps) else "failed"
        return plan

常见设计模式

1. 静态规划

LLM 生成完整计划，然后逐步执行。

2. 动态规划

每步执行后，LLM 根据结果调整后续计划。

3. 并行步骤

独立的步骤可以并行执行。

常见坑

1. Plan 不可执行: 步骤描述太模糊，无法转换为工具调用
2. 不做 Plan 验证: 生成的 Plan 引用了不存在的工具
3. 不做错误恢复: 某步失败后整个 Plan 失败
4. Plan 太长: 步骤太多，执行时间过长

参考资料

• Plan-and-Execute Pattern
• Agent Engine