LangChain ReAct Agent 开发：推理行动循环实现完整指南

前言：从真实错误场景开始

记得有一次，我在凌晨三点部署一个重要的 AI Agent 项目，当时代码测试环境运行完美，但一上线就遇到了 ConnectionError: timeout after 30 seconds 的错误。排查了整整两个小时，最后发现是我复制代码时不小心把 base_url 写成了 api.openai.com 而不是 api.holysheep.ai/v1。这个经历让我深刻体会到配置细节的重要性。

在本文中，我将分享如何正确使用 HolySheep AI 的 API 来构建 LangChain ReAct Agent，整个实现基于我参与三个生产级 Agent 项目的经验总结。

什么是 ReAct 架构？

ReAct（Reasoning + Acting）是一种让 AI Agent 能够进行多步推理和行动的框架。核心思想是让模型在每一步都经历：思考当前状态 → 决定行动 → 观察结果 → 调整策略的循环过程。

项目设置与完整代码实现

首先安装必要的依赖包：

pip install langchain langchain-core requests python-dotenv

完整的 ReAct Agent 实现代码：

import os
import json
import requests
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field

HolySheep API 配置
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

@dataclass
class Tool:
    """工具定义类"""
    name: str
    description: str
    func: Callable

@dataclass
class ReActAgent:
    """ReAct Agent 核心实现"""
    api_key: str
    model: str = "gpt-4.1"
    max_iterations: int = 10
    tools: List[Tool] = field(default_factory=list)
    
    def __post_init__(self):
        self.tool_map = {tool.name: tool for tool in self.tools}
        self.reasoning_history: List[Dict] = []
        
    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        tools_desc = "\n".join([
            f"- {tool.name}: {tool.description}"
            for tool in self.tools
        ]) if self.tools else "无可用工具"
        
        return f"""你是一个智能助手，可以使用工具来完成任务。

可用工具：
{tools_desc}

请按照以下格式回复：
思考：{你的推理过程}
行动：tool_name
行动输入：{{"param": "value"}}

或者最终回复：
思考：{你的推理过程}
行动：完成
行动输入：{最终答案}
"""
    
    def _call_llm(self, messages: List[Dict]) -> Dict[str, Any]:
        """调用 HolySheep API"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": 2000
        }
        
        try:
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            raise TimeoutError("API 请求超时，请检查网络连接或增加超时时间")
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                raise PermissionError("API 密钥无效或已过期，请检查 HOLYSHEEP_API_KEY")
            raise
    
    def _execute_tool(self, tool_name: str, arguments: Dict) -> str:
        """执行工具"""
        if tool_name not in self.tool_map:
            return f"错误：未找到工具 '{tool_name}'"
        
        tool = self.tool_map[tool_name]
        try:
            result = tool.func(**arguments)
            return str(result)
        except TypeError as e:
            return f"错误：工具参数不匹配 - {str(e)}"
        except Exception as e:
            return f"错误：{str(e)}"
    
    def _parse_response(self, response_text: str) -> tuple[str, str, Any]:
        """解析 LLM 响应"""
        lines = response_text.strip().split("\n")
        thought = ""
        action = None
        action_input = None
        
        for line in lines:
            if line.startswith("思考："):
                thought = line[3:].strip()
            elif line.startswith("行动："):
                action = line[3:].strip()
            elif line.startswith("行动输入："):
                try:
                    action_input = json.loads(line[5:].strip())
                except json.JSONDecodeError:
                    action_input = line[5:].strip()
        
        return thought, action, action_input
    
    def run(self, query: str) -> Dict[str, Any]:
        """运行 ReAct Agent"""
        self.reasoning_history = []
        
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": query}
        ]
        
        for iteration in range(self.max_iterations):
            # 调用 LLM
            response = self._call_llm(messages)
            assistant_message = response["choices"][0]["message"]["content"]
            
            # 解析响应
            thought, action, action_input = self._parse_response(assistant_message)
            
            # 记录推理过程
            step_record = {
                "iteration": iteration + 1,
                "thought": thought,
                "action": action,
                "input": action_input
            }
            self.reasoning_history.append(step_record)
            
            # 添加助手消息到对话历史
            messages.append({"role": "assistant", "content": assistant_message})
            
            # 判断是否完成
            if action == "完成":
                return {
                    "success": True,
                    "answer": action_input,
                    "steps": self.reasoning_history
                }
            
            # 执行工具
            if action and action in self.tool_map:
                tool_result = self._execute_tool(action, action_input)
                
                # 将工具执行结果添加到对话
                messages.append({
                    "role": "user",
                    "content": f"工具 '{action}' 执行结果：{tool_result}"
                })
                step_record["result"] = tool_result
        
        return {
            "success": False,
            "error": "达到最大迭代次数",
            "steps": self.reasoning_history
        }

定义工具函数
def calculator(expression: str) -> float:
    """计算数学表达式"""
    try:
        # 安全计算（仅支持基本运算）
        allowed_chars = set("0123456789+-*/.() ")
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return result
        return "错误：表达式包含非法字符"
    except Exception as e:
        return f"计算错误：{str(e)}"

def get_weather(city: str) -> str:
    """获取天气信息（模拟）"""
    weather_data = {
        "曼谷": "晴，32°C，湿度 75%",
        "北京": "多云，15°C，湿度 45%",
        "东京": "小雨，18°C，湿度 80%"
    }
    return weather_data.get(city, f"未找到 {city} 的天气数据")

使用示例
if __name__ == "__main__":
    # 创建工具
    tools = [
        Tool(name="calculator", description="计算数学表达式", func=calculator),
        Tool(name="get_weather", description="获取城市天气", func=get_weather)
    ]
    
    # 创建 Agent
    agent = ReActAgent(
        api_key=API_KEY,
        model="gpt-4.1",
        tools=tools
    )
    
    # 运行查询
    result = agent.run("曼谷今天的天气怎么样？如果气温高于30度，给我计算一下开空调的电费（假设每度电5泰铢，每天运行8小时）。")
    
    print(f"成功: {result['success']}")
    print(f"答案: {result.get('answer', result.get('error'))}")
    print(f"推理步骤数: {len(result['steps'])}")

使用 LangChain 的 ReAct 实现

如果你更倾向于使用 LangChain 框架，HolySheep 也完全兼容。以下是使用 LangChain 实现 ReAct Agent 的方式：

import os
from langchain.agents import AgentType, initialize_agent
from langchain.tools import Tool
from langchain.chat_models import ChatOpenAI
from langchain.schema import SystemMessage

HolySheep API 配置
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

创建 Chat 模型（使用 HolySheep）
llm = ChatOpenAI(
    model_name="gpt-4.1",
    temperature=0.7,
    max_tokens=2000
)

定义工具
def search_web(query: str) -> str:
    """模拟网络搜索"""
    return f"搜索结果：{query} 相关信息 - 这是一个模拟结果"

def get_date() -> str:
    """获取当前日期"""
    from datetime import datetime
    return datetime.now().strftime("%Y年%m月%d日")

tools = [
    Tool(
        name="search",
        func=search_web,
        description="搜索网络信息，输入搜索关键词"
    ),
    Tool(
        name="get_date",
        func=get_date,
        description="获取当前日期"
    )
]

初始化 ReAct Agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

运行查询
result = agent.run("今天是什么日期？帮我搜索一下泰国旅游的最新信息。")

print(f"Agent 结果: {result}")

实际应用案例：多步骤数据处理 Agent

在实际项目中，我曾使用 HolySheep 构建一个数据分析 Agent，它可以自动从多个数据源获取数据、清洗处理、然后生成报告：

import os
import json
from typing import List, Dict

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

class DataProcessingAgent:
    """数据分析处理 Agent"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.tools = {
            "fetch_data": self._fetch_data,
            "clean_data": self._clean_data,
            "analyze": self._analyze,
            "generate_report": self._generate_report
        }
    
    def _fetch_data(self, source: str, query: str) -> Dict:
        """获取数据"""
        # 模拟数据获取
        return {
            "status": "success",
            "records": 1500,
            "sample": [{"id": i, "value": i * 10} for i in range(5)]
        }
    
    def _clean_data(self, data: str) -> str:
        """清洗数据"""
        # 移除空值和异常值
        return "已清洗数据，移除了 23 条无效记录"
    
    def _analyze(self, criteria: str) -> str:
        """分析数据"""
        return f"分析完成：发现 {criteria} 相关数据占比 67.5%"
    
    def _generate_report(self, format: str) -> str:
        """生成报告"""
        return f"报告已生成，格式：{format}，包含 5 个章节"
    
    def execute(self, task: str) -> str:
        """执行任务"""
        steps = [
            ("fetch_data", {"source": "database", "query": task}),
            ("clean_data", {"data": "raw_data"}),
            ("analyze
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
Agent Tool Selection Strategy: การเพิ่มประสิทธิภาพ Function 
Spring Boot รวม AI API ผ่าน HolySheep พร้อม Production-Ready
ลิขสิทธิ์โค้ด AI: ความเสี่ยงทางกฎหมายและแนวทางจัดการสำหรับนั

前言：从真实错误场景开始

什么是 ReAct 架构？

项目设置与完整代码实现

HolySheep API 配置

定义工具函数

使用示例

使用 LangChain 的 ReAct 实现

HolySheep API 配置

创建 Chat 模型（使用 HolySheep）

定义工具

初始化 ReAct Agent

运行查询

实际应用案例：多步骤数据处理 Agent

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง

🔥 ลอง HolySheep AI