5行Python代码给Agent加上熔断保护

场景

你部署了一个AI Agent服务，用户传了一段长文本让你处理。Agent调LLM——OpenAI刚好挂了。连续超时、连续重试、连续报错。

没有熔断保护的话，程序会傻乎乎地一直重试。每次等30秒超时，用户等了两分钟，最终收获一个不明不白的异常。

一个更体面的做法：连续失败3次，直接切断，走本地降级。 核心代码不超过10行。

核心代码

import time
from functools import wraps

def circuit_breaker(failure_threshold=3, recovery_timeout=30):
    """连续失败超过阈值就熔断，超时后自动恢复"""
    state = {"failures": 0, "last_failure": 0, "open": False}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 熔断中就先检查要不要恢复
            if state["open"]:
                if time.time() - state["last_failure"] > recovery_timeout:
                    state["open"] = False
                else:
                    print("🔴 熔断中，走降级")
                    return fallback(*args, **kwargs)
            
            try:
                result = func(*args, **kwargs)
                state["failures"] = 0
                return result
            except Exception as e:
                state["failures"] += 1
                state["last_failure"] = time.time()
                if state["failures"] >= failure_threshold:
                    state["open"] = True
                    print(f"⚠️ 连续失败{failure_threshold}次，熔断")
                raise e
        return wrapper
    return decorator

def fallback(*args, **kwargs):
    return "[降级] 服务暂不可用，稍后再试"

效果

@circuit_breaker(failure_threshold=3, recovery_timeout=30)
def call_llm(prompt):
    return openai_client.chat(prompt)

for i in range(5):
    try:
        result = call_llm("你好")
        print(f"✅ {result}")
    except Exception:
        print(f"❌ 第{i+1}次失败")

输出：

❌ 第1次失败
❌ 第2次失败
❌ 第3次失败
⚠️ 连续失败3次，熔断
🔴 熔断中，走降级
🔴 熔断中，走降级

如果用的是异步

def async_circuit_breaker(failure_threshold=3, recovery_timeout=30):
    state = {"failures": 0, "last_failure": 0, "open": False}
    
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if state["open"]:
                if time.time() - state["last_failure"] > recovery_timeout:
                    state["open"] = False
                else:
                    return await async_fallback(*args, **kwargs)
            try:
                result = await func(*args, **kwargs)
                state["failures"] = 0
                return result
            except Exception as e:
                state["failures"] += 1
                state["last_failure"] = time.time()
                if state["failures"] >= failure_threshold:
                    state["open"] = True
                raise e
        return wrapper
    return decorator

两个小改进

1. 区分异常类型——认证错误不该触发熔断，超时和连接错误才该
2. 暴露熔断状态——加个 /health 接口给Prometheus采集，线上出了状况一目了然

一句话： 熔断不是为了防止出错，是出错不可避免的时候，快速失败、保系统不崩。