Agent容错自愈：六大韧性模式生产实战对比

import asyncio
import random
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=60, jitter=5),
    retry=retry_if_exception_type((RateLimitError, TimeoutError))
)
async def call_llm_with_retry(prompt: str, tools: list) -> dict:
    """带抖动的指数退避重试——仅对瞬态错误生效"""
    response = await client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}],
        tools=tools
    )
    if response.choices[0].finish_reason == "content_filter":
        raise ContentFilterError("内容过滤触发，不适用重试")
    return response

from circuitbreaker import circuit

@circuit(failure_threshold=3, recovery_timeout=30, expected_exception=Exception)
async def search_web(query: str) -> list:
    """Web搜索工具——熔断保护"""
    result = await search_api.search(query)
    return result

熔断器的三状态模型（Closed → Open → Half-Open）与Agent的执行循环天然契合。在Half-Open状态下，允许一个探测请求通过——如果成功，说明下游已恢复，关闭熔断器。

**实战经验**：对LLM API调用设置熔断器时，`failure_threshold`建议设为5次而非3次，因为LLM的延迟波动较大，3次失败可能只是暂时性波动。`recovery_timeout`建议设为60秒以上，给下游服务足够的恢复时间。

### 模式三：降级链（Fallback Chain）

将多个可替代的工具或模型按优先级排列，当首选方案失败时自动降级到次选，形成一条"降级链"。这是构建高可用Agent系统的核心模式。

```python
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

class FallbackLLM:
    def __init__(self):
        self.chain = [
            {"model": ChatOpenAI(model="gpt-4.1"), "name": "GPT-4.1", "cost": "high"},
            {"model": ChatAnthropic(model="claude-sonnet-4-6"), "name": "Claude Sonnet", "cost": "medium"},
            {"model": ChatGoogleGenerativeAI(model="gemini-2.5-pro"), "name": "Gemini", "cost": "low"},
        ]
        self.circuit_states = {m["name"]: "closed" for m in self.chain}

    async def invoke(self, prompt: str) -> str:
        for model_config in self.chain:
            if self.circuit_states[model_config["name"]] == "open":
                continue
            try:
                result = await model_config["model"].ainvoke(prompt)
                return result.content
            except Exception as e:
                logger.warning(f"{model_config['name']} failed: {e}")
                self.circuit_states[model_config["name"]] = "open"
                continue
        raise AllModelsFailedError("所有模型均不可用")

from datetime import datetime
import json

class AgentDeadLetterQueue:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.dlq_key = "agent:dlq"

    async def enqueue(self, task_id: str, error_context: dict):
        entry = {
            "task_id": task_id,
            "error": error_context,
            "state_snapshot": error_context.get("agent_state"),
            "timestamp": datetime.utcnow().isoformat(),
            "retry_count": error_context.get("retry_count", 0),
            "priority": self._calculate_priority(error_context)
        }
        await self.redis.zadd(
            self.dlq_key,
            {json.dumps(entry): entry["priority"]}
        )

    def _calculate_priority(self, ctx: dict) -> int:
        """高价值任务优先级更高"""
        base = 100
        if ctx.get("customer_tier") == "enterprise":
            base += 500
        if ctx.get("sla_deadline_hours", 24) < 4:
            base += 300
        return base

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# 构建带检查点的Agent图
graph = StateGraph(AgentState)
graph.add_node("analyze", analyze_request)
graph.add_node("plan", plan_steps)
graph.add_node("execute", execute_step)
graph.add_node("validate", validate_result)

checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute"]  # 在执行前暂停，等待人工审批
)

# 从检查点恢复
config = {"configurable": {"thread_id": "task-12345"}}
state = await app.ainvoke(None, config=config)  # 自动恢复到上次中断点

# OpenAI Agents SDK的审批模式
from openai import Agent, Runner

agent = Agent(
    name="finance_agent",
    instructions="处理用户退款请求",
    tools=[process_refund, check_order, send_email]
)

# 设置工具级审批
agent.add_guardrail(
    tool_name="process_refund",
    approval_required=True,  # 退款操作需人工确认
    max_auto_retries=2       # 自动重试上限
)

# LangGraph的中断点模式
from langgraph.graph import StateGraph

graph = StateGraph(AgentState)
graph.add_node("review", human_review_node)

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["process_payment"]  # 支付前暂停
)

# 人工审批后继续
state = await app.ainvoke(
    {"approval": "approved", "reviewer": "manager_zhang"},
    config={"configurable": {"thread_id": thread_id}}
)

# Temporal Agent Workflow示例
from temporalio import workflow, activity

@activity.defn
async def call_llm(prompt: str) -> str:
    # Activity自动重试+持久化
    response = await client.chat(prompt)
    return response

@workflow.defn
class AgentWorkflow:
    @workflow.run
    async def run(self, task: str) -> str:
        # Step 1: 分析任务
        analysis = await workflow.execute_activity(
            call_llm, f"分析任务: {task}",
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(
                maximum_attempts=5,
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=60),
            )
        )
        # Step 2: 如果分析失败，Workflow会自动从检查点恢复
        plan = await workflow.execute_activity(
            call_llm, f"制定计划: {analysis}",
            start_to_close_timeout=timedelta(seconds=60)
        )
        return plan

# 组合式容错Agent示例
class ResilientAgent:
    def __init__(self):
        self.llm_fallback = FallbackLLM()           # 模式三：降级链
        self.circuit_breaker = CircuitBreaker()      # 模式二：熔断器
        self.dlq = AgentDeadLetterQueue(redis)       # 模式四：死信队列
        self.checkpointer = PostgresSaver(...)       # 模式五：检查点
        self.max_retries = 3
        self.max_self_heal_rounds = 3

    async def run(self, task: str, thread_id: str):
        for attempt in range(self.max_retries):
            try:
                # 从检查点恢复
                state = await self.checkpointer.load(thread_id)

                # 执行Agent循环
                result = await self._execute_with_circuit_breaker(
                    task, state
                )

                # 输出校验
                if self._validate_output(result):
                    return result
                else:
                    # 触发自愈
                    result = await self._self_heal(result, task)
                    if result:
                        return result

            except CircuitBreakerError:
                result = await self.llm_fallback.invoke(task)
                if result:
                    return result
            except MaxRetriesExceeded:
                await self.dlq.enqueue(thread_id, {
                    "task": task, "agent_state": state,
                    "error": "max_retries_exceeded"
                })
                raise

        await self.dlq.enqueue(thread_id, {"task": task, "error": "all_recovery_failed"})