摘要:通过亲手"造轮子",我们不仅掌握了 ReAct、Plan-and-Solve 和 Reflection 的核心原理,更深刻理解了它们各自的适用场景和工程挑战。本文包含完整代码实现和调试经验分享,适合对 Agent 开发感兴趣的开发者阅读。

前言:为什么要"重复造轮子"?

在 LangChain、LlamaIndex 等成熟框架大行其道的今天,很多人会问:既然有现成的工具,为什么还要从零实现智能体?

我的答案是:框架能提高效率,但理解原理才能让你成为创造者

当你亲手处理过模型输出格式解析、工具调用失败重试、防止智能体陷入死循环等问题后,你才能真正理解框架背后的设计哲学。更重要的是,当标准组件无法满足你的复杂需求时,你将拥有深度定制乃至从零构建一个全新智能体的能力。

本文将带你完整体验这三种范式从理论到实践的完整过程。

一、环境准备与基础设施

1.1 核心依赖安装

pip install openai python-dotenv google-search-results

1.2 封装通用的 LLM 客户端

为了让代码更模块化,我们首先封装一个通用的 LLM 客户端类,它将作为所有智能体的"大脑":

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

load_dotenv()

class HelloAgentsLLM:
    """为本书定制的LLM客户端,兼容任何OpenAI接口"""
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None):
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        
        if not all([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供")
        
        self.client = OpenAI(api_key=apiKey, base_url=baseUrl)
    
    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """调用大语言模型进行思考,支持流式响应"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            stream=True,
        )
        
        collected_content = []
        for chunk in response:
            content = chunk.choices[0].delta.content or ""
            print(content, end="", flush=True)
            collected_content.append(content)
        
        return "".join(collected_content)

设计亮点

  • 使用 stream=True 实现流式输出,提升用户体验
  • 通过 .env 文件管理敏感密钥,符合安全最佳实践
  • 统一的接口设计,方便后续智能体复用

二、ReAct 范式:边想边做的智能体

2.1 核心思想

ReAct (Reasoning + Acting) 由 Shunyu Yao 于 2022 年提出,其核心是模仿人类解决问题的方式:思考与行动相辅相成

思考 (Thought) → 行动 (Action) → 观察 (Observation) → 思考 (Thought) → ...

2.2 工具定义:让智能体拥有"手和脚"

我们以 SerpApi 搜索引擎为例,实现一个智能解析工具:

from serpapi import SerpApiClient

def search(query: str) -> str:
    """基于SerpApi的网页搜索引擎工具"""
    api_key = os.getenv("SERPAPI_API_KEY")
    params = {
        "engine": "google",
        "q": query,
        "api_key": api_key,
        "gl": "cn",
        "hl": "zh-cn",
    }
    
    client = SerpApiClient(params)
    results = client.get_dict()
    
    # 智能解析:优先返回直接答案
    if "answer_box" in results and "answer" in results["answer_box"]:
        return results["answer_box"]["answer"]
    if "knowledge_graph" in results:
        return results["knowledge_graph"].get("description", "")
    
    # 退而求其次,返回前三个搜索结果
    snippets = [
        f"[{i+1}] {res.get('title', '')}\n{res.get('snippet', '')}"
        for i, res in enumerate(results.get("organic_results", [])[:3])
    ]
    return "\n\n".join(snippets)

2.3 完整的 ReAct 智能体实现

import re
from typing import List

REACT_PROMPT_TEMPLATE = """
你是一个有能力调用外部工具的智能助手。

可用工具如下:
{tools}

请严格按照以下格式进行回应:
Thought: 你的思考过程
Action: 
- `{{tool_name}}[{{tool_input}}]`: 调用工具
- `Finish[最终答案]`: 任务完成

问题: {question}
历史: {history}
"""

class ReActAgent:
    def __init__(self, llm_client, tool_executor, max_steps: int = 5):
        self.llm_client = llm_client
        self.tool_executor = tool_executor
        self.max_steps = max_steps
        self.history = []
    
    def run(self, question: str):
        self.history = []
        
        for step in range(self.max_steps):
            print(f"\n--- 第 {step + 1} 步 ---")
            
            # 1. 构建提示词
            prompt = REACT_PROMPT_TEMPLATE.format(
                tools=self.tool_executor.getAvailableTools(),
                question=question,
                history="\n".join(self.history)
            )
            
            # 2. 调用 LLM
            response = self.llm_client.think([{"role": "user", "content": prompt}])
            
            # 3. 解析输出(关键难点!)
            thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)", response, re.DOTALL)
            action_match = re.search(r"Action:\s*(.*?)$", response, re.DOTALL)
            
            thought = thought_match.group(1).strip() if thought_match else None
            action = action_match.group(1).strip() if action_match else None
            
            print(f"\n🤔 思考: {thought}")
            
            if not action:
                print("⚠️ 未能解析出Action,流程终止")
                break
            
            # 4. 执行动作
            if action.lower().startswith("finish"):
                final_answer = re.search(r"Finish[\[::]\s*(.*)", action, re.DOTALL)
                if final_answer:
                    print(f"\n🎉 最终答案: {final_answer.group(1).strip().rstrip(']')}")
                    return final_answer.group(1)
            
            # 5. 调用工具并记录观察
            tool_name, tool_input = self._parse_action(action)
            if tool_name and tool_input:
                observation = self.tool_executor.getTool(tool_name)(tool_input)
                print(f"\n👀 观察: {observation}")
                
                self.history.append(f"Action: {action}")
                self.history.append(f"Observation: {observation}")
        
        print("\n⚠️ 达到最大步数,流程终止")
        return None
    
    def _parse_action(self, action_text: str):
        """解析 Action 字符串,兼容多种格式"""
        match = re.match(r"(\w+)\[(.*)\]", action_text, re.DOTALL)
        if match:
            return match.group(1), match.group(2)
        
        # 兼容 [Search]: input 格式
        match = re.match(r"\[(\w+)\]:?\s*(.*)", action_text, re.DOTALL)
        if match:
            return match.group(1), match.group(2)
        
        return None, None

2.4 真实运行案例

问题:"华为最新的手机是哪一款?它的主要卖点是什么?"

执行过程

还挺有意思的叭~~~ 在这里,大模型需要用最合理的步数来完成自己的任务,所以会有一些search和thought上的报错,这个不是你写代码的问题,是为了做适配!!!
 

2.5 ReAct 的优缺点分析

 优势

  • 高可解释性:通过 Thought 链可以看到智能体的"心路历程"
  • 动态纠错能力:根据 Observation 随时调整策略
  • 工具协同能力强:天然适合需要外部信息的任务

 局限

  • 对 LLM 能力依赖强:格式输出不稳定会导致流程中断
  • 执行效率低:每一步都需要调用 LLM,成本高
  • 提示词脆弱性:模板微调可能影响整体行为

 实战踩坑记录
在实际调试中,我遇到了模型输出格式不稳定的问题。比如模型会输出中文冒号 Finish:答案 而不是 Finish[答案],甚至直接省略 Action: 前缀。解决方案是:

  1. 使用 re.DOTALL 标志处理多行内容
  2. 增加兜底逻辑,全文搜索 Finish 关键字
  3. 兼容多种括号和标点格式

三、Plan-and-Solve 范式:先谋后动的架构师

3.1 核心思想

如果说 ReAct 像一个经验丰富的侦探,那么 Plan-and-Solve 就像一位建筑师——先绘制完整蓝图,再严格按图施工

规划阶段 (Planning) → 执行阶段 (Solving)

3.2 规划器实现

import ast

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。将复杂问题分解成由多个简单步骤组成的行动计划。
输出必须是一个Python列表,每个元素是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出:
```python
["步骤1", "步骤2", "步骤3", ...]

"""

class Planner:
def init(self, llm_client):
self.llm_client = llm_client

3.3 执行器与状态管理

执行器的核心挑战是**状态管理**——必须记录每一步的结果并传递给后续步骤:

```python
EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。严格按照给定计划一步步解决问题。

原始问题: {question}
完整计划: {plan}
历史步骤与结果: {history}
当前步骤: {current_step}

请仅输出针对"当前步骤"的回答:
"""

class Executor:
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    def execute(self, question: str, plan: list[str]) -> str:
        history = ""
        
        for i, step in enumerate(plan):
            print(f"\n-> 执行步骤 {i+1}/{len(plan)}: {step}")
            
            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question=question,
                plan=plan,
                history=history if history else "无",
                current_step=step
            )
            
            response = self.llm_client.think([{"role": "user", "content": prompt}])
            history += f"步骤 {i+1}: {step}\n结果: {response}\n\n"
            
            print(f"✅ 步骤 {i+1} 完成,结果: {response}")
        
        return response

3.4 整合:PlanAndSolveAgent

class PlanAndSolveAgent:
    def __init__(self, llm_client):
        self.planner = Planner(llm_client)
        self.executor = Executor(llm_client)
    
    def run(self, question: str):
        print(f"\n--- 开始处理问题 ---\n问题: {question}")
        
        # 1. 生成计划
        plan = self.planner.plan(question)
        if not plan:
            print("\n--- 任务终止 --- 无法生成有效计划")
            return
        
        # 2. 执行计划
        final_answer = self.executor.execute(question, plan)
        print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")

3.5 真实运行案例

问题:"一个水果店周一卖出15个苹果。周二卖出的数量是周一的两倍。周三比周二少5个。三天总共卖出多少个?"

执行过程

3.6 适用场景

Plan-and-Solve 最适合

  • 多步数学应用题
  • 需要整合多个信息源的报告撰写
  • 代码生成任务(先构思结构再实现)

核心优势:结构性强、稳定性高、目标一致性好。

四、Reflection 范式:自我优化的迭代循环

4.1 核心思想

Reflection 机制为智能体引入了事后自我校正循环,使其能像人类一样审视自己的工作并迭代优化:

执行 (Execution) → 反思 (Reflection) → 优化 (Refinement) → ...

4.2 记忆模块设计

迭代的前提是记住之前的尝试:

from typing import List, Dict, Any

class Memory:
    """短期记忆模块,存储行动与反思轨迹"""
    def __init__(self):
        self.records: List[Dict[str, Any]] = []
    
    def add_record(self, record_type: str, content: str):
        self.records.append({"type": record_type, "content": content})
        print(f"📝 记忆已更新,新增一条 '{record_type}' 记录")
    
    def get_trajectory(self) -> str:
        """将记忆序列化为文本"""
        parts = []
        for record in self.records:
            if record['type'] == 'execution':
                parts.append(f"--- 上一轮尝试 (代码) ---\n{record['content']}")
            elif record['type'] == 'reflection':
                parts.append(f"--- 评审员反馈 ---\n{record['content']}")
        return "\n\n".join(parts)
    
    def get_last_execution(self) -> str:
        for record in reversed(self.records):
            if record['type'] == 'execution':
                return record['content']
        return None

4.3 三阶段提示词设计

这里只给一个示例,比较简单,小伙伴们可以自己来尝试填写一下~~~

4.4 Reflection Agent


class ReflectionAgent:
    def __init__(self, llm_client, max_iterations=3):
        self.llm_client = llm_client
        self.memory = Memory()
        self.max_iterations = max_iterations
    
    def run(self, task: str):
        print(f"\n--- 开始处理任务 ---\n任务: {task}")
        
        # 1. 初始执行
        print("\n--- 初始尝试 ---")
        initial_code = self._get_response(INITIAL_PROMPT.format(task=task))
        self.memory.add_record("execution", initial_code)
        
        # 2. 迭代循环:反思与优化
        for i in range(self.max_iterations):
            print(f"\n--- 第 {i+1}/{self.max_iterations} 轮迭代 ---")
            
            # a. 反思
            print("\n-> 正在反思...")
            last_code = self.memory.get_last_execution()
            feedback = self._get_response(
                REFLECT_PROMPT.format(task=task, code=last_code)
            )
            self.memory.add_record("reflection", feedback)
            
            # b. 检查终止条件
            if "无需改进" in feedback:
                print("\n✅ 反思认为代码已无需改进,任务完成")
                break
            
            # c. 优化
            print("\n-> 正在优化...")
            refined_code = self._get_response(
                REFINE_PROMPT.format(
                    task=task,
                    last_code_attempt=last_code,
                    feedback=feedback
                )
            )
            self.memory.add_record("execution", refined_code)
        
        final_code = self.memory.get_last_execution()
        print(f"\n--- 任务完成 ---\n最终代码:\n```python\n{final_code}\n```")
        return final_code
    
    def _get_response(self, prompt: str) -> str:
        return self.llm_client.think([{"role": "user", "content": prompt}])

4.5 真实运行案例

任务:"编写一个Python函数,找出1到n之间所有的素数"

迭代过程

第1轮 - 初始尝试

第1轮 - 反思反馈

这里忘记截图了,记得当时是说:当前代码的时间复杂度为 O(n * sqrt(n))。
当 n 非常大时,性能会显著下降。
建议使用埃拉托斯特尼筛法(Sieve of Eratosthenes),
时间复杂度为 O(n log log n),能显著提高效率。

第1轮 - 优化后

第2轮 - 反思反馈

这里忘记截图了,记得当时是说:当前代码使用了 Eratosthenes 筛法,时间复杂度为 O(n log log n)。后面就不记得了~~~

✅ 任务完成!

4.6 成本收益分析

 主要成本

  • API 调用开销:每轮迭代至少额外调用 2 次 LLM
  • 任务延迟:串行过程导致总耗时显著延长
  • 提示工程复杂度:需要为三个阶段分别设计高质量提示词

核心收益

  • 解决方案质量跃迁:从"合格"到"优秀"
  • 鲁棒性增强:发现并修复逻辑漏洞、边界情况处理不当等问题

适用场景:对结果质量、准确性要求极高,且对实时性要求宽松的任务(如生成关键业务代码、技术报告、科学研究推演)。

五、三种范式的对比与选择策略

范式 核心特点 优势 局限 适用场景
ReAct 边想边做,动态循环 高可解释性、动态纠错、工具协同强 对 LLM 依赖强、效率低、提示词脆弱 探索性任务、需要外部信息、实时查询
Plan-and-Solve 先规划后执行 结构性强、稳定性高、目标一致性好 计划一旦生成不可修改、缺乏动态调整 多步数学题、报告撰写、代码生成
Reflection 执行-反思-优化循环 显著提升质量、鲁棒性强 API 成本高、延迟大、提示工程复杂 关键业务代码、学术研究、深度分析

💡 选择建议

  • 需要实时搜索或动态调整? → ReAct
  • 逻辑清晰、步骤固定? → Plan-and-Solve
  • 对质量要求极高、可以接受高成本? → Reflection
  • 复杂场景? → 考虑混合架构(如 ReAct + Reflection)

 六、实战经验与调试技巧

6.1 常见坑点与解决方案

问题1:模型输出格式不稳定

问题2:多行内容导致解析失败

问题3:模型偷懒不输出 Action 前缀

6.2 调试技巧总结

  1. 打印完整提示词:追溯 LLM 决策源头的最直接方式
  2. 分析原始输出:判断是 LLM 问题还是解析逻辑问题
  3. 验证工具输入输出:确保格式匹配
  4. 添加 Few-shot 示例:显著提升格式遵循能力
  5. 调整 temperature=0:保证输出确定性

七、学习心得与总结

通过本章"从零构建智能体"的实战,我深刻理解了为什么教材强调"不要依赖框架,要亲手造轮子"。

核心收获

  1. 掌握三种核心范式:ReAct 的动态性、Plan-and-Solve 的结构性、Reflection 的迭代优化能力
  2. 理解工程挑战:模型输出格式不稳定、正则解析失败、工具调用容错等真实开发中的坑
  3. 建立架构思维:不同场景需要不同架构,没有"银弹",只有"最适合"
  4. 模块化设计意识:LLM 客户端、工具管理器、智能体逻辑应该解耦,方便复用

💡 设计哲学启示

  • ReAct 教会我"走一步看一步"的灵活性
  • Plan-and-Solve 教会我"谋定而后动"的结构性
  • Reflection 教会我"以成本换质量"的权衡思维

这三种范式不是互斥的,而是可以组合使用的"武器库"。在实际项目中,我们完全可以根据任务需求,灵活选择或混合使用。

八、延伸阅读与资源

  • 原始论文
  • 开源项目
    • LangChain Agent 模块
    • AutoGPT / BabyAGI
  • 推荐工具
    • SerpApi(网页搜索)
    • Tavily(AI 优化搜索)
    • ModelScope / AIHubmix(LLM API)

结语

智能体开发是一场"与不确定性共舞"的工程实践。大语言模型的强大能力让我们看到了无限可能,但其输出的不稳定性也带来了诸多挑战。

通过亲手实现这三种经典范式,我们不仅掌握了技术细节,更重要的是培养了系统设计能力问题拆解思维。当你真正理解了这些底层原理后,无论是使用 LangChain 这样的成熟框架,还是针对特殊需求定制自己的 Agent,你都将游刃有余。

最后送给大家一句话:框架是工具,原理是根基。只有根基扎实,才能在智能体开发的浪潮中立于不败之地!

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐