Agent 的工具不是你写的本地函数，就是外部 API、数据库、搜索引擎。它们会超时、会限流、会偶发 500。如果工具一挂整个 Agent 就崩，那它根本没法上线。本教程给每个工具调用套上三道防线和一道可选熔断：超时、指数退避重试、失败降级，全写成一个能直接复用的装饰器。
💡 演示用 Python 标准库加 tenacity（重试库）。没装的话 pip install tenacity。核心思想与语言无关，JS / Go 同理。
Step 1：先给工具加「超时」，别让它无限等
1 任何外部调用都必须有截止时间
没有超时的网络请求，遇到对方卡死会把你的 Agent 线程一直吊住。这里用 requests 的 timeout 做最小示范。把它想成给每次外部调用装一个闹钟：时间一到就放弃，立刻转去走后续的兜底逻辑，而不是干等对面回应。
import requests
def fetch_weather(city):
# timeout=(连接, 读取) 单位秒；任一步超过就抛异常
r = requests.get(f"https://api.example.com/weather?c={city}", timeout=(3, 10))
return r.json()别写 timeout=None（很多库的默认）。那等于「永远等」。Agent 跑长任务时，一个永不返回的调用会拖垮整条链路。
Step 2：加「指数退避重试」，只对可重试错误重试
2 网络抖动用重试救，逻辑错误别重试
超时、429 限流、临时 5xx 适合重试；但 400、鉴权失败、参数错误重试也没用，反而放大流量。tenacity 可以精确控制。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=8),
retry=retry_if_exception_type((requests.Timeout, requests.HTTPError)),
)
def fetch_weather(city):
r = requests.get(f"https://api.example.com/weather?c={city}", timeout=(3, 10))
r.raise_for_status() # 4xx/5xx 转异常，但只有 Timeout/HTTPError 会触发重试
return r.json()💡 wait_exponential 等的是 1、2、4 秒（封顶 8 秒），这叫指数退避。再叠加一点随机抖动（jitter）更好，能避免一堆请求同时重试把对端打爆（惊群效应）。
Step 3：加「降级」，主工具挂了用备胎
3 拿不到实时数据，就给一个能用的替代
实时接口挂了，与其报错，不如返回缓存值、默认值，或切到另一个数据源。关键是：降级返回值要明确标注「非实时」，别让下游误以为是最新数据。
CACHE = {"beijing": {"temp": 22, "fresh": False}}
def fetch_weather_safe(city):
try:
return fetch_weather(city) # 带重试的主流程
except Exception:
fallback = CACHE.get(city)
if fallback:
return {**fallback, "note": "降级：返回缓存，非实时"}
return {"note": "降级：暂无数据", "temp": None}降级不是「假装成功」。一定要在返回里带上 note 标记，否则上游把过期数据当成实时数据用，比直接报错更危险。
Step 4：加「熔断」，连续失败就先别打了
4 对端已经着火时，停止发请求
如果接口连续 5 次都失败，多半是它挂了或在维护。继续打只会雪上加霜。熔断器在「打开」期间直接快速失败，过一阵再半开试探。
import time
class CircuitBreaker:
def __init__(self, fail_limit=5, cooldown=30):
self.fail_limit = fail_limit
self.cooldown = cooldown
self.failures = 0
self.opened_at = 0
def allow(self):
if self.failures >= self.fail_limit:
if time.time() - self.opened_at < self.cooldown:
return False
self.failures = 0 # 冷却结束，半开重试
return True
def on_fail(self):
self.failures += 1
if self.failures >= self.fail_limit:
self.opened_at = time.time()Step 5：把四道收成一个装饰器 with_tool_guard
5 一行套到任意工具上
把超时、重试、降级、熔断拼起来，做成一个装饰器。以后给 Agent 加工具，顶部加一行即可。
def with_tool_guard(fallback=None):
def deco(fn):
@retry(stop=stop_after_attempt(3), wait=wait_exponential(1, 1, 8),
retry=retry_if_exception_type((requests.Timeout, requests.HTTPError)))
def wrapped(*a, **k):
return fn(*a, **k)
def safe(*a, **k):
if not breaker.allow():
return fallback() if fallback else {"note": "熔断中"}
try:
return wrapped(*a, **k)
except Exception:
breaker.on_fail()
return fallback() if fallback else {"note": "调用失败"}
return safe
return deco
@with_tool_guard(fallback=lambda: {"note": "降级：缓存", "temp": 22})
def fetch_weather(city):
...💡 装饰器把「可靠性」和「业务逻辑」解耦。工具函数只管干活，防护统一在外层，可读也好测。
Step 6：塞进 Agent 循环里跑一遍
6 模拟对端抽风，看兜底是否生效
故意让 fetch_weather 偶发抛错，验证重试生效；再让它一直错，验证熔断与降级最终返回可用值而非崩溃。
import random
def flaky(city):
if random.random() < 0.7:
raise requests.Timeout("抽风")
return {"temp": 20}
breaker = CircuitBreaker()
guarded = with_tool_guard(fallback=lambda: {"temp": None, "note": "降级"})
tool = guarded(flaky)
for _ in range(10):
print(tool("beijing")) # 多数走重试/降级，不会抛异常Step 7：记住三条红线
7 防护也有代价，别乱加
三道防线和熔断不是越多越好，用错地方反而制造麻烦。
红线清单：
1. 写操作（下单、发消息、删数据）不要盲目重试 —— 可能重复执行
2. 降级返回值必须标注「非实时 / 缓存」，禁止伪装成真实结果
3. 熔断阈值和冷却时间要按业务调，别照抄默认值最危险的 bug 是「悄悄降级」：用户以为操作成功，其实走的是兜底分支。所有降级与熔断事件都要记日志，最好上报监控，让人能看见。