外观
从无限循环到 8 步兜底：ReAct 重试避坑实录
问题背景：ReAct 不是“多调几次工具”那么简单
在基于 function-calling 的智能体中，ReAct 循环通常被简化为：模型思考、选择工具、执行工具、观察结果、继续思考。这个模式在 Demo 中非常顺畅，但一旦接入真实业务系统，失败路径会迅速变多：天气接口超时、数据库返回空、模型把订单号写成中文、工具权限不足、模型误把错误响应当成新线索继续调用。
如果只写一个 while True，让模型不断“推理—行动”，最常见的结果不是智能体变聪明，而是 token 账单上涨、工具被重复打爆、用户长时间等待，最后仍然没有答案。因此，ReAct 工程化的关键不只是提示词里写“请思考”，而是要显式设计失败重试、步数预算、终止条件和可观测状态。
一个稳定的 agent-workflow 至少要回答四个问题：哪些错误值得重试，重试几次；哪些错误应立即反馈给模型修正；达到最大步数后如何兜底；每一步留下哪些日志，便于复盘和评测。
方案对比：三种 ReAct 失败处理策略
第一种是“无限循环 + 模型自愈”。实现最简单，把工具异常直接塞回 observation，期待模型自己发现问题。它适合只读、低频、低成本的玩具场景，但在生产环境中风险很高，因为模型可能围绕同一错误反复规划。
第二种是“固定次数硬重试”。对所有工具异常都重试 3 次，简单粗暴，能覆盖偶发网络抖动，却无法处理参数错误。例如订单号格式非法时，重试 100 次也不会成功，反而放大下游压力。
第三种是“分类重试 + 步数预算 + 降级兜底”。它把工具失败分成可重试错误、可修正错误和不可恢复错误：网络抖动使用指数退避；参数错误把校验信息返回给模型；权限不足或预算耗尽立即终止。与此同时，整个 ReAct 循环拥有最大步数和最大工具调用次数，达到阈值后进入总结、澄清或人工转接。
实操步骤：实现可中断的 ReAct 循环
下面示例不依赖特定厂商 SDK，用一个可替换的 fake_llm 演示完整控制流。真实项目中可以把 fake_llm 替换为 OpenAI、Anthropic 或内部模型的 function-calling 接口，把 TOOLS 映射成实际 API、MCP 工具或数据库查询。
python
import time
import json
from dataclasses import dataclass, field
from typing import Callable, Any
# 可重试错误：通常是临时性问题，等待后可能恢复
class RetryableToolError(Exception):
pass
# 可修正错误：模型参数有问题，应把原因反馈给模型
class InvalidArgumentError(Exception):
pass
# 不可恢复错误：权限、越权、业务明确拒绝
class FatalToolError(Exception):
pass
@dataclass
class AgentConfig:
max_steps: int = 8 # ReAct 最大推理步数
max_tool_calls: int = 6 # 实际工具调用上限
max_retries: int = 2 # 同一个工具动作的临时失败重试次数
base_delay: float = 0.5 # 指数退避起始秒数
@dataclass
class RunState:
step: int = 0
tool_calls: int = 0
repeated_errors: dict = field(default_factory=dict)
messages: list = field(default_factory=list)
TOOLS = {
"query_order": {
"description": "根据订单号查询订单状态",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string", "pattern": "^ORD-[0-9]{6,}$"}
},
"required": ["order_id"]
}
}
}
def validate_and_call(name: str, arguments: dict[str, Any]) -> str:
if name not in TOOLS:
raise InvalidArgumentError(f"未知工具：{name}")
if name == "query_order":
order_id = arguments.get("order_id", "")
if not order_id.startswith("ORD-"):
raise InvalidArgumentError(
"order_id 必须形如 ORD-100001，不要使用中文或 # 号"
)
if order_id == "ORD-500000":
raise RetryableToolError("订单服务超时")
if order_id == "ORD-999999":
raise FatalToolError("无权访问该订单")
return json.dumps({"order_id": order_id, "status": "shipped"}, ensure_ascii=False)
raise InvalidArgumentError("工具未实现")
def call_tool_with_retry(name: str, arguments: dict, config: AgentConfig, state: RunState) -> str:
for attempt in range(config.max_retries + 1):
try:
state.tool_calls += 1
return validate_and_call(name, arguments)
except RetryableToolError as exc:
if attempt == config.max_retries:
return f"TOOL_RETRY_EXHAUSTED: {exc}。请换一种方案或告知用户稍后再试。"
# 指数退避并加入抖动，避免多个请求同时打爆下游
delay = config.base_delay * (2 ** attempt)
time.sleep(delay)
except InvalidArgumentError as exc:
# 参数错误不要盲目重试，直接让模型根据反馈修正
return f"INVALID_ARGUMENT: {exc}"
except FatalToolError as exc:
raise
def fake_llm(messages: list, tools: dict) -> dict:
# 测试用模型：先构造一次错误参数，再查询可恢复订单，最后回答
joined = json.dumps(messages, ensure_ascii=False)
if "INVALID_ARGUMENT" not in joined:
return {
"role": "assistant",
"tool_calls": [{
"name": "query_order",
"arguments": {"order_id": "订单100001"}
}]
}
if "shipped" not in joined:
return {
"role": "assistant",
"tool_calls": [{
"name": "query_order",
"arguments": {"order_id": "ORD-100001"}
}]
}
return {"role": "assistant", "content": "订单 ORD-100001 当前状态为已发货。"}
def run_react(user_input: str, config: AgentConfig | None = None) -> dict:
config = config or AgentConfig()
state = RunState(messages=[{"role": "user", "content": user_input}])
while state.step < config.max_steps:
state.step += 1
response = fake_llm(state.messages, TOOLS)
state.messages.append(response)
tool_calls = response.get("tool_calls")
if not tool_calls:
return {"status": "success", "answer": response.get("content", ""), "state": state}
if state.tool_calls + len(tool_calls) > config.max_tool_calls:
return fallback(state, "工具调用次数即将超过预算")
for call in tool_calls:
name = call["name"]
arguments = call.get("arguments", {})
try:
observation = call_tool_with_retry(name, arguments, config, state)
except FatalToolError as exc:
return fallback(state, f"工具拒绝执行：{exc}")
state.messages.append({
"role": "tool",
"name": name,
"content": observation
})
# 同一错误连续出现时及时止损，避免模型围绕错误空转
key = f"{name}:{observation}"
state.repeated_errors[key] = state.repeated_errors.get(key, 0) + 1
if state.repeated_errors[key] >= 2:
return fallback(state, f"重复失败：{observation}")
return fallback(state, "达到最大推理步数")
def fallback(state: RunState, reason: str) -> dict:
# 生产中可改为：向用户澄清、转人工、返回已收集的部分结果
return {
"status": "fallback",
"answer": f"暂时无法完成任务。原因：{reason}，请补充信息或稍后重试。",
"reason": reason,
"steps_used": state.step,
"tool_calls_used": state.tool_calls
}
if __name__ == "__main__":
result = run_react("帮我查一下订单 100001 的状态")
print(json.dumps(result, ensure_ascii=False, indent=2, default=str))这段代码的重点不是“捕获异常”本身，而是把异常转换成智能体可以理解的状态信号。INVALID_ARGUMENT 表示你可以改，TOOL_RETRY_EXHAUSTED 表示临时故障已耗尽预算，FatalToolError 则表示不要继续尝试。
提示词配合：让模型知道边界在哪里
仅有代码控制还不够，function-calling 的系统提示词也应明确工具使用规则。可以复用下面的模板：
text
你是一个订单助手。你可以使用工具，但必须遵守：
1. 调用 query_order 前，order_id 必须转换为 ORD-数字 格式。
2. 如果工具返回 INVALID_ARGUMENT，请阅读错误原因，修正参数后最多再试一次。
3. 如果工具返回 TOOL_RETRY_EXHAUSTED，不要重复调用同一工具，应说明服务暂时不可用或换用其他允许的方案。
4. 如果已有信息足以回答，就直接回答，不要为了“更确定”而重复查询。
5. 无法获得必要信息时，明确告诉用户缺什么，而不是编造订单状态。避坑总结：生产环境最容易踩的 5 个坑
第一，不要把所有异常都返回给模型无限重试。常见现象是日志里反复出现同一行 timeout，模型却不断生成相同参数。修正方法是区分临时错误与确定性错误，只对网络抖动、限流、5xx 使用指数退避；对 4xx 参数错误执行零重试并反馈修正建议。
第二，最大步数不能只限制 LLM 轮次，还要限制工具调用次数。有些模型一轮会并行发起多个 function-calling，如果只判断循环轮数，8 轮可能产生几十个外部请求。建议同时配置 max_steps、max_tool_calls、单工具调用上限和总超时时间。
第三，避免“错误观察”被模型误读为业务事实。例如工具返回“订单不存在”，模型可能回答“订单已取消”。错误消息应使用机器可识别前缀和稳定结构，如 {"error_type":"NOT_FOUND"}，并在提示词中要求不得把错误推断成不存在的状态。
第四，重试必须有幂等性保障。查询类工具可以安全重试，但下单、退款、发券等写操作不能因超时就盲目再次调用，否则可能重复扣款或重复发货。写操作应使用幂等键、请求 ID 或先查询事务状态，再决定是否补偿。
第五，兜底不等于返回一句生硬的失败。更好的做法是根据上下文分级：缺少参数时向用户澄清；外部服务超限时给出稍后重试建议；权限不足时引导申请权限；已有部分结果时展示已确认信息和未完成环节。同时记录步数、工具名、参数哈希、错误类型与最终终止原因，这些字段也是后续离线评测和提示词优化的核心数据。
结语
ReAct 的可靠性来自“有限自治”：模型负责规划和纠错，运行时负责预算、隔离风险和强制执行边界。把失败分类、退避重试、最大步数、重复错误检测和分级兜底做成框架能力，智能体才不会从 Demo 里的聪明助手，变成生产环境里停不下来的循环程序。
免责声明
本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。