第2章 Agent 规划（Planning）¶

学习目标¶

理解任务分解（Task Decomposition）的原理与方法
掌握 Plan-and-Execute 与 ReAct 两种主流规划范式
学会工具与子 Agent 的分派策略
理解动态重规划的触发条件与实现
能够为复杂业务设计多步规划 Agent

问题导入¶

某团队让一个 Agent 处理"帮我分析上季度华东区销售数据，找出下滑品类，生成报告并发邮件给总监"这种复合任务。结果 Agent 第一步就去调"发邮件"工具，邮件内容空白；或者反复查询同一个数据，始终无法推进。

问题在于：这类任务包含多个有依赖关系的子任务（取数→分析→找下滑品类→生成报告→发邮件），必须先规划再执行。直接让 Agent 边想边做，容易丢失全局视角、步骤错乱、无法回溯。规划（Planning）是 Agent 从"单步反应"走向"多步推理"的关键能力。本章讲解如何让 Agent 像项目经理一样拆解任务、生成计划、动态调整。

核心概念¶

1. 任务分解（Task Decomposition）¶

把复杂目标拆成可执行的原子子任务。原则：

MECE：子任务互不重叠、合起来覆盖目标
可执行：每个子任务对应一个工具或子 Agent
有依赖：标注前后顺序，便于串/并行调度
粒度适中：太粗无法执行，太细则规划成本高

例如"分析季度销售并出报告"可分解为：①取销售数据 → ②按品类聚合 → ③计算环比 → ④筛选下滑品类 → ⑤生成报告 → ⑥发送邮件。其中 ① 必须先于 ②。

2. 计划生成¶

用 LLM 一次性生成结构化计划（通常是 JSON 列表），每项包含：

{
  "step": 1,
  "task": "查询华东区上季度销售数据",
  "tool": "query_sales",
  "depends_on": [],
  "expected_output": "销售明细列表"
}

计划生成后可校验依赖关系是否成立、工具是否存在，再进入执行。

3. Plan-and-Execute 模式¶

经典两阶段模式：

Plan 阶段：Planner LLM 一次性生成完整计划
Execute 阶段：Executor 按计划依次调用工具，把结果回填

优点：全局视角好、步骤可控、易审计；缺点：缺乏中途灵活性，环境变化时计划易失效。适合步骤明确、环境稳定的任务（如 ETL 流程）。

4. ReAct 规划¶

ReAct 是"边规划边执行"：每轮先 Thought（思考下一步），再 Action（执行），再 Observation（观察），根据观察结果动态决定下一步。优点：灵活、能应对环境变化；缺点：缺乏全局规划，长任务易跑偏。适合探索性强、步骤不可预判的任务。

5. 工具与子 Agent 分派¶

复杂计划中，每个子任务可分派给：

工具：原子操作（查询、计算）
子 Agent：需要推理的子任务（如"分析数据趋势"交给分析 Agent）

分派策略：能工具化的就用工具，需要 LLM 推理的才用子 Agent，避免过度抽象。

6. 动态重规划¶

执行中发现计划失效（工具返回异常、中间结果与预期不符）时触发重规划：

触发条件：工具失败、数据缺失、用户追加需求
重规划范围：仅重做受影响的后续步骤，而非推翻全部
状态保留：已完成步骤的结果保留，作为重规划的输入

原理图¶

flowchart TB
    Start([用户目标]) --> Planner[Planner LLM<br/>生成结构化计划]

    Planner --> Plan[计划: step1→step2→step3→step4]

    Plan --> Exec1[执行 step1<br/>query_sales]
    Exec1 --> Obs1{结果符合预期?}
    Obs1 -->|是| Exec2[执行 step2<br/>aggregate]
    Obs1 -->|否| Replan[动态重规划<br/>调整后续步骤]

    Replan --> Plan
    Exec2 --> Exec3[执行 step3<br/>分析趋势]
    Exec3 --> Exec4[执行 step4<br/>发邮件]
    Exec4 --> Done([任务完成])

    style Planner fill:#fff9c4
    style Plan fill:#ffe0b2
    style Replan fill:#ffcdd2
    style Done fill:#c8e6c9

图：Plan-and-Execute 任务分解流程——Planner 生成计划，按序执行，遇异常触发动态重规划

示例代码¶

1. Plan-and-Execute 任务分解实现¶

"""
第2章示例：Plan-and-Execute 规划 Agent
用 Planner 生成结构化计划，Executor 按计划执行，支持动态重规划
"""
import asyncio
import json
from agents import Agent, Runner, function_tool


# ========== 工具定义 ==========
@function_tool
def query_sales(region: str, quarter: str) -> str:
    """查询指定区域和季度的销售数据"""
    data = {
        ("华东", "Q3"): "华东Q3销售：电子 1200万, 服装 800万, 食品 600万",
        ("华东", "Q2"): "华东Q2销售：电子 1500万, 服装 750万, 食品 620万",
    }
    return data.get((region, quarter), f"无 {region}{quarter} 数据")


@function_tool
def analyze_trend(data: str) -> str:
    """分析销售趋势，找出下滑品类"""
    if "电子 1200" in data and "电子 1500" in data:
        return "电子产品环比下滑 20%（1500→1200），为下滑最大品类"
    return "未检测到明显下滑"


@function_tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送至 {to}，主题：{subject}"


# ========== Planner Agent ==========
planner_agent = Agent(
    name="Planner",
    instructions=(
        "你是项目规划师。把用户目标分解为 3-6 个有序子任务，输出 JSON 数组。"
        "每项包含：step(序号), task(描述), tool(工具名), depends_on(前置步骤序号列表)。"
        "只输出 JSON，不要解释。"
    ),
    model="gpt-4o-mini",
)


# ========== Executor Agent ==========
executor_agent = Agent(
    name="Executor",
    instructions=(
        "你是执行者。根据给定的任务描述和历史结果，调用合适的工具完成当前步骤。"
        "若工具失败，说明原因并请求重规划。"
    ),
    tools=[query_sales, analyze_trend, send_email],
    model="gpt-4o-mini",
)


async def plan_and_execute(goal: str, max_steps: int = 8) -> str:
    """Plan-and-Execute 主流程"""
    # 1. 生成计划
    plan_result = await Runner.run(planner_agent, f"目标：{goal}")
    try:
        plan = json.loads(plan_result.final_output)
    except json.JSONDecodeError:
        # 容错：提取 JSON 片段
        text = plan_result.final_output
        plan = json.loads(text[text.find("[") : text.rfind("]") + 1])

    print("=== 生成的计划 ===")
    for item in plan:
        print(f"  步骤{item['step']}: {item['task']} (依赖: {item.get('depends_on', [])})")

    # 2. 按序执行
    results = {}
    for item in plan:
        step = item["step"]
        task = item["task"]
        deps = item.get("depends_on", [])
        context = "\n".join(f"步骤{d}结果：{results[d]}" for d in deps)

        print(f"\n=== 执行步骤 {step}: {task} ===")
        prompt = f"当前任务：{task}\n前置结果：{context}\n请调用工具完成。"
        exec_result = await Runner.run(executor_agent, prompt, max_turns=4)
        results[step] = exec_result.final_output
        print(f"结果：{results[step]}")

    return "\n".join(f"步骤{k}：{v}" for k, v in results.items())


if __name__ == "__main__":
    goal = "分析华东区Q3相比Q2的销售下滑品类，发邮件给总监报告结论"
    summary = asyncio.run(plan_and_execute(goal))
    print("\n=== 最终汇总 ===")
    print(summary)

运行结果示例：

=== 生成的计划 ===
  步骤1: 查询华东Q3销售数据 (依赖: [])
  步骤2: 查询华东Q2销售数据 (依赖: [])
  步骤3: 分析下滑品类 (依赖: [1, 2])
  步骤4: 发邮件给总监 (依赖: [3])

=== 执行步骤 1: 查询华东Q3销售数据 ===
结果：华东Q3销售：电子 1200万, 服装 800万, 食品 600万
=== 执行步骤 2: 查询华东Q2销售数据 ===
结果：华东Q2销售：电子 1500万, 服装 750万, 食品 620万
=== 执行步骤 3: 分析下滑品类 ===
结果：电子产品环比下滑 20%（1500→1200），为下滑最大品类
=== 执行步骤 4: 发邮件给总监 ===
结果：邮件已发送至 总监，主题：销售分析报告

2. 动态重规划¶

"""动态重规划：执行失败时调用 Planner 重新生成后续步骤"""
from agents import Agent, Runner


replanner = Agent(
    name="Replanner",
    instructions=(
        "你是重规划师。给定原计划、已完成步骤结果、失败信息，"
        "重新生成未完成步骤的 JSON 计划。保留已完成步骤，只调整后续。"
    ),
    model="gpt-4o-mini",
)


async def execute_with_replan(plan, executor, max_replans=2):
    results = {}
    for item in plan:
        try:
            r = await Runner.run(executor, item["task"], max_turns=3)
            results[item["step"]] = r.final_output
        except Exception as e:
            if max_replans <= 0:
                raise
            print(f"步骤 {item['step']} 失败：{e}，触发重规划")
            replan_prompt = f"原计划：{plan}\n已完成：{results}\n失败：{item} 错误：{e}"
            new_plan = await Runner.run(replanner, replan_prompt)
            return await execute_with_replan(
                json.loads(new_plan.final_output), executor, max_replans - 1
            )
    return results

企业案例¶

案例 1：复杂项目自动规划¶

某 SaaS 公司的"运营 Agent"接收运营人员的自然语言指令（如"做一次双 11 预热活动，覆盖老客和新客"），用 Plan-and-Execute 拆解为：①筛选老客群体 → ②筛选新客群体 → ③生成预热文案 → ④配置推送任务 → ⑤启动 A/B 测试 → ⑥汇总效果。Planner 输出结构化计划供人工审核，确认后自动执行。重规划机制应对"用户群体接口变更"等异常，失败步骤自动调整。上线后活动配置时长从 4 小时降到 20 分钟。

案例 2：ReAct 规划用于探索性任务¶

某数据团队的"探索分析 Agent"用 ReAct 模式：面对开放问题（"为什么上周留存下降？"），Agent 不知道该查哪些表，于是边查边想——先查留存数据，发现下降，再查分渠道留存，定位到某渠道异常，再查该渠道投放变化。这种"无法预先规划"的探索性任务，ReAct 比固定计划更有效。

常见问题¶

Q1：Plan-and-Execute 和 ReAct 怎么选？

A：步骤明确、环境稳定用 Plan-and-Execute（如 ETL、流程自动化）；探索性强、步骤不可预判用 ReAct（如根因分析）。复杂系统常混合使用：顶层 Plan-and-Execute 拆解，叶子节点用 ReAct 探索。

Q2：计划生成不准确怎么办？

A：①用 Few-shot 示例引导 Planner 输出正确格式；②加校验层（依赖是否成立、工具是否存在）；③允许人工审核计划后再执行；④执行后复盘，把好计划加入示例库。

Q3：重规划会不会导致无限循环？

A：会。必须设 max_replans 上限（通常 2-3 次），超过即降级人工或返回兜底结果。同时记录重规划原因，避免同类失败反复触发。

Q4：子任务之间能否并行？

A：可以。无依赖关系的子任务应并行执行以降低延迟。如示例中步骤1、步骤2 可并行查询。实现上用 asyncio.gather 调度无依赖任务。

Q5：规划粒度怎么把握？

A：经验法则：每个子任务对应一次工具调用或一次子 Agent 调用。3-6 步为宜，超过 8 步应考虑分层规划（大计划套子计划）。

面试题¶

简答题：任务分解的 MECE 原则是什么？为什么对 Agent 规划重要？

参考要点：互不重叠、完全覆盖；保证子任务不重复执行且目标不遗漏。

对比题：Plan-and-Execute 与 ReAct 各自的优缺点及适用场景。

参考要点：P&E 全局视角好但僵化，适合稳定流程；ReAct 灵活但缺全局，适合探索任务。

设计题：为"对比两个竞品的口碑"设计规划方案，标出可并行的步骤。
实现题：如何在 Plan-and-Execute 中实现并行执行无依赖步骤？

参考要点：拓扑排序找无依赖节点，用 asyncio.gather 并发调度。

原理题：动态重规划为什么只调整后续步骤而不推翻全部？

参考要点：保留已完成结果降低成本、保证任务连续性、避免重复执行副作用工具。

练习题¶

计划生成：让 Planner Agent 为"组织一次线上技术分享会"生成 6 步计划，校验依赖关系是否合理。
并行优化：改造示例 1，使步骤 1、步骤 2 并行执行，对比串行与并行的总耗时。
重规划模拟：故意让 analyze_trend 工具返回错误，验证重规划流程是否正确触发并调整后续步骤。

小结¶

规划是 Agent 处理复杂任务的核心能力。本章核心：任务分解把大目标拆成可执行子任务；Plan-and-Execute 适合稳定流程，ReAct 适合探索任务；动态重规划应对环境变化，但需设上限防失控；子任务可并行调度以降延迟。规划解决了"做什么、什么顺序"，下一章解决"工具如何被调用与执行"。