Agent基础：错误处理与模型适配器

本文是 LLM API 基础系列第四篇。上一篇：结构化输出与工具调用；系列入口：LLM API 模型调用服务。

API 与 SDK 行为核对于 2026-09-21。本文不构建完整 Agent 框架，只建立必要的失败边界。

1. 请求成功，不等于任务成功

一次模型调用至少有三层判断：

传输与服务层：连接是否成功，HTTP 状态是否正常？
生成层：响应是否完成，是否拒答、截断或提出工具请求？
应用层：内容是否符合 schema、权限与任务要求？

Chat Completions 和 Responses 都可能返回 HTTP 错误。Responses 还需要检查响应对象中的 status、error、incomplete_details。不能简单理解成“Chat 只看 HTTP，Responses 只看 error”。[1][2]

graph TD
    A[发起模型请求] --> B{请求成功}
    B -->|是| C[检查生成状态和业务契约]
    C --> D[返回结果或按原因处理]
    B -->|否| E{临时故障且预算充足}
    E -->|是| F[按服务建议退避]
    F --> A
    E -->|否| G[停止并报告原因]

“重试同一个请求”和“修改预算/上下文后重新生成”是不同操作，应该分别统计。

2. 失败分类与处置

类别	常见信号	合理处置
参数或能力不匹配	400、非法 schema、不支持的参数	修复请求，不要原样重复
身份或权限	401、403	检查密钥与授权，避免泄露密钥到日志
资源/模型不存在	404 等	检查端点、模型名称和账户可用性
网络连接/超时	SDK 连接或超时异常	在预算内重试；服务端可能已经处理原请求
请求速率限制	429 且错误体表明限流	降低并发，遵守退避提示
额度/余额不足	429 的某些错误码，或厂商自定义状态	补充额度或处理账户，不应忙等重试
服务临时异常	部分 5xx	有限重试，观察持续时间和服务状态
模型拒答	拒答字段、内容过滤信号	停止或请求用户澄清，不绕过安全限制
输出不完整	`length`、`incomplete` 等	检查具体原因，调整预算或拆分任务
结构/业务校验失败	JSON/Pydantic/权限校验错误	有限修复；权限拒绝不能靠换模型绕过

HTTP 状态码只是起点，同一个 429 可能意味着限流，也可能意味着额度不足。要结合供应商错误码和响应体判断；不要把所有 5xx 都当成必定可恢复。[1][3]

“用户要求不明确”也不必然等于“安全拒答”。业务上可以请求澄清，但不要把不同原因归到一个拒答类别。

3. 重试只保留一个负责层

3.1 最小做法：使用 SDK 的内置重试

OpenAI Python SDK 对连接问题、408、409、429 和 5xx 等错误提供内置重试；当前文档默认重试 2 次，可通过 max_retries 配置。[4]

import os

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
    timeout=30.0,
    max_retries=2,  # 最多追加 2 次重试，即最多尝试 3 次
)

这是最少代码的起点，但不等于完成了上表的所有业务策略：SDK 的通用重试不理解你全部的账户错误码、任务时限或降级要求。需要精细策略时，设 max_retries=0，由应用统一分类、退避和记录。

不要同时启用 SDK、应用包装器和任务队列的多层重试，而不计算总尝试次数。 例如外层尝试 3 次、每次 SDK 又尝试 3 次，最坏可能达到 9 次网络请求。

捕获异常时应使用 SDK 的 APIConnectionError、APITimeoutError、APIStatusError 等实际类型，不要假定内置 TimeoutError、ConnectionError 一定能捕获 SDK 异常。原文的 RateLimitError 未导入、ServiceOverloaded 未定义，不应当作可运行实现。[4]

3.2 自己实现退避时，要明确所有预算

一个常见的 full jitter 计算方式是：

1	delay = uniform(0, min(cap, base × 2^retry_index))

先计算上限，再在范围内随机取值，可以让等待时长始终不超过 cap。原文的 min(8, ...) + uniform(0, 0.2) 实际最大为 8.2 秒，并不是 8 秒。[5]

至少需要定义：

最大总尝试次数，不要混淆“尝试”与“重试”。
每次请求的连接/读取等超时策略。
整个任务的总时限、token/费用预算和取消机制。
服务端 Retry-After 等提示的处理方式；若建议等待超过任务剩余时限，应停止，而不是提前猛烈重试。
主模型失败后的降级条件，以及哪些失败禁止降级。

timeout=30.0 不等于“整个 Agent 任务最多运行 30 秒”。SDK 超时通常作用于网络阶段，重试和工具执行也会继续消耗时间。[4]

3.3 重试和幂等是两回事

连接超时并不证明服务端没处理请求。重新生成可能重复计费，也可能得到不同答案。不能默认所有模型端点都有通用幂等键，更不能把模型调用 ID 当作支付或写文件操作的完整去重方案。

对于有副作用的工具，由业务系统使用可靠的操作标识、事务或去重记录保证幂等。流式连接中断时，也不要简单把重试生成的文本拼接到旧片段后面。

4. 可观测性：记录足够信息，不记录不该记录的内容

字段	用途
`trace_id`、业务任务 ID	关联整条 Agent 执行过程
provider、endpoint、model	确认实际供应商、端点和模型/别名
SDK 版本、参数配置版本	复现请求配置，识别接口差异
request/response ID	与服务端排障信息关联，按供应商暴露方式读取
latency、attempt_count	区分慢请求与多次重试
HTTP 状态、供应商错误码	分析网络和服务失败
finish_reason / status	分析完成、截断和拒答等情况
token 用量、缓存/推理明细	估算成本和上下文压力
schema 错误类型	区分生成格式与业务规则问题
tool 名称、结果状态	审计执行过程，不必记录全部敏感内容

不要把完整 prompt、密钥、个人信息和工具原始输出无条件打进日志。 用量缺失应记录为未知，而不是 0。费用是根据对应模型计价规则计算的估算值，不是所有 API 都会直接返回。

5. Model Adapter：只统一目前真正用到的能力

适配器用于把供应商字段映射到应用自己的接口。它不能凭空让两个模型具备相同能力，也不能让所有工具、推理状态和多模态内容无损互换。

如果只有一个简单调用点，先写一个函数即可。当确实要接入两个供应商，或者需要离线替换模型时，再引入小型接口。

graph TD
    A[文本应用] --> B[TextModel 契约]
    B --> C[DeepSeek Chat 适配器]
    B --> D[OpenAI Responses 适配器]
    B --> E[FakeModel 离线实现]
    C --> F[DeepSeek API]
    D --> G[OpenAI API]

下面只实现单轮、非流式、纯文本接口。这不是完整 Agent Loop，也不支持把工具调用悄悄转换为空字符串。

5.1 完整的文本适配器示例

以下代码可放在同一个 Python 文件中运行。成功结果只包含应用自己的类型；异常仍由上层统一处理。

import os
from dataclasses import dataclass
from typing import Protocol

from dotenv import load_dotenv
from openai import OpenAI


@dataclass(frozen=True)
class TextResult:
    text: str
    model: str
    response_id: str
    input_tokens: int | None
    output_tokens: int | None


class TextModel(Protocol):
    def generate(self, task: str, *, output_budget: int) -> TextResult:
        ...


@dataclass
class DeepSeekTextAdapter:
    client: OpenAI
    model: str

    def generate(self, task: str, *, output_budget: int) -> TextResult:
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": task}],
            max_tokens=output_budget,
            extra_body={"thinking": {"type": "disabled"}},
        )
        if not response.choices:
            raise RuntimeError("未返回候选结果")
        choice = response.choices[0]
        message = choice.message
        if choice.finish_reason != "stop" or message.refusal or message.tool_calls:
            raise RuntimeError(f"不是完整的纯文本回答：{choice.finish_reason}")
        if not message.content:
            raise RuntimeError("文本为空")
        usage = response.usage
        return TextResult(
            message.content, response.model, response.id,
            usage.prompt_tokens if usage else None,
            usage.completion_tokens if usage else None,
        )


@dataclass
class OpenAIResponsesTextAdapter:
    client: OpenAI
    model: str

    def generate(self, task: str, *, output_budget: int) -> TextResult:
        response = self.client.responses.create(
            model=self.model,
            input=task,
            max_output_tokens=output_budget,
        )
        if response.status != "completed":
            raise RuntimeError(f"响应未完成：{response.status}")
        for item in response.output:
            if item.type == "message":
                if any(part.type == "refusal" for part in item.content):
                    raise RuntimeError("模型拒答")
            elif item.type != "reasoning":
                raise RuntimeError(f"文本接口不处理：{item.type}")
        if not response.output_text:
            raise RuntimeError("没有最终文本")
        usage = response.usage
        return TextResult(
            response.output_text, response.model, response.id,
            usage.input_tokens if usage else None,
            usage.output_tokens if usage else None,
        )


class TextApp:
    def __init__(self, model: TextModel):
        self.model = model

    def run(self, task: str) -> str:
        return self.model.generate(task, output_budget=2000).text


load_dotenv()
model = DeepSeekTextAdapter(
    client=OpenAI(
        api_key=os.environ["DEEPSEEK_API_KEY"],
        base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
        timeout=30.0,
        max_retries=0,
    ),
    model=os.environ["DEEPSEEK_MODEL_NAME"],
)
app = TextApp(model)
print(app.run("用三句话解释：模型调用为什么不等于完整 Agent？"))

更换供应商只替换构造代码，TextApp 不变。OpenAI 的模型须支持所用端点；推理模型的预算还要覆盖推理开销：

model = OpenAIResponsesTextAdapter(
    client=OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        base_url="https://api.openai.com/v1",
        timeout=60.0,
        max_retries=0,
    ),
    model=os.environ["OPENAI_MODEL"],
)
app = TextApp(model)

上面的适配器在异常场景下使用 RuntimeError 以缩短示例。生产中应按本篇前面的分类定义应用异常，保留原始异常为 cause，并保留供应商诊断信息；不要让重试层把所有 RuntimeError 当作临时故障。

5.2 离线测试不需要真实模型

在前面的类定义之后，可用下列代码验证应用依赖的是契约，而不是某个 SDK：

class FakeModel:
    def generate(self, task: str, *, output_budget: int) -> TextResult:
        return TextResult("固定回答", "fake", "offline-1", 0, 0)


assert TextApp(FakeModel()).run("任意输入") == "固定回答"

这里只验证了应用连接逻辑，不证明真实模型质量。供应商集成测试与任务回归评测仍然需要单独运行。

5.3 什么时候扩展为 Agent 的返回契约

真正需要工具循环或流式处理时，再把接口扩展为能表达下列信息的 ModelResult：

文本与结构化结果。
工具调用列表，保留调用 ID、工具名称和参数。
完成、截断、拒答等结果类别，以及原始结束原因。
token 用量、响应 ID、模型标识等元数据。
供应商要求回传的必要状态；必要时使用受控的、不透明状态对象，而不是丢弃。

“业务层不依赖 SDK 内部类型”不等于“必须删除所有供应商信息”。通用契约要保留运行所必需的语义，并对不支持的能力明确报错，不能假装兼容。

6. 路由与 fallback 不必写进单个适配器

适配器负责协议转换；路由负责选择模型；重试策略负责判断是否再试。 小项目可以把路由写在调用方，大项目再考虑独立策略层，不必一开始就设计工厂和复杂调度框架。

切换模型时还应检查：

目标模型是否支持所需工具、schema、上下文和模态。
是否满足数据区域、隐私及合规约束。
多轮状态能否继续，是否需要重建上下文。
降级后是否仍达到回归集要求，而不仅是返回 HTTP 200。

拒答、权限限制和数据区域要求不能通过 fallback 绕过；缺少关键上下文时，也不能不加说明地换成较小窗口模型继续执行。

7. 系列总结

到这里可以串起四篇文章：

API 结构：读懂请求、响应和兼容边界。
预算管理：确认参数生效，维护有限上下文。
可执行契约：校验模型建议，再决定是否执行工具。
可靠性边界：分类处理失败，以最小接口隔离供应商。

学习 Agent 时，比“先接入一个庞大框架”更重要的是弄清楚：谁管理状态、谁批准动作、谁执行工具、谁决定停止。

返回系列入口：LLM API 模型调用服务。

Jean's Blog