Agent系统天然依赖大量外部资源:大模型API、检索服务、工具接口、数据库,任何一环出问题都可能拖垮整个任务流程。一次LLM调用超时可能导致用户等待数十秒,一个工具接口的连锁失败可能耗尽线程池和连接池,进而引发整个服务的雪崩。容错设计的目标很明确:在依赖方故障时,快速失败、快速恢复,并尽可能提供有价值的兜底结果。熔断器和降级策略正是实现这一目标的两大核心手段。

一、Agent系统为什么特别需要容错设计
传统的微服务调用链通常一次请求只涉及两三个下游,而Agent的执行链路要复杂得多。一个典型的ReAct风格Agent,单次任务可能触发十几轮LLM调用,每一轮还可能附带工具执行、记忆检索、结果解析。假设单次依赖调用的可用性是99.9%,一次任务串行调用20次,整体成功率就只剩98%,叠加重试和超时的不合理配置,实际表现会更差。
更麻烦的是Agent的故障具有放大效应。LLM接口变慢时,如果Agent无限制等待,会话会堆积;会话堆积会占满工作线程;线程占满后整个服务无法响应新请求,这就是典型的雪崩过程。如果不做任何保护,一次上游抖动就可能演变成全局故障。
容错设计要解决三个层面的问题:一是快速失败,不要让慢请求拖住资源;二是故障隔离,某个工具挂掉不应该影响其他工具和主流程;三是优雅降级,主链路不可用时给出有价值的替代结果,而不是直接抛出错误让对话中断。
二、熔断器的原理与实现
熔断器的思想来自电路保险丝:当检测到下游持续故障时,主动切断请求,让调用方立即失败,避免资源被持续消耗。同时周期性地放行少量探测请求,下游恢复后自动闭合,恢复正常调用。
经典的熔断器有三个状态:关闭(Closed)、打开(Open)和半开(Half-Open)。关闭状态下请求正常放行,同时统计滑动窗口内的失败率;当失败率超过阈值(例如50%)且请求量达到最小样本数,熔断器打开,后续请求不再发往下游,直接走降级逻辑;打开状态持续一个冷却时间(比如30秒)后进入半开状态,放行少量探测请求,探测成功则回到关闭状态,失败则重新打开。
下面是一个用Python实现的简化版熔断器,可以直接嵌入Agent的工具调用层:
import time
import threading
class CircuitBreaker:
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
def __init__(self, failure_threshold=5, timeout=30, half_open_probes=3):
self.failure_threshold = failure_threshold # 连续失败阈值
self.timeout = timeout # 打开后冷却时间(秒)
self.half_open_probes = half_open_probes # 半开状态探测请求数
self.state = self.CLOSED
self.failure_count = 0
self.opened_at = 0
self.probe_success = 0
self.lock = threading.Lock()
def allow_request(self):
with self.lock:
if self.state == self.CLOSED:
return True
if self.state == self.OPEN:
# 冷却期结束,转入半开状态
if time.time() - self.opened_at >= self.timeout:
self.state = self.HALF_OPEN
self.probe_success = 0
return True
return False
# 半开状态:允许少量探测请求
return self.probe_success < self.half_open_probes
def record_success(self):
with self.lock:
if self.state == self.HALF_OPEN:
self.probe_success += 1
if self.probe_success >= self.half_open_probes:
self._reset()
else:
self._reset()
def record_failure(self):
with self.lock:
if self.state == self.HALF_OPEN:
# 探测失败,重新打开
self._trip()
else:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self._trip()
def _trip(self):
self.state = self.OPEN
self.opened_at = time.time()
self.failure_count = 0
def _reset(self):
self.state = self.CLOSED
self.failure_count = 0在Agent中使用时,建议为每个依赖单独创建一个熔断器实例,用一个字典管理。这样搜索工具挂了只会触发搜索的熔断,不会影响代码执行工具和主LLM调用。参数配置上也有讲究:对LLM这种高延迟依赖,失败阈值可以设低一些(连续3到5次失败就熔断),冷却时间设长一点(60秒以上),因为模型服务恢复通常需要时间;对内部轻量接口,阈值可以适当放宽。
生产环境不必手写,可以直接使用成熟库。Python生态里有pybreaker,Java生态里有Resilience4j,都支持注解或装饰器方式的接入,还提供滑动窗口统计、慢调用熔断等进阶能力。慢调用熔断特别适合LLM场景:如果P95延迟从2秒飙到30秒,即使最终都成功,也应该熔断,因为慢本身就是一种故障。
三、降级策略:熔断之后做什么
熔断只是切断了请求,但用户体验还需要降级逻辑来兜底。降级的本质是:承认拿不到最优结果,但给出一个次优但可用的结果。Agent场景下常用的降级策略有以下几种。
第一种是静态兜底话术。当工具熔断时,Agent直接返回类似“当前搜索服务暂时不可用,我可以基于已有知识回答您的问题”的回复,并继续走纯LLM生成路径。实现最简单,但至少保证对话不中断。
第二种是缓存降级。对搜索、检索类工具,可以缓存最近的成功结果,熔断期间直接返回缓存数据,并明确告知用户数据可能不是最新的。缓存可以用简单的LRU字典,也可以接入Redis。对于时效性要求不高的问答场景,几小时前的检索结果通常也够用。
第三种是多模型降级。主LLM不可用或触发限流时,切换到备用模型。很多团队会配置主用GPT-4级别的模型,降级到成本更低的小模型,虽然复杂任务能力会打折扣,但简单任务完全能胜任。可以在系统提示词中维护一个模型优先级列表,按顺序尝试。
第四种是功能降级。多步规划任务失败时,退化为单轮直答模式;联网搜索不可用时,提示用户自行确认时效性信息。功能降级的核心原则是保住核心对话能力,牺牲增值功能。
把熔断和降级组合起来,一个健壮的工具调用封装大致如下:
import functools
breakers = {}
def with_breaker(name, fallback=None):
if name not in breakers:
breakers[name] = CircuitBreaker(failure_threshold=3, timeout=60)
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
cb = breakers[name]
if not cb.allow_request():
# 熔断打开,直接走降级逻辑
if fallback:
return fallback(*args, **kwargs)
raise ToolUnavailableError(f"工具 {name} 熔断中")
try:
result = func(*args, **kwargs)
cb.record_success()
return result
except Exception:
cb.record_failure()
if fallback:
return fallback(*args, **kwargs)
raise
return wrapper
return decorator
# 使用示例:搜索工具带缓存降级
search_cache = {}
def search_fallback(query):
if query in search_cache:
return {"data": search_cache[query], "stale": True}
return {"data": "搜索服务暂不可用,请稍后再试", "stale": False}
@with_breaker("web_search", fallback=search_fallback)
def web_search(query):
result = call_search_api(query)
search_cache[query] = result # 成功时更新缓存
return {"data": result, "stale": False}四、监控与参数调优建议
容错机制上线后,如果没有配套的监控,你很难知道熔断器有没有误伤正常请求。必须监控的核心指标包括:每个依赖的熔断器当前状态、熔断触发次数、降级触发次数、降级响应占比。如果发现某个工具每天降级占比超过20%,说明这不是偶发故障,需要从根本上排查下游稳定性或更换供应商。
参数调优方面有几个经验值可供参考。超时时间要小于用户可忍受的等待时间,LLM调用建议设置15到30秒的硬超时,流式调用可以放宽到60秒;重试次数控制在2次以内,且只对幂等操作重试,重试之间加指数退避和随机抖动,避免重试风暴;熔断阈值建议基于滑动窗口统计失败率而非简单连续计数,样本数至少20个请求,避免低流量时误判。
还有一个容易忽视的点:降级逻辑本身也需要测试。很多团队的降级代码从未被执行过,真出事时才发现降级路径里有bug,比如降级返回的数据格式和正常路径不一致,导致下游解析报错。建议用混沌工程的方式,定期在测试环境主动注入故障(断开某个工具的mock服务),验证熔断和降级的完整链路。可以用一个简单的开关让Agent在“故障演练模式”下强制所有工具走降级分支,跑通全流程回归。
最后强调一点原则:容错设计是分层的。请求层做超时和重试,依赖层做熔断和隔离,业务层做降级和兜底,三层配合才能真正构建出一个在依赖方各种异常下依然可用的Agent系统。只做其中一层,故障总会在没有保护的地方穿透进来。