Agent_03_Plan-Execute模式

Agent2026年1月·agentPython
zbhgis 浩瀚地学1596 字8 分钟
创建于 更新于

为什么需要 Plan-Execute

ReAct 是"边想边做"，每执行一步就要调一次 LLM 决定下一步。在步骤明确、流程固定的场景里，这种频繁的 LLM 调用既费钱又费时间。Plan-Execute 的思路是把 LLM 的决策集中在前面：一次调用生成完整计划，后面按步骤执行，不需要 LLM 参与。

简单说，就是把 LLM 从"每一步的决策者"变成"全局计划的制定者"。计划生成完之后，执行就是普通代码——没有额外成本。

Python 实现

Plan-Execute Agent 完整实现

python
"""
Plan-Execute Agent —— 先规划，再执行

核心设计：
1. Plan 阶段：LLM 生成步骤列表（一次调用）
2. Execute 阶段：按步骤执行工具（不需要 LLM）
3. 可选 Replan：执行失败或需要动态调整时，重新规划
"""

from dataclasses import dataclass
from typing import Any


@dataclass
class PlanStep:
    """计划中的单一步骤"""
    step_number: int
    description: str
    tool_name: str
    tool_args_template: str  # 模板，执行时填充实际参数


@dataclass
class Plan:
    """完整的执行计划"""
    goal: str
    steps: list[PlanStep]
    estimated_steps: int  # 预估步骤数


class PlanExecutor:
    """
    计划执行器 —— 按步骤执行，收集结果
    执行阶段不需要 LLM，就是普通的代码执行。
    但如果某步失败了，可以选择跳过、重试、或者重新规划。
    """

    def __init__(self, tool_registry, on_failure: str = "skip"):
        self.tools = tool_registry
        self.on_failure = on_failure  # "skip" | "retry" | "replan"
        self.results: list[dict] = []

    def execute(self, plan: Plan, context: dict = None) -> dict:
        """
        执行整个计划

        返回：所有步骤的结果汇总
        """
        context = context or {}

        for step in plan.steps:
            # 填充参数模板（从上下文中取值）
            args = self._fill_args(step.tool_args_template, context)

            # 执行工具
            result = self.tools.execute(step.tool_name, **args)

            step_result = {
                "step": step.step_number,
                "description": step.description,
                "tool": step.tool_name,
                "success": result.success,
                "output": result.output,
                "error": result.error,
            }
            self.results.append(step_result)

            # 把结果存入上下文（后续步骤可以引用）
            context[f"step_{step.step_number}_result"] = result.output

            if not result.success:
                if self.on_failure == "skip":
                    print(f"  步骤 {step.step_number} 失败，跳过: {result.error}")
                    continue
                elif self.on_failure == "retry":
                    # 重试一次
                    retry_result = self.tools.execute(step.tool_name, **args)
                    if retry_result.success:
                        step_result["success"] = True
                        step_result["output"] = retry_result.output
                        context[f"step_{step.step_number}_result"] = retry_result.output
                    else:
                        print(f"  步骤 {step.step_number} 重试失败: {retry_result.error}")
                elif self.on_failure == "replan":
                    # 需要重新规划（返回当前进度，让调用方处理）
                    return {
                        "status": "replan_needed",
                        "results": self.results,
                        "failed_step": step.step_number,
                        "error": result.error,
                    }

        return {
            "status": "completed",
            "results": self.results,
            "context": context,
        }

    def _fill_args(self, template: str, context: dict) -> dict:
        """
        填充参数模板

        模板格式: '{"path": "{{step_1_result.path}}", "format": "bar"}'
        从上下文中替换变量
        """
        import re
        import json

        filled = template
        for key, value in context.items():
            filled = filled.replace(f"{{{{{key}}}}}", str(value))

        try:
            return json.loads(filled)
        except json.JSONDecodeError:
            return {}


class PlanExecuteAgent:
    """
    Plan-Execute Agent —— 规划 + 执行的组合

    核心逻辑：
    1. 接收目标 -> LLM 生成计划
    2. 按计划执行 -> 收集结果
    3. LLM 综合结果生成最终回复
    """

    def __init__(self, llm, tool_registry, config: dict = None):
        self.llm = llm
        self.tools = tool_registry
        self.config = config or {
            "max_replans": 1,       # 最多重新规划次数
            "on_failure": "replan", # 失败策略
            "verbose": True,
        }

    def run(self, goal: str) -> str:
        replan_count = 0

        while replan_count <= self.config["max_replans"]:
            if self.config["verbose"]:
                print(f"\n=== 第 {replan_count + 1} 次规划 ===")

            # Phase 1：规划
            plan = self._plan(goal)

            if self.config["verbose"]:
                print(f"  生成计划，共 {len(plan.steps)} 步：")
                for step in plan.steps:
                    print(f"    {step.step_number}. {step.description} (工具: {step.tool_name})")

            # Phase 2：执行
            executor = PlanExecutor(
                self.tools,
                on_failure=self.config.get("on_failure", "replan"),
            )
            exec_result = executor.execute(plan)

            if exec_result["status"] == "completed":
                # Phase 3：综合输出
                if self.config["verbose"]:
                    print(f"\n=== 执行完成，共 {len(exec_result['results'])} 步 ===")

                return self._synthesize(goal, exec_result)

            elif exec_result["status"] == "replan_needed":
                replan_count += 1
                if self.config["verbose"]:
                    print(f"\n=== 执行失败，需要重新规划（第 {replan_count} 次）===")
                    print(f"  失败步骤: {exec_result['failed_step']}, 错误: {exec_result['error']}")

                # 把失败信息传给 LLM，让它调整计划
                goal = self._update_goal_with_error(
                    goal, exec_result["results"], exec_result["error"]
                )
                continue

        return "多次规划后仍未能完成。"

    def _plan(self, goal: str) -> Plan:
        """调用 LLM 生成计划"""
        tool_schemas = "\n".join(
            f"- {t['name']}: {t['description']}" for t in self.tools.get_schemas()
        )

        prompt = f"""
目标: {goal}

可用工具:
{tool_schemas}

请将目标拆解为可执行的步骤列表。每步包含：
1. step_number: 步骤序号
2. description: 步骤描述
3. tool_name: 使用的工具名称
4. tool_args_template: 工具的 JSON 参数模板

返回 JSON 格式:
{{"steps": [{{"step_number": 1, "description": "...", "tool_name": "...", "tool_args_template": "..."}}]}}
"""

        response = self.llm.generate(prompt)
        # 解析 LLM 输出的 JSON 计划
        plan_data = self.llm.parse_json(response)
        steps = [PlanStep(**s) for s in plan_data["steps"]]
        return Plan(goal=goal, steps=steps, estimated_steps=len(steps))

    def _synthesize(self, goal: str, exec_result: dict) -> str:
        """LLM 综合执行结果，生成最终回复"""
        results_summary = "\n".join(
            f"步骤 {r['step']} ({r['tool']}): {'成功' if r['success'] else '失败'}"
            for r in exec_result["results"]
        )

        prompt = f"""
目标: {goal}

执行结果:
{results_summary}

请根据以上结果，生成一份完整的回复。
"""
        return self.llm.generate(prompt)

    def _update_goal_with_error(self, original_goal: str, results: list, error: str) -> str:
        """把失败信息整合到目标中，让 LLM 重新规划"""
        completed_steps = "\n".join(
            f"步骤 {r['step']} ({r['tool']}): {'成功' if r['success'] else '失败'}"
            for r in results
        )
        return (
            f"原始目标: {original_goal}\n"
            f"已完成的步骤:\n{completed_steps}\n"
            f"失败原因: {error}\n"
            f"请重新规划剩余步骤，避免同样的错误。"
        )

几个值得注意的点：

  • 规划和执行完全解耦。plan() 只负责生成计划，PlanExecutor 只负责执行，两者通过 Plan 数据结构传递信息。
  • 失败策略可配置。onfailure 支持 skip（跳过）、retry（重试一次）、replan（重新规划）。实际项目里我一般默认用 replan。
  • 上下文传递很重要。每步结果存入 context，后续步骤的参数模板可以引用前面的结果（比如 "path": "{{step1result}}"）。
  • Replan 不是"每步都重新想"，而是"出了问题才重新想"。这个区别直接影响了 LLM 调用成本。

两种规划策略

Plan 阶段不一定每次都要调 LLM。我习惯先尝试模板匹配，匹配不到再走 LLM。

python
class PlannerStrategy:
    """
    规划策略的两种实现
    不是所有任务都需要 LLM 来规划。
    固定任务用模板就行，灵活任务才需要 LLM。
    """

    @staticmethod
    def template(goal: str) -> Plan:
        """
        模板规划 —— 预定义步骤，按目标类型匹配

        适用场景：任务类型固定、步骤标准化的场景。
        优点：零 LLM 调用、零延迟、100% 可预测。
        缺点：只能处理预定义的任务类型。
        """
        templates = {
            "数据分析": Plan(
                goal="数据分析",
                steps=[
                    PlanStep(1, "读取数据", "read_csv", '{"path": "{{file_path}}"}'),
                    PlanStep(2, "数据清洗", "clean_data", '{"data": "{{step_1_result}}"}'),
                    PlanStep(3, "统计分析", "analyze", '{"data": "{{step_2_result}}"}'),
                    PlanStep(4, "生成图表", "plot_chart", '{"data": "{{step_3_result}}", "type": "bar"}'),
                    PlanStep(5, "生成报告", "generate_report", '{"summary": "{{step_3_result}}", "chart": "{{step_4_result}}"}'),
                ],
                estimated_steps=5,
            ),
            "文件处理": Plan(
                goal="文件处理",
                steps=[
                    PlanStep(1, "读取文件", "read_file", '{"path": "{{file_path}}"}'),
                    PlanStep(2, "处理内容", "process_content", '{"content": "{{step_1_result}}"}'),
                    PlanStep(3, "保存结果", "save_file", '{"content": "{{step_2_result}}", "path": "{{output_path}}"}'),
                ],
                estimated_steps=3,
            ),
        }

        # 简单关键词匹配（实际应该用更好的分类）
        for keyword, plan in templates.items():
            if keyword in goal:
                return plan

        # 没有匹配的模板，返回空计划
        return Plan(goal=goal, steps=[], estimated_steps=0)

    @staticmethod
    def llm(llm_instance, goal: str, tools: list[dict]) -> Plan:
        """
        LLM 规划 —— 动态生成步骤

        适用场景：任务类型多样、无法预定义模板的场景。
        优点：灵活，能处理新任务。
        缺点：一次 LLM 调用，延迟 2-5 秒，结果不可预测。
        """
        tool_schemas = "\n".join(f"- {t['name']}: {t['description']}" for t in tools)

        prompt = f"""
请将以下目标拆解为可执行步骤。

目标: {goal}

可用工具:
{tool_schemas}

规则:
1. 每步使用一个工具
2. 步骤之间有清晰的依赖关系
3. 使用模板语法引用前一步的结果: {{{{step_N_result}}}}

返回 JSON 格式: {{"steps": [...]}}
"""
        response = llm_instance.generate(prompt)
        plan_data = llm_instance.parse_json(response)
        steps = [PlanStep(**s) for s in plan_data["steps"]]
        return Plan(goal=goal, steps=steps, estimated_steps=len(steps))


class AdaptivePlanner:
    """
    自适应规划器 —— 先尝试模板匹配，匹配不到再调 LLM
    90% 的任务可能都是重复类型，不需要每次调 LLM。
    先用模板匹配（零成本），匹配不到再调 LLM（灵活但贵）。
    """

    def __init__(self, llm_instance=None):
        self.llm = llm_instance

    def plan(self, goal: str, tools: list[dict] = None) -> Plan:
        # 尝试模板规划（零成本）
        plan = PlannerStrategy.template(goal)
        if plan.steps:
            return plan

        # 模板匹配不到，用 LLM 规划
        if self.llm is None:
            raise ValueError("没有 LLM 实例，无法进行动态规划")
        return PlannerStrategy.llm(self.llm, goal, tools)

大部分任务是重复类型——数据分析、文件处理、代码执行，模板足够覆盖。只有遇到新任务类型时才需要 LLM。这样可以把 LLM 调用次数从"每次必调"降到"偶尔调"。

计划质量评估

Plan-Execute 最大的风险是计划本身有问题。与其盲目执行，不如先评估再决定要不要让 LLM 重新生成。

python
"""
计划质量评估 —— 完整性、可执行性、依赖正确性
"""


class PlanQualityEvaluator:
    """
    三个维度：
    1. 完整性（completeness）：计划是否覆盖了目标的关键方面？
    2. 可执行性（executability）：每一步的工具和参数是否合法？
    3. 依赖正确性（dependency）：步骤之间的引用是否有效？
    """

    @staticmethod
    def evaluate(plan: Plan, tools: ToolRegistry, goal: str) -> dict:
        """评估计划质量，返回评分和问题列表"""
        issues = []
        scores = {}

        # 维度一：完整性
        completeness = PlanQualityEvaluator._check_completeness(plan, goal)
        scores["completeness"] = completeness["score"]
        issues.extend(completeness["issues"])

        # 维度二：可执行性
        executability = PlanQualityEvaluator._check_executability(plan, tools)
        scores["executability"] = executability["score"]
        issues.extend(executability["issues"])

        # 维度三：依赖正确性
        dependency = PlanQualityEvaluator._check_dependency(plan)
        scores["dependency"] = dependency["score"]
        issues.extend(dependency["issues"])

        # 综合评分
        scores["overall"] = sum(scores.values()) / len(scores)

        return {"scores": scores, "issues": issues}

    @staticmethod
    def _check_completeness(plan: Plan, goal: str) -> dict:
        """
        完整性检查

        策略：
        1. 步骤数是否在合理范围内（太少可能遗漏，太多可能冗余）
        2. 步骤描述中是否包含关键动词（读取、分析、生成、保存等）
        3. 是否有明确的输出步骤（Final Answer / Report）
        """
        issues = []
        score = 100

        # 步骤数检查
        if len(plan.steps) < 2:
            score -= 30
            issues.append(f"计划只有 {len(plan.steps)} 步，可能不完整")
        elif len(plan.steps) > 15:
            score -= 15
            issues.append(f"计划有 {len(plan.steps)} 步，过于复杂，建议拆分")

        # 关键动词检查
        descriptions = " ".join(s.description for s in plan.steps)
        action_verbs = ["读取", "读取", "获取", "加载", "分析", "计算", "生成", "保存", "输出"]
        found_verbs = [v for v in action_verbs if v in descriptions]
        if not found_verbs:
            score -= 20
            issues.append("步骤描述中缺少关键操作动词")

        # 输出步骤检查
        has_output = any("输出" in s.description or "报告" in s.description or "答案" in s.description
                        for s in plan.steps)
        if not has_output and len(plan.steps) > 1:
            score -= 15
            issues.append("计划没有明确的输出/报告步骤")

        return {"score": max(score, 0), "issues": issues}

    @staticmethod
    def _check_executability(plan: Plan, tools: ToolRegistry) -> dict:
        """
        可执行性检查

        策略：
        1. 每步引用的工具是否存在
        2. 参数模板是否是合法 JSON
        3. 参数中引用的前置步骤是否存在
        """
        import json
        import re

        issues = []
        score = 100

        for step in plan.steps:
            # 工具存在性
            if step.tool_name not in tools._tools:
                score -= 25
                issues.append(f"步骤 {step.step_number}: 工具 '{step.tool_name}' 不存在")

            # 参数模板
            try:
                args = json.loads(step.tool_args_template)
                # 检查引用
                for key, value in args.items():
                    refs = re.findall(r'\{\{(\w+)\}\}', str(value))
                    for ref in refs:
                        ref_step = int(ref.split("_")[1]) if ref.startswith("step_") else None
                        if ref_step and ref_step >= step.step_number:
                            score -= 15
                            issues.append(
                                f"步骤 {step.step_number}: 引用了未来步骤的结果 '{ref}'"
                            )
            except json.JSONDecodeError:
                score -= 20
                issues.append(f"步骤 {step.step_number}: 参数模板不是合法 JSON")

        return {"score": max(score, 0), "issues": issues}

    @staticmethod
    def _check_dependency(plan: Plan) -> dict:
        """
        依赖正确性检查

        策略：
        1. 构建依赖图：步骤 A 引用步骤 B 的结果 -> A 依赖 B
        2. 检查是否存在循环依赖（A 依赖 B，B 依赖 A）
        3. 检查是否存在孤立步骤（既不依赖任何步骤，也不被任何步骤依赖）
        """
        import re

        issues = []
        score = 100

        # 构建依赖图
        dependencies: dict[int, set[int]] = {s.step_number: set() for s in plan.steps}

        for step in plan.steps:
            refs = re.findall(r'\{\{step_(\d+)_result\}\}', step.tool_args_template)
            for ref in refs:
                ref_num = int(ref)
                if ref_num < step.step_number:
                    dependencies[step.step_number].add(ref_num)
                elif ref_num == step.step_number:
                    score -= 20
                    issues.append(f"步骤 {step.step_number}: 自引用（引用自己的结果）")
                else:
                    score -= 20
                    issues.append(f"步骤 {step.step_number}: 前向引用了未来步骤 {ref_num}")

        # 检查孤立步骤（不依赖任何步骤且不被任何步骤依赖）
        depended_by: dict[int, set[int]] = {s.step_number: set() for s in plan.steps}
        for step_num, deps in dependencies.items():
            for dep in deps:
                depended_by[dep].add(step_num)

        for step in plan.steps:
            if not dependencies[step.step_number] and not depended_by[step.step_number]:
                if len(plan.steps) > 1:
                    # 第一步或最后一步可以是孤立的
                    if step.step_number not in (1, len(plan.steps)):
                        score -= 10
                        issues.append(f"步骤 {step.step_number}: 孤立步骤，与其他步骤无依赖关系")

        return {"score": max(score, 0), "issues": issues}

使用方式，在 _plan 方法里加一个质量门控：

python
def _plan(self, goal: str) -> Plan:
    plan = self._generate_plan(goal)

    # 质量评估
    evaluation = PlanQualityEvaluator.evaluate(plan, self.tools, goal)
    if evaluation["scores"]["overall"] < 60:
        # 质量不达标，要求 LLM 重新规划
        issues_text = "\n".join(f"- {issue}" for issue in evaluation["issues"])
        return self._plan_with_feedback(goal, issues_text)

    return plan

Replan 的三种策略

执行失败时，不是所有情况都需要"从头规划"。根据失败原因和位置选择合适的策略，可以节省 LLM 调用和时间。

python
"""
Replan 的三种策略
执行失败时，不是所有情况都需要"从头规划"。
根据失败原因选择合适的 Replan 策略，可以节省 LLM 调用次数和时间。
"""


class Replanner:
    """
    重规划器 —— 根据失败原因选择合适的 Replan 策略
    """

    @staticmethod
    def incremental(plan: Plan, failed_step: int, error: str, results: list[dict]) -> Plan:
        """
        策略一：增量修补

        保留已完成步骤的结果，只修复失败步骤及其后续步骤。
        不重新调用 LLM，只修改参数或替换工具。

        适用场景：
        - 工具不存在 -> 替换为等价工具
        - 参数格式错误 -> 修正参数格式
        - 权限/路径错误 -> 修正路径

        优点：零 LLM 调用、速度快。
        缺点：只能修复局部问题，无法解决全局计划错误。
        """
        # 找到失败步骤
        failed_step_obj = next((s for s in plan.steps if s.step_number == failed_step), None)
        if not failed_step_obj:
            return plan

        # 简单的错误-修复映射
        fixes = {
            "未知工具": lambda s: s,  # 实际场景应该有工具映射表
            "文件不存在": lambda s: PlanStep(
                s.step_number,
                f"检查{failed_step_obj.description}的文件路径",
                "check_file",
                '{"path": "..."}',
            ),
            "权限不足": lambda s: PlanStep(
                s.step_number,
                f"使用管理员权限执行{failed_step_obj.description}",
                "execute_as_admin",
                s.tool_args_template,
            ),
        }

        for error_keyword, fix_func in fixes.items():
            if error_keyword in error:
                plan.steps[failed_step - 1] = fix_func(failed_step_obj)
                return plan

        # 没有匹配的修复规则，返回原计划
        return plan

    @staticmethod
    def partial_replan(
        llm, plan: Plan, failed_step: int, error: str, results: list[dict], tools: list[dict]
    ) -> Plan:
        """
        策略二：部分重规划

        保留已完成步骤的结果，从失败步骤开始重新规划后续步骤。
        需要调用一次 LLM，但只规划剩余步骤。

        适用场景：
        - 失败步骤之后的步骤不再适用（比如中间结果格式变了）
        - 发现新的信息，需要调整后续策略

        优点：保留了已完成工作的成果，只重新规划剩余部分。
        缺点：LLM 需要理解已完成步骤的上下文，prompt 更复杂。
        """
        completed_results = "\n".join(
            f"步骤 {r['step']} ({r['tool']}): {'成功' if r['success'] else '失败'} - {r.get('output', '')}"
            for r in results
        )

        prompt = f"""
原始计划从步骤 {failed_step} 开始执行失败。
失败原因: {error}

已完成的步骤结果:
{completed_results}

可用工具:
{tools}

请从步骤 {failed_step} 开始重新规划剩余步骤。
考虑已完成步骤的实际结果（可能与原计划预期不同）。

返回 JSON: {{"steps": [...]}}
"""
        response = llm.generate(prompt)
        plan_data = llm.parse_json(response)

        # 替换失败步骤及之后的步骤
        new_steps = [PlanStep(**s) for s in plan_data["steps"]]
        # 重新编号
        remaining_start_idx = failed_step - 1
        plan.steps = plan.steps[:remaining_start_idx] + new_steps

        # 重新编号所有步骤
        for i, step in enumerate(plan.steps):
            step.step_number = i + 1

        return plan

    @staticmethod
    def full_replan(llm, goal: str, error: str, results: list[dict], tools: list[dict]) -> Plan:
        """
        策略三：全盘重规划

        放弃原计划，从零开始生成全新的计划。
        需要调用一次 LLM。

        适用场景：
        - 原计划本身有根本性错误（方向不对）
        - 失败步骤太早（大部分步骤都还没执行）
        - 增量修补和部分重规划都失败了

        优点：最彻底，最可能找到正确的方案。
        缺点：成本最高（一次完整 LLM 调用），已完成的工作被丢弃。
        """
        completed_summary = "\n".join(
            f"步骤 {r['step']}: {r.get('output', '')[:200]}" for r in results if r.get("success")
        )

        prompt = f"""
目标: {goal}

之前的执行尝试失败了:
{error}

已完成的部分工作:
{completed_summary}

请重新规划，避免同样的错误。

可用工具:
{tools}

返回 JSON: {{"steps": [...]}}
"""
        response = llm.generate(prompt)
        plan_data = llm.parse_json(response)
        steps = [PlanStep(**s) for s in plan_data["steps"]]
        return Plan(goal=goal, steps=steps, estimated_steps=len(steps))


class AdaptiveReplanner:
    """
    自适应 Replan 策略选择器
    根据失败步骤的位置和错误类型，自动选择最合适的 Replan 策略。
    """

    @staticmethod
    def choose(failed_step: int, total_steps: int, error: str) -> str:
        """
        选择 Replan 策略

        已知可修复的错误 -> incremental
        失败步骤在后期（> 70%）-> partial_replan（已完成大部分，不值得全盘重来）
        失败步骤在前期（< 30%）-> full_replan（反正没做多少，重做不亏）
        其他 -> partial_replan（默认最安全的选择）
        """
        # 已知可修复的错误
        incremental_errors = ["未知工具", "文件不存在", "权限不足", "参数格式错误"]
        if any(e in error for e in incremental_errors):
            return "incremental"

        # 根据失败位置
        progress = failed_step / total_steps if total_steps > 0 else 0

        if progress > 0.7:
            return "partial_replan"  # 已完成大部分，只修剩余
        elif progress < 0.3:
            return "full_replan"  # 才刚开始，重做不亏
        else:
            return "partial_replan"  # 中期，部分重规划

三种策略的成本对比：

策略LLM 调用保留进度修复范围适合场景
增量修补0 次100%单步参数错误、工具替换
部分重规划1 次失败前的步骤失败步骤及之后中间结果变化
全盘重规划1 次0%全部计划方向错误

依赖图与拓扑排序

Plan 中步骤之间的参数引用形成了一个有向图。如果图有环或前向引用，执行一定会失败。在执行前做拓扑排序验证，可以提前发现问题。

python
"""
参数依赖图 —— 用拓扑排序验证计划的可执行性
步骤之间的参数引用形成有向边。
如果图有环或前向引用，计划不可执行。
在执行前先做拓扑排序验证，提前发现问题。
"""

from collections import defaultdict, deque
import re


def build_dependency_graph(steps: list) -> tuple[dict, dict]:
    """从计划步骤构建依赖图"""
    graph = defaultdict(set)
    reverse_graph = defaultdict(set)

    for step in steps:
        refs = re.findall(r'\{\{step_(\d+)_result\}\}', step.tool_args_template)
        for ref in refs:
            ref_num = int(ref)
            graph[step.step_number].add(ref_num)
            reverse_graph[ref_num].add(step.step_number)

    return graph, reverse_graph


def topological_sort(steps: list) -> tuple[list[int], list[str]]:
    """拓扑排序——验证步骤依赖是否合理"""
    graph, reverse_graph = build_dependency_graph(steps)
    all_steps = {s.step_number for s in steps}

    in_degree = {s: len(graph.get(s, set()) & all_steps) for s in all_steps}
    queue = deque([s for s in all_steps if in_degree[s] == 0])
    sorted_order = []
    issues = []

    while queue:
        node = queue.popleft()
        sorted_order.append(node)
        for dependent in reverse_graph.get(node, set()):
            if dependent in in_degree:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)

    if len(sorted_order) < len(all_steps):
        remaining = all_steps - set(sorted_order)
        issues.append(f"存在循环依赖，涉及步骤: {remaining}")

    # 检查前向引用
    for step in steps:
        refs = re.findall(r'\{\{step_(\d+)_result\}\}', step.tool_args_template)
        for ref in refs:
            ref_num = int(ref)
            if ref_num >= step.step_number:
                issues.append(
                    f"步骤 {step.step_number}: 前向引用了步骤 {ref_num}"
                )

    return sorted_order, issues

对于有 10+ 步的计划，提前发现依赖问题可以省掉大量执行时间和 LLM 调用。

与 ReAct 的混合模式

Plan-Execute 和 ReAct 不是替代关系，可以组合使用。

python
class HybridAgent:
    """
    混合 Agent —— 先 Plan-Execute，遇到问题切换 ReAct

    核心逻辑：
    1. 先尝试 Plan-Execute（快、便宜）
    2. 如果执行失败或步骤不足，切换 ReAct（灵活、贵）
    3. ReAct 的结果可以更新 Plan-Execute 的模板

    适用场景：大部分任务可以模板化，但偶尔有复杂任务需要灵活推理。
    """

    def __init__(self, llm, tool_registry):
        self.llm = llm
        self.tools = tool_registry
        self.plan_templates = {}  # 积累的模板库

    def run(self, goal: str) -> str:
        # Phase 1：尝试 Plan-Execute
        plan = self._find_or_create_plan(goal)

        if plan.steps:
            executor = PlanExecutor(self.tools, on_failure="replan")
            result = executor.execute(plan)

            if result["status"] == "completed":
                return self._synthesize(goal, result)

            # Plan-Execute 失败，进入 Phase 2

        # Phase 2：切换 ReAct
        if self._should_fallback_to_react(goal):
            react_agent = ReActAgent(self.llm, self.tools)
            return react_agent.run(goal)

        return "无法完成任务。"

    def _find_or_create_plan(self, goal: str) -> Plan:
        """查找已有模板，或创建新计划"""
        # 先查找匹配的模板
        for keyword, plan in self.plan_templates.items():
            if keyword in goal:
                return plan
        return Plan(goal=goal, steps=[], estimated_steps=0)

    def _should_fallback_to_react(self, goal: str) -> bool:
        """判断是否需要回退到 ReAct"""
        # 简单规则：目标中包含"探索"、"分析"、"比较"等词时，ReAct 更合适
        react_keywords = ["探索", "分析", "比较", "调查", "研究", "为什么"]
        return any(kw in goal for kw in react_keywords)

这个模式解决了实际问题——你不可能预先知道所有任务类型。Plan-Execute 处理已知的、重复的任务，ReAct 处理新的、复杂的任务。ReAct 执行成功的经验还可以沉淀为新的 Plan-Execute 模板。

Plan-Execute vs ReAct 怎么选

不是"哪个好"的问题，是适合什么场景。

Plan-Execute 适合步骤明确、流程固定、追求效率和成本的场景，比如数据处理 pipeline、自动化测试、批量文件处理。ReAct 适合步骤不确定、需要中间决策、涉及探索或搜索的场景，比如信息检索、多步推理、动态工具选择。

一个简单的判断标准：如果你能预先写出完整的步骤列表，用 Plan-Execute。如果写不出来（因为中间步骤依赖前面的结果），用 ReAct。

更本质的是看任务的"探索性"。"读取 CSV -> 清洗 -> 分析 -> 画图"这种任务步骤确定，Plan-Execute 效率远高于 ReAct。"调查一下这家公司为什么业绩下滑"这种任务，你不知道中间会发现什么信息，ReAct 的"边做边想"模式更合适。

计划不准确怎么办？这是 Plan-Execute 最大的风险。我的做法：尽量用模板替代 LLM 规划，模板是人写的，比 LLM 可靠。执行失败时触发 Replan，不要一条路走到黑。执行前加一步计划审查，让质量评估过滤掉明显不靠谱的计划。

Plan-Execute 的代价也要清楚。LLM 生成的计划可能遗漏步骤、选错工具、参数不对，执行阶段无法自我纠正（除非有 Replan）。一旦计划生成完毕，执行过程中遇到意外情况，Agent 不知道怎么处理。如果用模板规划，每种任务类型需要一个模板，任务类型多的时候模板库不好维护。步骤之间通过 {{stepNresult}} 传递数据，步骤多了之后引用链很长，调试困难。

什么时候不该用：任务就是"调一个工具 -> 输出结果"，直接调就行。步骤是写死的代码永远不变，不需要 LLM 规划。每一步都依赖上一步的结果来决定下一步做什么，该用 ReAct。

框架中的实际应用

框架Plan-Execute 实现说明
LangGraph可以用 StateGraph 定义 Plan -> Execute 的流程LangGraph 本身就是状态机，Plan-Execute 是最简单的状态图
CrewAICrew 定义任务列表 -> kickoff() 执行CrewAI 的任务列表本质上就是一个 Plan，kickoff 是 Execute
AutoGenGroupChat + 规划 AgentAutoGen 中可以通过规划 Agent 生成计划，其他 Agent 执行
自定义 PipelineDAG/Workflow 引擎很多团队用 Airflow、Prefect 等 DAG 引擎做 Plan-Execute，LLM 只负责规划

演进路径

实际做 Plan-Execute 一般经历这几步：

起步阶段，直接硬编码步骤——data = readcsv("sales.csv")，cleaned = cleandata(data)，result = analyze(cleaned)。后来发现步骤需要可配置，就变成 pipeline 数组加循环执行。再后来发现任务类型变多了，硬编码不够用，引入 LLM 生成计划。最后发现计划经常不准，加上 Replan 机制，执行失败时重新规划。

写在最后

Plan-Execute 和 ReAct 互补，不是替代。优势是效率和可预测性，代价是灵活性和容错能力。最实用的方案是两者组合——先用 Plan-Execute 处理已知重复任务，遇到新任务或不准确时回退到 ReAct。