AI Agent 可靠性工程：从错误恢复到优雅降级的生产级实践

┌──────────────────────────────────────────────────────────────┐
│              传统软件 vs AI Agent 失败模式对比                  │
├──────────────────┬──────────────────┬────────────────────────┤
│     维度          │   传统软件        │     AI Agent            │
├──────────────────┼──────────────────┼────────────────────────┤
│ 失败可见性        │ 显式错误/崩溃     │ 静默输出错误结果         │
│ 可复现性         │ 同输入同输出       │ 非确定性，难以复现        │
│ 失败时机          │ 通常是即时的       │ 可能在 N 步后延迟出现    │
│ 错误传播          │ 异常中断执行       │ 错误悄无声息地向下游传播  │
│ 诊断方式          │ 查看错误码和堆栈   │ 需要重建完整推理链        │
│ 成本信号          │ 固定成本/请求      │ 失控 Agent 可能烧掉 20x 成本│
└──────────────────┴──────────────────┴────────────────────────┘

from enum import Enum
from dataclasses import dataclass
from typing import Optional

class ErrorCategory(Enum):
    INFRASTRUCTURE = "infrastructure"   # 网络超时、服务不可用、限流
    MODEL_OUTPUT = "model_output"       # 输出格式异常、空响应、幻觉
    LOGICAL = "logical"                 # 推理错误、目标漂移、上下文丢失
    SECURITY = "security"               # Prompt 注入、权限越界

@dataclass
class AgentError:
    category: ErrorCategory
    error_type: str
    message: str
    retryable: bool
    severity: str
    context: Optional[dict] = None

def classify_model_error(response: dict, expected_schema: dict) -> AgentError:
    if not response.get("content"):
        return AgentError(
            category=ErrorCategory.MODEL_OUTPUT,
            error_type="empty_response",
            message="Model returned empty content",
            retryable=True,
            severity="medium"
        )
    if not validate_schema(response["content"], expected_schema):
        return AgentError(
            category=ErrorCategory.MODEL_OUTPUT,
            error_type="schema_violation",
            message="Output does not match expected schema",
            retryable=True,
            severity="medium"
        )
    return AgentError(
        category=ErrorCategory.MODEL_OUTPUT,
        error_type="unknown",
        message="Unspecified model output error",
        retryable=True,
        severity="low"
    )

import time
import random

def retry_with_backoff(
    func,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter_factor: float = 0.3
):
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, jitter_factor * delay)
            time.sleep(delay + jitter)

class RetryStrategy:
    CONFIGS = {
        "timeout": {
            "max_retries": 3, "base_delay": 0.5, "max_delay": 5
        },
        "rate_limit": {
            "max_retries": 5, "base_delay": 60, "max_delay": 300
        },
        "server_error": {
            "max_retries": 2, "base_delay": 2, "max_delay": 10
        },
        "schema_violation": {
            "max_retries": 2, "base_delay": 0, "max_delay": 0
        },
        "auth_error": {
            "max_retries": 0
        },
    }

    @classmethod
    def get(cls, error_type: str) -> dict:
        return cls.CONFIGS.get(error_type, cls.CONFIGS["server_error"])

def build_retry_prompt(original_prompt: str, error: AgentError, attempt: int) -> str:
    if error.error_type == "schema_violation":
        return (
            f"{original_prompt}\n\n"
            f"【重要】你的上一次回复未符合要求的格式。"
            f"请严格按照以下 JSON Schema 输出：\n"
            f"{error.context.get('expected_schema')}\n"
            f"不要添加任何额外字段或文本说明。"
        )
    if error.error_type == "empty_response":
        return (
            f"{original_prompt}\n\n"
            f"【重要】请务必生成实际内容作为回复，"
            f"不要返回空内容。"
        )
    return original_prompt

┌─────────────────────────────────────────────────────────────┐
│                   熔断器状态机                                │
│                                                             │
│  ┌──────────┐    失败超阈值    ┌──────────┐                 │
│  │  CLOSED   │ ──────────────→ │   OPEN   │                 │
│  │ (正常放行) │                 │ (拒绝请求)│                 │
│  └──────────┘                  └─────┬────┘                 │
│       ↑                              │                       │
│       │ 探测成功                      │ 冷却时间到期           │
│       │                              ▼                       │
│       │                        ┌────────────┐               │
│       └─────────────────────── │ HALF-OPEN  │               │
│         探测失败 → 回到 OPEN    │ (允许单次探测)│              │
│                                └────────────┘               │
└─────────────────────────────────────────────────────────────┘

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class LLMCircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        detection_window: float = 300,
        cooldown_period: float = 300,
        extended_cooldown: float = 900
    ):
        self.failure_threshold = failure_threshold
        self.detection_window = detection_window
        self.cooldown_period = cooldown_period
        self.extended_cooldown = extended_cooldown
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.last_state_change = time.time()
        self.consecutive_reopen_count = 0

    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            cooldown = (
                self.extended_cooldown
                if self.consecutive_reopen_count > 0
                else self.cooldown_period
            )
            if time.time() - self.last_state_change > cooldown:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        return True

    def record_success(self):
        self.failure_count = 0
        self.consecutive_reopen_count = 0
        if self.state != CircuitState.CLOSED:
            self.state = CircuitState.CLOSED
            self.last_state_change = time.time()

    def record_failure(self):
        now = time.time()
        if now - self.last_failure_time > self.detection_window:
            self.failure_count = 0
        self.failure_count += 1
        self.last_failure_time = now
        if self.failure_count >= self.failure_threshold:
            if self.state == CircuitState.HALF_OPEN:
                self.consecutive_reopen_count += 1
            self.state = CircuitState.OPEN
            self.last_state_change = now

class AdaptiveRecovery:
    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.current_workers = 1
        self.scale_up_interval = 300

    def on_success(self):
        if self.current_workers < self.max_workers:
            self.current_workers = min(
                self.current_workers + 1,
                self.max_workers
            )

    def on_failure(self):
        self.current_workers = max(1, self.current_workers // 2)

    def get_concurrency(self) -> int:
        return self.current_workers

┌──────────────────────────────────────────────────────┐
│                   优雅降级层级                         │
│                                                      │
│  Level 1: 主路径（完整功能）                           │
│  ├─ 主力模型 + 完整上下文 + 全部工具                    │
│  │                                                   │
│  Level 2: 降级模型（功能降级）                         │
│  ├─ 备选模型 + 精简上下文 + 核心工具                    │
│  │                                                   │
│  Level 3: 缓存响应（数据复用）                         │
│  ├─ 基于语义相似度匹配的历史响应                        │
│  │                                                   │
│  Level 4: 规则引擎（最小可用）                         │
│  ├─ 基于模板的确定性响应                              │
│  │                                                   │
│  Level 5: 人工升级（兜底方案）                         │
│  └─ 转交人工处理 + 通知用户                           │
└──────────────────────────────────────────────────────┘

class GracefulAgent:
    def __init__(self):
        self.primary_llm = LLMClient(model="claude-opus-4")
        self.fallback_llm = LLMClient(model="claude-sonnet-4")
        self.minimal_llm = LLMClient(model="claude-haiku")
        self.cache = SemanticCache()
        self.rule_engine = RuleEngine()

    async def process(self, user_input: str) -> str:
        try:
            return await self.primary_llm.generate(
                prompt=self._build_full_prompt(user_input),
                temperature=0.1
            )
        except (TimeoutError, RateLimitError):
            pass
        try:
            return await self.fallback_llm.generate(
                prompt=self._build_simple_prompt(user_input),
                temperature=0.2
            )
        except Exception:
            pass
        cached = await self.cache.find_similar(user_input)
        if cached:
            return self._adapt_cached_response(cached, user_input)
        template_result = self.rule_engine.process(user_input)
        if template_result:
            return template_result
        return await self._escalate_to_human(user_input)

class BudgetCircuitBreaker:
    def __init__(self, max_cost_per_task: float = 1.0):
        self.max_cost = max_cost_per_task
        self.current_cost = 0.0

    def check_budget(self, estimated_tokens: int, model: str) -> bool:
        estimated_cost = self._estimate_cost(estimated_tokens, model)
        if self.current_cost + estimated_cost > self.max_cost:
            raise BudgetExceededError(
                f"Task budget exceeded: {self.current_cost:.4f} "
                f"/ {self.max_cost:.4f}"
            )
        return True

    def record_usage(self, input_tokens: int, output_tokens: int, model: str):
        self.current_cost += self._calculate_cost(
            input_tokens, output_tokens, model
        )

import json
import hashlib
from datetime import datetime

class AgentCheckpoint:
    def __init__(self, task_id: str, storage_path: str):
        self.task_id = task_id
        self.storage_path = storage_path

    def save(self, state: dict, step: int):
        checkpoint = {
            "task_id": self.task_id,
            "step": step,
            "state": state,
            "timestamp": datetime.utcnow().isoformat(),
            "checksum": hashlib.sha256(
                json.dumps(state, sort_keys=True).encode()
            ).hexdigest()
        }
        path = f"{self.storage_path}/{self.task_id}_step{step}.json"
        with open(path, "w") as f:
            json.dump(checkpoint, f)

    def load_latest(self) -> dict:
        import glob
        pattern = f"{self.storage_path}/{self.task_id}_step*.json"
        files = sorted(glob.glob(pattern))
        if not files:
            return None
        with open(files[-1]) as f:
            checkpoint = json.load(f)
        expected = hashlib.sha256(
            json.dumps(checkpoint["state"], sort_keys=True).encode()
        ).hexdigest()
        if checkpoint["checksum"] != expected:
            raise CheckpointCorruptionError("Checkpoint integrity check failed")
        return checkpoint

from opentelemetry import trace

tracer = trace.get_tracer("agent.reliability")

def traced_llm_call(prompt: str, model: str) -> str:
    with tracer.start_as_current_span("llm.generate") as span:
        span.set_attribute("gen_ai.system", "anthropic")
        span.set_attribute("gen_ai.request.model", model)
        response = llm_client.generate(prompt, model=model)
        span.set_attribute("gen_ai.usage.input_tokens", response.input_tokens)
        span.set_attribute("gen_ai.usage.output_tokens", response.output_tokens)
        span.set_attribute("gen_ai.response.finish_reason", response.finish_reason)
        return response.content

def traced_tool_call(tool_name: str, args: dict) -> dict:
    with tracer.start_as_current_span(f"tool.{tool_name}") as span:
        span.set_attribute("gen_ai.tool.name", tool_name)
        span.set_attribute("gen_ai.tool.call.id", generate_call_id())
        result = execute_tool(tool_name, args)
        span.set_attribute("gen_ai.tool.call.status", "success")
        return result

class HumanEscalationPolicy:
    ESCALATION_TRIGGERS = {
        "max_tool_failures": 3,
        "max_consecutive_retries": 5,
        "confidence_threshold": 0.4,
        "cost_multiplier": 10,
        "security_violation": True,
    }

    def should_escalate(self, agent_state: dict) -> bool:
        if agent_state["tool_failure_count"] >= self.ESCALATION_TRIGGERS["max_tool_failures"]:
            return True
        if agent_state["retry_count"] >= self.ESCALATION_TRIGGERS["max_consecutive_retries"]:
            return True
        if agent_state.get("confidence", 1.0) < self.ESCALATION_TRIGGERS["confidence_threshold"]:
            return True
        if agent_state["current_cost"] > agent_state["median_cost"] * self.ESCALATION_TRIGGERS["cost_multiplier"]:
            return True
        if agent_state.get("security_flag"):
            return True
        return False

┌─────────────────────────────────────────────────────────────────┐
│                    Agent 可靠性技术栈                             │
├─────────────────────────────────────────────────────────────────┤
│  Layer 5: 可观测性与持续改进                                      │
│  ├─ OpenTelemetry GenAI Tracing                                 │
│  ├─ 成本归因与异常检测                                           │
│  └─ Trace → Dataset → Eval 闭环                                 │
├─────────────────────────────────────────────────────────────────┤
│  Layer 4: Human-in-the-Loop                                    │
│  ├─ 升级策略与触发条件                                           │
│  ├─ 置信度阈值                                                  │
│  └─ 敏感操作审批                                                │
├─────────────────────────────────────────────────────────────────┤
│  Layer 3: 优雅降级                                              │
│  ├─ 多层回退（模型→缓存→规则→人工）                               │
│  ├─ 成本预算熔断                                                │
│  └─ 检查点与状态恢复                                             │
├─────────────────────────────────────────────────────────────────┤
│  Layer 2: 熔断器                                                │
│  ├─ 三态状态机（CLOSED→OPEN→HALF_OPEN）                         │
│  ├─ 自适应恢复                                                  │
│  └─ 分域隔离                                                    │
├─────────────────────────────────────────────────────────────────┤
│  Layer 1: 智能重试                                              │
│  ├─ 指数退避 + 抖动                                             │
│  ├─ 错误类型差异化策略                                           │
│  └─ 重试时 Prompt 修改                                          │
└─────────────────────────────────────────────────────────────────┘