AI Agent 错误处理：从工具调用失败到 LLM 幻觉的防御性设计

一、Agent 崩溃的 N 种方式：不只是 API 超时

AI Agent 的错误场景远比传统软件复杂。传统软件的错误主要来自网络超时、数据校验失败、资源不足等确定性原因。但 Agent 的错误来源多了一层不确定性——LLM 本身的输出不可控。

一个典型的工作流 Agent 执行路径：接收用户指令 → LLM 规划步骤 → 调用工具 A → 解析返回结果 → 调用工具 B → 生成最终回复。这条链路上至少有 5 个可能出错的节点：

LLM 输出格式错误。 要求返回 JSON，模型返回了带注释的 JSON 或格式错误的 JSON。解析失败，后续流程中断。

工具调用参数错误。 LLM 生成的函数参数类型不匹配或缺少必填参数，工具执行直接报错。

工具执行环境错误。 数据库连接断开、第三方 API 限流、文件权限不足——这些是传统意义上的运行时错误。

LLM 幻觉。 模型编造了不存在的工具名称、捏造了不存在的参数值、或者对工具返回结果做了错误解读。这类错误不会抛异常，但会导致业务逻辑错误。

死循环。 LLM 反复调用同一个工具、反复重试同一个失败的操作，消耗大量 Token 和时间而不产生有效进展。

如果 Agent 没有针对这些错误场景的防御性设计，任何一个节点的失败都会导致整个工作流崩溃，且错误信息对用户毫无意义。

二、Agent 错误分类与防御策略

graph TB
    subgraph 错误分类
        A[LLM 输出错误] --> A1[格式错误]
        A --> A2[幻觉输出]
        A --> A3[拒绝回答]
        B[工具调用错误] --> B1[参数校验失败]
        B --> B2[执行超时]
        B --> B3[环境异常]
        C[流程控制错误] --> C1[死循环]
        C --> C2[步骤超限]
        C --> C3[上下文溢出]
    end

    subgraph 防御策略
        D[输出校验 + 重试] --> A1
        E[结果交叉验证] --> A2
        F[降级回复] --> A3
        G[参数 Schema 校验] --> B1
        H[超时 + 重试] --> B2
        I[熔断 + 降级] --> B3
        J[步骤计数 + 强制终止] --> C1
        K[最大步数限制] --> C2
        L[上下文预算管理] --> C3
    end

三、生产级 Agent 错误处理框架

3.1 LLM 输出校验与自动修复

import json
import re
from dataclasses import dataclass
from typing import Optional, Any

@dataclass
class ParsedOutput:
    """LLM 输出解析结果"""
    success: bool
    data: Optional[dict] = None
    error: Optional[str] = None
    raw_output: str = ""

class LLMOutputParser:
    """LLM 输出解析器：带自动修复能力

    设计思路：
    - 优先严格解析，失败后尝试修复常见格式问题
    - 修复仍失败则要求 LLM 重新生成
    - 最多重试 2 次，避免无限循环
    """

    def __init__(self, llm_client=None, max_retries: int = 2):
        self.llm_client = llm_client
        self.max_retries = max_retries

    def parse_json(self, raw: str) -> ParsedOutput:
        """解析 LLM 输出为 JSON，带自动修复"""
        # 第一步：直接解析
        data = self._try_parse_json(raw)
        if data is not None:
            return ParsedOutput(success=True, data=data, raw_output=raw)

        # 第二步：尝试提取 JSON 块（模型可能包裹在 ```json ... ``` 中）
        extracted = self._extract_json_block(raw)
        if extracted:
            data = self._try_parse_json(extracted)
            if data is not None:
                return ParsedOutput(success=True, data=data, raw_output=raw)

        # 第三步：尝试修复常见格式问题
        fixed = self._try_fix_json(raw)
        if fixed:
            data = self._try_parse_json(fixed)
            if data is not None:
                return ParsedOutput(success=True, data=data, raw_output=raw)

        return ParsedOutput(
            success=False,
            error=f"JSON 解析失败，原始输出: {raw[:200]}",
            raw_output=raw,
        )

    def _try_parse_json(self, text: str) -> Optional[dict]:
        """尝试解析 JSON"""
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return None

    def _extract_json_block(self, text: str) -> Optional[str]:
        """从 Markdown 代码块中提取 JSON"""
        pattern = r'```(?:json)?\s*\n?(.*?)\n?```'
        match = re.search(pattern, text, re.DOTALL)
        if match:
            return match.group(1).strip()
        return None

    def _try_fix_json(self, text: str) -> Optional[str]:
        """修复常见的 JSON 格式问题"""
        # 移除行内注释：// comment
        fixed = re.sub(r'//.*?$', '', text, flags=re.MULTILINE)
        # 移除尾随逗号：, } 或 , ]
        fixed = re.sub(r',\s*([}\]])', r'\1', fixed)
        # 修复单引号为双引号
        fixed = fixed.replace("'", '"')
        return fixed if fixed != text else None

    def parse_with_retry(
        self, prompt: str, schema: dict = None
    ) -> ParsedOutput:
        """解析 LLM 输出，失败时自动重试

        重试时将错误信息反馈给 LLM，让它修正输出
        """
        for attempt in range(self.max_retries + 1):
            if attempt > 0 and self.llm_client:
                # 重试时将错误信息反馈给模型
                retry_prompt = (
                    f"上次的输出格式有误，请修正后重新输出。\n"
                    f"原始提示：{prompt}\n"
                    f"要求：严格输出 JSON 格式，不要添加注释或额外文本"
                )
                response = self.llm_client.chat.completions.create(
                    model="gpt-4o",
                    messages=[{"role": "user", "content": retry_prompt}],
                    max_tokens=1000,
                    temperature=0,
                    timeout=15,
                )
                raw = response.choices[0].message.content
            else:
                # 首次调用由外部完成，这里只解析
                raw = prompt

            result = self.parse_json(raw)
            if result.success:
                # 如果有 Schema，校验字段完整性
                if schema:
                    missing = self._validate_schema(result.data, schema)
                    if missing:
                        result.success = False
                        result.error = f"缺少必填字段: {missing}"
                        continue
                return result

        return result

3.2 工具调用错误处理与熔断

from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Optional
import time

class CircuitState(Enum):
    CLOSED = "closed"       # 正常状态
    OPEN = "open"           # 熔断状态：直接拒绝请求
    HALF_OPEN = "half_open" # 半开状态：允许少量请求探测

@dataclass
class CircuitBreaker:
    """熔断器：防止对故障工具的持续调用

    设计思路：
    - 连续失败 N 次后进入 OPEN 状态，直接拒绝请求
    - 冷却期后进入 HALF_OPEN 状态，允许 1 次探测
    - 探测成功则恢复 CLOSED，失败则重回 OPEN
    """
    name: str
    failure_threshold: int = 5      # 连续失败阈值
    recovery_timeout: float = 30.0  # 冷却期（秒）
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0

    def can_execute(self) -> bool:
        """判断是否允许执行"""
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            # 冷却期已过，进入半开状态
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        if self.state == CircuitState.HALF_OPEN:
            return True  # 半开状态允许 1 次探测
        return False

    def record_success(self):
        """记录成功：重置计数器"""
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self):
        """记录失败：累加计数，超阈值则熔断"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN


@dataclass
class ToolCallResult:
    """工具调用结果"""
    success: bool
    data: Any = None
    error: Optional[str] = None
    tool_name: str = ""
    execution_time: float = 0.0

class ToolExecutor:
    """工具执行器：带超时、重试和熔断

    设计思路：
    - 每个工具独立熔断，一个工具故障不影响其他工具
    - 超时保护：防止工具执行挂起
    - 参数预校验：在调用工具前校验参数，避免无效调用
    """

    def __init__(self, timeout: float = 10.0, max_retries: int = 2):
        self.timeout = timeout
        self.max_retries = max_retries
        self.circuit_breakers: dict[str, CircuitBreaker] = {}

    def execute(
        self,
        tool_name: str,
        tool_fn: Callable,
        params: dict,
        param_schema: dict = None,
    ) -> ToolCallResult:
        """执行工具调用，带完整的错误处理链"""

        # 第一步：参数预校验
        if param_schema:
            validation_error = self._validate_params(params, param_schema)
            if validation_error:
                return ToolCallResult(
                    success=False,
                    error=f"参数校验失败: {validation_error}",
                    tool_name=tool_name,
                )

        # 第二步：熔断检查
        cb = self._get_circuit_breaker(tool_name)
        if not cb.can_execute():
            return ToolCallResult(
                success=False,
                error=f"工具 {tool_name} 已熔断，请稍后重试",
                tool_name=tool_name,
            )

        # 第三步：带重试的执行
        last_error = None
        for attempt in range(self.max_retries + 1):
            start_time = time.time()
            try:
                result = tool_fn(**params)
                execution_time = time.time() - start_time

                cb.record_success()
                return ToolCallResult(
                    success=True,
                    data=result,
                    tool_name=tool_name,
                    execution_time=execution_time,
                )
            except Exception as e:
                execution_time = time.time() - start_time
                last_error = str(e)
                cb.record_failure()

        return ToolCallResult(
            success=False,
            error=f"工具 {tool_name} 执行失败（重试 {self.max_retries} 次）: {last_error}",
            tool_name=tool_name,
            execution_time=time.time() - start_time,
        )

    def _get_circuit_breaker(self, tool_name: str) -> CircuitBreaker:
        """获取或创建工具的熔断器"""
        if tool_name not in self.circuit_breakers:
            self.circuit_breakers[tool_name] = CircuitBreaker(name=tool_name)
        return self.circuit_breakers[tool_name]

    def _validate_params(self, params: dict, schema: dict) -> Optional[str]:
        """校验参数是否符合 Schema"""
        required = schema.get("required", [])
        for field_name in required:
            if field_name not in params:
                return f"缺少必填参数: {field_name}"

        properties = schema.get("properties", {})
        for key, value in params.items():
            if key not in properties:
                return f"未知参数: {key}"
            expected_type = properties[key].get("type")
            if expected_type and not self._check_type(value, expected_type):
                return f"参数 {key} 类型错误: 期望 {expected_type}"

        return None

    def _check_type(self, value: Any, expected_type: str) -> bool:
        """检查值类型"""
        type_map = {
            "string": str,
            "integer": int,
            "number": (int, float),
            "boolean": bool,
            "array": list,
            "object": dict,
        }
        expected = type_map.get(expected_type)
        if expected is None:
            return True
        return isinstance(value, expected)

3.3 Agent 执行循环的防护机制

class AgentExecutor:
    """Agent 执行器：带步骤限制和死循环检测

    设计思路：
    - 最大步数限制：防止 Agent 无限执行
    - 重复检测：检测 Agent 是否在重复相同的操作
    - 错误累积：连续错误超过阈值时终止执行
    """

    def __init__(
        self,
        max_steps: int = 15,
        max_consecutive_errors: int = 3,
        repeat_threshold: int = 3,
    ):
        self.max_steps = max_steps
        self.max_consecutive_errors = max_consecutive_errors
        self.repeat_threshold = repeat_threshold

    def execute(self, agent, user_input: str) -> dict:
        """执行 Agent 循环，带防护机制"""
        step_count = 0
        consecutive_errors = 0
        action_history = []  # 记录动作历史，用于检测重复
        result = {"success": False, "steps": [], "error": None}

        while step_count < self.max_steps:
            step_count += 1

            # 让 Agent 决定下一步动作
            action = agent.decide_next_action(user_input, result["steps"])

            # 检测死循环：连续执行相同动作
            action_key = f"{action.tool_name}:{json.dumps(action.params, sort_keys=True)}"
            action_history.append(action_key)
            if self._is_repeating(action_history):
                result["error"] = (
                    f"检测到重复操作: {action.tool_name}，"
                    f"可能陷入死循环，强制终止"
                )
                break

            # 执行动作
            tool_result = agent.execute_action(action)

            step_record = {
                "step": step_count,
                "action": action.tool_name,
                "success": tool_result.success,
                "error": tool_result.error,
            }
            result["steps"].append(step_record)

            if tool_result.success:
                consecutive_errors = 0
                # 检查 Agent 是否认为任务完成
                if agent.is_task_complete():
                    result["success"] = True
                    result["final_answer"] = agent.generate_final_answer()
                    break
            else:
                consecutive_errors += 1
                if consecutive_errors >= self.max_consecutive_errors:
                    result["error"] = (
                        f"连续 {consecutive_errors} 步执行失败，"
                        f"最后一次错误: {tool_result.error}"
                    )
                    break

        else:
            # 步数超限
            result["error"] = f"执行步数超过 {self.max_steps} 步限制"

        result["total_steps"] = step_count
        return result

    def _is_repeating(self, history: list[str]) -> bool:
        """检测是否在重复相同操作"""
        if len(history) < self.repeat_threshold:
            return False
        # 检查最近 N 次操作是否完全相同
        recent = history[-self.repeat_threshold:]
        return len(set(recent)) == 1

四、防御性设计的代价与适用边界

重试机制可能放大故障。 当下游服务过载时，Agent 的重试会加重下游压力。必须配合熔断器使用——熔断打开后不再重试，直接降级。

输出校验增加延迟。 JSON 解析、Schema 校验、重试生成都会增加单次调用的延迟。对于实时性要求极高的场景（如语音助手），需要权衡校验深度和响应速度。

死循环检测的误判。 某些合法场景需要多次调用同一工具（如分页查询），重复检测可能误判为死循环。对策：在动作中附加"调用原因"字段，只有原因也相同时才判定为重复。

适用场景： 所有生产级 Agent 系统，尤其是涉及工具调用和多步骤工作流的 Agent。不适用场景： 单轮问答、纯文本生成——这些场景的错误类型简单，不需要复杂的防御框架。

五、总结

AI Agent 的错误处理必须覆盖三个层面：LLM 输出的格式校验与自动修复、工具调用的超时重试与熔断、执行循环的步数限制与死循环检测。每一层防御都有其代价——重试增加延迟、熔断导致降级、步数限制可能中断合法操作。防御性设计的目标不是消除所有错误，而是在错误发生时保证系统不崩溃、用户有反馈、问题可追溯。生产级 Agent 的可靠性不是靠 LLM 的能力保证的，而是靠工程化的错误处理框架保证的。