Python AI Agent Tool Use: ReAct Pattern That Teaches LLMs to Actually Use Tools in 2026
AI与大数据
How Big Is the Gap Between LLMs That Can and Can't Use Tools?
GPT-4o can write code, analyze data, and answer questions, but it can't query real-time databases, call APIs, or operate file systems—until you give it "tools." But giving an LLM tools doesn't mean it knows how to use them. Models often pick the wrong tool, pass incorrect parameters, get stuck in loops, or even hallucinate non-existent tools.
The ReAct (Reasoning + Acting) pattern enables LLMs to truly coordinate reasoning and tool usage through a "Think-Act-Observe" loop. In 2026, this is the core paradigm for building production-grade AI Agents.
ReAct Pattern Core Concepts
| Concept | Description | Compared To |
|---|---|---|
| Reasoning | Model reasons about current state and next action | vs. direct tool calling |
| Acting | Execute tool call and get results | vs. pure text reasoning |
| Observation | Observe tool results and update reasoning | vs. ignoring results |
| Tool Definition | Define tool interface with JSON Schema | vs. natural language description |
| Function Calling | Model's native tool calling capability | vs. regex extraction |
| Tool Orchestration | Multi-tool coordination and dependency management | vs. single tool calling |
ReAct loop flow:
User Question → [Thought] Analyze problem → [Action] Select tool → [Observation] Get result
→ [Thought] Continue reasoning → [Action] Select tool → [Observation] Get result
→ [Thought] Sufficient info → [Answer] Final answer
Deep Analysis: Why Is Tool Calling So Hard?
| Problem | Cause | ReAct Solution |
|---|---|---|
| Wrong tool selected | Model doesn't understand tool boundaries | Thought step reasons before selecting |
| Parameter errors | Schema understanding gap | Strict Schema validation + retry |
| Loop calls | No termination condition | Max steps limit + observation judgment |
| Hallucinated tools | Model invents non-existent tools | Tool whitelist + forced selection |
| Misread results | Ignoring key info in tool returns | Observation step processes explicitly |
| Context loss | Long conversations forget early info | Structured memory management |
Step-by-Step: Building a ReAct Agent from Scratch
Step 1: Define Tool Schema
from typing import Any, Optional
from pydantic import BaseModel, Field
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="City name")
unit: str = Field(default="celsius", description="Temperature unit", pattern="^(celsius|fahrenheit)$")
class DatabaseQueryInput(BaseModel):
sql: str = Field(description="SQL query, SELECT only")
database: str = Field(default="production", description="Database name")
class CalculatorInput(BaseModel):
expression: str = Field(description="Math expression, e.g. '2 + 3 * 4'")
precision: int = Field(default=2, description="Decimal precision")
Step 2: Implement Tool Executor
import sqlite3
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 = {
"Beijing": {"temp": 28, "condition": "Sunny", "humidity": 45},
"Shanghai": {"temp": 32, "condition": "Cloudy", "humidity": 72},
"New York": {"temp": 75, "condition": "Sunny", "humidity": 55},
}
data = mock_data.get(city, {"temp": 25, "condition": "Unknown", "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": "Only SELECT queries are supported", "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, "Keyboard", 599.0, "completed"),
(2, "Monitor", 2999.0, "pending"),
(3, "Mouse", 199.0, "completed"),
])
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 contains unsafe characters", "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"Result: {query} - Article 1", "url": f"https://example.com/{query}/1", "snippet": f"Details about {query}..."},
{"title": f"Result: {query} - Documentation", "url": f"https://docs.example.com/{query}", "snippet": f"Technical specs for {query}..."},
]
return {"results": mock_results[:max_results], "query": query}
registry = ToolRegistry()
registry.register(
name="get_weather",
description="Get weather information for a specified city",
parameters=[
ToolParameter(name="city", type="string", description="City name"),
ToolParameter(name="unit", type="string", description="Temperature unit: celsius or fahrenheit", required=False, enum=["celsius", "fahrenheit"]),
],
executor=execute_weather,
)
registry.register(
name="query_database",
description="Execute SQL query to get database information, SELECT only",
parameters=[
ToolParameter(name="sql", type="string", description="SQL query statement"),
ToolParameter(name="database", type="string", description="Database name", required=False),
],
executor=execute_database_query,
)
registry.register(
name="calculate",
description="Calculate the value of a math expression",
parameters=[
ToolParameter(name="expression", type="string", description="Math expression"),
ToolParameter(name="precision", type="integer", description="Decimal precision", required=False),
],
executor=execute_calculator,
)
registry.register(
name="web_search",
description="Search the internet for information",
parameters=[
ToolParameter(name="query", type="string", description="Search keyword"),
ToolParameter(name="max_results", type="integer", description="Max results count", required=False),
],
executor=execute_web_search,
)
Step 3: Implement ReAct Agent Core
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 """You are an intelligent assistant that can use tools to answer user questions.
Follow the ReAct pattern:
1. Thought: Analyze the current situation and decide the next action
2. Action: Select and call the appropriate tool
3. Observation: Observe the tool's return result
4. Repeat until sufficient information is gathered
5. Answer: Provide the final answer based on collected information
Important rules:
- Call only one tool per step
- Read tool results carefully before deciding the next step
- If a tool returns an error, try correcting parameters and retry
- Do not fabricate non-existent tools
- When information is sufficient, provide the final answer directly"""
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 "Unable to generate answer"
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)})")
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 "Maximum steps reached, unable to complete reasoning."
if __name__ == "__main__":
agent = ReActAgent(registry)
answer = agent.run("What's the weather in Beijing? Also check the total amount of completed orders.")
print(f"\nFinal Answer: {answer}")
Complete Code: Multi-Tool Agent with Error Recovery
from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
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}' 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}' failed (attempt {attempt + 1}): {e}")
time.sleep(self.config.retry_delay)
else:
return json.dumps({"error": f"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 '{tool_call.function.name}' does not exist",
"available_tools": available,
})
try:
json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
return json.dumps({"error": f"Argument JSON parse failed: {e}"})
return None
def run(self, user_message: str, context: str = "") -> dict:
start_time = time.time()
self.state = AgentState()
system_prompt = f"""You are a professional AI assistant that uses tools to answer questions.
{context}
Workflow (ReAct pattern):
1. Thought: Analyze the problem, plan action
2. Action: Call the appropriate tool
3. Observation: Analyze tool results
4. Repeat until sufficient info
5. Answer: Provide complete answer
Rules:
- One tool per step
- Analyze results carefully
- Retry with corrected params on error
- Answer directly when info is sufficient"""
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 call failed: {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"Validation failed: {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("Check Beijing weather, search for latest Python version, and calculate (28 + 32) * 1.5")
print(f"\nAnswer: {result['answer']}")
print(f"Steps: {result['steps']}, Tool calls: {result['tool_calls']}, Time: {result['elapsed_seconds']}s")
Pitfall Guide
Pitfall 1: Model Hallucinates Non-existent Tools
Models sometimes "hallucinate" tool names that don't exist.
Solution: Tool whitelist validation, return available tools list, explicitly list tool names in system prompt.
Pitfall 2: Parameter Type Mismatch
Model passes strings to numeric parameters or misses required params.
Solution: Use enum in Schema, validate with Pydantic models, handle type conversion in executor.
Pitfall 3: Agent Stuck in Loop
Model repeatedly calls the same tool or bounces between tools.
Solution: Set max_steps hard limit (8-10 recommended), detect repeated calls in tool history, emphasize "answer directly when sufficient" in system prompt.
Pitfall 4: Tool Returns Excessively Long Results
Database query returns thousands of rows, exceeding context window.
Solution: Limit return rows in executor (max 20), summarize long results, use max_tokens to control response length.
Pitfall 5: Concurrent Tool Call Ordering Issues
Model calls multiple dependent tools simultaneously, causing missing parameters.
Solution: Limit one tool per step, emphasize "execute step by step" in system prompt, annotate prerequisites in tool descriptions.
Error Troubleshooting
| # | Error Message | Cause | Solution |
|---|---|---|---|
| 1 | Invalid function_call: function name not found |
Model called undefined tool | Add tool whitelist validation |
| 2 | JSON decode error in function arguments |
Model generated invalid JSON args | Add JSON parse error handling |
| 3 | Rate limit exceeded: 429 |
API rate limit exceeded | Add exponential backoff retry |
| 4 | Context length exceeded |
History + tool results too long | Truncate history or compress results |
| 5 | Tool execution timeout |
Tool execution timed out | Add timeout limit and async execution |
| 6 | Function argument missing required parameter |
Missing required parameter | Set required in Schema and validate |
| 7 | Circular tool call detected |
Tool loop detected | Add max_steps and repeat detection |
| 8 | Model refused to use tools |
Model refused tool usage | Check system prompt and tool_choice |
| 9 | Tool returned unexpected format |
Unexpected tool return format | Standardize JSON return format |
| 10 | Token usage exceeded budget |
Token usage over budget | Add token counting and budget control |
Advanced Optimization
1. Memory Management
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[Summary] {len(old_messages)} messages compressed"
self.messages = self.messages[len(self.messages) - 5:]
def get_messages(self) -> list[dict]:
result = []
if self.summary:
result.append({"role": "system", "content": f"Previous conversation summary: {self.summary}"})
result.extend(self.messages)
return result[-self.max_messages:]
2. Tool Orchestration DSL
class ToolPipeline:
def __init__(self, steps: list[dict]):
self.steps = steps
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
3. Token Budget Control
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)
Comparison Analysis
| Dimension | ReAct Agent | Plan-and-Execute | Reflexion | AutoGPT |
|---|---|---|---|---|
| Reasoning | Step-by-step | Plan first, then execute | Self-reflection | Autonomous goal-driven |
| Tool Calling | One per step | Batch by plan | With reflection retry | Free calling |
| Error Recovery | Adjust after observation | Re-plan | Self-critique | Trial and error |
| Context Mgmt | Sliding window | Plan cache | Reflection memory | Long-term memory |
| Best For | Multi-step reasoning | Complex task decomposition | High-quality output | Open-ended tasks |
| Token Cost | Medium | Higher | High | Very high |
| Controllability | High | Medium | Medium | Low |
| Implementation | Low | Medium | High | High |
| Stability | High | Medium | Medium | Low |
Summary: The ReAct pattern evolves LLMs from "passive answering" to "active reasoning + tool usage" through the "Think-Act-Observe" loop. Keys to building production-grade AI Agents in 2026: strict tool Schema definitions, reliable retry and error recovery mechanisms, reasonable step and Token budget controls. Start with single-tool Agents, gradually introduce multi-tool orchestration, memory management, and self-reflection—this is the best path to production AI Agents. Remember: good Agents don't let models run free; they guide models to make correct decisions through structured constraints.
Recommended Online Tools
- JSON Formatter: /en/json/format — Format Function Calling Schema and tool responses
- Base64 Encode/Decode: /en/encode/base64 — Encode/decode API keys and Agent configs
- Curl to Code: /en/dev/curl-to-code — Convert AI API debug curl to Python code
Try these browser-local tools — no sign-up required →
#Python#AI Agent#Tool Use#Function Calling#ReAct#工具编排#LangChain#智能体