前面的系列已经分别介绍了模型调用、消息历史、生成参数、结构化输出、错误处理、评估和可观测性。实际项目中还有一个经常被低估的问题:业务代码不应该到处散落着 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 的结构。
运行命令如下:
成功时会打印一个包含 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 应用更容易替换模型、集中处理异常,也让确定性的代码不必依赖一次真实生成结果。综合项目的下一步,应是为这条接口补充固定测试集和回归检查。