手撕源码：实现一个简单的ReAct智能体

核心知识点讲解

ReAct智能体的基本架构

一个完整的ReAct智能体通常由以下核心组件组成：

1. 推理引擎：负责生成思考过程和决策
2. 行动执行器：负责执行选定的行动
3. 环境交互器：负责与外部环境或工具交互
4. 记忆管理器：负责存储和管理历史信息
5. 决策控制器：负责控制整个执行流程

ReAct智能体的执行流程

1. 初始化：接收用户输入，设置初始状态
2. 循环执行：
   • 生成思考（Reasoning）
   • 选择行动（Action）
   • 执行行动并获取观察结果（Observation）
   • 更新状态和记忆
3. 终止条件：达到最大迭代次数或找到答案
4. 生成最终回答：基于所有思考和观察生成答案

实现ReAct智能体的关键技术

1. 提示词设计：设计有效的ReAct格式提示词
2. 工具集成：集成外部工具，如搜索、计算等
3. 状态管理：管理智能体的状态和历史信息
4. 错误处理：处理工具调用和执行过程中的错误
5. 终止条件：设置合理的终止条件

实用案例分析

案例1：实现一个基础的ReAct智能体

import openai
import json

class BasicReActAgent:
    def __init__(self, api_key, max_iterations=5):
        """初始化ReAct智能体"""
        openai.api_key = api_key
        self.max_iterations = max_iterations
        self.memory = []
    
    def generate_thought(self, question, context):
        """生成思考过程"""
        prompt = f"""你是一个使用ReAct范式的智能体。请分析以下问题并生成思考过程：

问题: {question}

当前上下文: {context}

思考: """
        
        response = openai.Completion.create(
            model="gpt-3.5-turbo-instruct",
            prompt=prompt,
            temperature=0.7,
            max_tokens=200
        )
        
        thought = response.choices[0].text.strip()
        return thought
    
    def generate_action(self, question, context, thought):
        """生成行动"""
        prompt = f"""你是一个使用ReAct范式的智能体。基于以下信息生成行动：

问题: {question}

当前上下文: {context}

思考: {thought}

可用行动：
1. Search[关键词] - 搜索相关信息
2. Finish[答案] - 完成任务并提供答案

行动: """
        
        response = openai.Completion.create(
            model="gpt-3.5-turbo-instruct",
            prompt=prompt,
            temperature=0.7,
            max_tokens=100
        )
        
        action = response.choices[0].text.strip()
        return action
    
    def execute_action(self, action):
        """执行行动"""
        if action.startswith("Search[") and action.endswith("]"):
            # 提取搜索关键词
            keyword = action[7:-1].strip()
            # 这里简化处理，实际应用中应调用真实的搜索API
            observation = f"搜索结果：关于'{keyword}'的相关信息..."
            return observation
        elif action.startswith("Finish[") and action.endswith("]"):
            # 提取答案
            answer = action[7:-1].strip()
            return f"完成：{answer}"
        else:
            return "无效行动，请重新选择"
    
    def run(self, question):
        """运行智能体"""
        context = ""
        self.memory = []
        
        for i in range(self.max_iterations):
            print(f"\n=== 迭代 {i+1} ===")
            
            # 1. 生成思考
            thought = self.generate_thought(question, context)
            print(f"思考: {thought}")
            
            # 2. 生成行动
            action = self.generate_action(question, context, thought)
            print(f"行动: {action}")
            
            # 3. 执行行动
            observation = self.execute_action(action)
            print(f"观察: {observation}")
            
            # 4. 更新上下文和记忆
            self.memory.append({
                "thought": thought,
                "action": action,
                "observation": observation
            })
            
            context = "\n".join([f"思考: {item['thought']}\n行动: {item['action']}\n观察: {item['observation']}" for item in self.memory])
            
            # 5. 检查是否完成
            if observation.startswith("完成："):
                answer = observation[4:]
                print(f"\n=== 最终答案 ===")
                print(answer)
                return answer
        
        # 达到最大迭代次数
        print(f"\n=== 达到最大迭代次数 ===")
        # 生成最终答案
        final_answer = self.generate_final_answer(question, context)
        print(final_answer)
        return final_answer
    
    def generate_final_answer(self, question, context):
        """生成最终答案"""
        prompt = f"""基于以下思考和观察，回答原始问题：

原始问题: {question}

思考和观察:
{context}

最终答案: """
        
        response = openai.Completion.create(
            model="gpt-3.5-turbo-instruct",
            prompt=prompt,
            temperature=0.7,
            max_tokens=300
        )
        
        answer = response.choices[0].text.strip()
        return answer

# 使用示例
if __name__ == "__main__":
    agent = BasicReActAgent(
        api_key="your-api-key",
        max_iterations=3
    )
    
    question = "如何制作一杯完美的意式浓缩咖啡？"
    agent.run(question)

案例2：集成真实工具的ReAct智能体

import openai
import requests
from serpapi import GoogleSearch

class ReActAgentWithTools:
    def __init__(self, openai_api_key, serpapi_api_key, max_iterations=5):
        """初始化带工具的ReAct智能体"""
        openai.api_key = openai_api_key
        self.serpapi_api_key = serpapi_api_key
        self.max_iterations = max_iterations
        self.memory = []
        self.tools = {
            "search": self.search_tool,
            "calculate": self.calculate_tool,
            "weather": self.weather_tool
        }
    
    def search_tool(self, query):
        """搜索工具"""
        try:
            params = {
                "q": query,
                "api_key": self.serpapi_api_key
            }
            search = GoogleSearch(params)
            results = search.get_dict()
            
            # 提取搜索结果
            if "organic_results" in results:
                snippets = []
                for result in results["organic_results"][:3]:  # 取前3个结果
                    if "snippet" in result:
                        snippets.append(result["snippet"])
                return "\n".join(snippets)
            else:
                return "未找到相关信息"
        except Exception as e:
            return f"搜索失败: {str(e)}"
    
    def calculate_tool(self, expression):
        """计算工具"""
        try:
            # 使用eval计算简单表达式，实际应用中应使用更安全的方法
            result = eval(expression)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算失败: {str(e)}"
    
    def weather_tool(self, location):
        """天气工具"""
        try:
            # 使用OpenWeatherMap API，需要替换为真实的API密钥
            api_key = "your-openweather-api-key"
            url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=metric&lang=zh_cn"
            response = requests.get(url)
            data = response.json()
            
            if data["cod"] == 200:
                weather = data["weather"][0]["description"]
                temp = data["main"]["temp"]
                humidity = data["main"]["humidity"]
                return f"{location}的天气: {weather}, 温度: {temp}°C, 湿度: {humidity}%"
            else:
                return "未找到该地点的天气信息"
        except Exception as e:
            return f"获取天气失败: {str(e)}"
    
    def generate_thought_and_action(self, question, context):
        """生成思考和行动"""
        prompt = f"""你是一个使用ReAct范式的智能体，可以使用以下工具：

工具列表：
1. search[查询词] - 搜索网络信息
2. calculate[表达式] - 进行数学计算
3. weather[地点] - 获取指定地点的天气信息
4. finish[答案] - 完成任务并提供答案

请按照以下格式输出：
思考: [你的思考过程]
行动: [你要执行的行动]

问题: {question}

当前上下文: {context}

思考: """
        
        response = openai.Completion.create(
            model="gpt-3.5-turbo-instruct",
            prompt=prompt,
            temperature=0.7,
            max_tokens=300
        )
        
        output = response.choices[0].text.strip()
        
        # 解析思考和行动
        if "行动: " in output:
            thought_part, action_part = output.split("行动: ", 1)
            thought = thought_part.replace("思考: ", "").strip()
            action = action_part.strip()
            return thought, action
        else:
            return "我需要分析这个问题并决定如何行动。", "search[如何回答这个问题]"
    
    def execute_action(self, action):
        """执行行动"""
        # 解析行动
        if action.startswith("search[") and action.endswith("]"):
            query = action[7:-1].strip()
            result = self.tools["search"](query)
            return f"搜索结果: {result}"
        elif action.startswith("calculate[") and action.endswith("]"):
            expression = action[10:-1].strip()
            result = self.tools["calculate"](expression)
            return f"计算结果: {result}"
        elif action.startswith("weather[") and action.endswith("]"):
            location = action[8:-1].strip()
            result = self.tools["weather"](location)
            return f"天气信息: {result}"
        elif action.startswith("finish[") and action.endswith("]"):
            answer = action[7:-1].strip()
            return f"完成: {answer}"
        else:
            return "无效行动，请使用正确的工具格式"
    
    def run(self, question):
        """运行智能体"""
        context = ""
        self.memory = []
        
        for i in range(self.max_iterations):
            print(f"\n=== 迭代 {i+1} ===")
            
            # 1. 生成思考和行动
            thought, action = self.generate_thought_and_action(question, context)
            print(f"思考: {thought}")
            print(f"行动: {action}")
            
            # 2. 执行行动
            observation = self.execute_action(action)
            print(f"观察: {observation}")
            
            # 3. 更新上下文和记忆
            self.memory.append({
                "thought": thought,
                "action": action,
                "observation": observation
            })
            
            context = "\n".join([f"思考: {item['thought']}\n行动: {item['action']}\n观察: {item['observation']}" for item in self.memory])
            
            # 4. 检查是否完成
            if observation.startswith("完成: "):
                answer = observation[4:]
                print(f"\n=== 最终答案 ===")
                print(answer)
                return answer
        
        # 达到最大迭代次数
        print(f"\n=== 达到最大迭代次数 ===")
        # 生成最终答案
        final_answer = self.generate_final_answer(question, context)
        print(final_answer)
        return final_answer
    
    def generate_final_answer(self, question, context):
        """生成最终答案"""
        prompt = f"""基于以下思考和观察，回答原始问题：

原始问题: {question}

思考和观察:
{context}

最终答案: """
        
        response = openai.Completion.create(
            model="gpt-3.5-turbo-instruct",
            prompt=prompt,
            temperature=0.7,
            max_tokens=300
        )
        
        answer = response.choices[0].text.strip()
        return answer

# 使用示例
if __name__ == "__main__":
    agent = ReActAgentWithTools(
        openai_api_key="your-api-key",
        serpapi_api_key="your-serpapi-key",
        max_iterations=5
    )
    
    question = "北京今天的天气如何？北京的人口大约是多少？"
    agent.run(question)

案例3：使用LangChain实现ReAct智能体

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.tools import Tool
from langchain.utilities import SerpAPIWrapper, Calculator
from langchain.agents import initialize_agent, AgentType

# 1. 初始化工具
search = SerpAPIWrapper()
calculator = Calculator()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索网络信息"
    ),
    Tool(
        name="Calculator",
        func=calculator.run,
        description="用于进行数学计算"
    )
]

# 2. 自定义ReAct提示词
react_prompt = PromptTemplate(
    template="""你是一个使用ReAct范式的智能体，能够通过思考和行动来解决问题。

可用工具:
{tools}

请按照以下格式回答：

问题: {input}

思考: [你的思考过程]
行动: [你要执行的行动，格式为工具名称后跟参数，例如：Search[关键词]]
观察: [行动的结果]
... (重复思考-行动-观察步骤)
思考: [最终思考]
回答: [你的最终答案]

开始回答：
""",
    input_variables=["input", "tools"]
)

# 3. 初始化语言模型
llm = OpenAI(temperature=0.7)

# 4. 创建LLMChain
react_chain = LLMChain(
    llm=llm,
    prompt=react_prompt,
    verbose=True
)

# 5. 使用LangChain内置的ReAct智能体
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 6. 运行自定义ReAct链
def run_custom_react(question):
    print("=== 使用自定义ReAct链 ===")
    tools_description = "\n".join([f"{tool.name}: {tool.description}" for tool in tools])
    result = react_chain.run(input=question, tools=tools_description)
    print(result)

# 7. 运行内置ReAct智能体
def run_builtin_react(question):
    print("\n=== 使用内置ReAct智能体 ===")
    result = agent.run(question)
    print(result)

# 使用示例
if __name__ == "__main__":
    question = "2023年世界杯冠军是谁？这个国家的人口大约是多少？"
    run_custom_react(question)
    run_builtin_react(question)

代码解析

案例1：基础ReAct智能体

1. 核心组件：

   • generate_thought：生成思考过程
   • generate_action：生成行动决策
   • execute_action：执行行动
   • run：控制整个执行流程

2. 执行流程：

   • 初始化上下文和记忆
   • 循环生成思考、行动和观察
   • 更新上下文和记忆
   • 检查终止条件
   • 生成最终答案

3. 简化处理：

   • 使用模拟的搜索功能
   • 简化的行动执行逻辑
   • 基础的记忆管理

案例2：带工具的ReAct智能体

1. 工具集成：

   • search_tool：使用SerpAPI进行搜索
   • calculate_tool：进行数学计算
   • weather_tool：获取天气信息

2. 增强功能：

   • 合并思考和行动生成，减少API调用
   • 完善的错误处理
   • 更复杂的工具调用逻辑

3. 实际应用：

   • 集成真实的第三方API
   • 处理多步骤任务
   • 管理复杂的上下文信息

案例3：LangChain实现

1. 内置工具：

   • 使用LangChain内置的SerpAPIWrapper和Calculator
   • 标准化的工具接口

2. 两种实现方式：

   • 自定义ReAct提示词和链
   • 使用LangChain内置的AgentType.ZERO_SHOT_REACT_DESCRIPTION

3. 优势：

   • 代码更简洁
   • 内置的智能体管理
   • 标准化的组件接口

高级技巧

1. 优化提示词

好的提示词是ReAct智能体成功的关键：

# 优化后的ReAct提示词
better_prompt = PromptTemplate(
    template="""你是一个专业的问题解决专家，使用ReAct范式来分析问题和执行行动。

你的目标是通过以下步骤解决问题：
1. 分析问题，理解需要什么信息
2. 决定使用什么工具来获取信息
3. 执行工具并分析结果
4. 基于所有信息生成最终答案

可用工具:
{tools}

请按照以下格式回答，确保思考过程清晰，行动具体：

问题: {input}

思考: [分析当前情况，说明为什么需要执行下一步行动]
行动: [工具名称[参数]]
观察: [工具执行结果]
... (重复上述步骤)
思考: [总结所有信息，说明如何得出最终答案]
回答: [清晰、准确的最终答案]

开始：
""",
    input_variables=["input", "tools"]
)

2. 改进记忆管理

有效的记忆管理可以提高智能体的性能：

class MemoryManager:
    def __init__(self, max_memory_size=10):
        self.max_memory_size = max_memory_size
        self.memory = []
    
    def add(self, item):
        """添加记忆项"""
        self.memory.append(item)
        # 保持记忆大小在限制范围内
        if len(self.memory) > self.max_memory_size:
            self.memory = self.memory[-self.max_memory_size:]
    
    def get_recent(self, n=3):
        """获取最近的n个记忆项"""
        return self.memory[-n:]
    
    def get_context(self):
        """获取上下文字符串"""
        context = "\n".join([
            f"思考: {item['thought']}\n行动: {item['action']}\n观察: {item['observation']}"
            for item in self.memory
        ])
        return context
    
    def clear(self):
        """清空记忆"""
        self.memory = []

# 在智能体中使用
class ImprovedReActAgent:
    def __init__(self, api_key, max_iterations=5):
        self.api_key = api_key
        self.max_iterations = max_iterations
        self.memory_manager = MemoryManager()
    
    # 其他方法...
    
    def run(self, question):
        context = ""
        self.memory_manager.clear()
        
        for i in range(self.max_iterations):
            # 生成思考和行动
            thought, action = self.generate_thought_and_action(question, context)
            
            # 执行行动
            observation = self.execute_action(action)
            
            # 添加到记忆
            self.memory_manager.add({
                "thought": thought,
                "action": action,
                "observation": observation
            })
            
            # 获取更新后的上下文
            context = self.memory_manager.get_context()
            
            # 检查是否完成
            if observation.startswith("完成: "):
                return observation[4:]
        
        # 生成最终答案
        return self.generate_final_answer(question, context)

3. 动态工具选择

根据任务类型动态选择工具：

class DynamicToolSelector:
    def __init__(self, tools):
        self.tools = tools
    
    def select_tools(self, question):
        """根据问题选择合适的工具"""
        selected_tools = []
        
        # 分析问题类型
        if any(keyword in question for keyword in ["天气", "温度", "下雨"]):
            if "weather" in self.tools:
                selected_tools.append(self.tools["weather"])
        
        if any(keyword in question for keyword in ["计算", "等于", "加", "减", "乘", "除"]):
            if "calculate" in self.tools:
                selected_tools.append(self.tools["calculate"])
        
        # 搜索工具几乎适用于所有问题
        if "search" in self.tools:
            selected_tools.append(self.tools["search"])
        
        return selected_tools

# 在智能体中使用
def generate_thought_and_action(self, question, context):
    # 动态选择工具
    selected_tools = self.tool_selector.select_tools(question)
    tools_description = "\n".join([f"{tool.name}: {tool.description}" for tool in selected_tools])
    
    # 生成思考和行动
    # ...

性能优化

1. 减少API调用

• 批量生成：一次生成多个思考和行动
• 缓存结果：缓存相同查询的结果
• 智能终止：提前识别可以直接回答的问题

2. 提高推理质量

• 链式思考：引导模型进行更深入的推理
• 示例引导：在提示词中包含示例
• 多步计划：鼓励模型制定详细的行动计划

3. 错误处理和恢复

• 重试机制：工具调用失败时自动重试
• 备用方案：当一个工具失败时尝试其他工具
• 自我纠正：识别和纠正错误的推理

总结

通过本集的学习，我们实现了一个简单的ReAct智能体，并深入理解了其工作原理：

1. 核心组件：推理引擎、行动执行器、环境交互器、记忆管理器、决策控制器
2. 执行流程：观察→思考→行动→观察→...→回答
3. 实现方式：
   • 基础实现：手动构建所有组件
   • 工具集成：集成外部工具和API
   • LangChain实现：使用LangChain的内置组件
4. 高级技巧：优化提示词、改进记忆管理、动态工具选择
5. 性能优化：减少API调用、提高推理质量、加强错误处理

ReAct范式的强大之处在于它结合了推理和行动，使智能体能够像人类一样思考和解决问题。通过与外部工具的集成，ReAct智能体可以获取最新信息，处理复杂任务，减少幻觉，提高可靠性。

课后思考

1. 如何设计更有效的ReAct提示词？
2. 如何平衡推理深度和执行效率？
3. 如何处理工具调用失败的情况？
4. 如何评估ReAct智能体的性能？
5. ReAct范式在哪些场景下特别适用？

在下一集中，我们将学习如何使用LangChain预置的Agent类型，如Zero-shot ReAct等，来更快速地构建智能体系统。