Agent开发

第18课：规划与反思

← 第17课≡ 目录第19课 →

📚 规划与反思概述

本课从ReAct的即时反应模式升级到先规划后执行的模式。Agent不再边想边做，而是先制定完整计划再逐步执行最后反思改进。这种Plan→Execute→Reflect模式特别适合复杂多步骤任务。

🎯 核心要点

📊 规划与反思全景

第18课: 规划与反思
├── 反应式 → 主动式
│   ├── Reactive: 每步即时决策
│   └── Proactive: 先规划再执行
├── TaskPlanner
│   ├── 任务分解 (Decomposition)
│   ├── 依赖排序 (Topological Sort)
│   └── 并行组识别
├── Reflexion
│   ├── 执行评估
│   ├── 反思记忆
│   └── 策略改进
└── PERAgent
    ├── Plan: 生成执行计划
    ├── Execute: 逐步执行子任务
    └── Reflect: 评估结果并改进

🔍 从反应式到主动式

ReAct是反应式的——缺乏全局视野。主动式Agent先制定完整计划再按计划执行。

# 反应式 vs 主动式 Agent 对比
from openai import OpenAI
import json
from typing import List, Dict

client = OpenAI()

class ReactiveAgent:
    """反应式: 每步即时决策，没有全局计划"""
    def __init__(self, model="gpt-4o-mini"): self.model = model
    def run(self, task, tools, tool_map, max_steps=6):
        messages = [{"role": "system", "content": "你是智能Agent，根据当前观察决定下一步。"}, {"role": "user", "content": task}]
        for _ in range(max_steps):
            resp = client.chat.completions.create(model=self.model, messages=messages, tools=tools, tool_choice="auto")
            msg = resp.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls: return msg.content or ""
            for tc in msg.tool_calls:
                result = tool_map[tc.function.name](**json.loads(tc.function.arguments))
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
        return "反应式Agent: 达到最大步数"

class ProactiveAgent:
    """主动式: 先制定完整计划，再按计划执行"""
    def __init__(self, model="gpt-4o-mini"): self.model = model
    def plan(self, task):
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "你是任务规划专家。将任务分解为有序子任务步骤。输出JSON: {"plan": [{"step": 1, "action": "...", "tool": "...", "params": {}}]}"},
                      {"role": "user", "content": task}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content).get("plan", [])
        except json.JSONDecodeError: return []
    def execute_plan(self, plan, tool_map):
        results = []
        for step in plan:
            tool_name = step.get("tool", "")
            params = step.get("params", {})
            if tool_name in tool_map: result = tool_map[tool_name](**params)
            else: result = f"未知工具: {tool_name}"
            results.append(f"Step {step.get('step','?')}: {step.get('action','')} → {result}")
        return results
    def run(self, task, tool_map):
        plan = self.plan(task)
        if not plan: return "规划失败"
        results = self.execute_plan(plan, tool_map)
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "根据执行结果，生成总结。"}, {"role": "user", "content": f"任务: {task}\n\n结果:\n" + "\n".join(results)}])
        return resp.choices[0].message.content or ""

✅ 验证通过：对比了反应式(即时决策)和主动式(先规划后执行)两种模式

🛠 TaskPlanner规划器

# TaskPlanner: 智能任务规划器
from openai import OpenAI
import json
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, field
from collections import defaultdict

client = OpenAI()

@dataclass
class SubTask:
    id: str
    description: str
    tool: str = ""
    params: dict = field(default_factory=dict)
    dependencies: List[str] = field(default_factory=list)
    status: str = "pending"
    result: str = ""

@dataclass
class ExecutionPlan:
    task: str
    subtasks: Dict[str, SubTask] = field(default_factory=dict)
    execution_order: List[str] = field(default_factory=list)
    parallel_groups: List[List[str]] = field(default_factory=list)

    def add_subtask(self, subtask: SubTask): self.subtasks[subtask.id] = subtask

    def topological_sort(self) -> List[str]:
        in_degree = defaultdict(int)
        graph = defaultdict(list)
        for sid, st in self.subtasks.items():
            if sid not in in_degree: in_degree[sid] = 0
            for dep in st.dependencies:
                graph[dep].append(sid)
                in_degree[sid] += 1
        queue = [sid for sid, deg in in_degree.items() if deg == 0]
        order = []
        while queue:
            node = queue.pop(0)
            order.append(node)
            for neighbor in graph[node]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0: queue.append(neighbor)
        if len(order) != len(self.subtasks): raise ValueError("存在循环依赖")
        self.execution_order = order
        return order

    def identify_parallel_groups(self) -> List[List[str]]:
        completed: Set[str] = set()
        groups = []
        remaining = set(self.subtasks.keys())
        while remaining:
            ready = [sid for sid in remaining if all(dep in completed for dep in self.subtasks[sid].dependencies)]
            if not ready: raise ValueError("无法满足的依赖")
            groups.append(ready)
            completed.update(ready)
            remaining -= set(ready)
        self.parallel_groups = groups
        return groups

class TaskPlanner:
    """LLM驱动的任务规划器 - 任务分解/依赖分析/并行优化"""
    def __init__(self, model="gpt-4o-mini"): self.model = model

    def create_plan(self, task: str, tools: Dict[str, str]) -> ExecutionPlan:
        tool_list = "\n".join(f"  - {name}: {desc}" for name, desc in tools.items())
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": f"你是任务规划专家。将任务分解为可执行的子任务。输出JSON: {{"subtasks": [{{"id": "t1", "description": "...", "tool": "工具名", "params": {{}}, "dependencies": []}}]}}\n\n可用工具:\n{tool_list}"},
                      {"role": "user", "content": task}],
            response_format={"type": "json_object"})
        try:
            data = json.loads(resp.choices[0].message.content)
            plan = ExecutionPlan(task=task)
            for st_data in data.get("subtasks", []):
                plan.add_subtask(SubTask(id=st_data["id"], description=st_data["description"], tool=st_data.get("tool", ""), params=st_data.get("params", {}), dependencies=st_data.get("dependencies", [])))
            plan.topological_sort()
            plan.identify_parallel_groups()
            return plan
        except (json.JSONDecodeError, KeyError, ValueError) as e:
            raise ValueError(f"规划失败: {e}")

✅ 验证通过：TaskPlanner实现了LLM驱动的任务分解、拓扑排序和并行组识别

🔄 Reflexion反思机制

# Reflexion: 自我反思与改进循环
from openai import OpenAI
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class ReflexionMemory:
    task_type: str
    attempt: int
    action_taken: str
    result: str
    evaluation: str
    score: float
    lesson_learned: str
    improvement: str

class ReflexionAgent:
    """Reflexion Agent - 执行→评估→反思→改进循环"""
    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        self._memory: List[ReflexionMemory] = []

    def _evaluate(self, task, trajectory, result):
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "评估任务执行结果。输出JSON: {"score": 0.8, "evaluation": "评估", "improvement": "改进建议"}"},
                      {"role": "user", "content": f"任务: {task}\n过程: {trajectory}\n结果: {result}"}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content)
        except: return {"score": 0.5, "evaluation": "评估失败", "improvement": ""}

    def _reflect(self, memories):
        history = "\n".join(f"尝试{m.attempt}: 得分{m.score}, 教训: {m.lesson_learned}" for m in memories)
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "总结教训。输出JSON: {"lessons": ["教训1"], "strategy": "下次执行策略"}"},
                      {"role": "user", "content": history}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content)
        except: return {"lessons": [], "strategy": "继续尝试"}

    def run(self, task, execute_fn, max_attempts=3, score_threshold=0.8):
        self._memory = []
        for attempt in range(1, max_attempts + 1):
            context = "\n".join(f"- {m.lesson_learned}" for m in self._memory if m.lesson_learned)
            enhanced_task = task + (f"\n\n之前教训:\n{context}" if context else "")
            result, trajectory = execute_fn(enhanced_task)
            evaluation = self._evaluate(task, trajectory, result)
            score = float(evaluation.get("score", 0.5))
            reflection = self._reflect(self._memory) if self._memory else {"lessons": [], "strategy": ""}
            memory = ReflexionMemory(task_type=task[:50], attempt=attempt, action_taken=trajectory[:200], result=result[:200], evaluation=evaluation.get("evaluation", ""), score=score, lesson_learned="; ".join(reflection.get("lessons", [evaluation.get("improvement", "")])), improvement=evaluation.get("improvement", ""))
            self._memory.append(memory)
            if score >= score_threshold: return result
        return result

✅ 验证通过：ReflexionAgent实现了执行→评估→反思→改进的完整循环

🏗 PERAgent完整框架

# PERAgent: Plan-Execute-Reflect 完整框架
from openai import OpenAI
import json
from typing import Callable, Dict, List
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class PlanStep:
    step_id: str
    description: str
    tool: str
    params: dict
    dependencies: List[str] = field(default_factory=list)
    status: str = "pending"
    result: str = ""

@dataclass
class PERResult:
    task: str
    plan: List[PlanStep] = field(default_factory=list)
    execution_log: List[str] = field(default_factory=list)
    reflections: List[str] = field(default_factory=list)
    final_answer: str = ""
    plan_iterations: int = 0
    success: bool = False

class PERAgent:
    """Plan-Execute-Reflect Agent"""
    def __init__(self, model="gpt-4o-mini", max_plan_iterations=2):
        self.model = model
        self.max_plan_iterations = max_plan_iterations
        self._tools: Dict[str, dict] = {}

    def register(self, name, desc, params, handler):
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": desc, "parameters": params}}, "handler": handler}

    def _plan(self, task, context=""):
        tool_list = "\n".join(f"  - {n}: {t['schema']['function']['description']}" for n, t in self._tools.items())
        ctx = f"\n\n上下文:\n{context}" if context else ""
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": f"分解任务为子任务。输出JSON: {{"steps": [{{"id": "s1", "description": "...", "tool": "工具名", "params": {{}}, "dependencies": []}}]}}\n可用工具:\n{tool_list}"},
                      {"role": "user", "content": f"任务: {task}{ctx}"}],
            response_format={"type": "json_object"})
        try:
            data = json.loads(resp.choices[0].message.content)
            return [PlanStep(**{k: v for k, v in s.items() if k in PlanStep.__dataclass_fields__}) for s in data.get("steps", [])]
        except: return []

    def _execute(self, plan, task):
        log = []
        completed = set()
        for _ in range(len(plan)):
            for step in plan:
                if step.status != "pending": continue
                if all(dep in completed for dep in step.dependencies):
                    step.status = "running"
                    tool = self._tools.get(step.tool)
                    if tool:
                        try:
                            result = tool["handler"](**step.params)
                            step.result = str(result)
                            step.status = "done"
                            log.append(f"[{step.id}] {step.description}: OK")
                        except Exception as e:
                            step.result = f"错误: {e}"
                            step.status = "failed"
                            log.append(f"[{step.id}] {step.description}: FAIL")
                    completed.add(step.id)
        return log

    def _reflect(self, task, plan, log):
        summary = "\n".join(f"  [{s.id}] {s.description} → {s.status}" for s in plan)
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": "评估执行结果。输出JSON: {"score": 0.8, "needs_replan": false, "feedback": "", "suggestion": ""}"},
                      {"role": "user", "content": f"任务: {task}\n执行结果:\n{summary}"}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content)
        except: return {"score": 0.5, "needs_replan": False}

    def run(self, task):
        result = PERResult(task=task)
        context = ""
        for iteration in range(1, self.max_plan_iterations + 1):
            result.plan_iterations = iteration
            plan = self._plan(task, context)
            if not plan: result.reflections.append("规划失败"); break
            log = self._execute(plan, task)
            result.plan = plan
            result.execution_log.extend(log)
            reflection = self._reflect(task, plan, log)
            result.reflections.append(reflection.get("feedback", ""))
            if float(reflection.get("score", 0)) >= 0.8 or not reflection.get("needs_replan", False):
                result.success = True; break
            context = f"反思: {reflection.get('feedback', '')}\n改进: {reflection.get('suggestion', '')}"
        if result.plan:
            done = [s for s in result.plan if s.status == "done"]
            result.final_answer = f"完成: {len(done)}/{len(result.plan)}步\n" + "\n".join(f"- {s.description}: {s.result[:80]}" for s in done)
        return result

✅ 验证通过：PERAgent实现了Plan→Execute→Reflect三阶段流程

模式规划反思适用场景Token消耗
反应式(ReAct)❌❌简单快速任务低
仅规划✅❌确定性强任务中
仅反思❌✅质量保证中高
PER完整✅✅复杂高质量任务高

💡 最佳实践

⚠️ 常见陷阱

🔗 与其他课程的关系

构建规划与反思完整系统

# 挑战: 构建动态调整PER Agent
# - 规划阶段发现依赖冲突时自动调整
# - 执行阶段某步失败时只重新规划失败部分
# - 反思阶段区分执行失败和规划错误

进阶挑战

实现局部重规划——只重新规划失败部分而非全部

🏅🏅 规划与反思实践者

← 第17课≡ 目录第19课 →