Plan-and-Execute 架构模式

┌─────────────────────────────────────────────────────────────┐
│                  Plan-and-Execute 设计哲学                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ReAct 模式：                                              │
│   ┌─────────┐     ┌─────────┐     ┌─────────┐             │
│   │ 思考    │ ──→ │ 行动    │ ──→ │ 观察    │             │
│   └─────────┘     └─────────┘     └─────────┘             │
│        ↑                                  │                │
│        └──────────────────────────────────┘                │
│        边思考边执行，每步都需要 LLM 推理                     │
│                                                             │
│   Plan-and-Execute 模式：                                   │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                Phase 1: Planning                     │  │
│   │   ┌─────────┐     ┌─────────┐     ┌─────────┐       │  │
│   │   │ 任务    │ ──→ │ 规划器  │ ──→ │ 计划    │       │  │
│   │   │ 输入    │     │Planner  │     │ Plan    │       │  │
│   │   └─────────┘     └─────────┘     └─────────┘       │  │
│   └─────────────────────────────────────────────────────┘  │
│                             │                               │
│                             ↓                               │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                Phase 2: Execution                    │  │
│   │   ┌─────────┐     ┌─────────┐     ┌─────────┐       │  │
│   │   │ 步骤1   │ ──→ │ 步骤2   │ ──→ │ 步骤N   │       │  │
│   │   │ 执行    │     │ 执行    │     │ 执行    │       │  │
│   │   └─────────┘     └─────────┘     └─────────┘       │  │
│   │        执行器（Executor）按计划逐步执行               │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                  两阶段工作流程详解                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Phase 1: Planning（规划阶段）                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                                                      │   │
│  │   1. 任务理解                                        │   │
│  │      • 分析用户意图                                  │   │
│  │      • 提取关键信息                                  │   │
│  │      • 识别约束条件                                  │   │
│  │                                                      │   │
│  │   2. 子任务分解                                      │   │
│  │      • 将大任务拆分为小任务                          │   │
│  │      • 确定任务粒度                                  │   │
│  │      • 定义任务边界                                  │   │
│  │                                                      │   │
│  │   3. 依赖分析                                        │   │
│  │      • 识别任务间依赖关系                            │   │
│  │      • 构建执行顺序                                  │   │
│  │      • 标记可并行任务                                │   │
│  │                                                      │   │
│  │   4. 工具匹配                                        │   │
│  │      • 为每个子任务分配工具                          │   │
│  │      • 预估资源需求                                  │   │
│  │      • 制定备选方案                                  │   │
│  │                                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  Phase 2: Execution（执行阶段）                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                                                      │   │
│  │   for each step in plan:                             │   │
│  │       1. 检查前置依赖是否完成                        │   │
│  │       2. 调用对应工具执行                            │   │
│  │       3. 记录结果 + 更新状态                         │   │
│  │       4. [可选] 动态调整后续计划                     │   │
│  │                                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                Plan-and-Execute 完整工作流程                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   ┌─────────────┐                                                  │
│   │  用户输入   │                                                  │
│   │  (Task)     │                                                  │
│   └──────┬──────┘                                                  │
│          │                                                         │
│          ↓                                                         │
│   ┌─────────────────────────────────────────────────────────────┐ │
│   │                    规划阶段 (Planning)                       │ │
│   │  ┌───────────────────────────────────────────────────────┐  │ │
│   │  │                     Planner                            │  │ │
│   │  │                                                       │  │ │
│   │  │   Task → 理解 → 分解 → 排序 → Plan                    │  │ │
│   │  │                                                       │  │ │
│   │  │   输出：                                              │  │ │
│   │  │   [Step1, Step2, Step3, ..., StepN]                  │  │ │
│   │  │   + 依赖关系                                          │  │ │
│   │  │   + 工具分配                                          │  │ │
│   │  └───────────────────────────────────────────────────────┘  │ │
│   └─────────────────────────────────────────────────────────────┘ │
│          │                                                         │
│          ↓                                                         │
│   ┌─────────────────────────────────────────────────────────────┐ │
│   │                    执行阶段 (Execution)                      │ │
│   │                                                              │ │
│   │   ┌─────────┐    ┌─────────┐    ┌─────────┐                │ │
│   │   │ Step 1  │───→│ Step 2  │───→│ Step N  │                │ │
│   │   │ 执行    │    │ 执行    │    │ 执行    │                │ │
│   │   └────┬────┘    └────┬────┘    └────┬────┘                │ │
│   │        │              │              │                      │ │
│   │        ↓              ↓              ↓                      │ │
│   │   ┌─────────┐    ┌─────────┐    ┌─────────┐                │ │
│   │   │ 检查    │    │ 检查    │    │ 检查    │                │ │
│   │   │ 结果    │    │ 结果    │    │ 结果    │                │ │
│   │   └────┬────┘    └────┬────┘    └────┬────┘                │ │
│   │        │              │              │                      │ │
│   │        └──────────────┼──────────────┘                      │ │
│   │                       │                                     │ │
│   │                       ↓                                     │ │
│   │              ┌─────────────────┐                           │ │
│   │              │   需要重规划？  │                           │ │
│   │              └────────┬────────┘                           │ │
│   │                       │                                     │ │
│   │         ┌─────────────┴─────────────┐                      │ │
│   │         ↓                           ↓                      │ │
│   │    ┌─────────┐               ┌───────────┐                │ │
│   │    │   是    │               │    否     │                │ │
│   │    │ Replan  │               │  继续执行  │                │ │
│   │    └─────────┘               └───────────┘                │ │
│   │                                                              │ │
│   └─────────────────────────────────────────────────────────────┘ │
│          │                                                         │
│          ↓                                                         │
│   ┌─────────────────────┐                                          │
│   │   生成最终结果      │                                          │
│   │   (Final Output)    │                                          │
│   └─────────────────────┘                                          │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# 计划的数据结构
@dataclass
class Plan:
    """执行计划"""
    goal: str                    # 总目标
    steps: List[PlanStep]        # 步骤列表
    dependencies: Dict[str, List[str]]  # 依赖关系
    
@dataclass
class PlanStep:
    """单个步骤"""
    id: str                      # 步骤ID
    description: str             # 步骤描述
    tool: str                    # 使用的工具
    tool_input: Dict[str, Any]   # 工具输入
    depends_on: List[str]        # 依赖的步骤ID
    status: StepStatus           # 状态
    
class StepStatus(Enum):
    """步骤状态"""
    PENDING = "pending"          # 待执行
    RUNNING = "running"          # 执行中
    COMPLETED = "completed"      # 已完成
    FAILED = "failed"            # 失败
    SKIPPED = "skipped"          # 已跳过

┌─────────────────────────────────────────────────────────────┐
│                    执行流程详解                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  示例任务：分析销售数据并生成报告                           │
│                                                             │
│  Plan:                                                      │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Step 1: load_data                                   │   │
│  │         - 工具: file_reader                         │   │
│  │         - 输入: "sales_2024.csv"                    │   │
│  │         - 依赖: 无                                  │   │
│  │                                                      │   │
│  │ Step 2: clean_data                                  │   │
│  │         - 工具: data_cleaner                        │   │
│  │         - 输入: {data: Step1.result}                │   │
│  │         - 依赖: [Step1]                             │   │
│  │                                                      │   │
│  │ Step 3: analyze_data                                │   │
│  │         - 工具: data_analyzer                       │   │
│  │         - 输入: {data: Step2.result}                │   │
│  │         - 依赖: [Step2]                             │   │
│  │                                                      │   │
│  │ Step 4: generate_report                             │   │
│  │         - 工具: report_generator                    │   │
│  │         - 输入: {analysis: Step3.result}            │   │
│  │         - 依赖: [Step3]                             │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  执行过程：                                                 │
│                                                             │
│  [Step 1] load_data                                         │
│  ├─ 检查依赖: ✓ 无依赖                                      │
│  ├─ 执行工具: file_reader("sales_2024.csv")                │
│  ├─ 结果: DataFrame(1000 rows, 10 columns)                 │
│  └─ 状态: completed ✓                                       │
│                                                             │
│  [Step 2] clean_data                                        │
│  ├─ 检查依赖: ✓ Step1 已完成                                │
│  ├─ 执行工具: data_cleaner(Step1.result)                   │
│  ├─ 结果: DataFrame(950 rows, 10 columns) # 去除空值       │
│  └─ 状态: completed ✓                                       │
│                                                             │
│  [Step 3] analyze_data                                      │
│  ├─ 检查依赖: ✓ Step2 已完成                                │
│  ├─ 执行工具: data_analyzer(Step2.result)                  │
│  ├─ 结果: {total: 1000000, top_product: "A", ...}          │
│  └─ 状态: completed ✓                                       │
│                                                             │
│  [Step 4] generate_report                                   │
│  ├─ 检查依赖: ✓ Step3 已完成                                │
│  ├─ 执行工具: report_generator(Step3.result)               │
│  ├─ 结果: "sales_report_2024.pdf"                          │
│  └─ 状态: completed ✓                                       │
│                                                             │
│  最终输出: sales_report_2024.pdf                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

"""
Plan-and-Execute Agent 基础实现
"""
 
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
 
# 导入 LangChain 相关模块
from langchain.llms import BaseLLM
from langchain.tools import BaseTool
from langchain.prompts import PromptTemplate
 
 
class StepStatus(Enum):
    """步骤状态"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
 
 
@dataclass
class PlanStep:
    """计划步骤"""
    id: str
    description: str
    tool: str
    tool_input: Dict[str, Any]
    depends_on: List[str] = field(default_factory=list)
    status: StepStatus = StepStatus.PENDING
    result: Optional[Any] = None
 
 
@dataclass
class Plan:
    """执行计划"""
    goal: str
    steps: List[PlanStep]
    
    def get_ready_steps(self) -> List[PlanStep]:
        """获取可以执行的步骤（依赖已满足）"""
        ready = []
        for step in self.steps:
            if step.status != StepStatus.PENDING:
                continue
            # 检查依赖
            deps_satisfied = all(
                self.get_step(dep_id).status == StepStatus.COMPLETED
                for dep_id in step.depends_on
            )
            if deps_satisfied:
                ready.append(step)
        return ready
    
    def get_step(self, step_id: str) -> Optional[PlanStep]:
        """获取指定步骤"""
        for step in self.steps:
            if step.id == step_id:
                return step
        return None
    
    def is_completed(self) -> bool:
        """检查计划是否完成"""
        return all(
            step.status == StepStatus.COMPLETED 
            for step in self.steps
        )
 
 
class Planner:
    """规划器：生成执行计划"""
    
    PLANNER_PROMPT = """
你是一个任务规划专家。请将用户的任务分解为具体的执行步骤。
 
可用工具：
\{tool_descriptions\}
 
任务：\{task\}
 
请生成执行计划，格式如下（JSON格式）：
\{
  "goal": "任务目标",
  "steps": [
    \{
      "id": "step_1",
      "description": "步骤描述",
      "tool": "工具名称",
      "tool_input": \{"param": "value"\},
      "depends_on": []
    \}
  ]
\}
 
要求：
1. 每个步骤使用一个工具
2. 明确标注步骤间的依赖关系
3. 步骤粒度适中，不要过大或过小
 
计划：
"""
    
    def __init__(self, llm: BaseLLM, tools: List[BaseTool]):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
    
    def plan(self, task: str) -> Plan:
        """生成执行计划"""
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])
        
        prompt = self.PLANNER_PROMPT.format(
            tool_descriptions=tool_descriptions,
            task=task
        )
        
        response = self.llm(prompt)
        
        # 解析 JSON
        plan_dict = self._parse_json(response)
        
        # 构建 Plan 对象
        steps = [
            PlanStep(
                id=s["id"],
                description=s["description"],
                tool=s["tool"],
                tool_input=s.get("tool_input", {}),
                depends_on=s.get("depends_on", [])
            )
            for s in plan_dict.get("steps", [])
        ]
        
        return Plan(goal=plan_dict.get("goal", task), steps=steps)
    
    def _parse_json(self, text: str) -> Dict:
        """从文本中解析 JSON"""
        import re
        json_match = re.search(r'```json\s*(.*?)\s*```', text, re.DOTALL)
        if json_match:
            return json.loads(json_match.group(1))
        return json.loads(text)
 
 
class Executor:
    """执行器：执行计划步骤"""
    
    def __init__(self, tools: List[BaseTool]):
        self.tools = {tool.name: tool for tool in tools}
        self.results: Dict[str, Any] = {}  # 存储步骤结果
    
    def execute_step(self, step: PlanStep, plan: Plan) -> Any:
        """执行单个步骤"""
        if step.tool not in self.tools:
            raise ValueError(f"未知工具: {step.tool}")
        
        tool = self.tools[step.tool]
        
        # 替换输入中的步骤引用
        tool_input = self._resolve_input(step.tool_input, plan)
        
        # 执行工具
        result = tool.run(tool_input)
        
        # 存储结果
        self.results[step.id] = result
        step.result = result
        step.status = StepStatus.COMPLETED
        
        return result
    
    def _resolve_input(self, tool_input: Any, plan: Plan) -> Any:
        """解析输入中的步骤引用"""
        if isinstance(tool_input, dict):
            return {
                k: self._resolve_input(v, plan)
                for k, v in tool_input.items()
            }
        elif isinstance(tool_input, str):
            # 检查是否是步骤引用，如 "Step1.result"
            if "." in tool_input:
                parts = tool_input.split(".")
                step_id = parts[0]
                if step_id in self.results:
                    return self.results[step_id]
        return tool_input
 
 
class Replanner:
    """重规划器：动态调整计划"""
    
    REPLAN_PROMPT = """
当前计划执行遇到问题，需要调整。
 
原计划：
\{original_plan\}
 
已完成的步骤：
\{completed_steps\}
 
失败的步骤：
\{failed_step\}
失败原因：\{error\}
 
请生成调整后的计划（JSON格式）：
\{
  "adjustment": "keep|modify|restart",
  "new_steps": [...],
  "reason": "调整原因"
\}
"""
    
    def __init__(self, llm: BaseLLM):
        self.llm = llm
    
    def replan(
        self, 
        plan: Plan, 
        failed_step: PlanStep, 
        error: str
    ) -> Plan:
        """重新规划"""
        prompt = self.REPLAN_PROMPT.format(
            original_plan=self._format_plan(plan),
            completed_steps=[s.id for s in plan.steps if s.status == StepStatus.COMPLETED],
            failed_step=failed_step.id,
            error=error
        )
        
        response = self.llm(prompt)
        # 解析并返回新计划
        # ... 实现略
        return plan
    
    def _format_plan(self, plan: Plan) -> str:
        return "\n".join([
            f"- {s.id}: {s.description} [{s.status.value}]"
            for s in plan.steps
        ])
 
 
class PlanExecuteAgent:
    """Plan-and-Execute Agent"""
    
    def __init__(
        self,
        llm: BaseLLM,
        tools: List[BaseTool],
        max_replans: int = 3
    ):
        self.llm = llm
        self.tools = tools
        self.planner = Planner(llm, tools)
        self.executor = Executor(tools)
        self.replanner = Replanner(llm)
        self.max_replans = max_replans
    
    def run(self, task: str) -> str:
        """运行 Agent"""
        # Phase 1: Planning
        print(f"[Planning] 正在规划任务: {task}")
        plan = self.planner.plan(task)
        print(f"[Planning] 生成计划，共 {len(plan.steps)} 个步骤")
        
        # Phase 2: Execution
        replan_count = 0
        
        while not plan.is_completed():
            # 获取可执行的步骤
            ready_steps = plan.get_ready_steps()
            
            if not ready_steps:
                # 没有可执行步骤，可能是有循环依赖或全部失败
                print("[Execution] 无法继续执行")
                break
            
            # 执行步骤
            for step in ready_steps:
                print(f"[Execution] 执行步骤: {step.id} - {step.description}")
                step.status = StepStatus.RUNNING
                
                try:
                    result = self.executor.execute_step(step, plan)
                    print(f"[Execution] 步骤 {step.id} 完成")
                except Exception as e:
                    print(f"[Execution] 步骤 {step.id} 失败: {e}")
                    step.status = StepStatus.FAILED
                    
                    # 尝试重规划
                    if replan_count < self.max_replans:
                        print(f"[Replanning] 正在重新规划...")
                        plan = self.replanner.replan(plan, step, str(e))
                        replan_count += 1
                    break
        
        # 生成最终结果
        return self._generate_final_output(plan)
    
    def _generate_final_output(self, plan: Plan) -> str:
        """生成最终输出"""
        completed = [s for s in plan.steps if s.status == StepStatus.COMPLETED]
        failed = [s for s in plan.steps if s.status == StepStatus.FAILED]
        
        output = f"任务执行完成\n"
        output += f"- 完成步骤: {len(completed)}/{len(plan.steps)}\n"
        
        if failed:
            output += f"- 失败步骤: {', '.join(s.id for s in failed)}\n"
        
        # 返回最后一个成功步骤的结果
        if completed:
            last_result = completed[-1].result
            output += f"\n结果: {last_result}"
        
        return output
 
 
# 使用示例
if __name__ == "__main__":
    from langchain_openai import OpenAI
    from langchain.tools import Tool
    
    # 定义工具
    def search_tool(query: str) -> str:
        return f"搜索结果: {query}"
    
    def analyze_tool(data: str) -> str:
        return f"分析结果: {data}"
    
    tools = [
        Tool(name="search", func=search_tool, description="搜索信息"),
        Tool(name="analyze", func=analyze_tool, description="分析数据")
    ]
    
    # 创建 Agent
    llm = OpenAI(temperature=0)
    agent = PlanExecuteAgent(llm=llm, tools=tools)
    
    # 运行
    result = agent.run("研究人工智能发展趋势并生成报告")
    print(result)

"""
使用 LangGraph 实现 Plan-and-Execute
推荐的生产环境实现方式
"""
 
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
import operator
 
 
# 定义状态
class PlanExecuteState(TypedDict):
    """Plan-Execute 状态"""
    input: str                              # 用户输入
    plan: List[str]                         # 执行计划
    past_steps: Annotated[List, operator.add]  # 已执行步骤
    response: str                           # 最终响应
 
 
def create_plan_execute_graph():
    """创建 Plan-Execute 图"""
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    # 规划节点
    def plan_step(state: PlanExecuteState):
        """生成执行计划"""
        task = state["input"]
        
        prompt = f"""将以下任务分解为步骤：
任务：\{task\}
 
输出格式：
1. 步骤一
2. 步骤二
...
"""
        response = llm.invoke(prompt)
        plan = response.content.strip().split("\n")
        return \{"plan": plan\}
    
    # 执行节点
    def execute_step(state: PlanExecuteState):
        """执行单个步骤"""
        plan = state["plan"]
        past_steps = state.get("past_steps", [])
        
        # 获取下一个要执行的步骤
        next_step = plan[len(past_steps)]
        
        # 执行步骤（这里简化处理）
        result = f"执行 {next_step} 完成"
        
        return \{"past_steps": [(next_step, result)]\}
    
    # 条件判断
    def should_continue(state: PlanExecuteState):
        """判断是否继续执行"""
        plan = state["plan"]
        past_steps = state.get("past_steps", [])
        
        if len(past_steps) < len(plan):
            return "execute"
        return "respond"
    
    # 响应节点
    def respond(state: PlanExecuteState):
        """生成最终响应"""
        past_steps = state.get("past_steps", [])
        
        response = "任务执行完成：\n"
        for step, result in past_steps:
            response += f"- {step}: {result}\n"
        
        return \{"response": response\}
    
    # 创建图
    workflow = StateGraph(PlanExecuteState)
    
    # 添加节点
    workflow.add_node("planner", plan_step)
    workflow.add_node("execute", execute_step)
    workflow.add_node("respond", respond)
    
    # 设置入口
    workflow.set_entry_point("planner")
    
    # 添加边
    workflow.add_edge("planner", "execute")
    workflow.add_conditional_edges(
        "execute",
        should_continue,
        {
            "execute": "execute",
            "respond": "respond"
        }
    )
    workflow.add_edge("respond", END)
    
    return workflow.compile()
 
 
# 使用示例
if __name__ == "__main__":
    graph = create_plan_execute_graph()
    
    result = graph.invoke({
        "input": "分析销售数据并生成报告"
    })
    
    print(result["response"])

┌─────────────────────────────────────────────────────────────┐
│                Plan-and-Execute 适用场景详解                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 数据分析工作流                                          │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 任务：分析用户行为数据，生成洞察报告                  │   │
│  │                                                      │   │
│  │ Plan:                                                │   │
│  │ Step1: 加载数据 (file_loader)                        │   │
│  │ Step2: 数据清洗 (data_cleaner)                       │   │
│  │ Step3: 统计分析 (statistical_analyzer)               │   │
│  │ Step4: 可视化图表 (chart_generator)                  │   │
│  │ Step5: 生成报告 (report_generator)                   │   │
│  │                                                      │   │
│  │ 优势：步骤清晰，每步结果可验证，支持断点续执行        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  2. 多步骤信息收集                                          │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 任务：调研竞品并生成分析报告                          │   │
│  │                                                      │   │
│  │ Plan:                                                │   │
│  │ Step1: 搜索竞品A信息 (search)                        │   │
│  │ Step2: 搜索竞品B信息 (search) ─┐ 可并行执行          │   │
│  │ Step3: 搜索竞品C信息 (search) ─┘                      │   │
│  │ Step4: 对比分析 (analyzer)                           │   │
│  │ Step5: 生成报告 (reporter)                           │   │
│  │                                                      │   │
│  │ 优势：无依赖步骤可并行，提升效率                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  3. 企业流程自动化                                          │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 任务：处理客户订单                                    │   │
│  │                                                      │   │
│  │ Plan:                                                │   │
│  │ Step1: 验证订单信息 (validator)                      │   │
│  │ Step2: 检查库存 (inventory_checker)                  │   │
│  │ Step3: 创建发货单 (shipping_creator)                 │   │
│  │ Step4: 发送确认邮件 (email_sender)                   │   │
│  │ Step5: 更新订单状态 (status_updater)                 │   │
│  │                                                      │   │
│  │ 优势：执行轨迹完整，满足审计要求                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                  Plan-and-Execute 优化策略                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 渐进式规划                                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 问题：一次性规划可能不准确                           │   │
│  │                                                      │   │
│  │ 优化方案：                                           │   │
│  │ • 先规划大致框架，细节逐步填充                       │   │
│  │ • 关键步骤详细规划，其他步骤粗略规划                 │   │
│  │ • 执行过程中按需细化下一步计划                       │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  2. 智能重规划                                              │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 问题：失败后完全重新规划成本高                       │   │
│  │                                                      │   │
│  │ 优化方案：                                           │   │
│  │ • 局部修复：只调整受影响的步骤                       │   │
│  │ • 备选路径：预设多个备选方案                         │   │
│  │ • 回滚重试：失败步骤尝试其他参数                     │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  3. 并行执行                                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 问题：串行执行效率低                                 │   │
│  │                                                      │   │
│  │ 优化方案：                                           │   │
│  │ • 识别无依赖的步骤                                   │   │
│  │ • 使用线程池并行执行                                 │   │
│  │ • 结果合并后继续后续步骤                             │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  4. 计划缓存                                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 问题：相似任务重复规划                               │   │
│  │                                                      │   │
│  │ 优化方案：                                           │   │
│  │ • 缓存历史计划模板                                   │   │
│  │ • 相似任务复用计划结构                               │   │
│  │ • 参数化模板提高复用性                               │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

"""
Plan-ReAct 混合架构
结合两者优势：规划的可控性 + 执行的灵活性
"""
 
class HybridAgent:
    """混合架构 Agent"""
    
    def run(self, task: str) -> str:
        # 1. 高层规划
        plan = self.planner.plan(task)
        
        for step in plan.steps:
            # 2. 每个步骤使用 ReAct 执行
            step_result = self.react_executor.execute(step)
            
            # 3. 关键步骤进行验证
            if step.is_critical:
                if not self.validator.validate(step_result):
                    # 4. 失败时重规划
                    plan = self.replanner.adjust(plan, step, step_result)
        
        return self.aggregate_results()

def handle_failure(step, error, plan):
    """处理执行失败"""
    
    # 1. 错误分析
    error_type = classify_error(error)
    
    # 2. 根据错误类型选择策略
    if error_type == "tool_error":
        # 工具错误：尝试备选工具
        return retry_with_alternative(step)
    
    elif error_type == "dependency_error":
        # 依赖错误：检查前置步骤
        return check_and_fix_dependencies(step, plan)
    
    elif error_type == "input_error":
        # 输入错误：修正输入参数
        return fix_input_and_retry(step)
    
    else:
        # 未知错误：重新规划
        return replan(plan, step, error)

import concurrent.futures
 
def execute_parallel(plan):
    """并行执行无依赖的步骤"""
    
    results = {}
    
    while not plan.is_completed():
        # 获取可执行的步骤
        ready_steps = get_ready_steps(plan, results)
        
        # 并行执行
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(execute_step, step): step
                for step in ready_steps
            }
            
            for future in concurrent.futures.as_completed(futures):
                step = futures[future]
                try:
                    results[step.id] = future.result()
                    step.status = "completed"
                except Exception as e:
                    step.status = "failed"
                    handle_failure(step, e)
    
    return results

@dataclass
class ExecutionRecord:
    """执行记录"""
    step_id: str
    step_description: str
    tool_name: str
    tool_input: Dict
    tool_output: Any
    start_time: datetime
    end_time: datetime
    status: str
    error: Optional[str] = None
 
 
class AuditableExecutor:
    """可审计的执行器"""
    
    def __init__(self):
        self.records: List[ExecutionRecord] = []
    
    def execute_step(self, step: PlanStep) -> Any:
        record = ExecutionRecord(
            step_id=step.id,
            step_description=step.description,
            tool_name=step.tool,
            tool_input=step.tool_input,
            start_time=datetime.now(),
            status="running"
        )
        
        try:
            result = self.tools[step.tool].run(step.tool_input)
            record.tool_output = result
            record.status = "completed"
        except Exception as e:
            record.error = str(e)
            record.status = "failed"
            raise
        finally:
            record.end_time = datetime.now()
            self.records.append(record)
        
        return result
    
    def export_audit_log(self) -> str:
        """导出审计日志"""
        return json.dumps([
            {
                "step_id": r.step_id,
                "tool": r.tool_name,
                "input": r.tool_input,
                "output": str(r.tool_output)[:100],  # 截断
                "duration": (r.end_time - r.start_time).total_seconds(),
                "status": r.status
            }
            for r in self.records
        ], indent=2)

任务是否可以预先结构化？
├── 是 → 任务是否有明确的依赖关系？
│        ├── 是 → Plan-and-Execute
│        └── 否 → 考虑 ReAct
│
└── 否 → 任务是否需要探索性执行？
         ├── 是 → ReAct
         └── 否 → 考虑 Plan-and-Execute（带重规划）

def evaluate_agent(agent, test_cases):
    """评估 Agent 性能"""
    results = []
    
    for case in test_cases:
        start_time = time.time()
        result = agent.run(case["input"])
        end_time = time.time()
        
        results.append({
            "success": result == case["expected"],
            "time": end_time - start_time,
            "steps": len(agent.executor.results),
            "replans": agent.replan_count,
            "tokens": count_tokens(agent.history)
        })
    
    return {
        "success_rate": mean([r["success"] for r in results]),
        "avg_time": mean([r["time"] for r in results]),
        "avg_steps": mean([r["steps"] for r in results]),
        "avg_replans": mean([r["replans"] for r in results])
    }