Python AI Agent工具调用：2026年ReAct模式让大模型真正学会使用工具
AI与大数据
大模型会"用工具"和不会"用工具"，差距有多大？
GPT-4o能写代码、做分析、回答问题，但它不会查询实时数据库、不会调用API、不会操作文件系统——直到你给它"工具"。但给大模型接上工具，并不意味着它就会用。模型经常选错工具、传错参数、陷入循环调用，甚至编造不存在的工具。
ReAct（Reasoning + Acting）模式通过"思考-行动-观察"循环，让大模型真正学会推理和工具使用的协调。2026年，这是构建生产级AI Agent的核心范式。
ReAct模式核心概念
| 概念 | 说明 | 对比 |
|---|---|---|
| Reasoning | 模型先推理当前状态和下一步行动 | 对比直接调用工具 |
| Acting | 执行工具调用并获取结果 | 对比纯文本推理 |
| Observation | 观察工具返回结果并更新推理 | 对比忽略结果 |
| Tool Definition | 用JSON Schema定义工具接口 | 对比自然语言描述 |
| Function Calling | 模型原生工具调用能力 | 对比正则提取 |
| Tool Orchestration | 多工具的编排和依赖管理 | 对比单工具调用 |
ReAct循环流程：
用户问题 → [Thought] 分析问题 → [Action] 选择工具 → [Observation] 获取结果
→ [Thought] 继续推理 → [Action] 选择工具 → [Observation] 获取结果
→ [Thought] 信息充分 → [Answer] 最终回答
问题深入分析：为什么工具调用这么难？
| 问题 | 原因 | ReAct解决方案 |
|---|---|---|
| 选错工具 | 模型不理解工具边界 | Thought步骤先推理再选择 |
| 参数错误 | Schema理解偏差 | 严格Schema验证+重试 |
| 循环调用 | 缺乏终止条件 | 最大步数限制+观察判断 |
| 幻觉工具 | 模型编造不存在的工具 | 工具白名单+强制选择 |
| 结果误读 | 忽略工具返回的关键信息 | Observation步骤显式处理 |
| 上下文丢失 | 长对话遗忘早期信息 | 结构化记忆管理 |
分步实操：从零构建ReAct Agent
第一步：定义工具Schema
from typing import Any, Optional
from pydantic import BaseModel, Field
from enum import Enum
import json
import datetime
class ToolParameter(BaseModel):
name: str
type: str
description: str
required: bool = True
enum: Optional[list[str]] = None
class ToolDefinition(BaseModel):
name: str
description: str
parameters: list[ToolParameter]
def to_openai_format(self) -> dict:
properties = {}
required = []
for param in self.parameters:
prop = {"type": param.type, "description": param.description}
if param.enum:
prop["enum"] = param.enum
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
}
class WeatherInput(BaseModel):
city: str = Field(description="城市名称")
unit: str = Field(default="celsius", description="温度单位", pattern="^(celsius|fahrenheit)$")
class DatabaseQueryInput(BaseModel):
sql: str = Field(description="SQL查询语句，仅支持SELECT")
database: str = Field(default="production", description="数据库名称")
class CalculatorInput(BaseModel):
expression: str = Field(description="数学表达式，如 '2 + 3 * 4'")
precision: int = Field(default=2, description="小数精度")
第二步：实现工具执行器
import sqlite3
import math
import re
from typing import Callable
class ToolRegistry:
def __init__(self):
self._tools: dict[str, dict] = {}
def register(self, name: str, description: str, parameters: list[ToolParameter], executor: Callable):
self._tools[name] = {
"definition": ToolDefinition(name=name, description=description, parameters=parameters),
"executor": executor,
}
def get_openai_tools(self) -> list[dict]:
return [tool["definition"].to_openai_format() for tool in self._tools.values()]
def execute(self, name: str, arguments: dict) -> str:
if name not in self._tools:
return json.dumps({"error": f"Tool '{name}' not found", "available": list(self._tools.keys())})
try:
result = self._tools[name]["executor"](**arguments)
return json.dumps(result, ensure_ascii=False, default=str)
except Exception as e:
return json.dumps({"error": str(e), "tool": name})
def execute_weather(city: str, unit: str = "celsius") -> dict:
mock_data = {
"北京": {"temp": 28, "condition": "晴", "humidity": 45},
"上海": {"temp": 32, "condition": "多云", "humidity": 72},
"深圳": {"temp": 35, "condition": "雷阵雨", "humidity": 85},
"New York": {"temp": 75, "condition": "Sunny", "humidity": 55},
}
data = mock_data.get(city, {"temp": 25, "condition": "未知", "humidity": 50})
if unit == "fahrenheit" and city in mock_data:
data["temp"] = data["temp"] * 9 / 5 + 32
return {"city": city, **data, "unit": unit, "updated_at": datetime.datetime.now().isoformat()}
def execute_database_query(sql: str, database: str = "production") -> dict:
forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
if any(kw in sql.upper() for kw in forbidden):
return {"error": "仅支持SELECT查询", "sql": sql}
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS orders (id INTEGER, product TEXT, amount REAL, status TEXT)")
cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", [
(1, "机械键盘", 599.0, "completed"),
(2, "4K显示器", 2999.0, "pending"),
(3, "无线鼠标", 199.0, "completed"),
(4, "笔记本支架", 89.0, "shipped"),
])
conn.commit()
try:
cursor.execute(sql)
columns = [desc[0] for desc in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
return {"results": rows, "row_count": len(rows), "database": database}
except Exception as e:
return {"error": str(e), "sql": sql}
finally:
conn.close()
def execute_calculator(expression: str, precision: int = 2) -> dict:
safe_chars = re.sub(r'[^0-9+\-*/().%\s]', '', expression)
if safe_chars != expression.strip():
return {"error": "表达式包含不安全字符", "expression": expression}
try:
result = eval(safe_chars, {"__builtins__": {}}, {"abs": abs, "round": round, "pow": pow})
return {"result": round(result, precision), "expression": expression}
except Exception as e:
return {"error": str(e), "expression": expression}
def execute_web_search(query: str, max_results: int = 3) -> dict:
mock_results = [
{"title": f"搜索结果: {query} - 相关文章1", "url": f"https://example.com/{query}/1", "snippet": f"关于{query}的详细信息..."},
{"title": f"搜索结果: {query} - 技术文档", "url": f"https://docs.example.com/{query}", "snippet": f"{query}的技术规范和用法..."},
{"title": f"搜索结果: {query} - 最新动态", "url": f"https://news.example.com/{query}", "snippet": f"{query}领域的最新进展..."},
]
return {"results": mock_results[:max_results], "query": query}
registry = ToolRegistry()
registry.register(
name="get_weather",
description="获取指定城市的天气信息，包括温度、天气状况和湿度",
parameters=[
ToolParameter(name="city", type="string", description="城市名称"),
ToolParameter(name="unit", type="string", description="温度单位：celsius或fahrenheit", required=False, enum=["celsius", "fahrenheit"]),
],
executor=execute_weather,
)
registry.register(
name="query_database",
description="执行SQL查询获取数据库信息，仅支持SELECT语句",
parameters=[
ToolParameter(name="sql", type="string", description="SQL查询语句"),
ToolParameter(name="database", type="string", description="数据库名称", required=False),
],
executor=execute_database_query,
)
registry.register(
name="calculate",
description="计算数学表达式的值",
parameters=[
ToolParameter(name="expression", type="string", description="数学表达式"),
ToolParameter(name="precision", type="integer", description="小数精度", required=False),
],
executor=execute_calculator,
)
registry.register(
name="web_search",
description="搜索互联网获取信息",
parameters=[
ToolParameter(name="query", type="string", description="搜索关键词"),
ToolParameter(name="max_results", type="integer", description="最大结果数", required=False),
],
executor=execute_web_search,
)
第三步：实现ReAct Agent核心
from openai import OpenAI
import os
class ReActAgent:
def __init__(self, tool_registry: ToolRegistry, model: str = "gpt-4o", max_steps: int = 8):
self.registry = tool_registry
self.model = model
self.max_steps = max_steps
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.conversation_history: list[dict] = []
def _build_system_prompt(self) -> str:
return """你是一个智能助手，能够使用工具来回答用户问题。
请按照ReAct模式工作：
1. Thought: 分析当前情况，决定下一步行动
2. Action: 选择合适的工具并调用
3. Observation: 观察工具返回结果
4. 重复以上步骤直到获得足够信息
5. Answer: 基于收集的信息给出最终回答
重要规则：
- 每次只调用一个工具
- 仔细阅读工具返回结果后再决定下一步
- 如果工具返回错误，尝试修正参数重试
- 不要编造不存在的工具
- 当信息足够时，直接给出最终回答"""
def run(self, user_message: str) -> str:
self.conversation_history = [
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_message},
]
tools = self.registry.get_openai_tools()
for step in range(self.max_steps):
print(f"\n--- Step {step + 1}/{self.max_steps} ---")
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation_history,
tools=tools,
tool_choice="auto",
temperature=0.1,
)
message = response.choices[0].message
self.conversation_history.append(message.to_dict())
if message.content:
print(f"Thought: {message.content}")
if not message.tool_calls:
return message.content or "无法生成回答"
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"Action: {func_name}({json.dumps(func_args, ensure_ascii=False)})")
result = self.registry.execute(func_name, func_args)
print(f"Observation: {result[:200]}...")
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
return "达到最大步数限制，未能完成推理。"
if __name__ == "__main__":
agent = ReActAgent(registry)
answer = agent.run("北京今天天气怎么样？帮我查一下最近完成的订单总金额。")
print(f"\n最终回答: {answer}")
完整代码：带错误恢复的多工具Agent
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional, Callable
import json
import os
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AgentConfig(BaseModel):
model: str = "gpt-4o"
max_steps: int = 10
retry_attempts: int = 2
retry_delay: float = 1.0
temperature: float = 0.1
verbose: bool = True
class AgentState(BaseModel):
step_count: int = 0
tool_calls_count: int = 0
errors: list[str] = []
tool_history: list[dict] = []
class ProductionAgent:
def __init__(self, tool_registry: ToolRegistry, config: AgentConfig | None = None):
self.registry = tool_registry
self.config = config or AgentConfig()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.state = AgentState()
def _execute_tool_with_retry(self, name: str, arguments: dict) -> str:
for attempt in range(self.config.retry_attempts):
try:
result = self.registry.execute(name, arguments)
parsed = json.loads(result)
if "error" in parsed:
if attempt < self.config.retry_attempts - 1:
logger.warning(f"Tool '{name}' returned error (attempt {attempt + 1}): {parsed['error']}")
time.sleep(self.config.retry_delay)
continue
return result
except Exception as e:
if attempt < self.config.retry_attempts - 1:
logger.warning(f"Tool '{name}' execution failed (attempt {attempt + 1}): {e}")
time.sleep(self.config.retry_delay)
else:
return json.dumps({"error": f"Tool execution failed after {self.config.retry_attempts} attempts: {e}"})
return json.dumps({"error": "Unexpected retry loop exit"})
def _validate_tool_call(self, tool_call) -> Optional[str]:
if tool_call.function.name not in self.registry._tools:
available = list(self.registry._tools.keys())
return json.dumps({
"error": f"工具 '{tool_call.function.name}' 不存在",
"available_tools": available,
"suggestion": f"请从以下工具中选择: {', '.join(available)}",
})
try:
json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
return json.dumps({"error": f"参数JSON解析失败: {e}", "raw_arguments": tool_call.function.arguments})
return None
def run(self, user_message: str, context: str = "") -> dict:
start_time = time.time()
self.state = AgentState()
system_prompt = f"""你是一个专业的AI助手，能够使用工具来回答用户问题。
{context}
工作流程（ReAct模式）：
1. Thought: 分析问题，规划行动
2. Action: 调用合适的工具
3. Observation: 分析工具结果
4. 重复直到信息充分
5. Answer: 给出完整回答
规则：
- 每步只调用一个工具
- 仔细分析返回结果
- 工具出错时修正参数重试
- 信息充分时直接回答"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
tools = self.registry.get_openai_tools()
for step in range(self.config.max_steps):
self.state.step_count = step + 1
try:
response = self.client.chat.completions.create(
model=self.config.model,
messages=messages,
tools=tools,
tool_choice="auto",
temperature=self.config.temperature,
)
except Exception as e:
self.state.errors.append(f"API调用失败: {e}")
break
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
break
for tool_call in message.tool_calls:
self.state.tool_calls_count += 1
func_name = tool_call.function.name
validation_error = self._validate_tool_call(tool_call)
if validation_error:
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": validation_error,
})
self.state.errors.append(f"验证失败: {func_name}")
continue
func_args = json.loads(tool_call.function.arguments)
self.state.tool_history.append({"tool": func_name, "args": func_args, "step": step + 1})
result = self._execute_tool_with_retry(func_name, func_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
final_message = messages[-1].get("content", "") if messages else ""
return {
"answer": final_message,
"steps": self.state.step_count,
"tool_calls": self.state.tool_calls_count,
"errors": self.state.errors,
"tool_history": self.state.tool_history,
"elapsed_seconds": round(time.time() - start_time, 2),
}
if __name__ == "__main__":
agent = ProductionAgent(registry, AgentConfig(max_steps=8, retry_attempts=2))
result = agent.run("帮我查一下北京的天气，然后搜索一下Python最新的版本信息，最后算一下 (28 + 32) * 1.5 的结果")
print(f"\n回答: {result['answer']}")
print(f"步数: {result['steps']}, 工具调用: {result['tool_calls']}, 耗时: {result['elapsed_seconds']}s")
避坑指南
坑1：模型编造不存在的工具
模型有时会"幻觉"出不存在的工具名，导致调用失败。
解决方案：
- 工具白名单验证，返回可用工具列表提示模型
- 使用
tool_choice: "auto"而非tool_choice: "required" - 在系统提示中明确列出可用工具名称
坑2：参数类型不匹配
模型传字符串给数字参数，或缺少必填参数。
解决方案：
- Schema中使用
enum限制可选值 - 使用Pydantic模型验证参数
- 在工具执行器中做类型转换和默认值处理
坑3：Agent陷入循环调用
模型反复调用同一工具，或在不同工具间来回切换。
解决方案：
- 设置
max_steps硬限制（推荐8-10步） - 在工具历史中检测重复调用
- 在系统提示中强调"信息充分时直接回答"
坑4：工具返回结果过长
数据库查询返回数千行结果，超出上下文窗口。
解决方案：
- 工具执行器中限制返回条数（如最多20行）
- 对长结果做摘要后再返回给模型
- 使用
max_tokens控制工具响应长度
坑5：并发工具调用顺序问题
模型同时调用多个有依赖关系的工具，导致参数缺失。
解决方案：
- 限制每步只调用一个工具
- 在系统提示中明确"逐步执行，等待结果后再决定下一步"
- 对有依赖的工具在描述中标注前置条件
报错排查
| 序号 | 报错信息 | 原因 | 解决方法 |
|---|---|---|---|
| 1 | Invalid function_call: function name not found |
模型调用了未定义的工具 | 添加工具白名单验证 |
| 2 | JSON decode error in function arguments |
模型生成的参数不是有效JSON | 添加JSON解析容错处理 |
| 3 | Rate limit exceeded: 429 |
API调用频率超限 | 添加指数退避重试 |
| 4 | Context length exceeded |
对话历史+工具结果超长 | 截断历史或压缩工具结果 |
| 5 | Tool execution timeout |
工具执行超时 | 添加超时限制和异步执行 |
| 6 | Function argument missing required parameter |
缺少必填参数 | Schema中设置required并验证 |
| 7 | Circular tool call detected |
工具循环调用 | 添加max_steps和重复检测 |
| 8 | Model refused to use tools |
模型拒绝使用工具 | 检查system prompt和tool_choice |
| 9 | Tool returned unexpected format |
工具返回格式异常 | 统一返回JSON格式 |
| 10 | Token usage exceeded budget |
Token用量超预算 | 添加token计数和预算控制 |
进阶优化
1. 记忆管理
class ConversationMemory:
def __init__(self, max_messages: int = 20, summary_threshold: int = 15):
self.messages: list[dict] = []
self.max_messages = max_messages
self.summary_threshold = summary_threshold
self.summary: str = ""
def add(self, message: dict):
self.messages.append(message)
if len(self.messages) > self.summary_threshold:
self._compress()
def _compress(self):
old_messages = self.messages[:len(self.messages) - 5]
self.summary += f"\n[历史摘要] 共{len(old_messages)}条消息"
self.messages = self.messages[len(self.messages) - 5:]
def get_messages(self) -> list[dict]:
result = []
if self.summary:
result.append({"role": "system", "content": f"之前对话的摘要: {self.summary}"})
result.extend(self.messages)
return result[-self.max_messages:]
2. 工具编排DSL
class ToolPipeline:
def __init__(self, steps: list[dict]):
self.steps = steps
@classmethod
def from_config(cls, config: list[dict]) -> "ToolPipeline":
return cls(steps=config)
def execute(self, registry: ToolRegistry, initial_input: dict) -> dict:
result = initial_input
for step in self.steps:
tool_name = step["tool"]
args_mapping = step.get("args", {})
args = {}
for key, value in args_mapping.items():
if isinstance(value, str) and value.startswith("$"):
args[key] = result.get(value[1:])
else:
args[key] = value
output = json.loads(registry.execute(tool_name, args))
result_name = step.get("output_as", tool_name)
result[result_name] = output
return result
pipeline = ToolPipeline.from_config([
{"tool": "get_weather", "args": {"city": "$city"}, "output_as": "weather"},
{"tool": "web_search", "args": {"query": "$city 旅游攻略"}, "output_as": "travel_info"},
])
3. Token预算控制
class TokenBudget:
def __init__(self, max_total: int = 100000, max_per_step: int = 8000):
self.max_total = max_total
self.max_per_step = max_per_step
self.used = 0
def can_proceed(self, estimated_tokens: int = 4000) -> bool:
return (self.used + estimated_tokens) < self.max_total
def consume(self, tokens: int):
self.used += tokens
def remaining(self) -> int:
return max(0, self.max_total - self.used)
def should_summarize(self) -> bool:
return self.used > self.max_total * 0.8
对比分析
| 维度 | ReAct Agent | Plan-and-Execute | Reflexion | AutoGPT |
|---|---|---|---|---|
| 推理方式 | 逐步推理+行动 | 先规划再执行 | 自我反思修正 | 自主目标驱动 |
| 工具调用 | 每步一个 | 按计划批量 | 带反思重试 | 自由调用 |
| 错误恢复 | 观察后调整 | 重新规划 | 自我批评修正 | 试错 |
| 上下文管理 | 滚动窗口 | 规划缓存 | 反思记忆 | 长期记忆 |
| 适用场景 | 多步推理 | 复杂任务分解 | 需要高质量输出 | 开放式任务 |
| Token消耗 | 中等 | 较高 | 高 | 很高 |
| 可控性 | 高 | 中 | 中 | 低 |
| 实现复杂度 | 低 | 中 | 高 | 高 |
| 稳定性 | 高 | 中 | 中 | 低 |
总结：ReAct模式通过"思考-行动-观察"循环，让大模型从"被动回答"进化为"主动推理+工具使用"。2026年构建生产级AI Agent的关键在于：严格的工具Schema定义、可靠的重试和错误恢复机制、合理的步数和Token预算控制。从单工具Agent开始，逐步引入多工具编排、记忆管理和自我反思，是落地AI Agent的最佳路径。记住：好的Agent不是让模型自由发挥，而是通过结构化约束引导模型做出正确决策。
在线工具推荐
- JSON格式化：/zh-CN/json/format — 格式化Function Calling的Schema和工具响应
- Base64编解码：/zh-CN/encode/base64 — 编解码API密钥和Agent配置
- Curl转代码：/zh-CN/dev/curl-to-code — 将AI API调试curl转为Python代码
本站提供浏览器本地工具，免注册即可试用 →
#Python#AI Agent#Tool Use#Function Calling#ReAct#工具编排#LangChain#智能体