Tool Schema Definition
A tool has three parts
- name — what the function is called / 函数叫什么名字
- description — when the LLM should call it / 模型在什么情况下调用它
- parameters — what inputs it needs / 它需要哪些输入参数
The 3 elements are the top-level structure
│
├── name ← 第1要素
├── description ← 第2要素
└── parameters ← 第3要素
│
├── properties ← parameters 的内部细节
└── required ← parameters 的内部细节e.g. — Annotated example: get_weather
# Universal JSON Schema structure (language-agnostic)
{
"name": "get_weather", # snake_case, no spaces
"description": "Get current weather # Tell model WHEN to call
for a given city.
Call this whenever the user asks about
weather, temperature, or climate.",
"parameters": {
"type": "object", # always "object" at top level
"properties": {
"city": {
"type": "string",
"description": "City name, e.g. 'Beijing'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"], # constrain choices
"description": "Temperature unit",
"default": "celsius"
}
},
"required": ["city"] # "unit" is optional
}
}# another simple example
{
"name": "get_weather",
"description": "Get weather for a city.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city name"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Temperature unit"
}
},
"required": ["city"] ← 只有 city 是必填 / only city is required
}
}The core question: After you register tools, how does the LLM decide which tool to call and what arguments to pass?
You send: [User Message] + [Tool Schemas]
↓
LLM thinks: “Which tool do I need? What args?”
↓
LLM returns: stop_reason = “tool_use”
content = [tool_use block]
↓
YOU execute: tool_name + input (arguments)
↓
Send result back → LLM continues
The LLM does NOT execute tools. It only decides and describes what to call. You execute it.
LLM 不执行工具。它只是决定并描述要调用什么。你来执行。
Key Concepts to Lock In / 必须掌握的核心概念
| Concept | What it means | 中文 |
|---|---|---|
| stop_reason = “tool_use” | LLM wants YOU to run a tool | LLM 要你执行工具 |
| stop_reason = “end_turn” | LLM is done, no more tools needed | LLM 完成，无需更多工具 |
| block.name | Which tool was selected | 选了哪个工具 |
| block.input | Arguments the LLM chose | LLM 选的参数 |
| block.id | Must be passed back with result | 返回结果时必须带上 |
Tool execution and result return means: the model does not directly run your function. It decides which tool to call and with what arguments. Your application executes the real function, then sends the tool result back to the model so the model can produce the final answer.
工具执行与结果返回”的意思是：模型本身不直接运行你的函数。它只判断“要调用哪个工具”和“参数是什么”。真正执行函数的是你的应用程序；执行完以后，你要把工具结果再写回给模型，让模型基于结果生成最终回答。
Why It Is Important
Tool result return is the part that makes AI reliable. Without it, the model may hallucinate data. With it, the model can ground its answer in real system output.
CN: 工具结果返回让 AI 变得可靠。没有它，模型可能编造数据；有了它，模型可以基于真实系统结果回答。
e.g. User asks: “Refund order 123.” The model should not invent the refund status. It should call get_order(123), then maybe call create_refund(123), then tell the user the real result.
Core Flow
EN: tool schema describes the function name, purpose, parameters, required fields, and JSON structure.
CN: tool schema 描述函数名、用途、参数、必填字段和 JSON 结构。
EN: tool call is the model’s request to your app: “Call this function with these arguments.”
CN: tool call 是模型对你的应用发出的请求：“请用这些参数调用这个函数。”
EN: tool execution happens in your code, not inside the model. This may call a database, API, file system, payment service, or internal business function.
CN: tool execution 发生在你的代码里，不是在模型内部。它可能调用数据库、API、文件系统、支付服务或内部业务函数。
EN: tool result is the real output returned by your function. It must be attached to the correct tool call ID.
CN: tool result 是你的函数返回的真实结果。它必须绑定到正确的 tool call ID。
EN: error handling means returning controlled errors to the model, such as invalid arguments, permission denied, timeout, or service unavailable.
CN: error handling 指把可控错误返回给模型，例如参数错误、权限不足、超时、服务不可用。
EN: full tool loop is the heart of an Agent. The model reasons, your code acts, the model observes the result, then continues.
CN: 完整 Tool Call 循环 是 Agent 的核心。模型负责推理，你的代码负责行动，模型观察结果后继续下一步。
%pip install openai
import json
import os
from openai import OpenAI
api_key=os.environ["OPENAI_API_KEY"]
# debug
print ("=" * 50)
print ("api_key = ",api_key)
==================================================
api_key = sk-proj-25_XulvDV5ns*****b-kB4KkkTGtSzbd-ABpNnDU58A
# define OpenAI Client
client = OpenAI(api_key = api_key)
# Real business function / 真实业务函数
def get_order_status(order_id: str) -> dict:
fake_db = {
"A1001": {"status": "shipped", "eta": "2026-06-20"},
"A1002": {"status": "processing", "eta": None},
}
if order_id not in fake_db:
return {
"ok": False,
"error": "ORDER_NOT_FOUND",
"message": f"Order {order_id} does not exist."
}
return {
"ok": True,
"order_id": order_id,
**fake_db[order_id],
}
# debug
print ( get_order_status('Refund order 123'))
{'ok': False, 'error': 'ORDER_NOT_FOUND', 'message': 'Order Refund order 123 does not exist.'}
print ( get_order_status('A1001'))
{'ok': True, 'order_id': 'A1001', 'status': 'shipped', 'eta': '2026-06-20'}
# define Tool
#define 3 core ： name, Description, parameter.
tools = [
{
"type": "function",
"name": "get_order_status",
"description": "Get the current status and estimated arrival date for an order.",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID, for example A1001."
}
},
"required": ["order_id"],
"additionalProperties": False
}
}
]
""" defines a Python function named.
Its job is to receive a user question, run the full “LLM + tool calling” loop, and finally return the model’s final answer.
CN: 它的作用是：接收用户问题，执行完整的“LLM + 工具调用”循环，最后返回模型的最终回答。
"""
def run_agent(user_text: str) -> str:
# Conversation state / 对话状态
input_items = [
{
"role": "user",
"content": user_text
}
]
while True:
response = client.responses.create( # Agent 的“模型请求入口”。
model="gpt-5.5",
input=input_items,
tools=tools,
instructions=( # <--System-level behavior instruction.
# <-- 系统级行为指令。
"You are a customer support assistant. "
"Use tools for order status questions. "
"If the tool returns an error, explain it politely."
)
)
# Save model output, including possible tool calls.
# 保存模型输出，包括可能出现的工具调用。
input_items += response.output # <--This appends the model’s response to
# the conversation history.
# 这行把模型的输出追加到对话历史里。
tool_outputs = [] #This creates an empty list to store results from executed tools.
#CN: 这里创建一个空列表，用来存放工具执行后的结果。
for item in response.output:
if item.type != "function_call":
continue
"""
The API provider defines it. API 提供商定义它。
The returned type is defined by the specific API provider’s response schema.
For OpenAI Responses API, OpenAI defines output item types
such as message, function_call, etc.
For Claude, Anthropic defines its own types like text, tool_use, tool_result.
For other APIs, the names may be different.
返回的 type 是由具体 API 提供商的“响应结构 schema”定义的。
OpenAI Responses API 里，OpenAI 定义了 message、function_call 等输出项类型。
Claude 里，Anthropic 定义了自己的类型，
比如 text、tool_use、tool_result。不同 API 名字可能不一样。
如果 类型 不是function_call,下面skipped 进入下个循环
"""
try:
args = json.loads(item.arguments)
if item.name == "get_order_status": # 如果model请求的工具是get_order_status， 我们
# 已经定义好了的工具， 就执行 这个工具
result = get_order_status(order_id=args["order_id"])
else: # 如果模型请求了一个你的代码不支持的工具名，
# 就不要执行任何东西，而是返回一个错误结果
result = {
"ok": False,
"error": "UNKNOWN_TOOL",
"message": f"Tool {item.name} is not supported."
}
except json.JSONDecodeError:
result = {
"ok": False,
"error": "BAD_JSON",
"message": "Tool arguments were not valid JSON."
}
except KeyError as exc:
result = {
"ok": False,
"error": "MISSING_ARGUMENT",
"message": f"Missing required argument: {exc}"
}
except Exception as exc:
result = {
"ok": False,
"error": "TOOL_EXECUTION_FAILED",
"message": str(exc)
}
# Return tool result to the model.
# 把工具结果写回给模型。
tool_outputs.append({
"type": "function_call_output",
"call_id": item.call_id,
"output": json.dumps(result)
})
# No tool call means the model has produced the final answer.
# 没有工具调用，说明模型已经生成最终回答。
if not tool_outputs: # tool_outputs 没有了,empty, tool_outputs就是 false
return response.output_text # 停止loop
input_items += tool_outputs
answer = run_agent("Where is my order A1001?")
print(answer)re-orchestrated implementation code
from openai import OpenAI
import json
from datetime import datetime
# EN: Create OpenAI client. It reads OPENAI_API_KEY from environment variables.
# CN: 创建 OpenAI 客户端。它会从环境变量 OPENAI_API_KEY 读取密钥。
client = OpenAI()
# ============================================================
# 1. Real functions: these are actually executed by your code
# 真实函数：这些函数才是你的代码真正会执行的东西
# ============================================================
def get_weather(city: str, date: str = "today"):
"""
EN:
Real weather tool.
In production, this function should call a real weather API.
CN:
真实天气工具。
在真实项目里，这里应该调用真正的天气 API。
"""
return {
"ok": True,
"tool": "get_weather",
"city": city,
"date": date,
"weather": "rainy",
"temperature_c": 18
}
def get_stock_price(ticker: str):
"""
EN:
Real stock price tool.
In production, this function should call a real stock market API.
CN:
真实股票价格工具。
在真实项目里，这里应该调用真正的股票行情 API。
"""
return {
"ok": True,
"tool": "get_stock_price",
"ticker": ticker,
"price": 213.56,
"currency": "USD"
}
def get_current_time(location: str = "New York"):
"""
EN:
Real current time tool.
This simple demo returns local machine time.
CN:
真实当前时间工具。
这个 demo 简单返回本机时间。
"""
return {
"ok": True,
"tool": "get_current_time",
"location": location,
"current_time": datetime.now().isoformat()
}
# ============================================================
# 2. Tool registry: allowed tools whitelist
# 工具注册表：允许被执行的工具白名单
# ============================================================
TOOL_REGISTRY = {
# EN: The key must match the tool schema "name".
# CN: key 必须和工具 schema 里的 "name" 一致。
"get_weather": get_weather,
"get_stock_price": get_stock_price,
"get_current_time": get_current_time,
}
# ============================================================
# 3. Tool schemas: descriptions shown to the model
# 工具说明书：给模型看的工具定义
# ============================================================
tools = [
{
# EN: This tells the API this tool is a function tool.
# CN: 这里告诉 API：这个工具是 function 类型。
"type": "function",
# EN: This name is what the model will return in item.name.
# CN: 这个名字之后会出现在模型返回的 item.name 里。
"name": "get_weather",
# EN: Description helps the model decide when to use this tool.
# CN: description 帮助模型判断什么时候该用这个工具。
"description": "Get weather for a city and date. Use this for live or forecast weather questions.",
# EN: parameters define what arguments the model must provide.
# CN: parameters 定义模型调用工具时必须提供哪些参数。
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name, for example New York"
},
"date": {
"type": "string",
"description": "Date, for example today, tomorrow, or 2026-06-19"
}
},
"required": ["city"]
}
},
{
"type": "function",
"name": "get_stock_price",
"description": "Get latest stock price for a ticker symbol. Use this for live stock price questions.",
"parameters": {
"type": "object",
"properties": {
"ticker": {
"type": "string",
"description": "Stock ticker symbol, for example AAPL, MSFT, TSLA"
}
},
"required": ["ticker"]
}
},
{
"type": "function",
"name": "get_current_time",
"description": "Get current time for a location. Use this when the user asks what time it is now.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Location name, for example New York or Shanghai"
}
},
"required": []
}
}
]
# ============================================================
# 4. Execute one function_call item
# 执行模型返回的一个 function_call
# ============================================================
def execute_tool_call(item):
"""
EN:
The model returns a function_call item like:
{
"type": "function_call",
"name": "get_weather",
"arguments": "{\"city\":\"New York\",\"date\":\"tomorrow\"}",
"call_id": "call_abc"
}
This function:
1. Parses item.arguments
2. Finds the real Python function from TOOL_REGISTRY
3. Executes the function
4. Returns the tool result
CN:
模型会返回一个 function_call item，例如：
{
"type": "function_call",
"name": "get_weather",
"arguments": "{\"city\":\"New York\",\"date\":\"tomorrow\"}",
"call_id": "call_abc"
}
这个函数做四件事：
1. 解析 item.arguments
2. 从 TOOL_REGISTRY 找到真实 Python 函数
3. 执行这个函数
4. 返回工具执行结果
"""
try:
# EN:
# item.arguments comes from the model.
# It is usually a JSON string, not a Python dict yet.
#
# CN:
# item.arguments 来自模型。
# 它通常是 JSON 字符串，还不是 Python 字典。
args = json.loads(item.arguments)
# EN:
# item.name also comes from the model.
# Example: "get_weather", "get_stock_price"
#
# CN:
# item.name 也来自模型。
# 例如："get_weather", "get_stock_price"
tool_name = item.name
# EN:
# Look up the real function by name.
# This is a whitelist. The model cannot execute arbitrary functions.
#
# CN:
# 根据工具名查找真实函数。
# 这是白名单机制。模型不能随便执行任意函数。
tool_func = TOOL_REGISTRY.get(tool_name)
# EN:
# If model requested a tool that our code does not support,
# return an error object instead of executing anything.
#
# CN:
# 如果模型请求了代码不支持的工具，
# 返回错误对象，不执行任何东西。
if tool_func is None:
return {
"ok": False,
"error": "UNKNOWN_TOOL",
"message": f"Tool {tool_name} is not supported."
}
# EN:
# Execute the real Python function.
# **args means convert dict into keyword arguments.
# Example: {"city": "New York"} -> get_weather(city="New York")
#
# CN:
# 执行真实 Python 函数。
# **args 表示把字典变成关键字参数。
# 例如：{"city": "New York"} -> get_weather(city="New York")
result = tool_func(**args)
# EN: Return successful tool result.
# CN: 返回工具成功执行的结果。
return result
except json.JSONDecodeError:
# EN:
# json.loads failed because item.arguments was not valid JSON.
#
# CN:
# json.loads 失败，说明 item.arguments 不是合法 JSON。
return {
"ok": False,
"error": "BAD_JSON",
"message": "Tool arguments were not valid JSON."
}
except TypeError as exc:
# EN:
# Usually happens when arguments do not match function parameters.
# Example: get_weather() got an unexpected keyword argument.
#
# CN:
# 通常发生在参数和函数签名不匹配时。
# 例如：传了函数不认识的参数。
return {
"ok": False,
"error": "BAD_ARGUMENTS",
"message": str(exc)
}
except Exception as exc:
# EN:
# Catch other unexpected failures inside the tool.
# Example: database error, network error, API error.
#
# CN:
# 捕获工具内部其他意外错误。
# 例如：数据库错误、网络错误、第三方 API 错误。
return {
"ok": False,
"error": "TOOL_EXECUTION_FAILED",
"message": str(exc)
}
# ============================================================
# 5. Full Agent loop: model -> tool -> model -> final answer
# 完整 Agent 循环：模型 -> 工具 -> 模型 -> 最终回答
# ============================================================
def ask_agent(user_input: str, max_steps: int = 5):
"""
EN:
This is the full Tool Calling loop.
The loop does this:
1. Send user input to the model.
2. Check whether model returned function_call items.
3. If yes, execute those tools.
4. Send function_call_output back to the model.
5. Repeat until model returns no function_call.
6. Return response.output_text as final answer.
CN:
这是完整的 Tool Calling 循环。
这个循环做这些事：
1. 把用户输入发给模型。
2. 检查模型是否返回 function_call。
3. 如果有，就执行这些工具。
4. 把 function_call_output 发回模型。
5. 重复，直到模型不再返回 function_call。
6. 返回 response.output_text 作为最终答案。
"""
# EN:
# First round input is the user's original question.
#
# CN:
# 第一轮 input 是用户原始问题。
next_input = user_input
# EN:
# previous_response_id connects multiple rounds into one conversation chain.
#
# CN:
# previous_response_id 用来把多轮调用接成同一个上下文链条。
previous_response_id = None
# EN:
# step counts how many model-tool rounds we have done.
#
# CN:
# step 记录已经进行了多少轮 “模型-工具” 循环。
step = 0
while True:
step += 1
# EN:
# Safety brake. Prevent accidental infinite loop.
#
# CN:
# 安全刹车。防止意外无限循环。
if step > max_steps:
return "Tool loop stopped: too many steps."
# EN:
# Send current input to the model.
# In first round, next_input is user text.
# In later rounds, next_input is tool_outputs.
#
# CN:
# 把当前 input 发给模型。
# 第一轮 next_input 是用户文本。
# 后续轮 next_input 是工具执行结果 tool_outputs。
response = client.responses.create(
model="gpt-4.1-mini",
input=next_input,
tools=tools,
previous_response_id=previous_response_id
)
# EN:
# Save response.id so the next API call continues from this response.
#
# CN:
# 保存 response.id，下一次 API 调用就能接着这个 response 继续。
previous_response_id = response.id
# EN:
# This list stores all tool results from this round.
# If the model requests no tools, it remains empty.
#
# CN:
# 这个列表保存本轮所有工具执行结果。
# 如果模型没有请求工具，它就保持空。
tool_outputs = []
# EN:
# response.output may contain different item types:
# - message
# - function_call
# - reasoning
# - other API-supported output items
#
# CN:
# response.output 里可能有不同类型的 item：
# - message 普通消息
# - function_call 工具调用请求
# - reasoning 推理相关内容
# - 其他 API 支持的输出项
for item in response.output:
# EN:
# We only execute tools for function_call items.
# If item is message/reasoning/etc., skip it.
#
# CN:
# 我们只对 function_call 类型执行工具。
# 如果是 message/reasoning 等，就跳过。
if item.type != "function_call":
continue
# EN:
# Execute the requested real tool.
#
# CN:
# 执行模型请求的真实工具。
result = execute_tool_call(item)
# EN:
# Build the tool result object required by Responses API.
# call_id must match the original function_call's call_id.
#
# CN:
# 构造 Responses API 要求的工具结果对象。
# call_id 必须和原始 function_call 的 call_id 对上。
tool_outputs.append({
"type": "function_call_output",
# EN: Links this result to the exact function_call.
# CN: 把这个结果绑定到对应的那次 function_call。
"call_id": item.call_id,
# EN:
# output should be a string, so we JSON-encode the result dict.
#
# CN:
# output 应该是字符串，所以把结果字典转成 JSON 字符串。
"output": json.dumps(result, ensure_ascii=False)
})
# EN:
# If no tool outputs were created, the model did not request tools.
# That means response.output_text is the final normal answer.
#
# CN:
# 如果没有产生 tool_outputs，说明模型没有请求工具。
# 这表示 response.output_text 就是最终普通回答。
if not tool_outputs:
return response.output_text
# EN:
# If tool_outputs is not empty, send tool results back to model
# in the next loop iteration.
#
# CN:
# 如果 tool_outputs 不为空，下一轮把工具结果发回模型。
next_input = tool_outputs
# ============================================================
# 6. Run demo
# 运行示例
# ============================================================
if __name__ == "__main__":
user_question = """
请回答：
1. 现在 New York 几点了？
2. Apple 当前股价多少？
3. 明天 New York 下雨吗？
"""
final_answer = ask_agent(user_question)
print(final_answer)A multi-turn tool loop is the repeated process where an LLM decides it needs a tool, your application executes that tool, sends the result back to the model, and the model either answers or asks for another tool call. This loop continues until the model can produce the final answer.
多轮 Tool 循环，就是模型反复经历这个过程：模型判断需要调用工具，你的程序执行真实函数，把结果写回对话，再让模型继续思考。模型可能最终回答，也可能继续请求下一个工具，直到任务完成。
Earlier, we discussed “Tool execution and result return”. That means: after the model asks for a tool, your program runs the real function and sends the result back to the model. AT here “Multi-turn Tool Loop” describes the full Tool Call loop, including before and after tool execution.
Difference
| Concept | EN | CN |
|---|---|---|
| Tool execution and result return | One key step: execute the real function and return the result to the model. | 一个关键步骤：执行真实函数，并把结果返回给模型。 |
| Full Tool Call Loop | The whole Agent cycle: model decides tool call → app executes tool → app returns result → model continues → maybe calls more tools → final answer. | 完整 Agent 循环：模型判断是否调用工具 → 程序执行工具 → 程序返回结果 → 模型继续思考 → 可能继续调用工具 → 最终回答。 |
Example,
EN: “What is the weather in New York today, and should I bring an umbrella?”
用户问：“今天纽约天气怎么样？我要不要带伞？”
The full loop is:
- Model decides: “I need weather data.” ：模型判断：“我需要天气数据。”
- Model outputs a tool call, for example
get_weather({ city: "New York" }).
模型输出工具调用，比如get_weather({ city: "New York" })。 - Your program executes the real weather function/API. 你的程序执行真实的天气函数/API。
- Your program writes the tool result back to the conversation. 你的程序把工具结果写回对话。
- Model reads the result and answers: “It may rain, bring an umbrella.” 模型读取结果后回答：“可能下雨，建议带伞。”
Step 3 and 4 are “Tool execution and result return.”, Steps 1 to 5 together are the “complete Tool Call loop.”
Full Tool Call Loop
完整 Tool Call 循环
= Model decides tool is needed
模型判断需要工具
+ Tool execution
工具执行
+ Tool result return
工具结果返回
+ Model continues reasoning
模型继续推理
+ Final answer or next tool call
最终回答或继续调用工具Imagine the model is a project manager. It cannot directly open your database, check the weather, or send an email. It can only say: “Please ask the database this.” Your program is the assistant who actually does it, brings back the result, and the manager decides the next step.
你可以把模型想成一个项目经理。它不能亲自查数据库、看天气、发邮件，只能说：“帮我查一下这个。”你的代码就像助理，真的去查，然后把结果交回来。经理看完结果，再决定继续查别的，还是直接回答用户。
One more example, ““Find the cheapest flight to Tokyo next Friday, check my calendar, and suggest a time to book.”
The model may need several tool turns: Produce final answer.
Call search_flights.
Call get_calendar.
Call compare_prices.
Call create_reminder.
Without a loop, the model gets only one chance. With a loop, the model can investigate step by step.
没有循环，模型只有一次机会。有了循环，模型可以一步一步调查。
Core Components
Previously, we knew a tool definition tells the model what tools exist, what each tool does, and what arguments are required. Usually this is described with JSON Schema.
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a city.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name, such as Beijing or New York."
}
},
"required": ["city"]
}
}
}The model does not execute the function. It only returns a structured request like: “Call get_weather with city = Beijing.”; Your application receives the tool call, validates the arguments, runs the actual function, catches errors, and gets a result.
def get_weather(city: str):
return {
"city": city,
"temperature": 22,
"unit": "C",
"condition": "Sunny"
}Repeat Until Final Answer / 重复直到最终回答
The model may request more tools after seeing the first result. Your app keeps looping until the response contains no tool calls and has a final answer.
Code Implementation
Below is a complete Python example using an OpenAI-style Chat Completions tool loop. DeepSeek and Azure OpenAI are very similar because they support OpenAI-compatible tool formats. Claude uses different message block names, but the loop idea is the same.
import json
import ast
import operator
from openai import OpenAI
client = OpenAI()
# ============================================================
# 1. Real backend functions
# 1. 真实后端函数
# ============================================================
def get_weather(city_name: str):
"""
EN: This is the real function executed by your program.
CN：这是你的程序真正执行的函数。
"""
# In production, call a real weather API here.
# 生产环境里，这里会调用真实天气 API。
return {
"cityName": city_name,
"temperature": "22°C",
"condition": "Rainy",
"umbrellaNeeded": True
}
def calculate(expression: str):
"""
EN: Safer demo calculator. Avoid using eval in production.
CN：更安全的演示计算器。生产环境不要直接用 eval。
"""
allowed_operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
}
def eval_node(node):
if isinstance(node, ast.Expression):
return eval_node(node.body)
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
if isinstance(node, ast.BinOp) and type(node.op) in allowed_operators:
left = eval_node(node.left)
right = eval_node(node.right)
return allowed_operators[type(node.op)](left, right)
raise ValueError("Unsupported expression")
tree = ast.parse(expression, mode="eval")
return {
"expression": expression,
"result": eval_node(tree)
}
def search_city(query: str):
"""
EN: Example tool: convert vague location into exact city.
CN：示例工具：把模糊地点转换成明确城市。
"""
if "capital of the United States" in query.lower():
return {"cityName": "Washington, D.C."}
if query.lower() in ["ny", "nyc"]:
return {"cityName": "New York"}
return {"cityName": query}
# ============================================================
# 2. Tool definitions shown to the model
# 2. 给模型看的 Tool 定义
# ============================================================
ALL_TOOLS = [
{
"type": "function",
"name": "weather",
"description": "Check current weather state for a city.",
"parameters": {
"type": "object",
"properties": {
"cityName": {
"type": "string",
"description": "City name, for example New York"
}
},
"required": ["cityName"],
"additionalProperties": False
}
},
{
"type": "function",
"name": "calculator",
"description": "Calculate a simple math expression.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Math expression, for example 18 * 25"
}
},
"required": ["expression"],
"additionalProperties": False
}
},
{
"type": "function",
"name": "search_city",
"description": "Find the exact city name from a vague location query.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Location query, for example capital of the United States"
}
},
"required": ["query"],
"additionalProperties": False
}
}
]
# ============================================================
# 3. Tool registry: tool name -> real backend function
# 3. 工具注册表：tool 名字 -> 真实后端函数
# ============================================================
TOOL_REGISTRY = {
"weather": get_weather,
"calculator": calculate,
"search_city": search_city
}
# ============================================================
# 4. Optional: select relevant tools from many tools
# 4. 可选：从很多工具中筛选本次相关工具
# ============================================================
def select_relevant_tools(user_input: str):
"""
EN: In real systems, you may have 1000 or 2000 tools.
You usually do not send all tools to the model.
You first select a smaller relevant subset.
CN：真实系统里，你可能有 1000 或 2000 个工具。
通常不会全部发给模型。
你会先筛选出本次相关的一小部分工具。
"""
text = user_input.lower()
selected = []
if "weather" in text or "天气" in text:
selected.append("weather")
if "capital" in text or "首都" in text or "ny" in text:
selected.append("search_city")
selected.append("weather")
if "*" in text or "+" in text or "calculate" in text or "计算" in text:
selected.append("calculator")
selected = list(dict.fromkeys(selected))
return [
tool for tool in ALL_TOOLS
if tool["name"] in selected
]
# ============================================================
# 5. Dispatch one function_call
# 5. 执行一个 function_call
# ============================================================
def execute_tool_call(tool_call):
"""
EN: This is where your program receives model's function_call
and executes the real backend function.
CN：这里就是你的程序收到模型的 function_call，
然后执行真实后端函数。
"""
# This is where the model's selected tool name appears.
# 这里体现模型选择了哪个工具。
tool_name = tool_call.name
# This is where the model's generated arguments appear.
# 这里体现模型生成的参数。
arguments = json.loads(tool_call.arguments)
if tool_name not in TOOL_REGISTRY:
return {
"error": f"Unknown tool: {tool_name}"
}
try:
real_function = TOOL_REGISTRY[tool_name]
# This is where your backend function is really executed.
# 这里才是真正执行你的后端函数。
result = real_function(**arguments)
return result
except Exception as error:
return {
"error": str(error),
"toolName": tool_name,
"arguments": arguments
}
# ============================================================
# 6. Multi-round Tool Calling loop
# 6. 多轮 Tool Calling 循环
# ============================================================
def run_agent(user_input: str):
"""
EN: This is the full Agent loop.
CN：这是完整 Agent 循环。
"""
tools = select_relevant_tools(user_input)
# -----------------------------
# First model call
# 第一次调用模型
# -----------------------------
response = client.responses.create(
model="gpt-5.5", # Replace with a model available to your account.
input=user_input,
# EN: This is where tools are sent to the model.
# CN：这里就是把 tools 传给模型的地方。
tools=tools,
# EN: auto means the model may answer directly or call tools.
# CN：auto 表示模型可以直接回答，也可以选择调用工具。
tool_choice="auto"
)
# -----------------------------
# Multi-round loop
# 多轮循环
# -----------------------------
while True:
function_calls = [
item for item in response.output
if item.type == "function_call"
]
# EN: If model returns no function_call, it is done.
# CN：如果模型没有返回 function_call，说明它已经完成最终回答。
if not function_calls:
return response.output_text
tool_outputs = []
for tool_call in function_calls:
# Example visible model decision:
# 示例：这里能看到模型的工具选择结果：
#
# tool_call.type == "function_call"
# tool_call.name == "weather"
# tool_call.arguments == "{\"cityName\":\"New York\"}"
tool_result = execute_tool_call(tool_call)
# EN: This is where tool result is written back.
# CN：这里就是把工具执行结果写回模型。
tool_outputs.append({
"type": "function_call_output",
"call_id": tool_call.call_id,
"output": json.dumps(tool_result)
})
# EN: Send tool outputs back to model.
# The model may now answer, or request another tool.
#
# CN：把工具结果发回模型。
# 模型可能现在回答，也可能继续请求下一个工具。
response = client.responses.create(
model="gpt-5.5",
input=tool_outputs,
previous_response_id=response.id,
tools=tools,
tool_choice="auto"
)
# ============================================================
# 7. Run examples
# 7. 运行示例
# ============================================================
if __name__ == "__main__":
answer = run_agent(
"What's the weather in the capital of the United States? Also calculate 18 * 25."
)
print(answer)Key Locations in this code:
Multiple available tools are here：多个可用工具在这里：
ALL_TOOLS = [...]The selected tools are sent to model here：筛选后的 tools 在这里发给模型：
response = client.responses.create(
input=user_input,
tools=tools,
tool_choice="auto"
)The model’s returned tool calls are read here：模型返回的工具调用在这里读取：
function_calls = [
item for item in response.output
if item.type == "function_call"
]The model’s chosen tool name and arguments are here：模型选择的工具名和参数在这里：
tool_name = tool_call.name
arguments = json.loads(tool_call.arguments)Your real backend function runs here：你的真实后端函数在这里执行：
real_function = TOOL_REGISTRY[tool_name]
result = real_function(**arguments)Tool result is written back here：工具结果在这里写回模型：
{
"type": "function_call_output",
"call_id": tool_call.call_id,
"output": json.dumps(tool_result)
}Multi-round Tool loop is here：多轮 Tool 循环在这里：
while True:
...
response = client.responses.create(...)Different between ”Execution & result return“ and “Multi-turn tool loop”
Earlier, we discussed ”Execution & result return“. It is one step. “Multi-round Tool loop” is the control structure that repeats that step until the model is done.
“Tool execution and result return” , model asked for one tool → your program executes it → returns result to model.
# executes the real backend function.
tool_result = execute_tool_call(tool_call)
# packages the result so it can be sent back to the model.把执行结果包装成模型能接收的格式。
tool_outputs.append({
"type": "function_call_output",
"call_id": tool_call.call_id,
"output": json.dumps(tool_result)
})
# function_call → real function → function_call_output
# 模型请求工具 → 执行真实函数 → 返回工具结果What is ReAct?
In AI and LLM engineering, ReAct stands for Reasoning + Acting. It is a prompting and design framework introduced by Yao et al. (2022) that allows Large Language Models to solve complex, multi-step tasks by interleaving chain-of-thought reasoning traces with tool execution actions in a closed-loop iterative process.
ReAct代表推理+行动（Reasoning + Acting）。它是Yao等人（2022）提出的一种提示和设计框架，允许大型语言模型通过将链式推理轨迹与工具执行行动在闭环迭代过程中交错进行，来解决复杂的多步骤任务。
The core innovation is the synergistic combination of:
- Reasoning: Step-by-step thinking that helps the model plan, track progress, and handle exceptions
逐步思考，帮助模型规划、跟踪进度和处理异常 - Acting: Executing external tools (APIs, databases, calculators, search engines) to gather real-world information
执行外部工具（API、数据库、计算器、搜索引擎）以获取真实世界信息
This creates a feedback loop where each observation from an action informs the next reasoning step, enabling the agent to dynamically adapt its plan based on new information.
这形成了一个反馈循环，每个行动的观察结果都会影响下一步的推理，使Agent能够根据新信息动态调整计划。
Before ReAct, LLMs had two separate paradigms:
- Pure Reasoning (CoT/Chain-of-Thought): Good for logic puzzles, but cannot access external data
擅长逻辑谜题，但无法访问外部数据 - Pure Action (Tool Calling): Can execute functions, but lacks planning and error recovery
可以执行函数，但缺乏规划和错误恢复
ReAct combines both into a single, unified loop that enables agents to:
- Break down complex problems into manageable steps
将复杂问题分解为可管理的步骤 - Gather information as needed
根据需要收集信息 - Revise plans based on observations
根据观察修订计划 - Explain their decision-making process
解释其决策过程
e.g.
User asks: “Find my latest unpaid invoice and draft a reminder email.”
A ReAct agent may do: call invoice search tool, inspect result, call customer profile tool, draft email, maybe ask for approval before sending.
Without ReAct, the model may hallucinate missing facts. With ReAct, the model can retrieve live data, execute real actions, and correct itself from tool results.
Core Components
A complete ReAct implementation consists of 6 core components:
| # | Component 组件 | Purpose 目的 |
|---|---|---|
| 1 | Prompt Template 提示词模板 | Instructs the LLM to output in structured format (Thought/Action/Observation) |
| 2 | Response Parser 响应解析器 | Extracts structured fields from LLM output using regex or specialized parsing 使用正则表达式或专门解析从LLM输出中提取结构化字段 |
| 3 | Tool Registry 工具注册表 | Centralized catalog of tools with their schemas and execution functions 集中式工具目录，包含其schema和执行函数 |
| 4 | Tool Executor 工具执行器 | Safely executes tools with validated arguments 使用验证后的参数安全执行工具 |
| 5 | State/Memory Manager 状态/记忆管理器 | Maintains conversation history, reasoning traces, and observations 维护对话历史、推理轨迹和观察结果 |
| 6 | Loop Controller 循环控制器 | Orchestrates iterations, applies max iteration limits, and detects termination 编排迭代次数、应用最大迭代限制并检测终止 |
Detailed Breakdown of Each Component
1. Prompt Template
This is the most critical component. It must explicitly instruct the LLM to:
- Use a specific format:
Thought: ...,Action: ...,Action Input: ...,Final Answer: ... - Explain what each field means
- List all available tools with their descriptions and parameter schemas
- Provide instructions on when to stop (when the final answer is ready)
2. Response Parser
This component uses regular expressions or structured parsing to extract:
# Critical: The parser must handle variations in LLM output
# Example of a robust parser:
thought_match = re.search(r"Thought:\s*(.*?)(?=Action:|Final Answer:|$)", response, re.DOTALL | re.IGNORECASE)
action_match = re.search(r"Action:\s*(\w+)", response, re.IGNORECASE)
action_input_match = re.search(r"Action Input:\s*(\{.*?\})", response, re.DOTALL | re.IGNORECASE)
final_answer_match = re.search(r"Final Answer:\s*(.*?)$", response, re.DOTALL | re.IGNORECASE)3. Tool Registry
Centralized management of tools. Each tool has:
{
"name": "query_database", # Unique identifier
"description": "Execute SQL query", # For LLM to understand / 供LLM理解
"parameters": { # JSON Schema for validation / 用于验证的JSON Schema
"type": "object",
"properties": {
"query": {"type": "string", "description": "SQL query"}
},
"required": ["query"]
},
"func": actual_python_function # The executable code / 可执行代码
}4. Tool Executor
Safely executes the tool with the provided arguments.
def execute(self, name: str, arguments: Dict[str, Any]) -> str:
if name not in self.tools:
return f"Error: Tool '{name}' not found"
try:
# Validate arguments against schema
# 根据schema验证参数
result = self.tools[name]["func"](**arguments)
return str(result) # Always return string for Observation
except Exception as e:
return f"Error executing {name}: {str(e)}" # Return error as observation6. Loop Controller
Orchestrates the entire ReAct loop:
def run(self, user_input: str) -> str:
self.iterations = 0
while self.iterations < self.max_iterations:
self.iterations += 1
# 1. Build prompt with history / 使用历史构建提示词
prompt = self._build_prompt(user_input)
# 2. Call LLM / 调用LLM
response = client.chat.completions.create(...)
# 3. Parse response / 解析响应
parsed = self._parse_response(response)
# 4. Check if final answer / 检查是否最终答案
if parsed["final_answer"]:
return parsed["final_answer"]
# 5. Execute action / 执行行动
if parsed["action"]:
observation = self.registry.execute(parsed["action"], parsed["action_input"])
self.messages.append({"role": "observation", "content": observation})
# 6. Continue loop / 继续循环
return "Max iterations reached"Complete Code Implementation
I’ll provide the complete code here with extensive Chinese and English comments. Please note that this is a self-contained implementation that you can copy and run.
#!/usr/bin/env python3
"""
A15: Complete ReAct Implementation with Full Loop Controller
This shows the Loop Controller as a separate concept with:
- Orchestration logic (the run method)
- Termination conditions
- State management
- Error handling at scale
- Flow control
这展示了循环控制器作为一个独立概念，包含：
- 编排逻辑（run方法）
- 终止条件
- 状态管理
- 大规模错误处理
- 流程控制
"""
import os
import json
import re
import logging
import time
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
logger = logging.getLogger(__name__)
# ============================================================================
# SECTION 1: STATE DEFINITIONS / 状态定义
# ============================================================================
class AgentStatus(Enum):
"""Agent execution status / Agent执行状态"""
RUNNING = "running"
COMPLETED = "completed"
MAX_ITERATIONS = "max_iterations"
ERROR = "error"
TOOL_FAILURE = "tool_failure"
@dataclass
class AgentState:
"""
Complete state of the agent at any point in the loop.
Agent在循环中任意点的完整状态。
This is the "memory" of the agent.
这是Agent的"记忆"。
"""
# Conversation history / 对话历史
messages: List[Dict[str, str]] = field(default_factory=list)
# Current iteration / 当前迭代次数
iteration: int = 0
# Status of the agent / Agent状态
status: AgentStatus = AgentStatus.RUNNING
# Accumulated reasoning traces for debugging / 用于调试的累积推理轨迹
reasoning_traces: List[str] = field(default_factory=list)
# Tool execution history / 工具执行历史
tool_history: List[Dict[str, Any]] = field(default_factory=list)
# Start time for timeout management / 用于超时管理的开始时间
start_time: float = field(default_factory=time.time)
def add_message(self, role: str, content: str) -> None:
"""Add a message to the conversation history."""
self.messages.append({"role": role, "content": content})
def add_reasoning_trace(self, thought: str) -> None:
"""Add a reasoning trace for debugging."""
self.reasoning_traces.append(f"Iteration {self.iteration}: {thought}")
def add_tool_execution(self, tool_name: str, arguments: Dict, result: str) -> None:
"""Record tool execution for auditing."""
self.tool_history.append({
"iteration": self.iteration,
"timestamp": datetime.now().isoformat(),
"tool": tool_name,
"arguments": arguments,
"result_preview": result[:200] # Truncate for storage
})
# ============================================================================
# SECTION 2: LOOP CONTROLLER (THE FULL CONCEPT) / 循环控制器（完整概念）
# ============================================================================
class LoopController:
"""
The Loop Controller - orchestrates the entire ReAct loop.
循环控制器 - 编排整个ReAct循环。
This is a SEPARATE CONCEPT from the run() method.
The Loop Controller encompasses:
1. Configuration (max iterations, timeouts, retry policies)
2. State management (AgentState)
3. Orchestration logic (run() method)
4. Termination detection (should_stop())
5. Error handling strategies (handle_error())
6. Flow control (conditional branching)
这是一个与run()方法不同的独立概念。
循环控制器包含：
1. 配置（最大迭代次数、超时、重试策略）
2. 状态管理（AgentState）
3. 编排逻辑（run()方法）
4. 终止检测（should_stop()）
5. 错误处理策略（handle_error()）
6. 流程控制（条件分支）
In production systems, this would be extended with:
- Circuit breakers / 断路器
- Retry with exponential backoff / 指数退避重试
- Parallel execution of sub-tasks / 并行执行子任务
- Human-in-the-loop checkpoints / 人机回圈检查点
"""
def __init__(
self,
registry: ToolRegistry,
model: str = "gpt-4o-mini",
max_iterations: int = 10,
max_timeout_seconds: int = 120,
temperature: float = 0.0,
enable_retry: bool = True,
max_retries: int = 3
):
"""
Initialize the Loop Controller with configuration.
使用配置初始化循环控制器。
Args / 参数:
registry: Tool registry with all available tools / 包含所有可用工具的工具注册表
model: LLM model to use / 使用的LLM模型
max_iterations: Maximum loop iterations / 最大循环迭代次数
max_timeout_seconds: Maximum execution time / 最大执行时间（秒）
temperature: LLM temperature for deterministic responses / LLM温度参数
enable_retry: Whether to retry on transient errors / 是否在瞬态错误时重试
max_retries: Maximum retry attempts / 最大重试次数
"""
self.registry = registry
self.model = model
self.max_iterations = max_iterations
self.max_timeout_seconds = max_timeout_seconds
self.temperature = temperature
self.enable_retry = enable_retry
self.max_retries = max_retries
# Initialize the agent state / 初始化Agent状态
self.state = AgentState()
# ReAct prompt template / ReAct提示词模板
self.react_prompt = """
You are an AI assistant using the ReAct framework (Reasoning + Acting).
You have access to the following tools:
{tool_descriptions}
IMPORTANT INSTRUCTIONS:
1. You MUST respond in EXACTLY this format, using these labels:
Thought: [Your reasoning about what to do next]
Action: [Tool name - MUST be from the list above]
Action Input: [JSON arguments for the tool]
OR
Thought: [Your reasoning]
Final Answer: [Your complete answer to the user]
2. You can use multiple tool calls across multiple iterations.
3. After each tool call, you will receive an Observation.
4. Use the Observation to inform your next Thought.
5. When you have enough information to answer, provide Final Answer.
Previous conversation:
{history}
Current user question: {user_input}
Now respond in the required format:
"""
logger.info("LoopController initialized")
logger.info(f"Configuration: max_iterations={max_iterations}, timeout={max_timeout_seconds}s")
# ========================================================================
# 1. ORCHESTRATION LOGIC (THE RUN METHOD) / 编排逻辑（run方法）
# ========================================================================
def run(self, user_input: str) -> str:
"""
Execute the complete ReAct loop.
执行完整的ReAct循环。
This is the main orchestration method that:
1. Initializes the state
2. Executes the loop
3. Checks termination conditions
4. Handles errors
5. Returns the final answer
这是主要的编排方法，它：
1. 初始化状态
2. 执行循环
3. 检查终止条件
4. 处理错误
5. 返回最终答案
"""
# Reset state for this run / 为这次运行重置状态
self.state = AgentState()
self.state.start_time = time.time()
self.state.add_message("user", user_input)
logger.info(f"Starting ReAct loop for query: '{user_input[:100]}...'")
# ====================================================================
# CORE LOOP - The actual iteration / 核心循环 - 实际迭代
# ====================================================================
while self.state.iteration < self.max_iterations:
self.state.iteration += 1
current_iter = self.state.iteration
logger.info(f"Iteration {current_iter}/{self.max_iterations}")
# -----------------------------------------------------------------
# Step 1: Check termination conditions / 步骤1：检查终止条件
# -----------------------------------------------------------------
if self._should_stop():
logger.info("Termination condition met. Stopping.")
break
# -----------------------------------------------------------------
# Step 2: Build prompt and call LLM / 步骤2：构建提示词并调用LLM
# -----------------------------------------------------------------
try:
response = self._call_llm_with_retry(user_input)
except Exception as e:
# Critical failure in LLM call / LLM调用中的关键失败
return self._handle_critical_error(str(e))
# -----------------------------------------------------------------
# Step 3: Parse LLM response / 步骤3：解析LLM响应
# -----------------------------------------------------------------
parsed = self._parse_response(response)
if parsed is None:
# Malformed response - can't proceed / 格式错误的响应 - 无法继续
self.state.add_message("assistant", "I apologize, but I need to reformat my response.")
continue
# -----------------------------------------------------------------
# Step 4: Check for Final Answer / 步骤4：检查最终答案
# -----------------------------------------------------------------
if parsed.get("final_answer"):
self.state.status = AgentStatus.COMPLETED
logger.info(f"Final answer reached after {current_iter} iterations")
return parsed["final_answer"]
# -----------------------------------------------------------------
# Step 5: Execute action if present / 步骤5：如果有行动则执行
# -----------------------------------------------------------------
if parsed.get("action"):
action = parsed["action"]
action_input = parsed.get("action_input", {})
# Validate the action / 验证行动
if not self._validate_action(action, action_input):
# Invalid action - let the agent recover / 无效行动 - 让Agent恢复
self.state.add_message(
"assistant",
f"Thought: {parsed.get('thought', '')}\nAction: {action}\nAction Input: {json.dumps(action_input)}"
)
self.state.add_message(
"observation",
f"ERROR: Tool '{action}' not found or invalid arguments. Available: {list(self.registry._tools.keys())}"
)
continue
# Execute the tool / 执行工具
observation = self.registry.execute(action, action_input)
# Record the execution / 记录执行
self.state.add_tool_execution(action, action_input, observation)
# Store in conversation history / 存储在对话历史中
self.state.add_message(
"assistant",
f"Thought: {parsed.get('thought', '')}\nAction: {action}\nAction Input: {json.dumps(action_input)}"
)
self.state.add_message("observation", observation)
# Store reasoning trace / 存储推理轨迹
if parsed.get("thought"):
self.state.add_reasoning_trace(parsed["thought"])
else:
# No action and no final answer - malformed / 没有行动也没有最终答案 - 格式错误
logger.warning(f"No action or final answer found in response. Parsed: {parsed}")
self.state.add_message(
"assistant",
f"Thought: {parsed.get('thought', 'I need to think about this.')}\n"
"I need to take an action or provide a final answer."
)
# ====================================================================
# Loop ended without final answer / 循环结束但没有最终答案
# ====================================================================
self.state.status = AgentStatus.MAX_ITERATIONS
return f"Maximum iterations ({self.max_iterations}) reached. Please refine your query."
# ========================================================================
# 2. TERMINATION DETECTION / 终止检测
# ========================================================================
def _should_stop(self) -> bool:
"""
Check all termination conditions.
检查所有终止条件。
This is a KEY part of the Loop Controller concept.
It centralizes all stop conditions in one place.
这是循环控制器概念的关键部分。
它将所有停止条件集中在一个地方。
Conditions checked / 检查的条件:
1. Max iterations reached / 达到最大迭代次数
2. Timeout exceeded / 超时
3. Tool failure threshold exceeded / 超过工具失败阈值
4. Status already set to completed / 状态已设置为完成
"""
# Condition 1: Max iterations / 条件1：最大迭代次数
if self.state.iteration >= self.max_iterations:
logger.info(f"Max iterations reached: {self.state.iteration}/{self.max_iterations}")
return True
# Condition 2: Timeout / 条件2：超时
elapsed = time.time() - self.state.start_time
if elapsed > self.max_timeout_seconds:
logger.info(f"Timeout reached: {elapsed:.1f}s > {self.max_timeout_seconds}s")
return True
# Condition 3: Tool failure threshold / 条件3：工具失败阈值
tool_failures = sum(
1 for entry in self.state.tool_history
if entry.get("result_preview", "").startswith("ERROR")
)
if tool_failures >= 3:
logger.info(f"Too many tool failures: {tool_failures}")
return True
# Condition 4: Status check / 条件4：状态检查
if self.state.status in [AgentStatus.COMPLETED, AgentStatus.ERROR]:
return True
return False
# ========================================================================
# 3. LLM CALL WITH RETRY LOGIC / 带重试逻辑的LLM调用
# ========================================================================
def _call_llm_with_retry(self, user_input: str) -> str:
"""
Call the LLM with retry logic for transient errors.
使用重试逻辑调用LLM以处理瞬态错误。
This is part of the Loop Controller's error handling strategy.
这是循环控制器错误处理策略的一部分。
"""
retry_count = 0
last_error = None
while retry_count <= self.max_retries:
try:
# Build the prompt with current history
# 使用当前历史构建提示词
prompt = self._build_prompt(user_input)
# Call LLM / 调用LLM
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a ReAct agent. Follow the format strictly."},
{"role": "user", "content": prompt}
],
temperature=self.temperature,
max_tokens=1024
)
return response.choices[0].message.content
except Exception as e:
last_error = e
retry_count += 1
if self.enable_retry and retry_count <= self.max_retries:
wait_time = 2 ** retry_count # Exponential backoff / 指数退避
logger.warning(f"LLM call failed (attempt {retry_count}), retrying in {wait_time}s: {e}")
time.sleep(wait_time)
else:
raise
raise last_error
# ========================================================================
# 4. RESPONSE PARSING / 响应解析
# ========================================================================
def _parse_response(self, response: str) -> Optional[Dict[str, Any]]:
"""
Parse the LLM response into structured fields.
将LLM响应解析为结构化字段。
Returns / 返回:
Dict with keys: thought, action, action_input, final_answer
OR None if parsing fails completely
"""
parsed = {
"thought": None,
"action": None,
"action_input": None,
"final_answer": None
}
# Extract Thought / 提取思考
thought_match = re.search(
r"Thought:\s*(.*?)(?=Action:|Final Answer:|$)",
response,
re.DOTALL | re.IGNORECASE
)
if thought_match:
parsed["thought"] = thought_match.group(1).strip()
# Extract Final Answer / 提取最终答案
final_match = re.search(
r"Final Answer:\s*(.*?)$",
response,
re.DOTALL | re.IGNORECASE
)
if final_match:
parsed["final_answer"] = final_match.group(1).strip()
# If we have final answer, no need to parse action
# 如果有最终答案，不需要解析行动
return parsed
# Extract Action / 提取行动
action_match = re.search(r"Action:\s*(\w+)", response, re.IGNORECASE)
if action_match:
parsed["action"] = action_match.group(1).strip()
# Extract Action Input (JSON) / 提取行动输入（JSON）
# Try multiple patterns to handle variations
# 尝试多种模式以处理变体
patterns = [
r"Action Input:\s*(\{.*?\})", # Standard JSON / 标准JSON
r"Action Input:\s*(\[.*?\])", # Array / 数组
r"Action Input:\s*(.*?)(?=Thought:|$)", # Fallback / 备用
]
for pattern in patterns:
match = re.search(pattern, response, re.DOTALL | re.IGNORECASE)
if match:
raw_input = match.group(1).strip()
try:
if raw_input.startswith('{'):
parsed["action_input"] = json.loads(raw_input)
elif raw_input.startswith('['):
parsed["action_input"] = json.loads(raw_input)
else:
# Try to parse as key-value pairs
# 尝试解析为键值对
parsed["action_input"] = {"raw": raw_input}
except json.JSONDecodeError:
parsed["action_input"] = {"raw": raw_input}
break
# Validate: we need both action and action_input to proceed
# 验证：我们需要同时有行动和行动输入才能继续
if parsed["action"] and parsed["action_input"] is None:
parsed["action_input"] = {}
return parsed
# ========================================================================
# 5. PROMPT BUILDING / 提示词构建
# ========================================================================
def _build_prompt(self, user_input: str) -> str:
"""
Build the full ReAct prompt with history.
使用历史构建完整的ReAct提示词。
"""
# Format history / 格式化历史
history_lines = []
for msg in self.state.messages:
role = msg["role"]
content = msg["content"]
if role == "user":
history_lines.append(f"User: {content}")
elif role == "assistant":
history_lines.append(f"Assistant: {content}")
elif role == "observation":
history_lines.append(f"Observation: {content}")
history_str = "\n".join(history_lines)
return self.react_prompt.format(
tool_descriptions=self.registry.get_tool_descriptions(),
history=history_str,
user_input=user_input
)
# ========================================================================
# 6. ACTION VALIDATION / 行动验证
# ========================================================================
def _validate_action(self, action: str, action_input: Dict[str, Any]) -> bool:
"""
Validate that the action is valid and has required arguments.
验证行动是否有效且具有所需参数。
This is a security measure.
这是一项安全措施。
"""
if action not in self.registry._tools:
logger.warning(f"Unknown tool: {action}")
return False
# Check required parameters / 检查必需参数
tool_info = self.registry._tools[action]
required = tool_info["parameters"].get("required", [])
for param in required:
if param not in action_input:
logger.warning(f"Missing required parameter '{param}' for tool '{action}'")
return False
return True
# ========================================================================
# 7. ERROR HANDLING / 错误处理
# ========================================================================
def _handle_critical_error(self, error_msg: str) -> str:
"""
Handle critical errors that cannot be recovered from.
处理无法恢复的关键错误。
"""
self.state.status = AgentStatus.ERROR
logger.error(f"Critical error: {error_msg}")
return f"I encountered a critical error: {error_msg}. Please try again."
# ========================================================================
# 8. FLOW CONTROL / 流程控制
# ========================================================================
def should_continue(self, parsed: Dict[str, Any]) -> bool:
"""
Flow control: decide if we should continue the loop.
流程控制：决定是否继续循环。
This allows for conditional branching based on the parsed response.
这允许基于解析响应的条件分支。
"""
if parsed.get("final_answer"):
return False
if parsed.get("action") is None:
return False
# Check if we have enough information (custom logic)
# 检查我们是否有足够的信息（自定义逻辑）
tool_count = len(self.state.tool_history)
if tool_count >= 5:
# After 5 tool calls, try to synthesize an answer
# 在5次工具调用后，尝试综合答案
return False
return True
# ========================================================================
# 9. STATE INSPECTION (for debugging) / 状态检查（用于调试）
# ========================================================================
def get_state_summary(self) -> Dict[str, Any]:
"""
Get a summary of the current agent state.
获取当前Agent状态的摘要。
Useful for debugging and observability.
对调试和可观测性很有用。
"""
return {
"iteration": self.state.iteration,
"status": self.state.status.value,
"total_messages": len(self.state.messages),
"tool_calls": len(self.state.tool_history),
"elapsed_seconds": time.time() - self.state.start_time,
"reasoning_traces": self.state.reasoning_traces[-3:], # Last 3
"last_tool": self.state.tool_history[-1] if self.state.tool_history else None
}
# ============================================================================
# SECTION 3: COMPLETE USAGE EXAMPLE / 完整使用示例
# ============================================================================
def main():
"""
Complete demonstration showing the Loop Controller in action.
完整的演示，展示循环控制器的实际运行。
"""
# Initialize tools / 初始化工具
registry = create_data_engineering_tools()
# Initialize the Loop Controller / 初始化循环控制器
controller = LoopController(
registry=registry,
model="gpt-4o-mini",
max_iterations=8,
max_timeout_seconds=60,
temperature=0.0,
enable_retry=True,
max_retries=3
)
# Example 1: Simple calculation / 示例1：简单计算
print("=" * 70)
print("EXAMPLE 1: Multi-step Calculation")
print("示例1：多步骤计算")
print("=" * 70)
result = controller.run("Calculate: (250 + 150) * 2, then subtract 100")
print(f"\nResult: {result}")
# Show state summary / 显示状态摘要
print(f"\nState Summary: {json.dumps(controller.get_state_summary(), indent=2)}")
print("\n" + "=" * 70)
print("EXAMPLE 2: Data Analysis")
print("示例2：数据分析")
print("=" * 70)
# Reset controller for new query / 重置控制器以处理新查询
controller = LoopController(
registry=registry,
model="gpt-4o-mini",
max_iterations=10,
max_timeout_seconds=60,
temperature=0.0
)
result = controller.run(
"Analyze sales data for Q1 and Q2. "
"Which quarter had better performance? "
"What was the growth percentage?"
)
print(f"\nResult: {result}")
print(f"\nState Summary: {json.dumps(controller.get_state_summary(), indent=2)}")
if __name__ == "__main__":
main()Code Index
| ReAct prompt template | self.react_prompt = """ |
| ReAct loop | def run(self, user_input: str) -> str: |
| Response Parser 解析器（提取Thought/Action） | def _parse_response |
| State/Memory Manager | @dataclass class AgentState |
| 终止检测 | def _should_stop |
| ToolRegistry | class ToolRegistry |
| component | included |
|---|---|
ToolRegistry 类 | ✅ |
AgentStatus 枚举 | ✅ |
AgentState 数据类 | ✅ |
LoopController 类 | ✅ |
__init__() 方法 | ✅ |
run() 方法 (ReAct主循环) | ✅ |
_should_stop() (终止检测) | ✅ |
_call_llm_with_retry() (LLM调用) | ✅ |
_parse_response() (ReAct解析器) | ✅ |
_build_prompt() (提示词构建) | ✅ |
_validate_action() (安全验证) | ✅ |
_handle_critical_error() (错误处理) | ✅ |
get_state_summary() (调试) | ✅ |
create_data_engineering_tools() | ✅ |
main() 演示函数 | ✅ |