02 模型 API 工程

第一部分 速记页

第二部分 易混对照

第三部分 面试口述稿

3.1 “调用大模型 API 要注意哪些工程问题？”

我按一次调用的生命周期来说。

发请求之前：API key 放在环境变量里，不进代码和 git；设置合理的超时，SDK 默认的超时往往很长，交互场景要设短一些；控制并发数，不要一下子把几百个请求同时打出去触发限流。

等响应的时候：交互场景用流式输出，首字能很快出来。流式有个容易踩的坑，工具调用的参数 JSON 会被切成好几段分别返回，要按调用的序号拼起来，全部收完再解析。

拿到响应之后：看 finish_reason，如果是 length 说明输出被截断了，JSON 很可能不完整；记录 usage，包括缓存命中的 token，用来算成本。

出错的时候：要分类。网络错误、超时、429 限流、500 和 503 可以重试，用指数退避加随机抖动；400、401、402、422 重试也没用，要立刻失败。主模型持续不可用，比如余额不足或者一直过载，就降级到提前评测过的备用模型。

我项目在这里吃过亏：评测跑到一半 DeepSeek 返回 402 余额不足，SDK 抛的是一个通用的状态码异常，评测脚本对每个项目单独捕获异常后继续跑，结果后面 4 个项目每个都是第一步就崩。应该对 402 这种全局性错误直接停止整批任务并告警。

3.2 “重试怎么设计？”

三个要点：只重试该重试的、等待时间指数增长、加随机抖动。

该重试的是临时性错误：连接失败、超时、429、500、502、503。不该重试的是请求本身有问题或者账户问题：400、401、402、422，重试只会浪费时间。

等待时间按指数增长，比如 0.5 秒、1 秒、2 秒、4 秒，设一个上限。但光有指数退避不够，如果很多客户端同时遇到限流，它们会在同一时刻一起重试，又一起被限流。AWS 有一篇讲退避和抖动的文章，推荐的做法叫 Full Jitter：每次在 0 到当前上限之间随机选一个等待时间，把重试时间打散。

另外如果服务端返回了 Retry-After 头，要按它说的时间等。OpenAI 的 Python SDK 已经内置了这些：默认重试 2 次，指数退避，会读 Retry-After。我用模拟的 HTTP 响应验证过，连续两次 429 之后成功，SDK 发了 3 次请求；402 和 400 都只发 1 次就抛异常。

最后要注意重试和幂等：模型调用本身没有副作用，重试是安全的；但如果 Agent 的一步里包含了写操作，就不能简单地把整步重试。

3.3 “流式输出怎么实现？工具调用在流式里有什么特别的？”

请求时设 stream=True，服务端返回 text/event-stream，每一块是 data: 开头的一行 JSON，里面的 delta 是增量，最后一块是 data: [DONE]。SDK 会把它包装成一个可迭代对象，for 循环逐块取就行。

文字内容直接拼接就好。工具调用麻烦一些：第一块给出工具调用的 id 和函数名，参数是空字符串，后面几块只给参数的片段，比如 {"comm、and": "cmake -B、 build"}，要按工具调用的 index 把片段拼起来，等 finish_reason 变成 tool_calls 再解析 JSON。中途解析一定会失败。

用量统计也要注意，流式默认不返回 usage，要加 stream_options={"include_usage": True}，只有最后一块带 usage，前面的块里是空的。

我项目的 Agent 调模型没有用流式，因为一次模型调用输出很短，主要是工具调用；推给前端的”流式”是 Agent 每一步的事件，用的是后端自己的 SSE，在 14 篇讲过。

3.4 “怎么让模型稳定输出 JSON？”

有三种做法，保证程度不一样。

第一种是 JSON 模式，response_format 设成 json_object，只保证输出是合法 JSON，不保证字段对。DeepSeek 文档还要求提示词里必须出现”json”这个词并给一个示例，而且提醒可能偶尔返回空内容，要把 max_tokens 设够，避免 JSON 被截断。

第二种是 JSON Schema 严格模式，OpenAI 叫 Structured Outputs，给一个 Schema 并设 strict，保证输出符合结构，代价是 Schema 有限制，比如所有字段都要列为必填、不允许额外字段。

第三种是强制调用一个工具，把想要的结构定义成工具参数，tool_choice 指定必须调用它。

不管用哪种，程序拿到之后都要再校验一遍，比如用 Pydantic 解析；解析失败要能处理：检查 finish_reason 是不是 length，可以把错误信息回给模型让它重新输出一次，重试次数要有上限。

3.5 “多个模型怎么切换？”

现在很多厂商提供 OpenAI 兼容的接口，请求和响应格式一样，改 base_url、api_key 和模型名就能换，我项目就是用 OpenAI 的 SDK 调 DeepSeek。

但格式兼容不等于行为一样：缓存命中的字段名不同，DeepSeek 是 prompt_cache_hit_tokens，OpenAI 是在 prompt_tokens_details 里的 cached_tokens；支持的参数和 Schema 限制不同；工具调用的稳定性不同。所以我会在自己的代码里包一层，把各家的 usage 统一成自己的字段，模型名、base_url 放配置里。

还要知道 OpenAI 自己已经往前走了一步：官方现在建议新项目用 Responses API，Chat Completions 仍然支持，但推理模型的推理状态保存、服务端压缩、内置工具这些新能力主要做在 Responses 上。国内厂商兼容的基本还是 Chat Completions 格式，所以我的封装层对外暴露自己的接口，底下哪种格式都能接。

降级的前提是备用模型在同一套评测集上测过，否则切过去可能效果差很多。降级策略可以是：余额不足、持续过载、主模型超时率过高时切换，同时告警。

第四部分 逐个详解

4.1 一次请求和响应长什么样

先说为什么会有 Chat Completions 这种格式。 OpenAI 在 2020 年开放 GPT-3 的 API 时，接口叫 Completions（补全）：传一段文本进去，模型接着往下写。要做对话，只能自己把历史拼成一整段文字，比如”用户：……\n助手：“，再让模型续写。哪一句是开发者定的规矩、哪一句是用户说的话，模型只能靠文字格式去猜，用户在输入里自己写一行”助手：“，格式就乱了。2022 年 11 月 ChatGPT 上线，2023 年 3 月 1 日 OpenAI 把 ChatGPT 用的 gpt-3.5-turbo 开放成 API，同时推出 Chat Completions 接口：输入不再是一段文本，而是一串带角色（system、user、assistant）的消息，由服务端按模型训练时用的格式拼好（OpenAI 当时叫它 ChatML）。这个模型的价格是之前 GPT-3.5 模型的十分之一，大批应用迁到了这个接口上，后来的工具调用也是在这个消息列表里加新的消息类型。其他厂商为了让开发者少改代码，纷纷提供和它一样的接口，这就是 4.9 节说的”OpenAI 兼容接口”。

Chat Completions：OpenAI 定义的对话接口格式，现在大多数厂商都提供兼容版本。我项目用 OpenAI 的 Python SDK 调 DeepSeek（agent/build_agent.py）：

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com")
resp = client.chat.completions.create(model=MODEL, messages=messages, tools=schemas)

请求的主要字段：

消息角色：

sequenceDiagram
    participant A as 应用
    participant M as 模型
    A->>M: system: 规则和角色<br/>user: 把项目交叉编译到 aarch64
    M-->>A: assistant: tool_calls=[run_command(cmake -B build)]
    Note over A: 执行工具
    A->>M: 前面全部消息<br/>+ assistant 那条<br/>+ tool: [exit=1] 找不到 zlib
    M-->>A: assistant: tool_calls=[...] 或最终回答

响应的主要字段（以 DeepSeek 接口文档为例）：

finish_reason 的取值（DeepSeek 文档）：

usage 的字段（DeepSeek 文档）：

各家字段名不一样：OpenAI 的缓存命中在 usage.prompt_tokens_details.cached_tokens 里（06 篇核对过）。这些厂商扩展字段在 OpenAI SDK 的对象上不一定有同名属性，要从 model_extra 里取，实验 1 里会看到。

我项目记了什么：trace.add_usage 只累加了 prompt_tokens 和 completion_tokens，没取缓存命中和思考 token，10 篇讲过这个缺口。

4.2 动手实验 1：SDK 的重试行为和错误分类

不调真实 API。用 MockTransport（模拟传输层）让 SDK 以为自己在和服务器通信，每次请求返回我们指定的状态码，看 SDK 重试了几次、抛了什么异常。保存为 sdk_retry.py，用项目虚拟环境运行。

import json
import time

import httpx2
import openai
from openai import OpenAI

OK_BODY = {
    "id": "c1", "object": "chat.completion", "created": 0, "model": "deepseek-flash",
    "choices": [{"index": 0, "finish_reason": "stop",
                 "message": {"role": "assistant", "content": "cmake -B build"}}],
    "usage": {"prompt_tokens": 1200, "completion_tokens": 8, "total_tokens": 1208,
              "prompt_cache_hit_tokens": 1000, "prompt_cache_miss_tokens": 200},
}


def make_client(plan, max_retries=2):
    calls = []

    def handler(request):
        calls.append(time.perf_counter())
        status = plan[min(len(calls), len(plan)) - 1]
        if status == 200:
            return httpx2.Response(200, json=OK_BODY)
        return httpx2.Response(status, json={"error": {"message": f"status {status}"}})

    client = OpenAI(api_key="test", base_url="https://api.example.test", max_retries=max_retries,
                    http_client=httpx2.Client(transport=httpx2.MockTransport(handler)))
    return client, calls


def run(title, plan, max_retries=2):
    client, calls = make_client(plan, max_retries)
    try:
        resp = client.chat.completions.create(model="deepseek-flash", messages=[{"role": "user", "content": "hi"}])
        u = resp.usage
        result = f"成功 finish_reason={resp.choices[0].finish_reason} 输入={u.prompt_tokens} 命中缓存={u.model_extra['prompt_cache_hit_tokens']}"
    except openai.APIStatusError as e:
        result = f"{type(e).__name__} status={e.status_code}"
    gaps = [round(b - a, 2) for a, b in zip(calls, calls[1:])]
    print(f"{title:<22} 请求 {len(calls)} 次，间隔秒数 {gaps} -> {result}")


run("429, 429, 200", [429, 429, 200])
run("503, 200", [503, 200])
run("402 余额不足", [402, 200])
run("400 参数错误", [400, 200])
run("429 x3（重试用完）", [429, 429, 429, 200])
run("429, 200 关掉重试", [429, 200], max_retries=0)

实际输出（Python 3.13，openai 3.13.0；间隔秒数带随机抖动，每次运行略有不同）：

429, 429, 200          请求 3 次，间隔秒数 [0.49, 0.76] -> 成功 finish_reason=stop 输入=1200 命中缓存=1000
503, 200               请求 2 次，间隔秒数 [0.46] -> 成功 finish_reason=stop 输入=1200 命中缓存=1000
402 余额不足               请求 1 次，间隔秒数 [] -> APIStatusError status=402
400 参数错误               请求 1 次，间隔秒数 [] -> BadRequestError status=400
429 x3（重试用完）           请求 3 次，间隔秒数 [0.45, 0.79] -> RateLimitError status=429
429, 200 关掉重试          请求 1 次，间隔秒数 [] -> RateLimitError status=429

逐段讲。

为什么是 httpx2？ 项目虚拟环境里的 openai 3.13.0 底层用的 HTTP 库是 httpx2（SDK 源码里的 import httpx2 可以看到），所以模拟传输层也要用它的 MockTransport。换了 SDK 版本，这里可能要改成 httpx。

make_client(plan, max_retries=2)。

• plan 是一个状态码列表，第 1 次请求返回 plan[0]，第 2 次返回 plan[1]……请求次数超过列表长度时一直返回最后一个
• def handler(request):：定义在 make_client 里的函数，每收到一个请求就被调用一次，返回一个模拟的响应。它能修改外层的 calls 列表（闭包，12 篇讲过）
• httpx2.Response(200, json=OK_BODY)：构造一个 HTTP 响应，json= 会把字典转成 JSON 作为响应体
• httpx2.MockTransport(handler)：一个假的传输层，不发网络请求，而是调用 handler
• OpenAI(..., http_client=httpx2.Client(transport=...))：让 SDK 用我们给的 HTTP 客户端。base_url 随便写一个，不会真的访问
• max_retries：SDK 的重试次数，默认 2

run。

• except openai.APIStatusError as e：SDK 对所有非 2xx 响应抛出的异常都继承自 APIStatusError。type(e).__name__ 取异常的类名，看 SDK 把它归成了哪一类
• u.model_extra['prompt_cache_hit_tokens']：usage 是一个 Pydantic 模型，OpenAI 标准里没有的字段（DeepSeek 扩展的缓存字段）会放进 model_extra 字典里
• zip(calls, calls[1:])：把相邻两次请求的时间配对，算出间隔

看结果。

1. 429 和 503 会自动重试：429 两次后成功，SDK 共发了 3 次请求；间隔约 0.5 秒和 0.8 秒，在增长，而且不是整数，说明有抖动。我读了 SDK 源码：初始等待 0.5 秒、每次翻倍、上限 8 秒，再乘以 0.75 到 1 之间的随机数；如果响应头有 Retry-After 就按它来（2 分钟以内）
2. 402 和 400 不重试，1 次就抛异常
3. 402 被抛成通用的 APIStatusError，不像 400 有专门的 BadRequestError、429 有 RateLimitError。我项目 runs.db 里 run 208 到 212 的状态就是 crash:APIStatusError。如果代码只 except RateLimitError，402 就会漏过去
4. 重试次数用完后抛出最后一次的异常：max_retries=2 意味着最多发 3 次请求
5. max_retries=0 关掉重试，第一次 429 就失败

SDK 默认会重试哪些（openai-python README）：连接错误、408 请求超时、409 冲突、429 限流、500 及以上的服务端错误，默认 2 次；默认超时 10 分钟，超时的请求也会重试。

自己改一改：

1. 在 handler 里给 429 响应加上 headers={"retry-after": "1"}，看间隔变成多少
2. 把 plan 改成 [500, 502, 200]
3. 在 run 里单独捕获 openai.RateLimitError 并打印”限流”，看 402 会不会被它捕获

4.3 错误分类：该重试的和不该重试的

DeepSeek 的错误码（官方文档）：

再加上网络层面的：连接失败、超时，一般都该重试。

flowchart TD
    E[调用失败] --> T{什么错误}
    T -->|连接失败 超时<br/>429 500 502 503| R{还有重试次数吗}
    R -->|有| W[指数退避 + 抖动<br/>有 Retry-After 按它等] --> AGAIN[重试]
    R -->|没有| F{有备用模型吗}
    T -->|402 余额不足| STOP[停止整批任务<br/>告警]
    STOP --> F
    T -->|400 401 422| BUG[不重试<br/>记录请求 修代码或配置]
    F -->|有| FB[降级到备用模型]
    F -->|没有| FAIL[向上报错]

为什么 402 要单独处理？ 它不是”这一次请求”的问题，而是”之后所有请求”都会失败。我项目的评测脚本 run_eval.py 对每个项目 try / except Exception，记录崩溃后继续跑下一个。这对”某个项目出了意外”是合理的，但 402 之后继续跑毫无意义。10 篇的实验里用”连续 3 次崩溃”规则能在 run 210 就发现，更直接的做法是识别出 402 立刻停止。

4.4 动手实验 2：自己写重试、抖动和降级

SDK 的重试只管”同一个模型再试”，不管”换模型”，也不会把 402 当成需要停下来的信号。下面用标准库写一个带降级的调用策略。保存为 retry_fallback.py。

import random


class APIError(Exception):
    def __init__(self, status):
        super().__init__(f"status {status}")
        self.status = status


RETRYABLE = {408, 429, 500, 502, 503}
FALLBACK = {402, 503}


def full_jitter(attempt, base=0.5, cap=8.0, rng=random):
    return rng.uniform(0, min(cap, base * 2 ** attempt))


def call_with_policy(models, send, max_attempts=4, rng=random, sleep=lambda s: None):
    log = []
    for model in models:
        for attempt in range(max_attempts):
            try:
                result = send(model)
                log.append(f"{model} 第{attempt + 1}次 成功")
                return result, log
            except APIError as e:
                if e.status in RETRYABLE and attempt < max_attempts - 1:
                    wait = full_jitter(attempt, rng=rng)
                    log.append(f"{model} 第{attempt + 1}次 {e.status} 等 {wait:.2f}s 重试")
                    sleep(wait)
                    continue
                log.append(f"{model} 第{attempt + 1}次 {e.status} 不重试")
                if e.status in FALLBACK:
                    break
                raise
    raise RuntimeError("所有模型都失败")


def scripted(plan):
    it = {m: iter(v) for m, v in plan.items()}

    def send(model):
        status = next(it[model])
        if status != 200:
            raise APIError(status)
        return f"{model} 的回答"
    return send


cases = {
    "限流两次后成功": {"flash": [429, 429, 200]},
    "余额不足换备用模型": {"flash": [402], "backup": [200]},
    "一直过载后换备用": {"flash": [503, 503, 503, 503], "backup": [200]},
    "参数错误直接失败": {"flash": [400], "backup": [200]},
}
for name, plan in cases.items():
    rng = random.Random(42)
    try:
        result, log = call_with_policy(["flash", "backup"], scripted(plan), rng=rng)
        print(f"{name}: {result}")
    except APIError as e:
        print(f"{name}: 抛出 {e}")
        log = []
    for line in log:
        print("   ", line)

rng = random.Random(1)
print("第 0-4 次重试的等待上限和一次抽样:",
      [(min(8.0, 0.5 * 2 ** a), round(full_jitter(a, rng=rng), 2)) for a in range(5)])

实际输出（Python 3.14，只用标准库）：

限流两次后成功: flash 的回答
    flash 第1次 429 等 0.32s 重试
    flash 第2次 429 等 0.03s 重试
    flash 第3次 成功
余额不足换备用模型: backup 的回答
    flash 第1次 402 不重试
    backup 第1次 成功
一直过载后换备用: backup 的回答
    flash 第1次 503 等 0.32s 重试
    flash 第2次 503 等 0.03s 重试
    flash 第3次 503 等 0.55s 重试
    flash 第4次 503 不重试
    backup 第1次 成功
参数错误直接失败: 抛出 status 400
第 0-4 次重试的等待上限和一次抽样: [(0.5, 0.07), (1.0, 0.85), (2.0, 1.53), (4.0, 1.02), (8.0, 3.96)]

逐段讲。

自定义异常。

• class APIError(Exception):：定义一个继承自 Exception 的异常类
• def __init__(self, status):：构造方法，创建对象时自动调用。self 代表正在创建的这个对象
• super().__init__(f"status {status}")：调用父类 Exception 的构造方法，设置异常的文字说明
• self.status = status：把状态码存成对象的属性，捕获异常后可以用 e.status 取

两个集合。 RETRYABLE 是可以重试的状态码，FALLBACK 是需要换模型的状态码。503 两边都有：先重试，重试用完还不行就换模型。402 只在 FALLBACK 里：不重试，直接换。

full_jitter：AWS 架构博客 Exponential Backoff And Jitter（Marc Brooker，2015）里的 Full Jitter 做法：等待时间 = 0 到 min(上限, 基数 × 2^第几次) 之间的随机数。rng.uniform(a, b) 返回 a 到 b 之间的随机小数。

call_with_policy。

• 外层循环按顺序尝试每个模型，内层循环是同一个模型的多次尝试
• sleep=lambda s: None：默认参数是一个什么都不做的函数。真实使用时传 time.sleep；实验里不真的等，跑得快
• 可重试且还有次数：记录、等待、continue 进入下一次尝试
• 不可重试或次数用完：如果在 FALLBACK 里，break 跳出内层循环，外层循环换下一个模型；否则 raise 把异常原样抛出，因为 400 这类错误换模型也没用
• 所有模型都失败，抛出 RuntimeError

scripted(plan)：生成一个假的”发送函数”。iter(列表) 把列表变成迭代器，next(迭代器) 每次取下一个元素，这样同一个模型第几次被调用就返回剧本里第几个状态码。

看结果。

1. 限流两次后第三次成功，没有换模型
2. 402 立刻换到备用模型，没有浪费时间重试
3. 503 重试到第 4 次（max_attempts=4）还不行，才换备用模型
4. 400 直接抛出，没有换模型，因为请求本身有问题
5. 最后一行：等待上限是 0.5、1、2、4、8 秒；实际抽到的是上限以内的随机数，而且不一定递增（第 4 次抽到 1.02 比第 3 次的 1.53 小）。这正是抖动的目的：让大量客户端的重试时间分散开，不在同一时刻一起撞上去

Brooker 那篇文章的模拟结论：只做指数退避不加抖动时，客户端还是会成群地在同一时刻重试；加了抖动后总工作量明显下降，Full Jitter 在他的模拟里做的总工作量最少。

还要考虑的：

自己改一改：

1. 给 call_with_policy 加一个 deadline 参数（总共最多等多少秒），累计等待超过就不再重试
2. 把 sleep 换成 time.sleep，感受实际等待
3. 加一个状态码 401，让它直接抛出，并想一想：401 需要告警吗

4.5 超时

一定要设超时。 不设超时的请求在网络异常时可能一直挂着，占着连接和线程。

OpenAI Python SDK 的默认值（README 和源码 _constants.py）：总超时 600 秒（10 分钟），连接超时 5 秒。10 分钟对交互场景太长了，用户早就走了。

client = OpenAI(timeout=30.0)
client.with_options(timeout=5.0).chat.completions.create(...)

第一行给整个客户端设默认超时，第二行只对这一次请求改。超时抛出 APITimeoutError，而且 SDK 默认会重试。

超时的几层：

流式请求的超时要特别注意：生成很长的回答可能要一两分钟，总超时设短了会被中途切断；应该主要依赖”两次数据之间的间隔”来判断是不是卡住了。

我项目的现状：OpenAI(api_key=..., base_url=...) 没有设置超时和重试次数，用的是 SDK 默认的 10 分钟和 2 次重试。评测场景可以接受，放到交互场景就不合适。

4.6 流式输出

流式输出（streaming）：模型每生成一小段就通过 SSE（14 篇）推送给客户端，而不是等全部生成完。客户端每次收到的是增量（delta）。

动手实验 3：模拟一个流式响应，其中既有文字又有工具调用，看参数片段怎么拼。保存为 stream_tools.py，用项目虚拟环境运行。

import json

import httpx2
from openai import OpenAI


def chunk(delta=None, finish=None, usage=None):
    body = {"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": "deepseek-flash",
            "choices": [] if delta is None else [{"index": 0, "delta": delta, "finish_reason": finish}],
            "usage": usage}
    return f"data: {json.dumps(body)}\n\n"


EVENTS = [
    chunk({"role": "assistant", "content": "先配置"}),
    chunk({"content": "项目。"}),
    chunk({"tool_calls": [{"index": 0, "id": "call_1", "type": "function",
                           "function": {"name": "run_command", "arguments": ""}}]}),
    chunk({"tool_calls": [{"index": 0, "function": {"arguments": "{\"comm"}}]}),
    chunk({"tool_calls": [{"index": 0, "function": {"arguments": "and\": \"cmake -B"}}]}),
    chunk({"tool_calls": [{"index": 0, "function": {"arguments": " build\"}"}}]}),
    chunk({}, finish="tool_calls"),
    chunk(usage={"prompt_tokens": 1500, "completion_tokens": 30, "total_tokens": 1530}),
    "data: [DONE]\n\n",
]


def handler(request):
    sent = json.loads(request.content)
    print("请求里的 stream_options:", sent.get("stream_options"))
    return httpx2.Response(200, headers={"content-type": "text/event-stream"},
                           content="".join(EVENTS).encode())


client = OpenAI(api_key="test", base_url="https://api.example.test",
                http_client=httpx2.Client(transport=httpx2.MockTransport(handler)))
stream = client.chat.completions.create(
    model="deepseek-flash", messages=[{"role": "user", "content": "编译"}],
    stream=True, stream_options={"include_usage": True})

text, calls, finish, usage = "", {}, None, None
for part in stream:
    if part.usage:
        usage = part.usage
    if not part.choices:
        continue
    choice = part.choices[0]
    delta = choice.delta
    if delta.content:
        text += delta.content
        print("文字片段:", repr(delta.content))
    for tc in delta.tool_calls or []:
        slot = calls.setdefault(tc.index, {"id": None, "name": "", "arguments": ""})
        if tc.id:
            slot["id"] = tc.id
        if tc.function.name:
            slot["name"] = tc.function.name
        if tc.function.arguments:
            slot["arguments"] += tc.function.arguments
            print("参数片段:", repr(tc.function.arguments))
    if choice.finish_reason:
        finish = choice.finish_reason

print("拼好的文字:", text)
print("拼好的工具调用:", calls[0]["name"], json.loads(calls[0]["arguments"]))
print("finish_reason:", finish, "| usage:", usage.prompt_tokens, usage.completion_tokens)

实际输出（Python 3.13，openai 3.13.0）：

请求里的 stream_options: {'include_usage': True}
文字片段: '先配置'
文字片段: '项目。'
参数片段: '{"comm'
参数片段: 'and": "cmake -B'
参数片段: ' build"}'
拼好的文字: 先配置项目。
拼好的工具调用: run_command {'command': 'cmake -B build'}
finish_reason: tool_calls | usage: 1500 30

逐段讲。

构造模拟的流。

• chunk(...)：拼出一块 SSE 数据。格式是 data: 加一行 JSON 加一个空行，和 14 篇讲的 SSE 格式一样
• EVENTS 模拟了真实流的结构：先是文字片段；然后工具调用的第一块带 id 和函数名、参数为空；接下来三块只有参数片段；然后一块 finish_reason 为 tool_calls；倒数第二块 choices 为空、只带 usage；最后 data: [DONE]
• 这个结构是按 DeepSeek 文档的描述构造的：开启 include_usage 时，每块都有 usage 字段，除了最后一块都是 null；流以 data: [DONE] 结束
• handler 里 json.loads(request.content) 读出 SDK 真正发出的请求体，确认 stream_options 被带上了

消费流。

• stream=True 时 create 返回的不是完整响应，而是一个可迭代对象，for part in stream 每次拿到一块
• if not part.choices: continue：只带 usage 的那块没有 choices，跳过后面的处理
• delta.tool_calls or []：没有工具调用时 tool_calls 是 None，or [] 让循环安全地什么都不做
• calls.setdefault(tc.index, {...})：字典的 setdefault，键不存在就放入默认值并返回它，存在就返回已有的值。按 index 分组，因为一次可能有多个并行工具调用，每个的片段交错到达
• slot["arguments"] += ...：参数片段逐个拼接
• 全部收完后才 json.loads

看结果。

1. 参数被切成了三段：{"comm、and": "cmake -B、 build"}。任何一段单独拿去解析 JSON 都会失败，必须拼完再解析
2. 片段的切分位置是任意的，可能切在键名中间、字符串中间
3. usage 只在流的最后出现，如果中途断开就拿不到用量，记账时要考虑

真实模型不保证总是这样切分，但”参数分多块到达”是流式工具调用的正常情况。也有框架（比如 LangChain）帮你拼好，拼不成合法 JSON 的放进 invalid_tool_calls（12 篇）。

自己改一改：

1. 再加一个 index: 1 的工具调用，让两个调用的参数片段交错到达，确认按 index 分组能正确拼出两个
2. 删掉最后那块 usage，看 usage 变量会是什么，程序会不会出错
3. 在参数片段第二块之后立刻 json.loads(slot["arguments"])，看报什么错

4.7 结构化输出

先说为什么会有结构化输出。 最早想让模型给出 JSON，只能在提示词里写”只输出 JSON”，再用 json.loads 去解析。常见的失败有：前面多一句”好的，以下是结果：“，JSON 外面包了一层 Markdown 代码块，少一个右括号，字段名写错。调用量一大，每天都有一批解析失败要重试。后来厂商一步步把这件事挪到了服务端。以 OpenAI 为例：2023 年 6 月推出函数调用，模型经过专门训练，能输出符合函数参数格式的 JSON（04 篇细讲）；2023 年 11 月 6 日第一届开发者大会上推出 JSON 模式，保证输出是合法 JSON，但不保证字段对；2024 年 8 月 6 日推出 Structured Outputs，给一个 JSON Schema 并打开 strict，服务端保证输出符合这个 Schema。OpenAI 在发布文章里给的数字是：复杂 Schema 的评测上，新模型加 Structured Outputs 得分 100%，旧的 gpt-4-0613 不到 40%。

要让模型输出程序能直接使用的数据，有三种做法：

OpenAI 文档（Structured Outputs）的说法：只有 Structured Outputs 能保证符合 Schema，JSON 模式只保证是合法 JSON；需要连接工具、函数、数据时用函数调用，只是想让回答给用户的内容有结构时用 response_format。严格模式要求所有字段都列为必填、设 additionalProperties: false。

DeepSeek 的 JSON 模式（JSON Output 文档）特别提醒两点：接口可能偶尔返回空内容；要把 max_tokens 设够，防止 JSON 被截断。

不管哪种，程序都要再校验：

flowchart TD
    R[拿到模型输出] --> F{finish_reason}
    F -->|length| TR[被截断<br/>调大 max_tokens 或让模型精简]
    F -->|stop / tool_calls| E{内容为空?}
    E -->|是| RETRY[重试一次]
    E -->|否| P{json.loads 成功?}
    P -->|否| FIX[把错误信息回给模型<br/>要求重新输出 限制次数]
    P -->|是| V{Pydantic / jsonschema<br/>校验通过?}
    V -->|否| FIX
    V -->|是| OK[使用]

04 篇有 Pydantic 校验工具参数的实验，这里的思路完全一样。

4.8 限流和并发：动手实验 4

限流（rate limit）：服务端限制你单位时间内的请求数、token 数或同时进行的请求数。超了返回 429。核对 DeepSeek 价格页时，deepseek-flash 标注的并发上限是 2500，deepseek-v4-pro 是 500。

批量任务（评测、数据处理）要主动控制并发，而不是全部发出去等着被 429 打回来。

asyncio 并发：14 篇讲过 async/await。调模型 API 大部分时间在等网络，适合用异步并发。OpenAI SDK 提供 AsyncOpenAI，用法和同步版一样，只是调用前加 await。

下面用假的异步函数模拟模型调用，看 Semaphore 怎么限制并发。保存为 async_limit.py。

import asyncio
import time

active = 0
peak = 0


async def fake_llm(i):
    global active, peak
    active += 1
    peak = max(peak, active)
    await asyncio.sleep(0.2)
    active -= 1
    return f"回答{i}"


async def limited(sem, i):
    async with sem:
        return await fake_llm(i)


async def main():
    global peak
    for label, limit in (("串行", 1), ("最多 3 个并发", 3), ("不限制", None)):
        peak = 0
        start = time.perf_counter()
        if limit is None:
            results = await asyncio.gather(*(fake_llm(i) for i in range(10)))
        else:
            sem = asyncio.Semaphore(limit)
            results = await asyncio.gather(*(limited(sem, i) for i in range(10)))
        print(f"{label:<10} 10 个请求 {time.perf_counter() - start:.2f}s，同时在跑最多 {peak} 个，结果顺序 {results[:3]}...")


asyncio.run(main())

实际输出（Python 3.14，只用标准库）：

串行         10 个请求 2.03s，同时在跑最多 1 个，结果顺序 ['回答0', '回答1', '回答2']...
最多 3 个并发   10 个请求 0.81s，同时在跑最多 3 个，结果顺序 ['回答0', '回答1', '回答2']...
不限制        10 个请求 0.20s，同时在跑最多 10 个，结果顺序 ['回答0', '回答1', '回答2']...

逐段讲。

• global active, peak：在函数里修改模块级变量时要用 global 声明，否则赋值会创建一个同名的局部变量
• active 记录此刻正在”调用模型”的数量，peak 记录出现过的最大值
• await asyncio.sleep(0.2)：模拟等待模型响应 0.2 秒，等待期间让出控制权
• asyncio.Semaphore(limit)：信号量，内部有一个计数器，最多允许 limit 个协程同时进入。async with sem: 进入时计数减一，满了就等待；离开时加一
• asyncio.gather(*协程们)：同时启动全部，等全部完成。返回结果的顺序和传入顺序一致，不是完成顺序，所以 results[:3] 总是回答 0、1、2

看结果。

1. 串行 10 个 × 0.2 秒 = 2 秒
2. 限制 3 个并发：10 个分成 3、3、3、1 四批，约 0.8 秒；同时在跑的最多 3 个
3. 不限制：0.2 秒全部完成，但同时有 10 个在跑。真实场景里这就是触发 429 的做法

实际项目里常见的组合：Semaphore 限制并发 + 4.4 节的重试退避 + 按 token 用量估算的速率控制。

我项目的做法：评测脚本 run_eval.py 在一个进程里串行跑项目；需要并行时，是开几个进程、用 WORKSPACE 环境变量给每组指定不同的工作目录（09 篇）。每个 Agent 内部的模型调用是串行的，因为每一步依赖上一步的结果。

自己改一改：

1. 把请求数改成 100、并发限制改成 10，估算时间再验证
2. 让 fake_llm 在 i 为 5 时抛出异常，看 gather 的行为；再试 asyncio.gather(..., return_exceptions=True)

4.9 OpenAI 兼容接口和多模型封装

OpenAI 兼容接口：很多厂商（DeepSeek 等）提供和 OpenAI Chat Completions 格式一致的接口。好处是同一个 SDK、同一套代码，改 base_url、api_key、model 就能切换。DeepSeek 价格页还写着两个模型也支持 Anthropic 的 API 格式。

但兼容不等于一样：

建议封装一层自己的调用函数：

flowchart LR
    APP[Agent / 业务代码] --> W[自己的 llm_call<br/>统一参数 统一 usage 字段<br/>超时 重试 降级 记录]
    W --> C1[DeepSeek<br/>OpenAI 兼容]
    W --> C2[备用模型]
    W --> TR[(链路追踪<br/>token 缓存命中 耗时 错误)]

封装里做：模型配置（名字、base_url、key 的环境变量名、价格）、统一 usage（输入、输出、缓存命中、思考 token）、超时和重试策略、降级顺序、记录到 10 篇的链路追踪。业务代码只调这一个函数，换模型不用改业务代码。

Chat Completions 还是 Responses API？（2026-09 查阅 OpenAI 迁移指南）

OpenAI 2025 年 3 月推出 Responses API，现在的官方说法是 Chat Completions 仍然支持，但新项目建议用 Responses。原来的 Assistants API 已经在 2026 年 8 月 26 日下线，替代品也是 Responses。

OpenAI 迁移指南里给的内部测试数字是：同样的推理模型，走 Responses 在 SWE-bench 上高 3%，缓存利用率提升 40% 到 80%，原因主要是推理条目能跨轮保留。

怎么选：只调 OpenAI、要用推理模型或内置工具，用 Responses；要在多家模型之间切换（尤其国内厂商），Chat Completions 格式仍然是公约数。不管选哪个，都在自己的封装层里屏蔽掉差异。

我项目的现状：build_agent.py 和 mini_agent.py 各自直接 OpenAI(...)，graph_agent.py 用 LangChain 的 ChatOpenAI，三处分别构造客户端，没有统一封装，也没有备用模型。

4.10 API key 和配置管理

原则：key 不写进代码、不进 git、不进日志、不进模型上下文（11 篇）。

我项目的做法（agent/mini_agent.py）：

def load_env(path=Path(__file__).resolve().parent.parent / ".env"):
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip())

逐段讲。

• 默认参数 path=Path(__file__).resolve().parent.parent / ".env"：__file__ 是当前脚本的路径，.parent.parent 往上两级到项目根目录。以脚本自己的位置为锚点，从任何目录运行都能找到 .env
• 文件不存在就直接返回，不报错，这样也可以只用系统环境变量
• 跳过空行、# 开头的注释、没有等号的行
• line.split("=", 1)：只按第一个等号切，值里本身带等号也不会被切坏
• os.environ.setdefault(key, value)：只在环境变量不存在时才设置。所以命令行里 MODEL=xxx python ... 临时指定的值优先于 .env 文件
• 仓库里提交的是 .env.example，内容只有 DEEPSEEK_API_KEY=，告诉别人需要哪些变量

这个简单实现的局限：不处理引号（KEY="abc" 会把引号也读进去）、不支持变量展开。常用的库是 python-dotenv。

生产环境：用密钥管理服务或部署平台的密钥配置注入环境变量；不同环境（开发、测试、生产）用不同的 key；给 Agent 用的 key 单独申请、设额度上限，出问题可以单独吊销。

配置项还有哪些：模型名（我项目 MODEL 环境变量，默认 deepseek-flash）、base_url、超时、重试次数、步数上限、是否开知识库（USE_KB）、实现选择（AGENT_IMPL）。评测结果里要记录这些配置，否则不知道结果是在什么设置下跑出来的（09 篇）。

第五部分 对照项目

对照开源实现：pi

pi（05 篇第五部分介绍过）有一个独立的包 packages/ai，专门做本篇 4.9 节说的”自己封装一层”，而且同时接了十几家服务商（OpenAI、Anthropic、Google、Bedrock、Mistral 等，每种接口格式一个实现文件）。源码以 commit 7b4cfd6 为准。

值得讲的几点：

1. 为什么按报错文字匹配，而不是按状态码？ 接了十几家服务商，各家 SDK 抛的错误形式不一样：有的有状态码，有的只有一句话；同样是 429，有的是”请求太快”（该重试），有的是”本月额度用完”（不该重试）。所以 pi 把两类的关键词各列一张表，先排除额度类，再匹配临时故障类。缺点是脆弱，服务商改了报错文字就可能判断错。源码里很多关键词旁边标着 GitHub issue 编号，说明这张表是一次次线上故障攒出来的。

2. 这正好对应我项目的 402 教训。 我的评测脚本在余额不足后又崩了 4 个项目。pi 的”不该重试”名单第一类就是额度和账单问题，而且优先级高于”429 可以重试”。区别在于 pi 只是不重试、把错误显示给人看；我的场景是无人值守的批量任务，还要再往前一步：识别出来就停掉整批任务。

3. 关掉 SDK 重试，和本篇追问清单里说的”防叠加”一致。 pi 文档给了一个更具体的理由：SDK 自己重试时，可能把”额度用完”也当成限流去等，Agent 就会一直卡到额度恢复，pi 这一层根本看不到错误。

4. 为什么没有抖动？ 本篇 4.4 节说抖动是为了避免”很多客户端同时重试”。pi 是一个人在用的命令行工具，同一时刻只有它自己在重试，不存在一群客户端撞在一起的问题。这条原则适用于服务端、批量任务这类多个请求同时失败的场景；我项目多进程跑评测时就属于这种情况。

5. 把错误当数据而不是异常。 我项目调模型出错直接抛异常，一路冒泡到评测脚本。pi 把错误变成一条普通回复，主循环看到 stopReason 是 error 就结束，上层再决定重试、压缩还是报给用户。好处是所有结局走同一条路径，事件、日志、会话记录都完整，不会因为一个异常漏记数据。

第六部分 追问清单

第七部分 闭卷自测

1. 说出四种消息角色和各自的用途。tool 消息必须带什么字段？

2. finish_reason 为 length 意味着什么？程序应该怎么处理？

3. DeepSeek 的 usage 里和缓存相关的字段叫什么？它们和 prompt_tokens 什么关系？在 OpenAI SDK 对象上怎么取到？

4. 实验 1 里，哪些状态码 SDK 自动重试了？402 抛出的是什么异常类？这对我项目有什么影响？

5. 为什么指数退避还要加抖动？Full Jitter 的等待时间怎么算？

6. 实验 2 的策略里，429、402、503、400 分别怎么处理？为什么 400 不换模型？

7. 自己实现重试策略时，为什么要把 SDK 的 max_retries 设为 0？

8. 流式输出里工具调用的参数是怎么到达的？实验 3 里怎么拼？usage 在哪一块？

9. JSON 模式、JSON Schema 严格模式、强制工具调用三种结构化输出分别保证什么？拿到输出后还要做哪些检查？

10. 实验 4 里限制 3 个并发时 10 个请求为什么约 0.8 秒？gather 返回结果的顺序是什么？

11. OpenAI 兼容接口能让切换模型”只改 base_url”吗？列出至少三处不同。

12. 我项目的 load_env 用了 os.environ.setdefault，这样写有什么效果？这个实现有什么局限？

13. pi 的自动重试默认没有随机抖动，本篇却强调要加抖动。两者矛盾吗？

14. pi 为什么把 SDK 自带的重试默认关掉？

延伸阅读

1. DeepSeek：Create Chat Completion — finish_reason、usage 字段、流式 usage、思考模式参数
2. DeepSeek：错误码
3. DeepSeek：JSON Output
4. OpenAI：Migrate to the Responses API — Responses 和 Chat Completions 的区别、为什么建议新项目迁移
5. OpenAI：Structured Outputs — 严格 Schema 与 JSON 模式的区别
6. openai-python — 重试、超时、异步、流式的默认行为
7. AWS：Exponential Backoff And Jitter（Marc Brooker，2015）
8. pi：ai/src/utils/retry.ts（commit 7b4cfd6）— 接了十几家服务商之后的错误分类表和重试实现；ai/src/types.ts 是统一的消息、用量和结束原因定义

下一篇：03 Prompt 工程——调用稳了，接下来是写给模型的那段话怎么写、怎么迭代、怎么评测。