前面的系列已经分别介绍了模型调用、消息历史、生成参数、结构化输出、错误处理、评估和可观测性。实际项目中还有一个经常被低估的问题:业务代码不应该到处散落着 SDK 调用。本文用一个很小的“文本分类器”把模型服务封装起来,让上层只关心输入和结果,后续更换模型服务或接入离线测试时,不必重写业务逻辑。

为什么需要适配器

直接在业务函数里调用 SDK 看起来很快,但随着项目变大,代码通常会变成这样:读取环境变量、创建客户端、拼接消息、处理响应、记录日志,全都和“分类一条文本”混在一起。测试业务规则时还会不小心发出真实请求,更换服务商时也要搜索并修改许多文件。

适配器(adapter)就是一个窄接口。本文把它定义为一个 classify 方法:输入文本,返回固定的 Python 字典。具体使用哪种 SDK、模型名和请求参数,都留在适配器内部。它不是新的 AI 技术,而是一种普通的软件设计手法,价值在于缩小变化范围。

准备环境和配置

创建虚拟环境并安装官方 Python SDK:

1
2
3
python -m venv .venv
source .venv/bin/activate
python -m pip install openai

只从环境变量读取密钥和模型名。下面的值是占位符,不要把真实密钥写进代码或提交到 Git:

1
2
3
4
export OPENAI_API_KEY="替换为你的真实密钥"
export MODEL_NAME="替换为你可用的模型名称"
# 如果使用兼容接口,再设置实际地址;官方服务可不设置
export OPENAI_BASE_URL=""

本例使用 JSON 模式请求一个固定字段。不同模型对结构化输出能力和参数支持可能不同,运行前应以所选服务的官方文档为准;如果服务不支持 response_format,可以先删除该参数,并在程序中增加 JSON 解析和校验。

编写模型适配器

新建 classifier.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import json
import os
from typing import Any

from openai import OpenAI


class ModelClassifier:
def __init__(self) -> None:
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("请先设置 OPENAI_API_KEY")
self.model = os.environ.get("MODEL_NAME")
if not self.model:
raise RuntimeError("请先设置 MODEL_NAME")
base_url = os.environ.get("OPENAI_BASE_URL") or None
self.client = OpenAI(api_key=api_key, base_url=base_url)

def classify(self, text: str) -> dict[str, Any]:
if not text.strip():
raise ValueError("text 不能为空")

response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": (
"你是文本分类器,只能返回 JSON。"
"category 只能是 positive、negative 或 neutral,"
"confidence 是 0 到 1 之间的数字。"
),
},
{"role": "user", "content": text},
],
response_format={"type": "json_object"},
)
content = response.choices[0].message.content
if not content:
raise RuntimeError("模型没有返回内容")

result = json.loads(content)
category = result.get("category")
confidence = result.get("confidence")
if category not in {"positive", "negative", "neutral"}:
raise ValueError("category 不符合约定")
if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
raise ValueError("confidence 不符合约定")
return {"category": category, "confidence": float(confidence)}


if __name__ == "__main__":
classifier = ModelClassifier()
print(classifier.classify("这个版本启动很快,使用体验明显变好了。"))

这里有三个重要边界。第一,ModelClassifier 负责把环境配置转换成客户端,并发出请求。第二,classify 对输入和输出做最低限度校验,不把模型返回的任意 JSON 直接交给业务层。第三,脚本入口只是演示;其它代码只需要调用 classify,不需要知道 messages 或 choices 的结构。

运行命令如下:

1
python classifier.py

成功时会打印一个包含 category 和 confidence 的字典。具体分类和置信度取决于模型输入,不应把某一次输出当作固定结果。若要验证程序本身,优先检查字段、类型和范围,而不是硬编码某个模型答案。

让业务层不依赖 SDK

再写一个只关心业务的函数:

1
2
3
4
5
6
7
def route_feedback(text: str, classifier: ModelClassifier) -> str:
result = classifier.classify(text)
if result["confidence"] < 0.6:
return "人工复核"
if result["category"] == "negative":
return "创建客服跟进任务"
return "自动归档"

route_feedback 没有读取密钥,也没有访问 SDK 响应对象。将来若把分类换成本地模型,只要提供同样的 classify 方法,路由逻辑就无需改变。这种依赖倒置也便于测试:可以传入一个假的分类器,而不产生网络请求。

1
2
3
4
5
6
class FakeClassifier:
def classify(self, text: str) -> dict[str, float | str]:
return {"category": "negative", "confidence": 0.9}


print(route_feedback("页面打不开", FakeClassifier()))

这个测试的确定性来自假对象,不依赖模型的随机性、网络或账户余额。真实项目还应把分类结果、模型名、请求耗时和异常记录到日志中,并对超时、限流和重试设置明确策略;适配器正是集中实现这些横切能力的合适位置。

常见问题

为什么不直接把客户端作为全局变量? 小脚本可以这样做,但全局对象会让测试和多配置运行更困难。把客户端放进适配器实例,生命周期和依赖更清楚。

JSON 解析成功就代表结果可信了吗? 不是。JSON 只说明格式可解析,仍需检查枚举值、数值范围和业务规则。低置信度结果应进入人工复核,而不是自动执行高风险操作。

更换模型只改一个变量就够了吗? 不一定。不同模型的上下文容量、参数和结构化输出支持可能不同。适配器减少了改动范围,但仍要重新运行测试集,并检查成本、延迟和结果质量。

小结

本文没有增加一个更复杂的调用流程,而是把模型 SDK 和业务逻辑隔开:适配器负责配置、请求和结果校验,业务层负责规则和决策,假对象负责离线测试。这样的边界能让 AI 应用更容易替换模型、集中处理异常,也让确定性的代码不必依赖一次真实生成结果。综合项目的下一步,应是为这条接口补充固定测试集和回归检查。