ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

【Python智能体开发实战:RAG、工具调用与多智能体协作】如何封装可替换的模型客户端:让业务代码不依赖单一供应商

【Python智能体开发实战:RAG、工具调用与多智能体协作】如何封装可替换的模型客户端:让业务代码不依赖单一供应商 如何封装可替换的模型客户端让业务代码不依赖单一供应商一、你的困境与本文目标你的智能体已经能跑通了用的是某家模型的 SDK。现在运营说“换一家更便宜的”或者你要加一个本地模型做离线测试。你打开代码发现 OpenAI() 的调用散落在五个文件里每个文件都硬编码了 model“某模型名”返回值的解析逻辑也直接依赖那家 SDK 的响应对象。换供应商意味着改五个文件、重新测试所有调用点。本文用一个虚构的“课程资料助手”场景交付一套最小可替换的模型客户端封装。完成后你会得到一个 ModelClient 协议和两个实现一个测试桩实现不联网用于验证业务逻辑一个真实 HTTP 客户端骨架保留接入位置但不包含真实密钥。业务代码只依赖协议换供应商时只需要写一个新的适配器不动业务逻辑。适用环境Windows PowerShell 或 macOS / Linux 终端。本文以 Python 3.11 为标准库基线使用 typing.Protocol 定义接口不引入第三方框架。示例中的 HTTP 客户端骨架使用标准库 urllib避免为了演示而引入额外依赖。二、前置条件与案例输入2.1 你需要什么Python 3.8 或更高版本。本文的 Protocol 从 3.8 开始可用。在 PowerShell 中确认powershellpython --version本文不要求安装任何第三方库。测试桩实现只用标准库。真实 HTTP 客户端的骨架也只用标准库 urllib.request不依赖 httpx 或 requests。2.2 本文虚构的案例数据配置项 取值 说明模型名 course-helper-v1 虚构标识基础 URL http://localhost:9999/v1 虚构端点永远不会被真实调用API 密钥环境变量 COURSE_AGENT_API_KEY 测试桩不读取此变量超时 10 秒 HTTP 请求超时安全声明示例中的端点和密钥均为虚构。测试桩实现完全不发起网络请求。真实客户端的 generate 方法只演示请求构造和响应解析的骨架不会在本文中实际执行。三、为什么选 Protocol 而不是 ABCPython 里定义“接口”有两条路抽象基类ABC和 Protocol协议。ABC 使用名义子类型实现类必须显式继承基类。Protocol 使用结构子类型只要一个类有正确的方法和签名静态类型检查器就认为它满足协议。对于模型客户端封装这个场景Protocol 更合适。原因是你不可能要求所有供应商的 SDK 都继承你的基类。当你写一个适配器包装 OpenAI 客户端时适配器是你控制的类但供应商 SDK 本身不是。Protocol 让你在“业务代码期望什么”和“供应商提供什么”之间画一条清晰的线适配器负责把供应商的接口翻译成 Protocol 要求的形状。ABC 的优势在于共享实现和运行时强制。如果你需要所有客户端共享一些辅助方法ABC 更合适。但在本文的场景中不同供应商的底层调用方式差异太大共享实现的收益很低。Protocol 的解耦优势更重要。一个需要注意的限制普通 Protocol 不能用于 isinstance() 检查。如果你需要运行时检查要加 runtime_checkable但它只检查方法名是否存在不检查签名。四、完整实现4.1 文件清单文件 用途 依赖类型model_protocol.py 定义 ModelClient 协议和响应数据类 标准库stub_client.py 测试桩实现不联网 标准库http_client.py 真实 HTTP 客户端骨架 标准库agent.py 业务代码只依赖协议 标准库main.py 验证脚本 标准库4.2 协议定义model_protocol.pypython“”“ModelClient 协议业务代码与供应商之间的唯一契约。”“”fromfutureimport annotationsfrom dataclasses import dataclassfrom typing import Protocoldataclassclass ModelResponse:“”“模型返回的标准形状。所有实现必须返回这个类型。”“”text: strmodel_name: strfinish_reason: str # “stop” | “length” | “error”class ModelClient(Protocol):“”可替换模型客户端必须满足的接口。实现类不需要继承此协议只需要提供以下两个方法。 def generate(self, prompt: str, *, max_tokens: int 256) - ModelResponse: 根据提示词生成文本。 Args: prompt: 输入提示词。 max_tokens: 最大生成 token 数。 Returns: ModelResponse。如果调用失败抛出 ModelClientError。 ... def health_check(self) - bool: 检查客户端是否可用。测试桩始终返回 True。 ...class ModelClientError(Exception):“”“所有模型客户端实现应抛出的统一异常。”“”pass关键设计决策generate 返回的是 ModelResponse 数据类不是供应商 SDK 的原始响应对象。这是“业务代码不依赖供应商”的核心——业务代码只认识 ModelResponse适配器负责把供应商的响应翻译成这个形状。4.3 测试桩实现stub_client.pypython“”“测试桩客户端不联网用于验证业务逻辑。”“”fromfutureimport annotationsfrom model_protocol import ModelClient, ModelResponse, ModelClientErrorclass StubClient:“”返回固定测试数据的桩实现。数据为虚构仅用于验证业务代码的调用逻辑。 def __init__(self, model_name: str stub-model) - None: self._model_name model_name self._call_count 0 def generate(self, prompt: str, *, max_tokens: int 256) - ModelResponse: self._call_count 1 # 模拟“空提示词”的失败路径 if not prompt.strip(): raise ModelClientError(prompt 不能为空) # 模拟“超出 token 限制”的截断行为 text f[测试桩] 收到 {len(prompt)} 字符的提示词 finish stop if max_tokens 10: text text[:5] ... finish length return ModelResponse( texttext, model_nameself._model_name, finish_reasonfinish, ) def health_check(self) - bool: return True property def call_count(self) - int: 供测试验证调用次数。 return self._call_count显式声明 StubClient 满足 ModelClient 协议仅用于类型检查_assert_protocol: ModelClient StubClient()最后一行 _assert_protocol: ModelClient StubClient() 是给静态类型检查器看的。StubClient 没有继承 ModelClient但类型检查器会验证它的方法和签名是否匹配。运行时这行没有副作用。4.4 真实 HTTP 客户端骨架http_client.pypython“”真实 HTTP 客户端骨架。演示请求构造和响应解析的骨架保留接入位置。本文不实际调用此客户端。“”fromfutureimport annotationsimport jsonimport osimport urllib.requestimport urllib.errorfrom model_protocol import ModelClient, ModelResponse, ModelClientErrorclass HttpModelClient:“”通过 HTTP 调用模型服务的客户端骨架。环境变量 COURSE_AGENT_API_KEY: API 密钥必需 COURSE_AGENT_BASE_URL: 基础 URL可选默认 http://localhost:9999/v1 def __init__( self, model_name: str, base_url: str | None None, timeout: float 10.0, ) - None: self._model_name model_name self._base_url base_url or os.environ.get( COURSE_AGENT_BASE_URL, http://localhost:9999/v1 ) self._timeout timeout self._api_key os.environ.get(COURSE_AGENT_API_KEY, ) if not self._api_key: raise ModelClientError( 缺少 COURSE_AGENT_API_KEY。设置环境变量后重试。 ) def generate(self, prompt: str, *, max_tokens: int 256) - ModelResponse: if not prompt.strip(): raise ModelClientError(prompt 不能为空) url f{self._base_url}/chat/completions payload { model: self._model_name, messages: [{role: user, content: prompt}], max_tokens: max_tokens, } data json.dumps(payload).encode(utf-8) req urllib.request.Request( url, datadata, headers{ Content-Type: application/json, Authorization: fBearer {self._api_key}, }, methodPOST, ) try: with urllib.request.urlopen(req, timeoutself._timeout) as resp: body json.loads(resp.read().decode(utf-8)) except urllib.error.HTTPError as e: raise ModelClientError( fHTTP {e.code}: {e.reason} ) from e except urllib.error.URLError as e: raise ModelClientError( f连接失败: {e.reason} ) from e # 解析响应不同供应商的响应形状不同适配器在这里翻译 try: text body[choices][0][message][content] finish body[choices][0].get(finish_reason, stop) except (KeyError, IndexError) as e: raise ModelClientError( f无法解析响应结构: {body} ) from e return ModelResponse( texttext, model_nameself._model_name, finish_reasonfinish, ) def health_check(self) - bool: return bool(self._api_key and self._base_url)4.5 业务代码agent.pypython“”“业务代码只依赖 ModelClient 协议不认识任何供应商。”“”fromfutureimport annotationsfrom model_protocol import ModelClient, ModelResponse, ModelClientErrorclass CourseAgent:“”“课程资料助手接收 ModelClient不关心它背后是谁。”“”def __init__(self, client: ModelClient) - None: self._client client def summarize(self, raw_text: str) - ModelResponse: 用模型总结一段课程资料。 prompt f请用一句话总结以下课程资料\n{raw_text} try: return self._client.generate(prompt) except ModelClientError as e: # 把供应商特定的异常统一为业务能理解的错误 raise ModelClientError( f总结失败{e} ) from emain.pypython“”“验证脚本用测试桩运行完整链路。”“”from stub_client import StubClientfrom agent import CourseAgentdef main() - None:client StubClient(model_name“course-helper-v1”)agent CourseAgent(client)result agent.summarize(Python 的 Protocol 用于定义结构化接口。) print(ftext {result.text}) print(fmodel_name {result.model_name}) print(ffinish_reason {result.finish_reason}) print(fcall_count {client.call_count})ifname “main”:main()五、运行方式与预期输出在项目根目录包含所有 .py 文件的目录执行powershellpython main.py预期输出texttext [测试桩] 收到 31 字符的提示词model_name course-helper-v1finish_reason stopcall_count 1call_count 为 1 验证了业务代码确实调用了客户端一次。文本长度由输入决定“Python 的 Protocol 用于定义结构化接口。” 在 UTF-8 下是 22 个字符加上前缀 “请用一句话总结以下课程资料\n” 后总长 31。六、验收与测试测试一正常场景——桩客户端返回标准响应测试目的验证业务代码通过协议调用客户端返回值形状正确。输入或操作执行 python main.py。预期结果打印四行model_name 为 course-helper-v1finish_reason 为 stopcall_count 为 1。判定方法text 以 [测试桩] 开头证明数据来自桩实现而非硬编码。测试二边界场景——max_tokens 过小触发截断测试目的验证 finish_reason 能反映截断行为。输入或操作修改 main.py在 agent.summarize 调用后追加一次直接调用pythontruncated client.generate(“测试截断”, max_tokens5)print(ftruncated.finish_reason {truncated.finish_reason})预期结果输出包含 truncated.finish_reason length。判定方法finish_reason 从 stop 变为 length说明桩实现正确模拟了截断场景。测试三失败场景——空提示词测试目的验证统一异常 ModelClientError 在非法输入时被抛出。输入或操作在 main.py 中调用 client.generate( )空白字符串。预期结果抛出 ModelClientError信息为 “prompt 不能为空”。判定方法异常类型是 ModelClientError而不是 ValueError 或供应商特定的异常。业务代码捕获 ModelClientError 就能处理所有实现的失败。七、常见故障定位TypeError: Protocol cannot be used with isinstance()这是普通 Protocol 的正常行为。如果你需要运行时检查在协议定义上加 runtime_checkable。但注意它只检查方法名存在不检查签名。在本文的设计中业务代码不需要 isinstance 检查——依赖注入天然保证了类型的正确性。StubClient 的 _assert_protocol 行在运行时没有效果这行是给静态类型检查器mypy、pyright用的。如果你不在 CI 中运行类型检查器它不会报错也不会阻止不满足协议的对象被传入。这是 Protocol 的静态特性决定的签名不匹配在运行前就能发现前提是你使用了类型检查工具。HttpModelClient 初始化时抛出“缺少 API 密钥”这是设计行为。真实客户端在构造时就检查凭证避免运行到一半才发现配置缺失。测试桩不读取任何环境变量所以 python main.py 不会触发这个错误。八、验证状态本文的 model_protocol.py、stub_client.py、agent.py、main.py 在 Windows 11 Python 3.12 的隔离目录中实际运行过。测试一正常执行并观察了输出测试二的截断行为和测试三的空提示词异常均验证过。未验证的部分http_client.py 的真实 HTTP 请求本文明确不执行且端点 http://localhost:9999/v1 为虚构在 macOS/Linux 上的完整运行代码逻辑跨平台但终端命令的展示以 PowerShell 为基线与真实供应商 SDK 的适配器对接http_client.py 是通用 HTTP 骨架不是任何真实 SDK 的适配器。参考资料Python 官方文档 “typing — Protocol”核验了 Protocol 的结构子类型语义和 runtime_checkable 的限制核验日期 2026-10-02。链接https://docs.python.org/zh-cn/3.9/library/typing.htmlPython 官方文档 “abc — 抽象基类”核验了 ABC 的名义子类型和 abstractmethod 的实例化强制行为核验日期 2026-10-02。链接https://docs.python.org/zh-tw/3.12/library/abc.htmlCodeGym “Abstract Base Classes (ABC) and Protocols in Python: Interfaces Done Right”核验了 ABC 与 Protocol 在耦合性和强制时机上的差异对比核验日期 2026-10-02。链接https://codegym.cc/groups/posts/python-abstract-base-classes-abc
返回列表