跳到主要内容

AI 应用中的依赖注入模式:编写经得起模型切换的代码

· 阅读需 11 分钟
Tian Pan
Software Engineer

当 OpenAI 在 2024 年 1 月停用 text-davinci-003 时,那些将该模型名称织入业务逻辑的团队花了数周时间才将其解耦。并不是因为更换模型在技术上有多难——毕竟只是一个字符串和一次 API 调用——而是因为该模型与一切都纠缠在一起:提示词构建、响应解析、错误处理、重试逻辑,所有这些都交织在一个特定的供应商会提供答案的假设中。对于中等规模的生产系统,这类迁移的工程成本估计在 5 万至 10 万美元之间,外加一个月或更长时间的工程注意力分散。

解决方案并不新奇。这是每个后端工程师都已经熟悉的模式:依赖注入。核心洞察是,你的业务逻辑应该依赖于语言模型的抽象,而不是来自 OpenAI 或 Anthropic 的具体客户端。在启动时注入具体的实现。代码的其余部分永远不需要知道接口背后是哪个供应商。

这在抽象层面上听起来显而易见。但在实践中,大多数团队在原型阶段会跳过它,之后也从未重新审视,然后在模型被弃用、供应商涨价三倍或竞争对手的模型在关键基准测试中击败自家模型时,才不得不面对这种“迁移税”。

为什么 AI 应用会与供应商耦合

耦合是自然发生的。你从 OpenAI Python SDK 开始,因为文档非常出色,而且 Playground 就在手边。你构建了问答功能、摘要生成器、代码审查器。每个功能都直接调用 client.chat.completions.create(model="gpt-4o", ...)。模型名称在六个文件中都是字符串字面量。响应解析假设了 OpenAI 的响应结构。重试逻辑包装了 OpenAI 的速率限制错误类。

三个月后,你想要评估 Claude,因为它在你的任务上表现更好。现在你不只是在更换供应商——你是在重构六个文件,重写响应解析器(Anthropic 的响应结构不同),替换错误处理,并更新你的提示词格式,因为 Anthropic 处理系统提示词的方式与 OpenAI 不同。原本应该是两小时的实验变成了一个两周的项目。

供应商 API 的碎片化使得情况比听起来更糟。Google Gemini 会拒绝带有空 items: {} 字段的结构,而 OpenAI 则接受。Anthropic 要求显式的缓存控制标记,而其他供应商则自动处理缓存。温度约束也各不相同:OpenAI 较新的推理模型不支持它,而 Anthropic 在扩展思维模式下要求必须设为 1。每个供应商都有自己的怪癖,这些怪癖会破坏那些基于另一个供应商假设编写的代码。

让你获得自由的接口

依赖倒置原则指出,高层模块应该依赖于抽象,而不是具体的实现。对于 AI 应用来说,这种抽象就是语言模型接口——一个约定“给定输入,产生输出”的合同,而不指定由哪个供应商来履行。

在 Python 中,这看起来像这样:

from abc import ABC, abstractmethod

class LanguageModel(ABC):
@abstractmethod
def invoke(self, prompt: str) -> str: ...

@abstractmethod
def invoke_messages(self, messages: list[dict]) -> str: ...

你的业务逻辑将此接口作为构造函数参数:

class DocumentSummarizer:
def __init__(self, model: LanguageModel):
self._model = model

def summarize(self, text: str) -> str:
return self._model.invoke(f"Summarize this document:\n\n{text}")

在应用启动时,你注入具体的实现:

# 生产环境
summarizer = DocumentSummarizer(model=OpenAIModel("gpt-4o"))

# 针对 Claude 的评估运行
summarizer = DocumentSummarizer(model=AnthropicModel("claude-opus-4-5"))

# 单元测试
summarizer = DocumentSummarizer(model=MockModel(fixed_response="summary text"))

DocumentSummarizer 类在上述三个场景中从未改变。你只需修改一行代码即可在模型之间进行 A/B 评估。你可以编写不需要网络调用的确定性单元测试。当一个模型被弃用时,你更新的是启动配置,而不是业务逻辑。

接口需要涵盖的内容

一个最小化的语言模型接口需要处理的不只是文本生成。生产级接口通常包括:

  • 同步和流式生成:某些用例需要完整响应;另一些则需要将 token 流式传输到 UI。
  • 结构化输出:许多供应商现在支持 JSON schema 强制执行;你的接口应该抽象这一点,以便调用者声明输出形状,而不是供应商特定的强制机制。
  • Token 计数:成本控制和上下文管理需要知道提示词在发送前消耗了多少 token。
  • 错误规范化:速率限制、上下文长度错误和安全拒绝在不同供应商之间的表现各不相同。你的接口应该将这些转换为统一的异常层次结构,这样调用者就不必处理特定于供应商的错误。

检索器(retriever)——即 RAG 中获取相关文档的组件——需要一个平行的接口。工具服务器(tool server)——执行函数调用的组件——也需要一个。这些组件都有特定于供应商的实现,且其变化独立于你的核心逻辑。

配置驱动的供应商选择

一旦你拥有了接口,下一步就是让供应商选择变成配置驱动,而非代码驱动。这种模式是一个供应商注册表 (provider registry):一个从字符串键到产生具体实现的工厂函数的映射。

PROVIDER_REGISTRY = {
"openai:gpt-4o": lambda: OpenAIModel("gpt-4o"),
"anthropic:claude-opus-4-5": lambda: AnthropicModel("claude-opus-4-5"),
"gemini:flash-2.0": lambda: GeminiModel("gemini-flash-2.0"),
}

def build_model(config_key: str) -> LanguageModel:
factory = PROVIDER_REGISTRY.get(config_key)
if not factory:
raise ValueError(f"Unknown model key: {config_key}")
return factory()

你的应用程序从环境变量或配置文件中读取模型键:

model = build_model(os.environ["LLM_MODEL_KEY"])

这种模式让你只需更改配置即可切换供应商,无需更改代码。它还能实现更复杂的行为:将不同任务路由到不同模型,实现回退链 (fallback chains)(例如:尝试 Claude,如果遇到速率限制则回退到 GPT-4o),或应用预算控制(当月度支出超过阈值时使用更便宜的模型)。

LiteLLM 已将这种模式工业化为一个代理服务器,通过统一的 OpenAI 兼容 API 支持超过 100 个供应商。它处理跨供应商的负载均衡、供应商不可用时的回退路由、用于多实例部署的基于 Redis 的速率限制跟踪,以及成本核算。运行高容量 AI 工作负载的团队使用它来实现与供应商无关的路由,且网关开销低于 15 微秒。对于规模较小的团队,它也提供 Python SDK,在没有服务器基础设施的情况下封装了相同的抽象。

测试红利

这种模式带来的测试收益是即时且具体的。如果没有依赖注入,测试 AI 应用程序逻辑要么需要调用真实的 API(速度慢、成本高、非确定性),要么需要在模块层级对 SDK 进行补丁 (patching)(脆弱,且需要对库的内部实现有深入了解)。

有了合适的模型接口,你只需编写一次模拟实现 (mock implementation):

class MockModel(LanguageModel):
def __init__(self, responses: dict[str, str] | str):
self._responses = responses

def invoke(self, prompt: str) -> str:
if isinstance(self._responses, str):
return self._responses
return self._responses.get(prompt, "default response")

现在,每个测试都能获得快速、确定性且零成本的执行。你可以测试你的摘要生成器是否发送了正确的提示词格式,你的重试逻辑是否在错误时触发,你的结构化输出解析器是否处理了边缘情况 —— 这一切都无需网络调用。原本需要几分钟(等待 API 响应)的测试套件现在只需几秒钟即可运行完毕。

模拟实现也成为了设计良好接口的强制机制 (forcing function)。如果模拟实现很难编写,说明接口过于宽泛。保持模拟实现简单的压力会促使你走向精简、专注的抽象。

框架提供的开箱即用功能

一些框架已经将这种抽象标准化,因此你不需要自己构建。

LangChain 的 ChatModel 接口通过通用的 .invoke() 方法封装了 50 多个供应商。从 ChatOpenAI 切换到 ChatAnthropicChatGoogleGenerativeAI 只需一行代码的改动。权衡之处在于 LangChain 的抽象具有较强的倾向性 (opinionated) —— 它在消息格式、回调系统和链式组合方面做出了选择,你必须接受这些选择或绕过它们。

LlamaIndex 采用了更模块化的方法,将生态系统拆分为专注的包。你可以只使用它的检索抽象而不采用其代理框架,反之亦然。这使得逐步采用变得更加容易。

Haystack 将应用程序构建为显式的管道 —— 每一个节点都是可替换组件的有向图。这在架构层级强制执行了抽象:如果不通过定义的接口,你根本无法将组件直接连接到供应商。希望通过护栏防止耦合的团队会发现这种结构非常有价值。

对于企业级 Java 应用程序,Spring AI 和 LangChain4j 都应用了传统的依赖注入 —— 就像 Java 工程师在数据库和消息队列中使用 Spring 的 @Bean@Service 模式一样 —— 来处理 AI 组件。语言模型变成了另一个注入的依赖,在 application.yml 中配置,并使用标准 Spring 测试工具在测试中进行替换。

这种模式何时会显得过度

对于生命周期只有 6 个月的原型或内部工具,构建完整的供应商抽象层可能并不值得。如果你确定会固定使用一个供应商,且选错的代价很低,那么这种抽象只会增加认知负荷而没有收益。

一个有用的启发式准则:如果你预计代码将在生产环境中运行一年以上,或者你预见到需要进行模型评估周期并在实际流量中对比供应商,那么请从一开始就构建抽象。早期添加它的增量成本很小;而在积累了 6 个月的耦合后再进行事后改造,成本将非常巨大。

另一个信号是团队规模。在个人项目中,你可以全凭记忆处理耦合并快速重构。但在一个十人团队中,每一个接触代码的新工程师都会对模型做出假设,而这些假设会逐渐固化。接口既是团队的沟通工具,也是一种技术机制。

你正在累积的迁移税

在当前市场中,模型的生命周期在面临弃用或促使切换的重大能力转型之前,通常只有 12–18 个月。OpenAI 的弃用周期、Anthropic 的 Claude 版本更迭以及 Google 的 Gemini 推广都已证明,你今天基于其构建的模型,两年后将不再是你运行的模型。关键在于,你是选择通过清晰的抽象来持续支付迁移成本(让每次升级都变得低廉),还是选择一次性支付(一个导致功能开发停滞数月的重构项目)。

那些在抽象层进行投入的团队报告称,模型升级只需要几小时而非几周——只需更改配置、针对评估集(eval set)进行验证,然后推送到生产环境。而那些没有这样做的团队,则是在工程博客上发表“为期六周的迁移历程”的那些。

这种模式并不新鲜。这与工程师在 ORM 时代应用到数据库、云存储提供商以及消息队列上的依赖倒置(dependency inversion)如出一辙。AI 组件只是最新的外部依赖项,将其视为“外部”会大有裨益:通过定义的接口注入,在不触动业务逻辑的情况下可随时更换,并且在测试中可被模拟(mockable)。

为期 2-3 个月的迁移税是可选的。这是你在项目开始时就做出的选择。

References:Let's stay in touch and Follow me for more thoughts and updates