测试调用大模型的代码：用 pytest Mock OpenAI 客户端的实战指南

调用大语言模型（LLM）的代码该怎么测试？这是许多 AI 功能开发者都会遇到的难题。原作者 Aman Kumar 构建了一个 OpenAI 兼容的 API 网关，长期阅读他人的集成代码，他发现一个反复出现的反模式：测试套件每次运行都调用真实模型。这样做又慢又不稳定，消耗请求额度，而且测试失败时你根本分不清是代码出了问题，还是模型今天恰好换了个说法回答。

本文整理了他的一套替代方案。它适用于官方 openai Python 包指向任意 OpenAI 兼容的 base URL 的场景，与具体使用哪家供应商无关。

核心原则：测试检查你的代码，而非模型

LLM 功能中大多数出错的地方其实不在模型本身，而在围绕它的代码：构建 messages 列表、选择正确的 model ID、从回复中解析 JSON、处理空的 choices 列表、在遇到 429 时重试、统计 token 数量。这些逻辑全都是确定性的，只要阻止网络调用发生，它们都是可测试的。

基于这个认识，作者把测试拆成两组：

1. 单元测试（占绝大多数）。客户端被替换成一个 fake，运行只需毫秒级，永不触碰网络。
2. 线上冒烟测试（极少数）。它们真实访问端点，默认被标记跳过，只检查响应的"形状"而非具体措辞。

这种划分的意义在于：把 99% 的确定性逻辑交给快速、免费、可靠的单元测试，把少量真实验证留给冒烟测试去捕捉 fake 无法发现的配置错误。

第一步：给客户端留出"接缝"

最难测试的代码，是那种在模块导入时就创建客户端的写法：

「接缝」（seam）这个概念来自 Michael Feathers 在《修改代码的艺术》中的定义：程序中可以在不修改代码本身的情况下改变其行为的地方。在面向对象代码里，最常见的接缝形式是依赖注入——把外部依赖（如网络客户端、数据库连接）从函数内部的硬编码改为通过参数传入。这样做的直接好处是：调用方可以在测试时传入 fake，在生产时传入真实对象，业务逻辑本身完全不感知差异。与之相对的反模式是在模块顶层直接实例化客户端（`client = OpenAI()`），此时客户端成为全局状态，测试要替换它就只能依赖 `unittest.mock.patch` 去猴子补丁模块属性，既脆弱又与模块路径强耦合。依赖注入是这套测试方案可行的根本前提。
# 难以测试
from openai import OpenAI
client = OpenAI()

def summarize(text):
    r = client.chat.completions.create(model="gpt-4o-mini", messages=[...])
    return r.choices[0].message.content

作者的做法是把客户端作为参数传入，并为生产环境提供一个小工厂函数：

# llm.py
import os
from openai import OpenAI

def make_client():
    return OpenAI(
        base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
        api_key=os.environ["LLM_API_KEY"],
        timeout=60,
        max_retries=2,
    )

def summarize(client, text, model):
    r = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "Summarize in 2 sentences."},
            {"role": "user", "content": text},
        ],
        temperature=0.2,
    )
    if not r.choices:
        raise ValueError("empty response")
    return (r.choices[0].message.content or "").strip()

这里有两个设计要点：model ID 和 base URL 都来自配置，这意味着切换供应商时根本不需要动业务函数；同时客户端通过参数注入，天然为测试留出了替换的"接缝"（seam）。

第二步：一个极简的 fake 客户端

做这件事其实不需要任何 mocking 库。OpenAI 客户端的访问路径是 client.chat.completions.create(...)，所以 fake 只需提供这条路径，并返回带有 .choices[0].message.content 的对象即可。

# tests/fakes.py
from types import SimpleNamespace as NS

def completion(content, finish_reason="stop", prompt_tokens=10, completion_tokens=5):
    msg = NS(content=content, role="assistant", tool_calls=None)
    return NS(
        choices=[NS(index=0, message=msg, finish_reason=finish_reason)],
        usage=NS(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens,
                 total_tokens=prompt_tokens + completion_tokens),
    )

class FakeCompletions:
    def __init__(self, replies):
        self.replies = list(replies)
        self.calls = []
    def create(self, **kwargs):
        self.calls.append(kwargs)
        reply = self.replies.pop(0)
        if isinstance(reply, Exception):
            raise reply
        return reply

class FakeClient:
    def __init__(self, *replies):
        self.chat = NS(completions=FakeCompletions(replies))

让这个 fake 真正有用的，有两点超越了"返回一个字符串"的层面：calls 列表记录了每一次请求，让你可以断言代码到底发送了什么；而回复可以是一个异常对象，让你能够脚本化地模拟一连串失败序列。这两点让 fake 从玩具变成了能捕捉真实 bug 的工具。

第三步：能真正抓到 bug 的测试

# tests/test_llm.py
import pytest
from llm import summarize
from tests.fakes import FakeClient, completion

def test_sends_expected_request():
    fake = FakeClient(completion("Short summary."))
    out = summarize(fake, "long text", model="my-model")
    assert out == "Short summary."
    call = fake.chat.completions.calls[0]
    assert call["model"] == "my-model"
    assert call["messages"][0]["role"] == "system"
    assert call["messages"][1]["content"] == "long text"

def test_none_content_becomes_empty_string():
    fake = FakeClient(completion(None))
    assert summarize(fake, "x", model="m") == ""

def test_empty_choices_raises():
    fake = FakeClient(completion("x"))
    fake.chat.completions.replies[0].choices = []
    with pytest.raises(ValueError):
        summarize(fake, "x", model="m")

其中 None content 的情况是真实存在的坑：当模型只返回一个 tool call，或者响应被过滤时，message.content 可能为 None。作者坦言，自己不止一次见过生产代码因为在 None 上调用 .strip() 而崩溃。这正是单元测试的价值所在——它逼你显式处理这些边界情况。

测试重试与错误处理

如果你用自己的重试逻辑包裹调用，就该脚本化地模拟失败。真实的异常类需要一个 response 对象，所以作者通常在单元测试里抛出自己的轻量级替身异常，并单独测试错误映射逻辑：

class FakeRateLimit(Exception):
    pass

def call_with_retry(client, attempts=3, **kw):
    for i in range(attempts):
        try:
            return client.chat.completions.create(**kw)
        except FakeRateLimit:
            if i == attempts - 1:
                raise

def test_retries_then_succeeds():
    fake = FakeClient(FakeRateLimit(), FakeRateLimit(), completion("ok"))
    r = call_with_retry(fake, model="m", messages=[])
    assert r.choices[0].message.content == "ok"
    assert len(fake.chat.completions.calls) == 3

在真实代码中，应当捕获 openai.RateLimitError 和 openai.APITimeoutError，并把异常类型或 sleep 函数注入进来，这样测试就不会真的去等待退避时间。

OpenAI Python SDK 中，openai.RateLimitError 和 openai.APITimeoutError 都继承自 openai.APIStatusError，其构造函数要求传入一个真实的 HTTP response 对象，这使得在单元测试中直接实例化它们相当繁琐。常见的解法有两种：一是如文中所示，定义轻量替身异常类（FakeRateLimit），并让重试逻辑捕获可注入的异常类型而非硬编码 openai.RateLimitError；二是使用 unittest.mock.MagicMock 构造一个伪造的 response 对象传入真实异常。前者更简洁，后者更贴近真实行为。此外，如果使用指数退避（exponential backoff），应将 time.sleep 也作为可注入参数或通过 unittest.mock.patch 替换，否则单元测试会因真实等待而变慢——这与「单元测试要在毫秒级完成」的目标背道而驰。

测试 JSON 解析

如果你的功能要求模型返回 JSON，那么解析器往往是 bug 最密集的地方。应该用模型实际会产出的各种"丑陋"输出去喂它：

import json, re

def parse_json_reply(text):
    text = text.strip()
    m = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
    if m:
        text = m.group(1)
    return json.loads(text)

作者给出一个极具价值的建议：从你的日志里收集真实的错误回复，把它们作为测试用例逐一加进去。这份列表最终会成为整个项目中最宝贵的测试文件——因为它承载了模型行为的真实分布，而不是你想象中的理想输出。

流式响应与线上冒烟测试

对于流式（streaming）响应，fake 需要返回一个由 chunk 组成的可迭代对象，形状为 chunk.choices[0].delta.content。测试要点在于断言你的代码能正确拼接这些片段，并能处理 delta.content 为 None 的 chunk（首个和末尾的 chunk 常常如此）。

由于 fake 可能逐渐偏离真实情况，作者保留了少量真实调用，默认跳过、按需运行：

流式（streaming）响应的工作原理是：OpenAI API 在 `stream=True` 时返回一个 Server-Sent Events 流，SDK 将其包装成一个可迭代对象，每次迭代产出一个 `ChatCompletionChunk` 对象。与非流式不同，完整文本需要通过拼接每个 chunk 的 `choices[0].delta.content` 字段获得。首个 chunk 的 `delta` 通常只含 `role` 字段而 `content` 为 `None`，末尾 chunk 的 `finish_reason` 会变为 `"stop"` 且 `content` 也为 `None`，因此流式处理代码必须对 `None` 做防御性检查（例如 `content = chunk.choices[0].delta.content or ""`）。在 fake 中模拟流式只需让 `create()` 返回一个普通的 Python 列表或生成器，其元素是用 `SimpleNamespace` 构造的 chunk 对象，业务代码的 `for chunk in stream` 循环无需感知任何差异。
# tests/test_live.py
import os, pytest
from llm import make_client, summarize

live = pytest.mark.skipif(not os.environ.get("RUN_LIVE"), reason="set RUN_LIVE=1")

@live
def test_live_summary_shape():
    client = make_client()
    out = summarize(client, "The cat sat on the mat. It was warm.",
                    model=os.environ["LLM_MODEL"])
    assert isinstance(out, str) and 0 < len(out) < 500

注意这里的断言：只检查类型和大致长度，绝不检查确切文本。作者只在发布前，或切换供应商、base URL、model ID 之后运行这些测试，而非每次提交都跑。当指向网关时，这也是确认 model ID 字符串是否正确的方式——错误的 ID 通常会在这里就以 404 或"model not found"报错，而不是拖到生产环境才暴露。

哪些做法应该避免

作者明确列出了几种他会跳过的反模式：

• 快照测试模型输出。模型一变它就坏，而且它对你的代码质量毫无反馈价值。
• 用 cassette 工具为每个测试录制真实响应。虽然可行，但录制的内容会过时，并掩盖响应形状的变化。作者只在少数复杂的 tool-calling 流程中使用录制。
• 在 SDK 底层 mock requests/httpx。这会让测试耦合到 SDK 内部实现。正确的做法是在客户端边界处 mock。

总结

整套方法可以浓缩为几句话：把客户端作为参数传入，fake 掉 chat.completions.create，既断言你收到了什么、也断言你发送了什么，脚本化地模拟失败，并保留一小组按需启用、只检查形状的线上测试。这样你的测试套件会变得快速、确定且免费运行，而少量的线上测试则负责捕捉 fake 无法发现的配置错误。

对于任何在生产环境中集成 LLM 的团队来说，这套"客户端边界 mock + 形状化冒烟测试"的分层思路，比盲目调用真实模型要稳健得多，也是值得借鉴的工程实践。