2.3 ReAct的代码实现与优化

读者读完这节，能掌握ReAct框架的完整代码实现，理解关键优化策略并能独立部署

学习目标

• 掌握ReAct框架的完整Python实现
• 理解关键优化策略和性能调优方法
• 学会使用主流ReAct工具链（LangChain、CrewAI）
• 能够根据具体场景优化ReAct智能体的表现

核心概念

ReAct框架的代码实现需要考虑以下几个核心要素：

1. 循环管理：控制推理-行动-观察循环的执行流程
2. 工具集成：与外部工具的无缝集成机制
3. 状态管理：维护智能体内部状态和上下文信息
4. 错误恢复：优雅处理各种异常情况
5. 性能优化：提升推理效率和响应速度

环境准备 / 前置知识

必需依赖

# 基础依赖
python>=3.8
torch>=1.9.0
transformers>=4.21.0
langchain>=0.1.0
openai>=1.3.0

# 可选优化依赖
accelerate>=0.20.0
bitsandbytes>=0.39.0
flash-attn>=2.3.0
​

推荐开发环境

• IDE: VS Code + Python插件
• 调试工具: ipython, pdb
• 测试框架: pytest, unittest
• 版本控制: git

分步实战

步骤 1：基础ReAct框架实现

import json
import logging
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod

@dataclass
class ReActState:
    """ReAct智能体状态管理"""
    current_thought: str = ""
    action_history: List[Dict] = None
    observations: List[Dict] = None
    final_answer: str = ""
    max_iterations: int = 10
    current_iteration: int = 0
    
    def __post_init__(self):
        if self.action_history is None:
            self.action_history = []
        if self.observations is None:
            self.observations = []

class Tool(ABC):
    """工具抽象基类"""
    
    @property
    @abstractmethod
    def name(self) -> str:
        pass
    
    @property
    @abstractmethod
    def description(self) -> str:
        pass
    
    @abstractmethod
    def run(self, query: str) -> str:
        pass
    
    @abstractmethod
    def run_with_params(self, **kwargs) -> str:
        pass

class SearchTool(Tool):
    """搜索工具实现"""
    
    @property
    def name(self) -> str:
        return "search"
    
    @property
    def description(self) -> str:
        return "Search the web for information"
    
    def run(self, query: str) -> str:
        # 实现搜索逻辑
        return f"Search results for: {query}"
    
    def run_with_params(self, **kwargs) -> str:
        query = kwargs.get('query', '')
        return self.run(query)

class CalculatorTool(Tool):
    """计算器工具实现"""
    
    @property
    def name(self) -> str:
        return "calculator"
    
    @property
    def description(self) -> str:
        return "Perform mathematical calculations"
    
    def run(self, query: str) -> str:
        try:
            # 安全计算实现
            result = eval(query)
            return f"Calculation result: {result}"
        except Exception as e:
            return f"Calculation error: {str(e)}"
    
    def run_with_params(self, **kwargs) -> str:
        return self.run(kwargs.get('expression', ''))
​

步骤 2：ReAct核心控制器实现

class ReActAgent:
    """ReAct智能体核心实现"""
    
    def __init__(self, tools: List[Tool], max_iterations: int = 10):
        self.tools = {tool.name: tool for tool in tools}
        self.state = ReActState(max_iterations=max_iterations)
        self.logger = logging.getLogger(__name__)
    
    def _generate_thought(self, task: str) -> str:
        """生成思考过程"""
        prompt = f"""当前任务：{task}
思考过程：
1. 分析问题的具体要求
2. 确定需要使用的工具
3. 规划执行步骤
4. 预测可能的困难和解决方案

请生成详细的思考过程："""
        
        # 这里使用LLM生成思考过程
        # 实际实现中可以使用OpenAI、Claude等API
        thought = f"基于任务'{task}'，我需要使用搜索工具获取相关信息，然后进行推理分析。"
        return thought
    
    def _select_action(self, thought: str) -> Dict:
        """根据思考选择行动"""
        # 解析思考内容，决定使用哪个工具
        if "搜索" in thought or "查找" in thought:
            return {
                "tool": "search",
                "query": thought,
                "reason": "需要搜索相关信息"
            }
        elif "计算" in thought or "算" in thought:
            return {
                "tool": "calculator",
                "expression": thought,
                "reason": "需要数学计算"
            }
        else:
            return {
                "tool": "default",
                "reason": "无法确定合适的工具"
            }
    
    def _execute_action(self, action: Dict) -> str:
        """执行行动并获取观察结果"""
        tool_name = action.get("tool")
        tool = self.tools.get(tool_name)
        
        if not tool:
            return f"工具'{tool_name}'不存在"
        
        try:
            if tool_name == "search":
                result = tool.run(action.get("query", ""))
            elif tool_name == "calculator":
                result = tool.run(action.get("expression", ""))
            else:
                result = tool.run(action.get("query", ""))
            
            return result
        except Exception as e:
            return f"执行错误: {str(e)}"
    
    def _update_observation(self, action: Dict, observation: str):
        """更新观察结果"""
        self.state.observations.append({
            "action": action,
            "observation": observation,
            "timestamp": self.state.current_iteration
        })
    
    def _generate_final_answer(self) -> str:
        """生成最终答案"""
        prompt = f"""基于以下思考过程和行动结果：
思考历史：{self.state.current_thought}
行动历史：{self.state.action_history}
观察结果：{self.state.observations}

请生成最终答案："""
        
        # 使用LLM生成最终答案
        answer = "基于分析，我已经完成了任务。"
        return answer
    
    def run(self, task: str) -> Dict[str, Any]:
        """执行ReAct循环"""
        self.logger.info(f"开始处理任务: {task}")
        
        for iteration in range(self.state.max_iterations):
            self.state.current_iteration = iteration + 1
            
            # 思考阶段
            thought = self._generate_thought(task)
            self.state.current_thought = thought
            self.logger.info(f"迭代 {iteration + 1} 思考: {thought}")
            
            # 行动阶段
            action = self._select_action(thought)
            self.state.action_history.append(action)
            self.logger.info(f"迭代 {iteration + 1} 行动: {action}")
            
            # 观察阶段
            observation = self._execute_action(action)
            self._update_observation(action, observation)
            self.logger.info(f"迭代 {iteration + 1} 观察: {observation}")
            
            # 检查是否完成
            if "完成" in observation or "答案" in observation:
                break
        
        # 生成最终答案
        self.state.final_answer = self._generate_final_answer()
        
        return {
            "task": task,
            "thoughts": [self.state.current_thought],
            "actions": self.state.action_history,
            "observations": self.state.observations,
            "final_answer": self.state.final_answer,
            "iterations": self.state.current_iteration
        }
​

步骤 3：高级优化实现

class OptimizedReActAgent(ReActAgent):
    """优化版的ReAct智能体"""
    
    def __init__(self, tools: List[Tool], max_iterations: int = 10):
        super().__init__(tools, max_iterations)
        self.cache = {}  # 思考缓存
        self.memory_buffer = []  # 记忆缓冲区
        self.optimization_strategies = {
            "early_stopping": True,
            "cache_enabled": True,
            "parallel_execution": False
        }
    
    def _generate_thought(self, task: str) -> str:
        """带缓存的思考生成"""
        # 检查缓存
        cache_key = hash(task + str(self.state.current_iteration))
        if cache_key in self.cache and self.optimization_strategies["cache_enabled"]:
            self.logger.info("使用缓存的思考过程")
            return self.cache[cache_key]
        
        # 生成新的思考
        thought = super()._generate_thought(task)
        
        # 缓存结果
        self.cache[cache_key] = thought
        return thought
    
    def _select_action(self, thought: str) -> Dict:
        """带优化的行动选择"""
        # 检查记忆缓冲区
        for memory in self.memory_buffer[-5:]:  # 检查最近5条记忆
            if memory["task"] == thought:
                self.logger.info("使用记忆中的行动选择")
                return memory["action"]
        
        # 正常行动选择
        action = super()._select_action(thought)
        
        # 添加到记忆缓冲区
        self.memory_buffer.append({
            "task": thought,
            "action": action,
            "timestamp": self.state.current_iteration
        })
        
        # 限制记忆大小
        if len(self.memory_buffer) > 20:
            self.memory_buffer = self.memory_buffer[-20:]
        
        return action
    
    def early_stopping_check(self, observation: str) -> bool:
        """提前停止检查"""
        if not self.optimization_strategies["early_stopping"]:
            return False
        
        # 检查是否已经找到答案
        if "答案" in observation or "完成" in observation:
            return True
        
        # 检查是否有重复模式
        if len(self.state.observations) >= 3:
            last_3 = obs["observation"] for obs in self.state.observations[-3:]
            if len(set(last_3)) == 1:  # 最后3次观察相同
                return True
        
        return False
​

步骤 4：LangChain集成实现

from langchain.agents import AgentExecutor, Tool
from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

def create_react_agent_with_langchain():
    """使用LangChain创建ReAct智能体"""
    
    # 定义工具
    tools = [
        Tool(
            name="search",
            description="Search the web for information",
            func=lambda query: f"Search results for: {query}"
        ),
        Tool(
            name="calculator",
            description="Perform mathematical calculations",
            func=lambda expression: f"Calculation result: {eval(expression)}"
        )
    ]
    
    # 创建提示模板
    react_prompt = PromptTemplate(
        input_variables=["input", "agent_scratchpad"],
        template="""Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Observation can repeat N times)
Thought: I know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:""",
        partial_variables={
            "tool_names": [tool.name for tool in tools]
        }
    )
    
    # 创建LLM
    llm = OpenAI(temperature=0)
    
    # 创建ReAct智能体
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=react_prompt,
        verbose=True
    )
    
    return agent

def create_react_agent_with_crewai():
    """使用CrewAI创建ReAct智能体"""
    
    from crewai import Agent, Task, Crew, Process
    
    # 定义智能体
    research_agent = Agent(
        role='Research Agent',
        goal='Conduct thorough research using search tools',
        backstory='You are an expert researcher with access to search tools',
        verbose=True,
        tools=["search"]
    )
    
    analysis_agent = Agent(
        role='Analysis Agent',
        goal='Analyze research results and draw conclusions',
        backstory='You are an expert analyst with critical thinking skills',
        verbose=True,
        tools=["calculator"]
    )
    
    # 定义任务
    research_task = Task(
        description='Research the topic and provide comprehensive information',
        agent=research_agent
    )
    
    analysis_task = Task(
        description='Analyze the research results and provide insights',
        agent=analysis_agent
    )
    
    # 创建团队
    crew = Crew(
        agents=[research_agent, analysis_agent],
        tasks=[research_task, analysis_task],
        process=Process.sequential,
        verbose=True
    )
    
    return crew
​

完整示例

# 完整的ReAct智能体使用示例

def main():
    """主函数演示完整的ReAct智能体使用"""
    
    # 1. 初始化工具
    tools = [
        SearchTool(),
        CalculatorTool()
    ]
    
    # 2. 创建智能体
    agent = ReActAgent(tools, max_iterations=5)
    
    # 3. 定义任务
    task = "搜索Python的最新版本并计算其发布年份距离现在的年数"
    
    print(f"执行任务: {task}")
    print("=" * 50)
    
    # 4. 执行任务
    result = agent.run(task)
    
    # 5. 输出结果
    print("\n最终结果:")
    print(f"任务: {result['task']}")
    print(f"思考: {result['thoughts'][0]}")
    print(f"行动次数: {result['iterations']}")
    print(f"最终答案: {result['final_answer']}")
    
    # 6. 使用优化版本
    optimized_agent = OptimizedReActAgent(tools, max_iterations=5)
    optimized_result = optimized_agent.run(task)
    
    print("\n优化版本结果:")
    print(f"迭代次数: {optimized_result['iterations']}")
    print(f"最终答案: {optimized_result['final_answer']}")

if __name__ == "__main__":
    main()
​

常见问题 FAQ

Q1：ReAct框架如何处理工具调用失败的情况？

A：ReAct框架通过多层次错误处理机制确保稳定性：

1. 异常捕获：在工具调用时使用try-catch捕获异常
2. 错误诊断：分析错误类型并生成诊断信息
3. 恢复策略：根据错误类型选择恢复策略
4. 重新规划：必要时重新规划后续行动

Q2：如何优化ReAct智能体的推理效率？

A：可以从以下几个方面进行优化：

1. 缓存机制：缓存重复的思考过程
2. 并行处理：支持多工具并行调用
3. 提前终止：在满足条件时提前停止循环
4. 记忆管理：优化记忆缓冲区的管理策略

Q3：ReAct与传统的链式推理有什么区别？

A：主要区别包括：

1. 循环性：ReAct是循环推理，链式推理是线性的
2. 工具调用：ReAct可以动态调用工具，链式推理通常固定
3. 观察反馈：ReAct基于观察结果调整策略，链式推理缺乏反馈
4. 灵活性：ReAct更加灵活适应不同场景

最佳实践与避坑

实践建议

• 合理设置最大迭代次数：避免无限循环
• 提供详细的工具描述：帮助智能体正确选择工具
• 实现完善的错误处理：确保系统稳定性
• 使用缓存机制：提升重复任务的效率

常见坑点

• 工具描述不清晰：导致智能体无法正确选择工具
• 缺乏错误处理：系统容易崩溃
• 迭代次数过多：影响响应速度
• 没有缓存机制：重复任务效率低下

性能优化技巧

• 使用更轻量级的LLM模型
• 实现工具结果缓存
• 优化提示模板设计
• 使用异步处理机制

本节小结

本节详细介绍了ReAct框架的完整代码实现，从基础架构到高级优化策略。主要内容包括：

1. 核心组件实现：工具系统、状态管理、循环控制
2. 优化策略：缓存机制、提前停止、记忆管理
3. 主流框架集成：LangChain和CrewAI的使用方法
4. 错误处理机制：多层次的异常处理和恢复策略

通过掌握这些实现方法，你可以根据具体场景定制和优化ReAct智能体，提升其性能和可靠性。

下一节我们将深入探讨ReAct在不同应用场景中的具体实践，包括搜索引擎优化、智能客服、数据分析等。

延伸阅读

• 官方文档：LangChain官方文档ReAct章节
• 相关章节：本教程3.1节工具系统设计
• 推荐论文：ReAct: Synergizing Reasoning and Acting in Language Models

关键词：Agent智能体开发实战,ReAct框架,代码实现,优化策略,LangChain,最佳实践
难度：进阶
预计阅读：30分钟