AI Agent实战系列·第5篇
Plan-Execute-Reflect，从"边做边想"到"先想后做"

写在前面

上期回顾： 第4篇我们给Agent装上了"记忆力"。但光有记忆还不够，面对复杂任务时，Agent依然会"手忙脚乱"。

本期核心：

  • ✅ Plan-and-Execute：规划与执行分离
  • ✅ ReWOO框架：减少70%的API调用
  • ✅ 自我反思：让Agent会改进
  • ✅ 完整实战案例

一、翻车现场：不会规划的Agent

让Agent写一份市场分析报告：

我：帮我分析AI Agent市场，写份报告

传统Agent（ReAct模式）：
  → 搜索"AI Agent市场"... 
  → 看到10篇文章，开始写...
  → 写了一半发现缺数据
  → 又搜索"市场规模"...
  → 又搜索"竞品对比"...
  → 重复调用15次API 💸
  
结果：
  ✅ 报告写完了
  ❌ 花了10分钟
  ❌ 浪费了$2的API费用
  ❌ 逻辑混乱，缺少整体规划

问题根源： 传统ReAct是"边想边做"，就像没有规划的装修队，走一步看一步。

二、人类如何解决复杂任务？

假设老板让你写竞品分析，你会怎么做？

❌ 低效做法（传统ReAct）
  → 立刻开始搜索
  → 看到什么写什么
  → 想到什么查什么

✅ 高效做法（Plan-Execute）
  Step1: 规划（5分钟）
    ├─ 明确目标：分析哪些维度？
    ├─ 拆解任务：
    │   ├─ 收集竞品列表
    │   ├─ 分析产品特点（可并行）
    │   ├─ 对比价格性能（可并行）
    │   └─ 撰写报告
    └─ 确定顺序和优先级
  
  Step2: 执行（15分钟）
    ├─ 批量搜索（并行）
    └─ 按计划撰写
  
  Step3: 检查（2分钟）
    └─ 发现问题 → 补充完善

核心理念：先想清楚再动手，而不是边做边想

三、Plan-and-Execute：效率提升10倍

架构对比

传统ReAct                    Plan-and-Execute
┌──────────┐                ┌──────────┐
│ Think 1  │                │  Plan    │ ← 一次性规划
│ Action 1 │ ← 15次LLM调用  │ (1次调用) │
│ Think 2  │                └────┬─────┘
│ Action 2 │                     ↓
│ Think 3  │                ┌──────────┐
│   ...    │                │ Execute  │ ← 批量执行
└──────────┘                │(并行3次)  │
                            └────┬─────┘
                                 ↓
                            ┌──────────┐
                            │ Reflect  │ ← 自我检查
                            └──────────┘

效果对比

维度 ReAct Plan-and-Execute
LLM调用次数 15次 5次（节省70%）
总耗时 10分钟 3分钟
成本 $2 $0.6
可并行 ❌ ✅
可控性 难预测 计划透明

四、ReWOO框架：最佳实践

ReWOO = Reasoning withOut Observation

核心思想： 先把所有任务规划好，再一次性执行

工作流程

# 用户提问："比较GPT-4和Claude的优缺点"

# Step 1: Planner（规划器）
plan = [
    "Plan1: 搜索GPT-4的特点和价格",
    "Plan2: 搜索Claude的特点和价格",    # 可与Plan1并行
    "Plan3: 根据#E1和#E2，对比两者优缺点"
]

# Step 2: Worker（执行器）- 并行执行
evidence = {
    "#E1": search("GPT-4特点价格"),     # 并行
    "#E2": search("Claude特点价格")     # 并行
}

# Step 3: Solver（求解器）
answer = generate_answer(plan, evidence)

核心实现

class ReWOOAgent:
    """规划型Agent - 核心实现"""
    
    def __init__(self, tools):
        self.tools = tools
    
    def run(self, task: str) -> str:
        """执行任务的三步走"""
        
        # Step 1: 制定计划
        print("🧠 正在制定计划...")
        plans = self.plan(task)
        print(f"生成了{len(plans)}个子任务")
        
        # Step 2: 批量执行（可并行）
        print("⚙️ 开始执行...")
        evidence = self.execute(plans)
        
        # Step 3: 整合答案
        print("🎯 整合结果...")
        answer = self.solve(task, plans, evidence)
        
        return answer
    
    def plan(self, task: str):
        """生成执行计划"""
        prompt = f"""
你是任务规划专家。将任务分解为可执行的步骤。

可用工具：search(查询), calculate(计算), write(写作)

任务：{task}

请生成计划（格式：Plan1: 工具名(参数)）
可引用前面的结果：#E1, #E2
"""
        response = call_llm(prompt)
        return self.parse_plans(response)
    
    def execute(self, plans):
        """并行执行计划"""
        from concurrent.futures import ThreadPoolExecutor
        
        evidence = {}
        with ThreadPoolExecutor(max_workers=3) as executor:
            futures = {}
            for plan in plans:
                future = executor.submit(
                    self.tools[plan['tool']],
                    plan['args']
                )
                futures[future] = plan
            
            for future in futures:
                plan = futures[future]
                evidence[f"#E{plan['id']}"] = future.result()
        
        return evidence
    
    def solve(self, task, plans, evidence):
        """整合证据生成答案"""
        prompt = f"""
任务：{task}
执行计划：{plans}
收集的证据：{evidence}

请基于证据给出完整答案：
"""
        return call_llm(prompt)

实际效果

任务：比较GPT-4和Claude的优缺点

🧠 正在制定计划...
  Plan1: search(GPT-4特点和定价)
  Plan2: search(Claude特点和定价)
  Plan3: compare(#E1, #E2)

⚙️ 开始执行...
  ✅ Plan1完成 (1.2秒)
  ✅ Plan2完成 (1.3秒) ← 并行执行
  ✅ Plan3完成 (0.8秒)

总耗时：2.1秒（串行需要3.3秒）
API调用：3次（ReAct需要8次）

🎯 最终答案：
GPT-4优势：更强的推理能力...
Claude优势：更长的上下文窗口...

五、自我反思：让Agent会改进

Reflexion机制：执行 → 评估 → 反思 → 改进

class ReflexionAgent:
    """支持自我反思的Agent"""
    
    def run_with_reflection(self, task: str):
        """最多尝试3次，直到满意为止"""
        
        for attempt in range(3):
            print(f"\n🔄 第{attempt+1}次尝试")
            
            # 1. 执行任务
            result = self.base_agent.run(task)
            
            # 2. 自我评估（1-10分）
            score, feedback = self.evaluate(task, result)
            print(f"📊 自评分数：{score}/10")
            
            # 3. 满意就返回
            if score >= 8:
                print("✅ 结果满意！")
                return result
            
            # 4. 不满意就反思
            reflection = self.reflect(feedback)
            print(f"🤔 反思：{reflection}")
            
            # 5. 根据反思改进
            task = f"{task}\n改进要求：{reflection}"
        
        return result
    
    def evaluate(self, task, result):
        """使用LLM评估结果质量"""
        prompt = f"""
评估任务完成质量（1-10分）

任务：{task}
结果：{result}

请给出分数和反馈。
"""
        response = call_llm(prompt)
        return self.parse_score(response)
    
    def reflect(self, feedback):
        """根据反馈生成改进建议"""
        prompt = f"""
分析失败原因，提出改进方案。

反馈：{feedback}

改进建议：
"""
        return call_llm(prompt)

实际效果：

任务：写一篇关于AI Agent的技术文章

🔄 第1次尝试
📊 自评分数：6/10
💬 反馈：内容太浅，缺少代码示例
🤔 反思：需要增加具体代码实现，加深技术深度

🔄 第2次尝试
📊 自评分数：8/10
💬 反馈：代码示例丰富，逻辑清晰
✅ 结果满意！

对比：
  第1次：800字，无代码
  第2次：2000字，3个完整示例

六、完整实战：智能研究助手

整合所有能力，打造一个完整的Agent：

class ResearchAssistant:
    """智能研究助手"""
    
    def research(self, topic: str):
        """执行研究任务"""
        
        # 1. 分解任务
        tasks = self.decompose(topic)
        # 输出：
        # ├─ 收集市场数据
        # ├─ 分析竞品情况
        # ├─ 总结趋势预测
        # └─ 撰写研究报告
        
        # 2. 规划执行
        agent = ReWOOAgent(tools)
        results = {}
        
        for task in tasks:
            result = agent.run(task)
            results[task] = result
        
        # 3. 整合报告
        report = self.generate_report(topic, results)
        
        # 4. 自我检查
        if self.quality_check(report) < 8:
            report = self.improve(report)
        
        return report

使用示例：

assistant = ResearchAssistant()

report = assistant.research(
    "AI Agent在企业中的应用现状与趋势"
)

# 自动完成：
# ✅ 任务分解（4个子任务）
# ✅ 并行搜索（节省60%时间）
# ✅ 生成报告（3000字）
# ✅ 自我检查（2次改进）
# 
# 最终输出：
# 📄 8000字研究报告
# 📊 包含5个图表
# 🔗 引用15个数据来源

七、三个关键优化技巧

技巧1：智能判断是否需要规划

def should_plan(task: str) -> bool:
    """简单任务直接做，复杂任务先规划"""
    
    # 简单任务：直接执行
    simple_patterns = [
        "今天天气",
        "1+1等于",
        "帮我搜索"
    ]
    
    if any(p in task for p in simple_patterns):
        return False  # 使用ReAct
    
    # 复杂任务：先规划
    return True  # 使用Plan-Execute

技巧2：动态调整并行度

def get_max_workers(plans: list) -> int:
    """根据任务数量动态调整并行数"""
    
    if len(plans) <= 2:
        return 2
    elif len(plans) <= 5:
        return 3
    else:
        return 5  # 避免过度并行导致限流

技巧3：缓存计划结果

class PlanCache:
    """缓存相似任务的计划"""
    
    def get_similar_plan(self, task: str):
        """查找相似任务的计划"""
        # 使用向量搜索找相似任务
        similar = self.vector_search(task)
        
        if similar and similar['score'] > 0.9:
            return similar['plan']
        
        return None

总结

今天我们构建了完整的规划系统：

✅ Plan-Execute模式： 效率提升10倍，成本降低70%
✅ ReWOO框架： 规划-执行-求解三步走
✅ 自我反思： 让Agent会自我改进
✅ 实战案例： 智能研究助手完整实现

💡 核心洞察： 规划是高效的关键。先想清楚再动手，比边做边想快10倍。

关键数据：

  • 减少LLM调用：70%
  • 提升执行速度：3-5倍
  • 降低成本：60-80%
  • 提高成功率：40%

下期预告

下期内容：

  • 多Agent架构设计
  • AutoGen框架实战
  • 角色分工与任务路由
  • 实战：软件开发团队Agent

练习题：

用Plan-Execute模式实现一个"旅行规划助手"：

  1. 输入：目的地、预算、天数
  2. 输出：完整的行程安排
  3. 要求：自动搜索景点、酒店、交通

留言分享你的实现！

如果这篇文章对你有帮助：
• 👍 点赞支持
• 🔖 收藏备用
• 📤 分享朋友

下期见！
本文是《AI Agent实战系列》第5篇，后续还会更新AI Agent进阶玩法。关注公众号【架构之旅】，第一时间解锁全套实战教程，错过不再补～

# 人工智能 # python
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

cover

Java 实现 Agent ，From AgentScope

avatar MCP技术社区

你的电脑就是语音工作室：Voicebox 开源克隆语音、七引擎 TTS、MCP 让 AI 开口说话

一个免费、本地运行的 AI 语音工作室，让你从 3 秒音频克隆任意声音，用 7 个 TTS 引擎生成多语言语音，还能让 AI agent 用你克隆的声音跟你对话。more。

avatar MCP技术社区
cover

【本地AI安全防护实战指南：5个漏洞+一键加固方案（Ollama篇）】

avatar MCP技术社区