Agent Fault Tolerance（Agent 容错回退）

Skill-Agent-Fault-Tolerance · 16-智能体工程

causalforecastingoptimizationmulti_agent供应链与补货MAS与智能体工程WF-A 智能补货

年化 ROI5-15 万元

实现难度⭐⭐☆☆☆

业务优先级⭐⭐⭐⭐☆

业务视角

适用角色CTO / 技术负责人 · 产品经理 · 数据工程师

适用平台跨境运营 AI Agent 工程落地 · Amazon SP API + LLM 集成 · 多平台数据采集 Agent

什么情况下用想把 AI 集成到业务系统，但 LLM 稳定性差、幻觉问题、成本控制都是挑战；Agent 任务失败了不知道哪步出了问题

成功是什么样的AI Agent 在生产环境稳定运行，失败可追踪，成本可控，复杂任务完成率 >85%

业务痛点

LLM 返回结果不稳定不可靠AI 幻觉导致业务决策错误Agent 任务失败了不知道哪步出问题AI 调用成本控制不住

1. 解决的问题

Agent 执行可能因 API 超时、LLM 输出格式错误、工具返回异常而失败。容错机制：(1) Retry with exponential backoff（$t_{retry} = \min(t_{base} \cdot 2^n, t_{max})$）；(2) Fallback 策略——LLM 失败切备选模型，工具失败用简化版；(3) Circuit Breaker——连续失败 N 次后停止尝试，避免雪崩。

2. 核心算法逻辑

3. 业务应用场景

WF-A 补货 Agent 调用库存 API 超时（3 次重试均失败）→ Circuit Breaker 打开→降级为"基于最新已知库存+需求预测"的保守建议 → 30 秒后半开测试→API 恢复→恢复正常。避免因 API 抖动导致补货决策完全中断。

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

ROI：避免 Agent 中断导致的决策延迟，年化隐性 5-15 万元
难度：⭐⭐☆☆☆ | 优先级：⭐⭐⭐⭐☆

7. 代码模板

代码块数量：1 · 路径：未检测到

import time, random

class CircuitBreaker:
    def __init__(self, failure_threshold=3, cooldown_sec=30):
        self.failures = 0; self.threshold = failure_threshold
        self.cooldown = cooldown_sec; self.last_failure = 0; self.open = False
    
    def call(self, fn, *args, fallback=None, **kwargs):
        if self.open:
            if time.time() - self.last_failure > self.cooldown:
                self.open = False; self.failures = 0  # half-open
            elif fallback:
                return fallback(*args, **kwargs)
        
        for attempt in range(3):
            try:
                result = fn(*args, **kwargs)
                self.failures = 0; return result
            except Exception:
                wait = min(2**attempt, 8); time.sleep(wait * 0.01)  # scaled for test
        
        self.failures += 1
        if self.failures >= self.threshold:
            self.open = True; self.last_failure = time.time()
        return fallback(*args, **kwargs) if fallback else None

# test
cb = CircuitBreaker(failure_threshold=2, cooldown_sec=0.1)
fail_fn = lambda: (_ for _ in ()).throw(Exception("timeout"))
fallback = lambda: "cached_result"
assert cb.call(fail_fn, fallback=fallback) == "cached_result"
assert cb.open  # circuit opened after 2 failures
print("[✓] Agent Fault Tolerance 测试通过")

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。