ReAct Agent实战指南：如何用Python快速搭建你的第一个智能代理（附代码）

ReAct Agent实战指南：如何用Python快速搭建你的第一个智能代理（附代码）

在人工智能技术快速迭代的今天，智能代理（Agent）正从实验室走向实际应用。不同于传统脚本的固定逻辑，ReAct Agent结合了推理（Reasoning）和行动（Action）的能力，让程序具备了类似人类的决策灵活性。本文将带你从零开始，用Python构建一个能理解任务、自主决策的智能代理。

1. 环境准备与基础配置

开始前需要确保你的开发环境满足以下条件：

• Python 3.8或更高版本
• 稳定的网络连接（用于模型下载）
• 至少8GB内存（处理基础模型的最低要求）

推荐使用conda创建独立环境：

conda create -n react_agent python=3.8 conda activate react_agent

核心依赖库安装：

pip install transformers==4.28.1 pip install torch==2.0.0 pip install langchain==0.0.198

提示：如果遇到CUDA相关错误，请根据显卡型号安装对应版本的torch（如pip install torch==2.0.0+cu117）

验证安装是否成功：

import torch print(torch.__version__) # 应输出2.0.0 from transformers import pipeline print(pipeline('text-generation')('Hello')[0]['generated_text']) # 测试基础NLP功能

2. ReAct Agent核心架构解析

一个典型的ReAct Agent包含三个关键组件：

1. 推理引擎：分析任务并制定策略
2. 行动模块：执行具体操作（如API调用、代码执行）
3. 记忆系统：存储历史交互记录

下面是一个最小化实现框架：

from typing import List, Dict, Any from abc import ABC, abstractmethod class ReActAgent: def __init__(self, model_name: str = "gpt-3.5-turbo"): self.model = self._load_model(model_name) self.memory = [] def _load_model(self, model_name: str): # 实际项目中替换为真实模型加载逻辑 return {"name": model_name} @abstractmethod def reason(self, observation: str) -> str: """生成推理过程""" pass @abstractmethod def act(self, reasoning: str) -> Any: """执行具体行动""" pass def run(self, input_text: str) -> Any: reasoning = self.reason(input_text) action_result = self.act(reasoning) self.memory.append({ "input": input_text, "reasoning": reasoning, "result": action_result }) return action_result

3. 实战：构建天气查询Agent

让我们实现一个能查询天气的实用Agent。这个Agent需要：

1. 理解用户关于天气的询问
2. 提取关键信息（地点、时间）
3. 调用天气API获取数据
4. 用自然语言回复

首先安装额外依赖：

pip install requests python-dotenv

创建.env文件存储API密钥：

WEATHER_API_KEY=your_key_here

完整实现代码：

import os import requests from dotenv import load_dotenv from typing import Optional load_dotenv() class WeatherAgent(ReActAgent): def __init__(self): super().__init__() self.api_key = os.getenv("WEATHER_API_KEY") self.base_url = "https://api.weatherapi.com/v1" def reason(self, observation: str) -> str: prompt = f""" 用户询问：{observation} 请分析： 1. 目标城市是哪里？ 2. 查询的时间是什么时候？（今天/明天/具体日期） 3. 用户最关心的天气要素是什么？ 按以下格式回复： 城市:<城市名> 时间:<时间描述> 要素:<温度/降水/风力等> """ # 实际项目中这里应调用LLM生成推理 # 简化示例直接使用规则提取 if "北京" in observation: return "城市:北京\ntime:今天\n要素:温度" return "城市:上海\ntime:今天\n要素:温度" def act(self, reasoning: str) -> str: lines = reasoning.split('\n') params = {line.split(':')[0]: line.split(':')[1] for line in lines} response = requests.get( f"{self.base_url}/current.json", params={ "key": self.api_key, "q": params["城市"], "lang": "zh" } ) data = response.json() current = data["current"] return f"{params['城市']}当前天气：{current['temp_c']}℃，{current['condition']['text']}" # 使用示例 agent = WeatherAgent() print(agent.run("北京现在天气怎么样？"))

4. 高级技巧与性能优化

当Agent需要处理复杂任务时，以下策略可以提升表现：

4.1 多步推理实现

def complex_reasoning(agent: ReActAgent, task: str, max_steps: int = 5): observation = task for _ in range(max_steps): reasoning = agent.reason(observation) print(f"推理过程：\n{reasoning}") result = agent.act(reasoning) if "最终答案" in result: return result observation = result return "达到最大步数仍未解决"

4.2 记忆优化方案

使用向量数据库存储历史交互：

from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import FAISS class MemoryEnhancedAgent(ReActAgent): def __init__(self): super().__init__() self.embeddings = OpenAIEmbeddings() self.vectorstore = FAISS.from_texts([""], self.embeddings) def retrieve_similar(self, query: str, k: int = 3) -> List[str]: docs = self.vectorstore.similarity_search(query, k=k) return [doc.page_content for doc in docs] def run(self, input_text: str) -> Any: similar = self.retrieve_similar(input_text) context = "\n".join(similar) reasoning = self.reason(f"历史上下文：{context}\n新问题：{input_text}") action_result = self.act(reasoning) self.vectorstore.add_texts([f"Q: {input_text}\nA: {action_result}"]) return action_result

4.3 性能对比测试

不同实现方式的响应时间对比：

5. 常见问题排查

问题1：API调用返回403错误

• 检查.env文件位置是否正确
• 确认API密钥是否有访问权限
• 验证网络代理设置

问题2：推理结果不符合预期

# 调试方法 agent = WeatherAgent() test_input = "上海明天会下雨吗" print("原始推理：", agent.reason(test_input))

问题3：内存泄漏处理

定期清理记忆存储：

def cleanup_memory(agent: MemoryEnhancedAgent, max_items: int = 100): if len(agent.vectorstore.index_to_docstore_id) > max_items: agent.vectorstore = FAISS.from_texts( agent.vectorstore.similarity_search("", k=max_items), agent.embeddings )

在实际项目中，我发现最影响Agent稳定性的往往是边缘案例的处理。比如当用户询问"下周三纽约的天气"时，需要正确解析相对日期和国际化城市名称。这需要结合专门的日期处理库（如dateparser）和地理位置服务。