一、大模型"翻车"的常见姿势

把AI Agent部署到跨境电商生产环境后，你会发现一个残酷的现实：大模型远比你想象的更不可靠。

它会编造不存在的商品ID、把美元价格当成人民币直接回复客户、在深夜突然切换成法语跟阿拉伯客户聊天、因为上下文超长而"失忆"忘记刚才查过的库存数字、在多轮工具调用后直接输出JSON格式的原始数据而非友好回复……这些都不是理论上的风险，而是真实发生过的事故。

更棘手的是，Agent的故障传播链比普通服务更长：一个工具调用异常 → 推理方向跑偏 → 后续所有步骤都失效 → 最终输出错误结果。相比传统服务的"一锤子买卖"，Agent的"多米诺骨牌"效应让问题排查和修复都变得更加困难。

所以，异常处理不是锦上添花，而是Agent上线的必要条件。本文将围绕三类核心问题展开：

1. 基础容错——LLM调用失败、超时、限流的应对策略
2. 输出质量保障——幻觉检测、格式校验、语义检查
3. 降级兜底——当Agent彻底"翻车"时，如何保住业务底线

---

二、基础容错：LLM调用层的稳定性保障

2.1 重试机制（含退避策略）

LLM服务不可能100%可用。限流、超时、服务端5xx错误，都是常态。因此，重试是Agent接入层的标配。

以下是一个带指数退避的重试装饰器：

import time
import random
from functools import wraps
from typing import Type, Tuple, Optional

class RetryConfig:
    def __init__(
        self,
        max_attempts: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 30.0,
        backoff_multiplier: float = 2.0,
        jitter: bool = True,
        retryable_exceptions: Optional[Tuple[Type[Exception], ...]] = None
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.backoff_multiplier = backoff_multiplier
        self.jitter = jitter
        self.retryable_exceptions = retryable_exceptions or (
            ConnectionError,
            TimeoutError,
            RateLimitError,  # 自定义限流异常
            ServiceUnavailableError,  # 自定义服务不可用异常
        )

def with_retry(config: RetryConfig = None):
    """带指数退避的重试装饰器"""
    config = config or RetryConfig()
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, config.max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except config.retryable_exceptions as e:
                    last_exception = e
                    
                    if attempt == config.max_attempts:
                        raise  # 最后一次重试失败，抛出异常
                    
                    # 计算退避延迟：base_delay * (backoff_multiplier ^ (attempt - 1))
                    delay = config.base_delay * (config.backoff_multiplier ** (attempt - 1))
                    delay = min(delay, config.max_delay)
                    
                    # 加入随机抖动，防止"雷鸣群"效应
                    if config.jitter:
                        delay *= (0.5 + random.random() * 0.5)
                    
                    print(f"[重试] 第{attempt}次失败，{delay:.2f}秒后重试... 错误: {e}")
                    time.sleep(delay)
            
            raise last_exception  # 理论上不会执行到这里
        return wrapper
    return decorator

# 使用示例
@with_retry()
def call_llm_with_retry(messages):
    """调用LLM，自动重试"""
    return llm.invoke(messages)

2.2 超时控制

Agent的一次调用可能包含多轮推理，整体耗时不可控。必须为每次LLM调用和整体任务设置超时：

import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError

class TimeoutManager:
    """LLM调用超时管理器"""
    
    def __init__(self, default_timeout: int = 30):
        self.default_timeout = default_timeout
        self.executor = ThreadPoolExecutor(max_workers=5)
    
    def call_with_timeout(self, func, timeout: int = None, *args, **kwargs):
        """在指定超时时间内执行函数"""
        timeout = timeout or self.default_timeout
        future = self.executor.submit(func, *args, **kwargs)
        try:
            return future.result(timeout=timeout)
        except FutureTimeoutError:
            future.cancel()
            raise TimeoutError(f"LLM调用超时（{timeout}秒）")
        except Exception as e:
            future.cancel()
            raise

# Agent整体执行超时
def execute_agent_with_timeout(executor, input_data, timeout: int = 60):
    """Agent整体执行超时控制"""
    try:
        result = timeout_manager.call_with_timeout(
            executor.invoke, 
            timeout,
            input_data
        )
        return result
    except TimeoutError:
        # 记录告警，返回降级结果
        return {"output": "抱歉，当前请求处理时间较长，请稍后重试或联系人工客服。"}

---

三、输出质量保障：检测"幻觉"与格式异常

LLM调用成功不等于结果可用。输出质量保障是Agent上线前最关键的一道防线。

3.1 幻觉检测：防止LLM编造数据

跨境电商场景中，编造价格、库存、物流信息会直接导致客诉和资损。我们通过规则校验 + 逻辑一致性检查来拦截幻觉：

from typing import Dict, Any, List, Tuple
import re

class HallucinationDetector:
    """幻觉检测器"""
    
    def __init__(self):
        # 定义危险模式：LLM不应生成的内容
        self.dangerous_patterns = [
            (r'\$\d+,\d+', '大额价格'),  # 防止编造高额价格
            (r'库存\s*[：:]\s*\d+', '库存数字'),  # 库存必须来自API
            (r'订单号\s*[：:]\s*[A-Z0-9]+', '订单号'),  # 订单号必须查询
            (r'已发货|已签收|已退款', '物流/售后状态'),  # 状态必须有依据
        ]
    
    def detect(self, response: str, tool_results: List[Dict]) -> Tuple[bool, List[str]]:
        """
        检测回复中是否存在幻觉
        返回: (是否安全, 风险列表)
        """
        risks = []
        
        # 1. 检查是否包含危险模式（且未在工具结果中找到依据）
        for pattern, label in self.dangerous_patterns:
            matches = re.findall(pattern, response)
            if matches:
                # 检查工具结果中是否有对应数据支撑
                if not self._has_support_in_tool_results(matches, tool_results):
                    risks.append(f"检测到未经工具验证的{label}: {matches[:3]}")
        
        # 2. 检查是否包含"我查到/我查询到"但实际没有调用工具
        if "我查" in response or "查询到" in response:
            if not any(r.get("tool_name") in ["search_products", "check_stock"] 
                       for r in tool_results):
                risks.append("LLM声称查询了数据，但实际没有工具调用记录")
        
        # 3. 检查回复中是否包含JSON结构（应被格式化，不应直接暴露）
        if response.strip().startswith("{") or response.strip().startswith("["):
            risks.append("回复包含未格式化的JSON数据")
        
        is_safe = len(risks) == 0
        return is_safe, risks
    
    def _has_support_in_tool_results(self, matches: List[str], tool_results: List[Dict]) -> bool:
        """检查工具结果中是否包含匹配内容"""
        all_tool_text = " ".join([
            str(r.get("output", "")) for r in tool_results
        ])
        for m in matches:
            if m in all_tool_text:
                return True
        return False

# 使用示例
detector = HallucinationDetector()

def post_process_agent_output(output: str, tool_results: List[Dict]) -> Dict:
    """Agent输出后处理"""
    is_safe, risks = detector.detect(output, tool_results)
    
    if is_safe:
        return {"status": "ok", "content": output}
    else:
        # 记录风险日志，触发告警
        log_alert("agent_hallucination", {"risks": risks, "output": output[:500]})
        return {
            "status": "risky",
            "content": output,
            "risks": risks,
            "suggestion": "建议人工复核"
        }

3.2 工具调用的"护栏"：防止危险操作

当Agent具备修改配置、调整预算、执行退款等能力时，需要在工具层设置硬性护栏：

from enum import Enum
from pydantic import BaseModel, Field

class ToolActionLevel(Enum):
    READ = "read"        # 只读操作，直接执行
    WRITE = "write"      # 写操作，需要二次确认
    DANGEROUS = "danger" # 危险操作，必须人工审批

class ToolGuardrail:
    """工具调用的安全护栏"""
    
    # 危险操作需要审批
    DANGEROUS_TOOLS = {"adjust_budget", "refund_order", "cancel_order", "update_price"}
    # 写操作需要确认
    WRITE_TOOLS = {"create_campaign", "modify_inventory", "add_coupon"}
    
    @classmethod
    def check_action(cls, tool_name: str, args: Dict, user_role: str = None) -> Tuple[bool, str]:
        """
        检查工具调用是否安全
        返回: (是否允许, 原因/要求)
        """
        if tool_name in cls.DANGEROUS_TOOLS:
            if user_role not in ["admin", "manager"]:
                return False, f"危险操作 {tool_name} 需要管理员权限，已转人工审批"
            return False, f"危险操作 {tool_name} 需要人工二次确认"
        
        if tool_name in cls.WRITE_TOOLS:
            return False, f"写操作 {tool_name} 需要用户确认，是否继续？"
        
        return True, "ok"

# 在Agent执行工具前插入
def tool_call_interceptor(tool_name: str, args: Dict, user_role: str) -> Dict:
    """工具调用拦截器"""
    allowed, message = ToolGuardrail.check_action(tool_name, args, user_role)
    
    if not allowed:
        return {
            "status": "blocked",
            "message": message,
            "requires_approval": True
        }
    
    return {"status": "allowed"}

---

四、降级兜底：当Agent彻底"翻车"时

当LLM无法解析用户意图、连续重试失败、输出质量检测不通过时，需要有一套降级策略保证业务底线。

4.1 三层降级阶梯

class FallbackStrategy:
    """降级策略管理器"""
    
    def __init__(self):
        self.fallback_handlers = []
    
    def execute_with_fallback(self, user_input: str, user_context: Dict) -> str:
        """
        按优先级尝试不同的处理链路
        直到有一个成功返回
        """
        # 第一层：完整Agent模式（有工具调用）
        try:
            result = self._try_full_agent(user_input, user_context)
            if result and self._is_valid_response(result):
                return result
        except Exception as e:
            log_error("full_agent_failed", str(e))
        
        # 第二层：纯LLM模式（无工具，只基于知识库）
        try:
            result = self._try_llm_only(user_input, user_context)
            if result and self._is_valid_response(result):
                return result
        except Exception as e:
            log_error("llm_only_failed", str(e))
        
        # 第三层：基于规则的关键词匹配
        try:
            result = self._try_rule_based(user_input)
            if result:
                return result
        except Exception as e:
            log_error("rule_based_failed", str(e))
        
        # 第四层：万能兜底（转人工或标准话术）
        return self._final_fallback(user_input, user_context)
    
    def _try_full_agent(self, input_: str, context: Dict) -> str:
        """尝试完整Agent"""
        return agent_executor.invoke({"input": input_, "context": context}).get("output")
    
    def _try_llm_only(self, input_: str, context: Dict) -> str:
        """纯LLM模式（检索知识库回答）"""
        docs = vector_store.similarity_search(input_, k=3)
        context_str = "\n".join([d.page_content for d in docs])
        return llm.invoke(f"基于以下知识回答用户问题：\n{context_str}\n\n问题：{input_}")
    
    def _try_rule_based(self, input_: str) -> str:
        """基于规则的快速匹配"""
        # 简单关键词匹配
        keywords_map = {
            "物流|快递|运输": "您好，物流信息请在【订单详情】页面查询，或联系物流公司。",
            "退货|退款|售后": "您好，售后问题请点击【联系客服】转人工处理。",
            "活动|促销|优惠": "您好，当前促销活动请查看首页活动专区。",
        }
        for pattern, response in keywords_map.items():
            if re.search(pattern, input_):
                return response
        return None
    
    def _final_fallback(self, input_: str, context: Dict) -> str:
        """最终兜底"""
        # 如果是工作时间转人工，否则返回话术
        if self._is_business_hours():
            return "您好，您的问题已转接人工客服，请稍候..."
        return "您好，客服已下班，请留言或在工作时间（9:00-18:00）联系我们。"

4.2 Agent状态机与异常流转

为Agent引入状态管理，在异常发生时可以回退到安全状态：

from enum import Enum

class AgentStatus(Enum):
    IDLE = "idle"
    UNDERSTANDING = "understanding"
    CALLING_TOOL = "calling_tool"
    GENERATING = "generating"
    COMPLETED = "completed"
    ERROR = "error"
    FALLBACK = "fallback"

class AgentStateMachine:
    """Agent状态机"""
    
    def __init__(self):
        self.status = AgentStatus.IDLE
        self.last_status = None
        self.error_count = 0
        self.max_errors = 3
    
    def transition(self, new_status: AgentStatus):
        self.last_status = self.status
        self.status = new_status
        print(f"[状态切换] {self.last_status} -> {new_status}")
    
    def handle_error(self, error: Exception) -> AgentStatus:
        """处理错误并决定下一步状态"""
        self.error_count += 1
        
        if self.error_count >= self.max_errors:
            return AgentStatus.FALLBACK
        
        if self.status == AgentStatus.UNDERSTANDING:
            return AgentStatus.UNDERSTANDING  # 重试理解
        elif self.status == AgentStatus.CALLING_TOOL:
            return AgentStatus.FALLBACK  # 工具调用失败直接降级
        else:
            return AgentStatus.ERROR
    
    def reset(self):
        """重置状态机"""
        self.status = AgentStatus.IDLE
        self.error_count = 0

---

五、监控与告警：提前发现"翻车"苗头

最后，用Prometheus埋点来量化Agent的"健康状况"：

from prometheus_client import Counter, Histogram, Gauge

# 定义告警指标
AGENT_TOTAL = Counter('agent_requests_total', '总请求', ['status'])
AGENT_DURATION = Histogram('agent_duration_seconds', 'Agent耗时', ['model'])
AGENT_FALLBACK = Counter('agent_fallback_total', '降级触发次数', ['fallback_level'])
AGENT_HALLUCINATION = Counter('agent_hallucination_total', '幻觉检测次数')
AGENT_LLM_ERROR = Counter('agent_llm_error_total', 'LLM错误次数', ['error_type'])

# 活跃会话数
ACTIVE_SESSIONS = Gauge('agent_active_sessions', '当前活跃会话数')

配合Grafana设置告警规则：

告警条件	阈值	严重级别
降级率 > 10%	5分钟内	P1
LLM超时率 > 5%	5分钟内	P2
幻觉检测 > 3次	1小时内	P2
Agent成功率 < 90%	5分钟内	P1

---

六、总结

回到最初的问题：当大模型"翻车"时怎么办？

防护层级	策略	目标
调用层	重试 + 超时 + 多模型切换	确保LLM服务可用
质量层	幻觉检测 + 格式校验 + 护栏拦截	确保输出可靠
业务层	多级降级 + 状态回退	确保业务不中断
观测层	指标监控 + 告警规则	提前发现问题

记住一个原则：把LLM当成一种"概率性可靠"的服务来对待——它大概率能正确工作，但小概率会出错。与其期待它永远不出错，不如建立一套让它"即使犯错，也能被拦截"的防御体系。