graph TD
    A[自主Agent生产故障谱系] --> B[模型层]
    A --> C[工具层]
    A --> D[状态层]
    A --> E[编排层]
    A --> F[成本层]

    B --> B1[返回格式错误]
    B --> B2[幻觉工具调用]
    B --> B3[拒绝执行]
    B --> B4[速率限制429]

    C --> C1[API超时]
    C --> C2[返回异常数据]
    C --> C3[权限拒绝403]
    C --> C4[服务不可用503]

    D --> D1[内存溢出OOM]
    D --> D2[状态损坏]
    D --> D3[检查点丢失]
    D --> D4[并发冲突]

    E --> E1[循环调用]
    E --> E2[级联失败]
    E --> E3[死锁等待]
    E --> E4[优先级反转]

    F --> F1[Token耗尽]
    F --> F2[成本飙升]
    F --> F3[预算超支]
    F --> F4[无用重试浪费]

from langgraph.pregel import RetryPolicy

# LangGraph: 基于输出验证的智能重试
def should_retry(state, response):
    """只有当工具调用格式错误时才重试，幻觉内容不重试"""
    if response.tool_calls:
        for tc in response.tool_calls:
            if not validate_tool_schema(tc):
                return True, "tool_call_schema_invalid"
    return False, None

agent = create_react_agent(
    model,
    tools,
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_interval=1.0,
        backoff_factor=2.0,
        retry_on=should_retry  # 自定义重试判断
    )
)

// Temporal: 跨进程崩溃的持久重试
const handle = await workflow.start(agentWorkflow, {
  taskQueue: 'agent-tasks',
  retry: {
    maximumAttempts: 5,
    initialInterval: '2s',
    maximumInterval: '60s',
    backoffCoefficient: 2.0,
    // Agent特有的非重试错误
    nonRetryableErrorTypes: [
      'TokenBudgetExhausted',
      'InvalidToolConfiguration',
      'HumanRejectedAction'
    ]
  }
});

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-5',
    retries=3,  # 输出验证失败自动重试
    result_type=ResearchReport  # Pydantic模型验证
)

@agent.tool(retries=2)
async def search_web(ctx, query: str) -> list[str]:
    """工具级别的独立重试配置"""
    return await search_client.search(query)

graph LR
    subgraph "三层熔断架构"
        direction TB
        L1["Layer 1: 工具级熔断<br>per-tool circuit breaker"]
        L2["Layer 2: 提供商级熔断<br>per-provider circuit breaker"]
        L3["Layer 3: 能力级熔断<br>per-capability circuit breaker"]
    end

    L1 --> |"工具A调用失败3次"| L1
    L1 --> |"触发工具A熔断<br>切换到工具B"| L2
    L2 --> |"OpenAI API错误率>50%"| L2
    L2 --> |"触发提供商熔断<br>切换到Anthropic"| L3
    L3 --> |"function calling退化"| L3
    L3 --> |"触发能力熔断<br>降级为纯文本响应"|

from datetime import datetime, timedelta
from dataclasses import dataclass, field
from enum import Enum
import threading

class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 熔断
    HALF_OPEN = "half_open"  # 探测

@dataclass
class AgentCircuitBreaker:
    """三层熔断器：工具→提供商→能力"""

    # Layer 1: 工具级
    tool_failures: dict[str, list[datetime]] = field(default_factory=dict)
    tool_threshold: int = 3
    tool_window: timedelta = timedelta(minutes=5)

    # Layer 2: 提供商级
    provider_errors: dict[str, list[datetime]] = field(default_factory=dict)
    provider_threshold: float = 0.5  # 错误率50%

    # Layer 3: 能力级
    capability_degradation: dict[str, float] = field(default_factory=dict)
    capability_threshold: float = 0.7  # 能力评分低于70%触发

    def check(self, tool: str, provider: str, capability: str) -> CircuitState:
        """三层级联检查"""
        # 检查能力级（最高优先级）
        if self.capability_degradation.get(capability, 1.0) < self.capability_threshold:
            return CircuitState.OPEN  # 能力级熔断，不可恢复

        # 检查提供商级
        provider_errs = self._count_recent(self.provider_errors.get(provider, []))
        if provider_errs > self.provider_threshold * 100:
            return CircuitState.OPEN  # 提供商级熔断

        # 检查工具级
        tool_fails = self._count_recent(self.tool_failures.get(tool, []))
        if tool_fails >= self.tool_threshold:
            return CircuitState.HALF_OPEN  # 工具级半开，允许探测

        return CircuitState.CLOSED

# 模型降级链示例
from typing import Optional
import httpx

class ModelFallbackChain:
    """Agent模型降级链"""

    CHAINS = {
        "complex_reasoning": [
            {"model": "claude-opus-4-6", "provider": "anthropic"},
            {"model": "gpt-5.4", "provider": "openai"},
            {"model": "gemini-3.1-pro", "provider": "google"},
        ],
        "tool_calling": [
            {"model": "gpt-5.4", "provider": "openai"},
            {"model": "claude-sonnet-4-6", "provider": "anthropic"},
            {"model": "gemini-3.1-flash", "provider": "google"},
        ],
        "simple_response": [
            {"model": "gpt-5-mini", "provider": "openai"},
            {"model": "claude-haiku-4-5", "provider": "anthropic"},
            {"model": "gemini-3.1-flash", "provider": "google"},
            {"model": "llama-4-maverick", "provider": "local"},  # 最终兜底
        ]
    }

    async def call_with_fallback(
        self, 
        chain_name: str, 
        prompt: str,
        tools: Optional[list] = None
    ):
        """沿降级链尝试，直到成功"""
        chain = self.CHAINS[chain_name]
        errors = []

        for model_config in chain:
            try:
                result = await self._call_model(model_config, prompt, tools)
                # 验证结果质量
                if self._validate_result(result, tools):
                    return result
                else:
                    errors.append(f"{model_config['model']}: quality_check_failed")
                    continue
            except Exception as e:
                errors.append(f"{model_config['model']}: {str(e)}")
                # 检查是否是提供商级故障
                if self._is_provider_error(e):
                    self._trip_provider_breaker(model_config['provider'])
                continue

        # 所有模型都失败
        raise AllModelsFailedError(chain_name, errors)

from langgraph.graph import StateGraph

def model_node_primary(state):
    try:
        return call_model(state, model="gpt-5.4")
    except ModelError:
        return {"fallback": True}

def model_node_fallback(state):
    return call_model(state, model="claude-opus-4-6")

graph = StateGraph(AgentState)
graph.add_conditional_edges("primary", 
    lambda s: "fallback" if s.get("fallback") else "next",
    {"fallback": "model_fallback", "next": "tool_execute"})

from pydantic_ai import Agent
from pydantic_ai.models import Model

# Pydantic AI: 运行时模型切换
agent = Agent()  # 延迟绑定模型

async def run_with_fallback(prompt: str):
    models = ['openai:gpt-5', 'anthropic:claude-opus-4-6', 'google:gemini-3.1']
    for model_name in models:
        try:
            result = await agent.run(prompt, model=model_name)
            return result.data
        except (ModelHTTPError, UnexpectedModelBehavior):
            continue
    raise RuntimeError("All models failed")

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# 开发环境：内存检查点
checkpointer = MemorySaver()

# 生产环境：PostgreSQL持久检查点
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_checkpoints"
)

# 创建带检查点的Agent
graph = create_react_agent(model, tools, checkpointer=checkpointer)

# 运行并获取检查点
config = {"configurable": {"thread_id": "research-task-001"}}
result = await graph.ainvoke({"messages": messages}, config)

# 从最近的检查点恢复
# 如果Agent在节点3崩溃，重启后自动从节点3的状态恢复
checkpoint = await checkpointer.aget(config)
print(f"恢复到步骤: {checkpoint['metadata']['source']}")

// Temporal: 事件溯源自动持久化
export async function agentWorkflow(task: AgentTask): Promise<AgentResult> {
  // 步骤1: 研究
  const research = await executeActivity(researchPhase, {
    task: task.description,
    // Activity失败自动重试，无需额外代码
    startToCloseTimeout: '5 minutes',
    retry: { maximumAttempts: 3 }
  });

  // 检查点隐式保存：research完成就是一个事件
  // 如果进程在这里崩溃，Temporal Worker重启后从research结果恢复

  // 步骤2: 分析
  const analysis = await executeActivity(analysisPhase, {
    research,
    startToCloseTimeout: '10 minutes'
  });

  // 步骤3: 报告
  return await executeActivity(reportPhase, {
    analysis,
    startToCloseTimeout: '3 minutes'
  });
}

flowchart TB
    subgraph "脆弱Agent：内存执行"
        A1[启动Agent] --> A2[调用LLM]
        A2 --> A3[调用工具A]
        A3 --> A4["💥 进程崩溃"]
        A4 --> A5["从头重新开始<br>已消耗$50 Token白费"]
    end

    subgraph "韧性Agent：持久执行"
        B1[启动Workflow] --> B2[调用LLM]
        B2 --> B3["事件记录：LLM完成"]
        B3 --> B4[调用工具A]
        B4 --> B5["事件记录：工具A完成"]
        B5 --> B6["💥 进程崩溃"]
        B6 --> B7["新Worker启动"]
        B7 --> B8["重放事件历史"]
        B8 --> B9["从工具A结果继续<br>零Token浪费"]
    end

# Temporal Python SDK: Agent工作流
from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def call_llm(prompt: str, model: str = "gpt-5") -> str:
    """Activity: 单次LLM调用，失败自动重试"""
    response = await client.chat.completions.create(
        model=model, messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@activity.defn
async def execute_tool(tool_name: str, params: dict) -> dict:
    """Activity: 工具执行，结果自动持久化"""
    return await tool_registry.execute(tool_name, params)

@workflow.defn
class ResearchAgent:
    """持久Agent工作流：即使服务器崩溃也能恢复"""

    @workflow.run
    async def run(self, task: str) -> str:
        # 每个Activity完成都是持久事件
        research = await workflow.execute_activity(
            call_llm, f"研究：{task}",
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3)
        )

        # 崩溃恢复点：research结果已持久化
        analysis = await workflow.execute_activity(
            call_llm, f"分析：{research}",
            start_to_close_timeout=timedelta(minutes=10)
        )

        report = await workflow.execute_activity(
            call_llm, f"生成报告：{analysis}",
            start_to_close_timeout=timedelta(minutes=5)
        )

        return report

// Inngest: 事件驱动的持久Agent
import { inngest } from "./client";

export const researchAgent = inngest.createFunction(
  { id: "research-agent", retries: 5 },
  { event: "agent/research.requested" },
  async ({ event, step }) => {
    // 每个step自动持久化
    const searchResults = await step.run("search", async () => {
      return await webSearch(event.data.query);
    });

    // 崩溃恢复点：searchResults已持久

    const analysis = await step.run("analyze", async () => {
      return await callLLM(`分析：${JSON.stringify(searchResults)}`);
    });

    const report = await step.run("report", async () => {
      return await callLLM(`生成报告：${analysis}`);
    });

    return { report };
  }
);

// Trigger.dev v4: 极简持久Agent
import { task } from "@trigger.dev/sdk/v4";

export const researchAgent = task({
  id: "research-agent",
  run: async (payload: { query: string }) => {
    // 每个operation自动持久化
    const results = await searchWeb(payload.query);
    const analysis = await analyzeWithLLM(results);
    return generateReport(analysis);
  },
  retry: { maxAttempts: 3 },
});

Level 0: 完整Agent能力（全部工具可用，最佳模型）
Level 1: 减少工具集（核心工具可用，模型降级）
Level 2: 缓存响应（返回历史相似任务的缓存结果）
Level 3: 纯文本回复（不调用任何工具，用模型内知识回复）

class GracefulDegradation:
    """Agent优雅降级控制器"""

    def __init__(self):
        self.health = {
            "tools": {},      # 每个工具的健康状态
            "models": {},     # 每个模型的健康状态  
            "services": {},   # 外部服务健康状态
        }
        self.cache = ResponseCache()  # 向量相似度缓存

    async def execute(self, task: str) -> AgentResponse:
        # Level 0: 完整能力
        if self._all_healthy():
            return await self._full_agent(task)

        # Level 1: 降级工具集
        available_tools = self._get_healthy_tools()
        if available_tools:
            return await self._limited_agent(task, available_tools)

        # Level 2: 缓存响应
        cached = await self.cache.find_similar(task)
        if cached and cached.confidence > 0.8:
            return AgentResponse(
                content=cached.content,
                metadata={"source": "cache", "confidence": cached.confidence},
                degraded=True
            )

        # Level 3: 纯文本回复
        return await self._text_only_fallback(task)

    async def _text_only_fallback(self, task: str) -> AgentResponse:
        """最终兜底：用最廉价模型的纯文本回复"""
        return AgentResponse(
            content="由于系统负载较高，我暂时无法使用工具。基于我的知识...",
            metadata={"source": "fallback", "level": 3},
            degraded=True,
            disclaimer="此回复未经过工具验证，仅供参考"
        )

graph TB
    subgraph "自主Agent六维韧性栈"
        direction TB

        subgraph "外部防护层"
            D1["第1维: 智能重试<br>格式感知+上下文压缩"]
            D2["第2维: 多层熔断<br>工具→提供商→能力"]
            D3["第3维: 模型降级链<br>GPT-5→Claude→Gemini→Local"]
        end

        subgraph "内部持久层"
            D4["第4维: 状态快照<br>LangGraph Checkpointer"]
            D5["第5维: 持久执行<br>Temporal/Inngest"]
        end

        subgraph "最终兜底层"
            D6["第6维: 优雅降级<br>完整Agent→缓存→纯文本"]
        end

        Request["用户请求"] --> D1
        D1 --> |"重试耗尽"| D2
        D2 --> |"工具熔断"| D3
        D3 --> |"模型切换"| Execute["执行Agent"]
        Execute --> D4
        D4 --> D5
        D5 --> |"不可恢复"| D6
        D6 --> Response["降级响应"]
    end

"""
生产级自主Agent韧性栈集成示例
框架：LangGraph + Temporal + 自定义降级
"""

from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class ResilientAgentConfig:
    # 第1维：重试
    max_retries: int = 3
    retry_backoff: float = 2.0

    # 第2维：熔断
    tool_failure_threshold: int = 3
    provider_error_rate_threshold: float = 0.5

    # 第3维：模型降级链
    model_chains: dict = None  # 按能力类型配置

    # 第4维：检查点
    checkpoint_backend: str = "postgres"
    checkpoint_conn: str = "postgresql://..."

    # 第5维：持久执行
    durable_runtime: str = "temporal"  # temporal/inngest/none

    # 第6维：降级
    cache_enabled: bool = True
    fallback_to_text: bool = True

class ResilientAgent:
    def __init__(self, config: ResilientAgentConfig):
        self.config = config
        self.circuit_breaker = AgentCircuitBreaker(config)
        self.model_fallback = ModelFallbackChain()
        self.checkpointer = self._init_checkpointer()
        self.degradation = GracefulDegradation()

    async def run(self, task: str) -> AgentResponse:
        """六维韧性保护的Agent执行"""
        config = {"configurable": {"thread_id": task.id}}

        # 检查是否有历史检查点可恢复
        checkpoint = await self.checkpointer.aget(config)
        if checkpoint:
            state = checkpoint["state"]
        else:
            state = {"task": task, "steps_completed": []}

        # 沿模型降级链尝试执行
        for model in self.model_fallback.get_chain("complex_reasoning"):
            # 检查模型级熔断
            if self.circuit_breaker.check("llm", model.provider, "reasoning") == CircuitState.OPEN:
                continue

            try:
                result = await self._execute_with_retry(
                    state, model, max_retries=self.config.max_retries
                )

                # 成功：保存检查点并返回
                await self.checkpointer.asave(config, result)
                return AgentResponse(content=result, degraded=False)

            except (ModelError, ToolError) as e:
                self.circuit_breaker.record_failure(
                    tool=e.tool, provider=model.provider, 
                    capability=e.capability
                )
                continue

        # 所有模型失败：触发降级
        return await self.degradation.execute(task)

Agent执行时长<5分钟？
├── 是 → 需要检查点？
│   ├── 否 → Pydantic AI（最简）或 LangGraph（灵活）
│   └── 是 → LangGraph + PostgresSaver
└── 否 → 需要跨进程持久？
    ├── 是 → Temporal（大规模）或 Inngest（TypeScript小团队）
    └── 否 → LangGraph + PostgresSaver + 自定义降级