一、前言

在 Demo 阶段，AI Agent 的异常处理往往很简单：try/except 包一层，打条日志就完事。

但一旦进入企业级生产环境，问题会迅速暴露：

• 模型限流了，该不该重试？
• 上下文超长，重试 3 次有意义吗？
• 工具调用失败，是参数问题还是下游服务挂了？
• 用户输入触发了 Prompt 注入，怎么和普通业务异常区分？
• 告警应该发给算法团队、平台团队还是安全团队？

通用异常不够用，Agent 需要一套专属异常体系。

本文将从设计原则、分层架构到完整代码，手把手搭建一套可直接落地企业级 AI Agent 的异常体系。

---

二、为什么通用异常不够用？

先看几个生产环境常见场景：

场景	通用异常	问题
模型返回 429	HTTPError	不知道是否该重试、等多久
上下文超长	BadRequestError	容易被无脑重试，浪费 token
工具超时	TimeoutError	无法定位是哪个工具、什么参数
Prompt 注入	ValueError	安全团队无法单独路由告警
达到最大迭代	RuntimeError	无法判断是规划问题还是模型问题

核心痛点：异常缺少「归因 + 可恢复性 + 上下文 + 降级策略」。

---

三、设计原则

原则	说明
分层归因	异常必须能定位到「基础设施 / 模型 / 工具 / 记忆 / 规划 / 安全 / 业务」中的具体层
可恢复性标记	retryable 决定是否重试，避免对「上下文超长」这类错误做无意义重试
上下文自携带	trace_id、session_id、tool_name、token 用量等随异常一起抛出，便于排障
可降级	每个异常携带 severity + fallback 能力，支撑熔断/降级
用户态/开发态分离	message 给开发看，user_message 给终端用户看，防止内部信息泄露
可序列化	to_dict() 直接对接日志、告警、API 响应

---

四、异常体系整体架构

AgentError                       ← 全局基类
├── InfrastructureError          ← 网络/存储/限流
│   ├── NetworkError
│   ├── UpstreamTimeoutError
│   ├── RateLimitError
│   └── StorageError
├── ModelError                   ← LLM 调用
│   ├── ModelAPIError
│   ├── ModelTimeoutError
│   ├── ContextLengthExceededError
│   ├── ContentFilterError
│   ├── TokenQuotaExceededError
│   └── InvalidModelResponseError
├── ToolError                    ← 工具/函数调用
│   ├── ToolNotFoundError
│   ├── ToolExecutionError
│   ├── ToolTimeoutError
│   ├── ToolValidationError
│   └── ToolPermissionError
├── RetrievalError               ← 记忆/向量检索
│   ├── VectorStoreError
│   ├── EmbeddingError
│   └── MemoryOverflowError
├── PlanningError                ← 规划/循环控制
│   ├── PlanGenerationError
│   ├── MaxIterationExceededError
│   └── LoopDetectedError
├── SecurityError                ← 安全合规
│   ├── PromptInjectionDetectedError
│   ├── DataLeakageError
│   └── UnauthorizedAccessError
└── BusinessError                ← 业务语义
    ├── ValidationError
    └── WorkflowError

---

五、完整代码实现

目录结构建议：

agent/
└── errors/
    ├── __init__.py
    ├── base.py
    ├── codes.py
    ├── context.py
    ├── handler.py
    ├── layers.py
    └── translator.py

5.1 错误码与枚举

# agent/errors/codes.py
from __future__ import annotations
import enum


class ErrorCode(str, enum.Enum):
    # ---- 基础设施 AGENT-INFRA-xxx ----
    INFRA_NETWORK        = "AGENT-INFRA-001"
    INFRA_TIMEOUT        = "AGENT-INFRA-002"
    INFRA_RATE_LIMIT     = "AGENT-INFRA-003"
    INFRA_STORAGE        = "AGENT-INFRA-004"

    # ---- 模型 AGENT-MODEL-xxx ----
    MODEL_API            = "AGENT-MODEL-001"
    MODEL_TIMEOUT        = "AGENT-MODEL-002"
    MODEL_CTX_EXCEEDED   = "AGENT-MODEL-003"
    MODEL_CONTENT_FILTER = "AGENT-MODEL-004"
    MODEL_QUOTA_EXCEEDED = "AGENT-MODEL-005"
    MODEL_INVALID_RESP   = "AGENT-MODEL-006"

    # ---- 工具 AGENT-TOOL-xxx ----
    TOOL_NOT_FOUND       = "AGENT-TOOL-001"
    TOOL_EXECUTION       = "AGENT-TOOL-002"
    TOOL_TIMEOUT         = "AGENT-TOOL-003"
    TOOL_VALIDATION      = "AGENT-TOOL-004"
    TOOL_PERMISSION      = "AGENT-TOOL-005"

    # ---- 记忆/检索 AGENT-MEM-xxx ----
    MEM_VECTOR_STORE     = "AGENT-MEM-001"
    MEM_EMBEDDING        = "AGENT-MEM-002"
    MEM_OVERFLOW         = "AGENT-MEM-003"

    # ---- 规划 AGENT-PLAN-xxx ----
    PLAN_GENERATION      = "AGENT-PLAN-001"
    PLAN_MAX_ITERATION   = "AGENT-PLAN-002"
    PLAN_LOOP_DETECTED   = "AGENT-PLAN-003"

    # ---- 安全 AGENT-SEC-xxx ----
    SEC_PROMPT_INJECTION = "AGENT-SEC-001"
    SEC_DATA_LEAKAGE     = "AGENT-SEC-002"
    SEC_UNAUTHORIZED     = "AGENT-SEC-003"

    # ---- 业务 AGENT-BIZ-xxx ----
    BIZ_VALIDATION       = "AGENT-BIZ-001"
    BIZ_WORKFLOW         = "AGENT-BIZ-002"


class Severity(str, enum.Enum):
    LOW      = "low"
    MEDIUM   = "medium"
    HIGH     = "high"
    CRITICAL = "critical"

5.2 上下文传播（Trace / Session）

# agent/errors/context.py
from contextvars import ContextVar

trace_id_var:   ContextVar[str] = ContextVar("trace_id", default="")
session_id_var: ContextVar[str] = ContextVar("session_id", default="")
tenant_id_var:  ContextVar[str] = ContextVar("tenant_id", default="")

5.3 异常基类

# agent/errors/base.py
from __future__ import annotations
from typing import Any, Optional
from .codes import ErrorCode, Severity
from .context import trace_id_var, session_id_var, tenant_id_var


class AgentError(Exception):
    """所有 Agent 异常的基类。"""

    code: ErrorCode       = ErrorCode.BIZ_WORKFLOW
    retryable: bool       = False
    severity: Severity    = Severity.MEDIUM
    http_status: int      = 500

    def __init__(
        self,
        message: str,
        *,
        context: Optional[dict[str, Any]] = None,
        user_message: Optional[str] = None,
        cause: Optional[BaseException] = None,
        retryable: Optional[bool] = None,
        severity: Optional[Severity] = None,
    ) -> None:
        super().__init__(message)
        self.message       = message
        self.user_message  = user_message or "系统繁忙，请稍后重试"

        # 自动附带链路上下文
        self.context: dict[str, Any] = {
            "trace_id":   trace_id_var.get(),
            "session_id": session_id_var.get(),
            "tenant_id":  tenant_id_var.get(),
            **(context or {}),
        }

        if retryable is not None:
            self.retryable = retryable
        if severity is not None:
            self.severity = severity
        if cause is not None:
            self.__cause__ = cause

    def to_dict(self) -> dict[str, Any]:
        return {
            "type":         type(self).__name__,
            "code":         self.code.value,
            "message":      self.message,
            "user_message": self.user_message,
            "retryable":    self.retryable,
            "severity":     self.severity.value,
            "http_status":  self.http_status,
            "context":      self.context,
            "cause":        repr(self.__cause__) if self.__cause__ else None,
        }

    def __str__(self) -> str:
        return f"[{self.code.value}] {self.message} | ctx={self.context}"

5.4 各层异常

# agent/errors/layers.py
from __future__ import annotations
from typing import Any, Optional
from .base import AgentError
from .codes import ErrorCode, Severity


# ============ 基础设施 ============
class InfrastructureError(AgentError):
    code = ErrorCode.INFRA_NETWORK


class NetworkError(InfrastructureError):
    retryable = True


class UpstreamTimeoutError(InfrastructureError):
    code = ErrorCode.INFRA_TIMEOUT
    retryable = True
    http_status = 504


class RateLimitError(InfrastructureError):
    code = ErrorCode.INFRA_RATE_LIMIT
    retryable = True
    http_status = 429

    def __init__(
        self,
        message: str,
        *,
        retry_after: Optional[float] = None,
        model: Optional[str] = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.retry_after = retry_after
        if retry_after is not None:
            self.context["retry_after"] = retry_after
        if model:
            self.context["model"] = model


class StorageError(InfrastructureError):
    code = ErrorCode.INFRA_STORAGE
    retryable = True


# ============ 模型 ============
class ModelError(AgentError):
    code = ErrorCode.MODEL_API

    def __init__(self, message: str, *, model: Optional[str] = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        if model:
            self.context["model"] = model


class ModelAPIError(ModelError):
    retryable = True
    http_status = 502


class ModelTimeoutError(ModelError):
    code = ErrorCode.MODEL_TIMEOUT
    retryable = True
    http_status = 504


class ContextLengthExceededError(ModelError):
    code = ErrorCode.MODEL_CTX_EXCEEDED
    retryable = False            # 重试无意义，必须走压缩/截断
    http_status = 400

    def __init__(
        self,
        message: str,
        *,
        max_tokens: Optional[int] = None,
        actual_tokens: Optional[int] = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.context.update({"max_tokens": max_tokens, "actual_tokens": actual_tokens})


class ContentFilterError(ModelError):
    code = ErrorCode.MODEL_CONTENT_FILTER
    severity = Severity.HIGH
    retryable = False
    http_status = 422


class TokenQuotaExceededError(ModelError):
    code = ErrorCode.MODEL_QUOTA_EXCEEDED
    severity = Severity.HIGH
    retryable = False
    http_status = 402

    def __init__(
        self,
        message: str,
        *,
        used: Optional[int] = None,
        quota: Optional[int] = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, **kw)
        self.context.update({"used": used, "quota": quota})


class InvalidModelResponseError(ModelError):
    """模型返回结构不可解析（JSON 截断、tool_call 缺失等）。"""
    code = ErrorCode.MODEL_INVALID_RESP
    retryable = True


# ============ 工具 ============
class ToolError(AgentError):
    code = ErrorCode.TOOL_EXECUTION

    def __init__(self, message: str, *, tool_name: str, **kw: Any) -> None:
        ctx = kw.pop("context", {}) or {}
        ctx["tool_name"] = tool_name
        super().__init__(message, context=ctx, **kw)
        self.tool_name = tool_name


class ToolNotFoundError(ToolError):
    code = ErrorCode.TOOL_NOT_FOUND
    http_status = 404


class ToolExecutionError(ToolError):
    retryable = True

    def __init__(
        self,
        message: str,
        *,
        tool_name: str,
        arguments: Optional[dict] = None,
        **kw: Any,
    ) -> None:
        super().__init__(message, tool_name=tool_name, **kw)
        if arguments is not None:
            # ⚠️ 注意脱敏：不要把密钥/token 写进 context
            self.context["arguments"] = arguments


class ToolTimeoutError(ToolError):
    code = ErrorCode.TOOL_TIMEOUT
    retryable = True
    http_status = 504


class ToolValidationError(ToolError):
    code = ErrorCode.TOOL_VALIDATION
    retryable = False
    http_status = 400


class ToolPermissionError(ToolError):
    code = ErrorCode.TOOL_PERMISSION
    severity = Severity.HIGH
    retryable = False
    http_status = 403


# ============ 记忆/检索 ============
class RetrievalError(AgentError):
    code = ErrorCode.MEM_VECTOR_STORE


class VectorStoreError(RetrievalError):
    retryable = True


class EmbeddingError(RetrievalError):
    code = ErrorCode.MEM_EMBEDDING
    retryable = True


class MemoryOverflowError(RetrievalError):
    code = ErrorCode.MEM_OVERFLOW
    retryable = False
    severity = Severity.HIGH


# ============ 规划 ============
class PlanningError(AgentError):
    code = ErrorCode.PLAN_GENERATION


class PlanGenerationError(PlanningError):
    retryable = True


class MaxIterationExceededError(PlanningError):
    code = ErrorCode.PLAN_MAX_ITERATION
    retryable = False
    severity = Severity.HIGH

    def __init__(self, message: str, *, max_iter: int, **kw: Any) -> None:
        super().__init__(message, **kw)
        self.context["max_iter"] = max_iter


class LoopDetectedError(PlanningError):
    code = ErrorCode.PLAN_LOOP_DETECTED
    retryable = False
    severity = Severity.HIGH


# ============ 安全 ============
class SecurityError(AgentError):
    code = ErrorCode.SEC_UNAUTHORIZED
    severity = Severity.CRITICAL
    retryable = False
    http_status = 403


class PromptInjectionDetectedError(SecurityError):
    code = ErrorCode.SEC_PROMPT_INJECTION


class DataLeakageError(SecurityError):
    code = ErrorCode.SEC_DATA_LEAKAGE


class UnauthorizedAccessError(SecurityError):
    code = ErrorCode.SEC_UNAUTHORIZED
    http_status = 401


# ============ 业务 ============
class BusinessError(AgentError):
    code = ErrorCode.BIZ_WORKFLOW
    severity = Severity.LOW
    http_status = 400


class ValidationError(BusinessError):
    code = ErrorCode.BIZ_VALIDATION

    def __init__(self, message: str, *, field: Optional[str] = None, **kw: Any) -> None:
        super().__init__(message, **kw)
        if field:
            self.context["field"] = field


class WorkflowError(BusinessError):
    code = ErrorCode.BIZ_WORKFLOW

5.5 异常转换器（把三方 SDK 异常归一化）

# agent/errors/translator.py
from __future__ import annotations
from typing import Any
from .layers import (
    ModelAPIError,
    ModelTimeoutError,
    RateLimitError,
    ContextLengthExceededError,
    ToolExecutionError,
    ToolTimeoutError,
)


def translate_openai_error(exc: BaseException, *, model: str | None = None) -> Exception:
    """把 OpenAI SDK 的异常翻译成 Agent 自己的异常体系。"""
    name = type(exc).__name__
    msg = str(exc)

    if name == "RateLimitError":
        return RateLimitError(f"模型限流: {msg}", model=model, cause=exc)

    if name in ("APITimeoutError", "Timeout"):
        return ModelTimeoutError(f"模型调用超时: {msg}", model=model, cause=exc)

    if name == "BadRequestError" and "context_length" in msg:
        return ContextLengthExceededError(f"上下文超长: {msg}", model=model, cause=exc)

    if name == "APIError":
        return ModelAPIError(f"模型服务异常: {msg}", model=model, cause=exc)

    return exc


def translate_tool_error(
    exc: BaseException,
    *,
    tool_name: str,
    arguments: dict[str, Any] | None = None,
) -> Exception:
    name = type(exc).__name__
    if name in ("TimeoutError", "asyncio.TimeoutError"):
        return ToolTimeoutError(
            "工具执行超时",
            tool_name=tool_name,
            arguments=arguments,
            cause=exc,
        )
    return ToolExecutionError(
        f"工具执行失败: {exc}",
        tool_name=tool_name,
        arguments=arguments,
        cause=exc,
    )

5.6 统一处理装饰器（重试 + 降级 + 上报）

# agent/errors/handler.py
from __future__ import annotations
import asyncio
import functools
import inspect
import logging
from typing import Any, Awaitable, Callable, Optional, TypeVar

from .base import AgentError
from .layers import RateLimitError

logger = logging.getLogger("agent.errors")
T = TypeVar("T")


def with_error_handling(
    *,
    retries: int = 3,
    backoff: float = 0.5,
    fallback: Optional[Callable[..., Awaitable[Any] | Any]] = None,
    wrap_as: Optional[type[AgentError]] = None,
):
    """
    统一异常处理：
      - AgentError 且 retryable=True → 指数退避重试
      - 其它未捕获异常 → 包装成 AgentError（可选 wrap_as）
      - 重试耗尽 → 调用 fallback（降级），否则抛出
    """

    def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exc: BaseException | None = None

            for attempt in range(1, retries + 1):
                try:
                    return await func(*args, **kwargs)

                except AgentError as e:
                    last_exc = e
                    logger.warning(
                        "agent_error attempt=%d/%d code=%s retryable=%s ctx=%s",
                        attempt, retries, e.code.value, e.retryable, e.context,
                    )

                    if not e.retryable or attempt == retries:
                        break

                    delay = backoff * (2 ** (attempt - 1))
                    if isinstance(e, RateLimitError) and e.retry_after:
                        delay = max(delay, e.retry_after)
                    await asyncio.sleep(delay)

                except Exception as e:  # 未预期异常
                    logger.exception("unexpected_error in %s", func.__name__)
                    wrapped_cls = wrap_as or AgentError
                    last_exc = wrapped_cls(str(e), cause=e)
                    break

            if fallback is not None:
                logger.info("using fallback for %s", func.__name__)
                result = fallback(*args, **kwargs)
                if inspect.isawaitable(result):
                    result = await result
                return result

            assert last_exc is not None
            raise last_exc

        return wrapper

    return decorator

5.7 导出模块

# agent/errors/__init__.py
from .base import AgentError
from .codes import ErrorCode, Severity
from .context import trace_id_var, session_id_var, tenant_id_var
from .handler import with_error_handling
from .translator import translate_openai_error, translate_tool_error
from .layers import (
    InfrastructureError, NetworkError, UpstreamTimeoutError, RateLimitError, StorageError,
    ModelError, ModelAPIError, ModelTimeoutError, ContextLengthExceededError,
    ContentFilterError, TokenQuotaExceededError, InvalidModelResponseError,
    ToolError, ToolNotFoundError, ToolExecutionError, ToolTimeoutError,
    ToolValidationError, ToolPermissionError,
    RetrievalError, VectorStoreError, EmbeddingError, MemoryOverflowError,
    PlanningError, PlanGenerationError, MaxIterationExceededError, LoopDetectedError,
    SecurityError, PromptInjectionDetectedError, DataLeakageError, UnauthorizedAccessError,
    BusinessError, ValidationError, WorkflowError,
)

---

六、使用示例

6.1 LLM 调用

import openai
from agent.errors import with_error_handling, translate_openai_error

openai_client = openai.AsyncOpenAI()


@with_error_handling(
    retries=3,
    backoff=0.5,
    fallback=lambda *a, **kw: {"content": "服务暂时不可用，请稍后重试"},
)
async def call_llm(prompt: str, *, model: str = "gpt-4o") -> dict:
    try:
        return await openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
    except openai.OpenAIError as e:
        raise translate_openai_error(e, model=model)

6.2 工具执行

from agent.errors import ToolNotFoundError, translate_tool_error

TOOL_REGISTRY: dict = {}


async def run_tool(name: str, args: dict) -> any:
    if name not in TOOL_REGISTRY:
        raise ToolNotFoundError("工具不存在", tool_name=name)

    try:
        return await TOOL_REGISTRY[name](**args)
    except Exception as e:
        raise translate_tool_error(e, tool_name=name, arguments=args)

6.3 Agent 主循环

from agent.errors import LoopDetectedError, MaxIterationExceededError


async def agent_loop(goal: str, max_iter: int = 15):
    seen: set[str] = set()

    for i in range(max_iter):
        step = await plan(goal)
        sig = f"{step.tool}:{step.args}"

        if sig in seen:
            raise LoopDetectedError(
                "检测到重复调用循环",
                context={"step": sig, "iteration": i},
                user_message="任务进入循环，已自动中止",
            )

        seen.add(sig)

    raise MaxIterationExceededError(
        "超过最大迭代次数",
        max_iter=max_iter,
        user_message="任务过于复杂，已中止",
    )

6.4 FastAPI 全局异常处理

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from agent.errors import AgentError, SecurityError

app = FastAPI()


@app.exception_handler(AgentError)
async def agent_error_handler(request: Request, exc: AgentError):
    if isinstance(exc, SecurityError):
        await alert(exc.to_dict())

    return JSONResponse(
        status_code=exc.http_status,
        content={
            "code":         exc.code.value,
            "user_message": exc.user_message,
            "trace_id":     exc.context.get("trace_id"),
        },
    )

6.5 入口注入 Trace

from uuid import uuid4
from agent.errors.context import trace_id_var, session_id_var, tenant_id_var


async def handle_request(req):
    trace_id_var.set(req.headers.get("X-Trace-Id") or uuid4().hex)
    session_id_var.set(req.session_id)
    tenant_id_var.set(req.tenant_id)
    ...

---

七、异常输出示例

{
  "type": "ContextLengthExceededError",
  "code": "AGENT-MODEL-003",
  "message": "上下文超长: maximum context length is 128000 tokens",
  "user_message": "系统繁忙，请稍后重试",
  "retryable": false,
  "severity": "medium",
  "http_status": 400,
  "context": {
    "trace_id": "a1b2c3d4",
    "session_id": "sess-001",
    "tenant_id": "tenant-a",
    "model": "gpt-4o",
    "max_tokens": 128000,
    "actual_tokens": 135000
  },
  "cause": "BadRequestError('maximum context length...')"
}

---

八、工程化落地建议

1. 错误码即契约
   AGENT-XXX-NNN 全局唯一，前后端/告警系统按前缀路由：SEC- 直通安全团队、MODEL- 直通算法团队。

2. 重试要克制

   • 只对 retryable=True 重试。
   • ContextLengthExceeded、ContentFilter、SecurityError 一律不重试。
   • 使用「全抖动指数退避 + retry_after」避免惊群。

3. 降级链路显式化
   fallback 不要写成裸 except，而是按 severity 分层：

   • LOW：用默认值
   • HIGH：走人工介入
   • CRITICAL：直接熔断

4. 脱敏
   context 会进日志/告警，工具 arguments、用户输入、token 必须过滤敏感字段。

5. 可观测性
   在 with_error_handling 里统一打点，输出 OTel 的 error.type、error.code，配合 trace_id 关联模型调用链。

6. 不要在 Agent 内部吞异常
   所有工具/LLM 调用必须转换成 AgentError 子类再抛出，禁止裸 raise Exception，否则规划层无法判断是否重试。

7. 配合熔断器
   以 (code, tool_name/model) 为 key 统计滑动窗口失败率，达到阈值触发 CircuitOpenError（可作为 InfrastructureError 的子类补充）。

---

九、总结

这套异常体系的核心思想是：

异常不是错误信息，而是携带「可恢复性 + 归因 + 上下文 + 降级策略」的结构化契约。

它让规划器、监控、告警、前端都能基于同一份数据做出正确决策，而不是各自猜测。

如果你正在构建企业级 AI Agent，建议尽早把异常体系独立成模块，后续接入重试、熔断、降级、可观测性都会顺理成章。