多模型接入怎么设计：一个独立 AI 调用层的架构与取舍

接入第一个大模型时，事情往往很简单：安装 SDK，填入 API Key，把消息发出去，再把返回的文字显示出来。

接入第二家以后，问题才开始变得有意思。

两家都支持聊天，也都支持流式输出，但消息怎么传、工具调用怎么返回、思考强度怎么设置，可能各有一套写法。即使都兼容 OpenAI API，细节也未必相同。业务代码里很容易开始出现这样的判断：“如果是这家，就多传一个字段；如果是那家，就从另一个位置取结果。”

我在设计一个对话应用时，希望把这些差异收在一个独立的 AI 层里。应用负责组织对话，AI 层负责完成模型调用。以后换一个应用，这部分代码仍然能拿过去使用。

本文讨论的就是这个边界：如何给调用方一个稳定的接口，同时保留不同模型真实存在的能力差异。设计参考了 pi 的 AI 包，实现采用 Python。下面会用具体接口解释设计，但重点放在这些接口为什么需要存在，以及它们分别解决什么问题。

1. 多模型接入，难在哪里？

假设应用要让模型回答一句“解释一下递归”。只看最终文字，几家服务似乎没有什么区别。可一旦把完整过程展开，差异就出来了。

输入可能包含系统指令、图片、历史回复和工具结果。输出可能先出现文字，再出现一个工具调用；工具参数还是分几个片段传回来的。一次请求也可能在已经返回一半内容后断开。

调用方需要知道的不只是“返回了什么”，还包括“现在发生到哪一步”和“最后为什么停止”。

我把需要处理的问题分成三组：

表达差异。 同一段对话，在不同 API 里可能使用不同的消息结构；同一个思考强度，也可能放在不同字段中。
能力差异。 某个模型能接收图片，另一个只能接收文字；某个模型允许关闭思考，另一个始终会思考。
运行过程。 流式结果如何消费，请求如何取消，客户端何时关闭，失败以后能否重试。

这三组问题不能用同一种办法处理。表达差异适合转换，能力差异需要明确展示，运行过程则需要一套稳定的约定。

如果把它们都塞进一个“通用参数字典”，调用方虽然少看见几个类型，却要自己记住每家供应商的例外。复杂度并没有消失。

2. 先划清边界：这一层负责到哪里？

在写接口之前，我先确定了一件事：AI 层收到的内容，应该已经是上层决定要交给模型看的内容。

例如，一个模拟面试应用可能有身份设定、面试要求、用户简历、历史消息和压缩后的摘要。选择哪些材料、按什么顺序组合、什么时候压缩，属于上下文管理。什么时候开始一次生成、工具执行后是否继续调用，属于上层运行时的调度。

AI 层接收组织好的输入，把它转换成目标供应商的请求，然后把输出整理成调用方能处理的结果。

它仍然会处理消息格式。例如，系统指令应该放进哪个字段，工具结果怎样关联之前的工具调用，图片怎样编码。这属于请求适配，不意味着它要替应用决定保留哪段历史。

整体边界可以画成下面这样。图中的箭头表示使用关系，具体请求过程会在后面单独展开。

flowchart TB
    subgraph APP[应用]
        direction LR
        C[上下文管理]
        R[runtime 调度]
        T[工具执行]
        R --> C
        R --> T
    end

    subgraph AI[独立 AI 层]
        M[Models：查询与调用入口]
        P[Provider：供应商组合]
        A[认证解析]
        D[API 协议适配器]
        M --> P
        P --> A
        P --> D
    end

    R --> M
    D --> SDK[供应商 SDK]
    SDK --> S[模型服务]

这个边界对工具调用尤其重要。模型返回“调用某个工具”时，AI 层负责整理工具名称、调用 ID 和参数，也可以校验参数是否符合声明。真正执行工具、处理权限、决定是否继续下一轮，仍然由应用负责。

这样做以后，AI 包就不必依赖应用的数据库、Web 框架或语音模块。它可以被命令行程序、后台任务或另一个 Web 应用复用。独立性首先来自职责，放进单独的目录只是最后一步。

3. 用 Model、Provider、Models 组织接入能力

这一层的核心结构围绕三个对象展开：Model、Provider 和 Models。名字看起来接近，但它们解决的是不同问题。

Model：描述这次要调用什么

Model 是一份模型描述，包含模型 ID、所属供应商、使用的 API 协议、服务地址，以及图片、工具、思考档位等能力信息。

它本身不持有连接，也不保存密钥。

把这些信息放在一起，是为了让一次调用的目标足够明确。调用方拿到一个 Model，就能知道要用什么模型，以及它提供哪些选项。实际发送时，也不用再到几个全局变量里拼凑模型 ID 和地址。

这里有两个字段需要特别区分：

model.provider  # 供应商身份，例如 deepseek
model.api       # API 协议，例如 openai-completions

DeepSeek 可以使用 OpenAI 兼容协议，但请求仍然发往 DeepSeek，使用的也仍然是 DeepSeek 的凭据。协议名称不能替代供应商身份。

Provider：把一家供应商需要的能力组合起来

一个 Provider 提供模型目录、认证方式和可使用的协议适配器。

例如，创建 DeepSeek Provider 时，就把它的模型描述、密钥读取方式、默认服务地址和兼容配置组合好。应用通常只需要配置 API Key，再从目录中选择模型，不必每次手写这些信息。

provider = create_deepseek_provider()

创建 Provider 和查询目录都不需要向供应商请求模型列表。这里的目录是包内维护的静态数据；真正发起生成请求时，才需要有效凭据。

认证也单独处理。请求可以明确提供凭据，否则从凭据存储或供应商对应的环境变量读取。这样，模型目录可以安全地展示给应用，而不用把密钥混进模型配置。

Models：给调用方一个稳定入口

Models 管理当前应用注册的 Provider，并提供模型查询、认证解析和调用入口。

models.set_provider(provider, owns_provider=True)

available = models.get_models("deepseek")
model = models.get_model("deepseek", "deepseek-flash")

这里的 "deepseek" 是供应商 ID。先找到对应 Provider，再从它的目录中查找模型。

模型目录服务于“发现与选择”。发起调用时，Models 根据传入 Model 的供应商身份找到 Provider，Provider 再根据 API 协议选择适配器。目录不承担调用白名单的职责；应用也可以自行构造一个完整的 Model，再交给已经注册的 Provider。

这个安排保留了两个独立的扩展方向：新增一家使用现有协议的供应商，通常只需要组合配置；新增一种协议，则需要实现新的适配器。如果新供应商存在独有的认证或请求行为，也要补相应支持，不能仅凭“兼容”两个字就假定全部可用。

供应商与协议分开以后，这两种变化就不必都挤进同一批条件分支。

4. 一次模型调用，应该提供怎样的接口？

先统一输入和结果

调用方传入的是 Context：系统指令、消息和工具声明。消息使用 UserMessage、AssistantMessage、ToolResultMessage 等明确的类型，内容可以包含文字或图片。

这份输入表达的是“要交给模型的信息”。它还不是可以直接发送的 HTTP 请求。不同协议对消息、系统指令和工具定义的要求，留给适配器处理。

返回结果也不能只有一个字符串。AssistantMessage 需要容纳文字、思考内容、工具调用、用量和停止原因。否则，应用遇到工具调用或中途失败时，仍然得重新钻进 SDK 的返回对象里查找细节。

与此同时，有些供应商返回的信息只有再次调用同一个模型时才有意义，例如思考内容对应的签名。这些信息也不能简单丢掉。统一消息结构可以保留协议相关的附加数据，让后续调用继续使用；换模型时，再由适配器判断哪些内容可以重放。

流式事件与最终结果，分别服务于什么？

应用通常有两种需要：界面要尽快显示新文字，保存记录时又要拿到一份完整结果。

因此，流对象同时提供异步迭代和 result()。前者观察过程，后者取得最终消息。

下面是一段完整的使用示例。它需要安装相应 SDK，并提前配置 DEEPSEEK_API_KEY。

import asyncio

from ai import Context, Models, SimpleOptions, UserMessage
from ai.providers.deepseek import create_deepseek_provider


async def main():
    async with Models() as models:
        models.set_provider(
            create_deepseek_provider(),
            owns_provider=True,
        )
        model = models.get_model("deepseek", "deepseek-flash")
        if model is None:
            raise ValueError("模型不在当前目录中")

        stream = models.stream_simple(
            model,
            Context(messages=[UserMessage("解释一下递归。")]),
            SimpleOptions(reasoning="low"),
        )

        async for event in stream:
            if event.type == "text_delta":
                print(event.delta, end="", flush=True)

        result = await stream.result()
        print("\n停止原因：", result.stop_reason)
        if result.error is not None:
            print("调用失败：", result.error.message)


asyncio.run(main())

result() 等待的是同一次请求的结果，不会为了得到完整消息再调用一次模型。如果应用只需要最终结果，可以直接使用 complete_simple()。

一次调用的主要过程如下。图中把 SDK 合并在适配器一侧，避免让传输细节遮住调用关系。

sequenceDiagram
    participant App as 调用方
    participant Models as Models
    participant Provider as Provider
    participant Adapter as 协议适配器与 SDK
    participant Service as 模型服务

    App->>Models: Model、Context、调用选项
    Models-->>App: 返回流对象
    Models->>Provider: 解析认证并发起调用
    Provider->>Adapter: 选择协议，传入模型与凭据
    Adapter->>Service: 发送协议请求
    loop 接收流式输出
        Service-->>Adapter: 上游事件
        Adapter-->>Models: 统一事件
        Models-->>App: 从流中读取事件
    end
    Models-->>App: result() 返回最终 AssistantMessage

如果供应商提供的工具参数是分片 JSON，适配器会累计原始片段，并在参数完整后提供解析结果。应用可以观察工具调用正在生成，但不能拿半截 JSON 当成已经可执行的参数。

为什么还要保留协议专用入口？

统一参数适合常见调用，但供应商经常提供独有设置。例如缓存方式、推理摘要，以及某些特定的生成参数，很难都塞进一个通用选项类。

所以接口分成两组：

使用方式	流式结果	完整结果
通用参数	`stream_simple`	`complete_simple`
协议专用参数	`stream`	`complete`

两组入口使用相同的输入和返回契约，差别主要在调用选项。这样，应用在使用某个原生功能时，不必连消息处理、取消和结果消费也一起绕开。

对于选项类尚未覆盖的请求字段，还保留 on_payload 这样的 Hook。它允许调用方在发送前修改请求。这个入口能提供灵活性，但使用者需要了解目标协议，因此它更适合少量明确的扩展需求。

5. 统一到什么程度，才不会损失模型能力？

我认为，多模型接入里最容易过度设计的地方，就是试图把所有能力统一成完全一样的操作。

思考强度是一个很好的例子。一个模型可能提供 low、high、max，另一个可能还有 medium、xhigh。如果所有模型都显示同一套档位，库就必须决定：用户选了一个不支持的值，到底应该怎么替换？

替换成更低的档位，可能影响回答质量；替换成更高的档位，又可能增加耗时和费用。即使请求成功，用户真正选中的设置也已经变了。

让选项跟着模型走

这套设计选择把实际选项交给调用方：

model.reasoning_levels
# DeepSeek 示例：('off', 'low', 'high', 'max')

应用根据当前模型显示选项。用户选哪个强度，就发送哪个强度。换模型以后，仍可用的选择可以保留；已经不可用的选择，由应用恢复为默认设置，并让用户看见变化。

AI 层只处理本次调用，不替应用记住或调整上次的选择。如果直接调用代码传入了模型不提供的值，就在发送请求前返回明确错误。

这里仍然需要协议转换。例如，同样选择 high，OpenAI Responses 使用 reasoning.effort，DeepSeek Chat Completions 使用 reasoning_effort，Anthropic Messages 使用 output_config.effort。字段位置不同，但强度值不变。

这种方式比较容易解释，也比较容易排查请求。代价是，上层不能假设所有模型都拥有同一组控制项。未来如果接入只接受思考预算的模型，就要提供预算设置，或者明确使用协议专用参数。

默认设置和关闭思考，要分开表达

这两个意思经常被一个空值混在一起：

SimpleOptions()                 # 使用默认设置
SimpleOptions(reasoning="off")  # 明确关闭思考
SimpleOptions(reasoning="max")  # 明确选择 max

不填写时，先采用 Model 配置的默认值。如果模型配置也没有默认值，就不发送相关设置，交给供应商处理。供应商的默认行为可能仍然会思考。

off 则是一条明确指令。只有允许关闭的模型才提供这个选项。适配器负责把它写成对应 API 接受的形式：有的接口使用 none，有的使用 thinking.type="disabled"。

我们沿用了 pi 的 off 选项，但内部传递方式有所不同。pi 的 Agent 会把 off 转成未设置 reasoning，再由适配器处理；这里始终保留显式的 off，直到构造请求时才转换。这样，不填写和明确关闭就不会混用。

pi 还提供了统一档位到不同模型控制方式的映射。它适合希望跨模型保留一种强度偏好的调用方，例如程序自动切换模型时，仍然尽量维持“多思考一些”的设置。

直接使用模型原生选项，和维护一套跨模型映射，都有各自的用途。这里更看重用户选择与实际请求一致，因此没有把映射做成默认行为。如果以后需要自动路由多个模型，可以在那个明确的使用场景中设计转换策略。

模型目录是一份声明，不是一张通行证

目录能告诉应用有哪些模型、模型声明了哪些能力，以及应该怎样调用。它不能保证当前账户已经获得权限，也不能证明某个密钥仍然有效。

容量和价格同样如此。知道一个模型的上下文窗口，并不意味着调用层已经具备准确计算 token 或压缩上下文的能力。价格随条件变化时，也不应该随便填入一个固定值，再给应用一种精确计费的错觉。

把声明的能力、账户的权限和实际调用结果分开，很多含糊的“模型可用”判断就能解释清楚。

6. 流式调用背后的运行机制

模型请求开始以后，还会出现一组与供应商参数无关、但很影响应用体验的问题。

停止读取，不等于停止生成

假设用户关闭了聊天窗口，负责显示文字的循环退出了。这个动作是否应该顺便取消模型请求？

答案取决于应用。有些应用希望关闭窗口后继续生成并保存结果，有些应用则希望立即停止。因此，流对象不根据“有没有人继续读”猜测调用方的意图。

调用方使用单独的取消信号控制生成。下面的片段假定 models、model 和 context 已经准备好：

from ai import AbortController, SimpleOptions

controller = AbortController()
stream = models.stream_simple(
    model,
    context,
    SimpleOptions(signal=controller.signal),
)

# 用户点击停止时，由应用调用：
controller.abort()

result = await stream.result()

只从异步循环里 break，不会自动取消生成。取消等待 result() 的那个任务，也不等于取消底层请求。这样，停止显示、停止等待和停止生成就有了各自明确的含义。

取消信号会传到实际请求任务，并关闭对应的本地响应资源。不过，本地结束请求不构成供应商停止计算或退款的保证，这部分仍由服务端处理。

当前流使用主动生产的事件队列，请求不等待调用方开始迭代才启动。这个方式适合同时消费过程和最终结果，但也有明确的成本：队列没有容量上限，消费者长期落后会积累事件。它也不是广播，同一条流上的多个迭代器会竞争读取。需要多个订阅者时，应用应安排一个消费者，再自行分发。

谁创建资源，谁负责关闭？

SDK 客户端通常持有连接资源。多个调用可能复用它，因此不能在每次请求结束后都关闭，也不能让任何一个使用者随意关闭共享对象。

这里采用明确的所有权规则。注册 Provider 时，默认只是借用；传入 owns_provider=True，才表示将关闭责任交给 Models。

这就是前面例子中那句代码的含义：

models.set_provider(provider, owns_provider=True)

退出 async with Models() 时，Models 会取消自己的活动请求，并关闭自己拥有的 Provider。Provider 再关闭它拥有的适配器资源。适配器自己创建的 SDK 客户端由它关闭，外部注入的客户端默认由外部管理。

这样，一个 Provider 可以被多个 Models 借用。共享它的应用必须保证：所有使用者结束之后，真正的拥有者才能关闭它。关闭责任是明确约定，不靠引用关系猜测。

错误与重试，需要保留过程信息

一次流式请求可能已经返回部分文字，随后才出现网络错误。如果把错误简单转成一个空字符串，应用既不知道出了什么问题，也不知道之前收到的内容该怎么处理。

因此，请求失败会形成带错误信息的最终消息，并保留已经收到的内容。调用方查看停止原因，就能区分正常结束、达到输出上限、工具调用、取消和失败。错误信息中保留状态码、请求 ID 等可用线索，避免要求业务代码直接依赖某家 SDK 的异常类型。

这也不意味着任何错误都被吞进消息。例如，已经关闭的 Models 被再次使用，或者输入结构不满足入口契约，仍可能直接抛出异常。调用过程失败与接口使用错误，需要分别处理。

重试则更谨慎。尚未进入流式响应时，连接建立失败或明确允许重试的拒绝，可以按配置重试。已经返回了一半内容后，库不会自动重新生成，再把两次结果拼成一条回答。

整次生成是否重试，应由应用明确决定。否则，重复文字还只是表面问题；如果已经执行过工具，重新生成可能改变后续动作。

7. 如何判断这套设计真的有效？

我会从两个方向验证：调用行为是否正确，扩展时是否仍能维持原来的边界。

请求和结果，都要看得到

测试多模型接入时，只验证一个假的 SDK 返回了预设文本，证明不了适配器真的构造了正确请求。

更有价值的做法是运行真实的 Models、Provider、适配器和 SDK，只在 HTTP 服务边界提供可控响应。这样可以直接检查发出去的地址、认证头、消息结构和控制参数，也可以观察上游事件怎样变成统一结果。

例如，验证关闭思考时，应该检查实际请求里有没有关闭指令，以及是否错误地带上了开启时的强度。验证不支持的档位时，除了看错误消息，还应该确认请求根本没有发出去。

对于流式输出，普通文字只是起点。工具参数分片、工具结果重放、收到部分内容后失败、取消发生时的资源释放，都值得用具体场景验证。测试场景来自接口承诺，而不是为了把每个内部方法都调用一遍。

离线测试与真实联调，回答不同的问题

离线测试能稳定验证请求构造和事件处理，但不能证明账户权限、真实服务对字段的支持程度，以及网络中的取消行为都符合预期。

因此，真实模型联调仍然必要。它应该显式启用，选择具体供应商和模型，并控制调用次数与成本。离线测试负责让代码行为可重复，真实联调负责确认代码与服务能够协作。

看新增供应商时，改动落在哪里

还有一种检验设计的办法：观察新增能力时需要修改哪些模块。

如果新增一家使用现有协议的供应商，主要变化落在模型目录、认证和兼容配置中，而统一调用入口不需要增加供应商判断，说明结构达到了预期。

如果新增一种协议，则应该在适配器里处理新的请求和事件形式。上层仍然读取统一消息、观察流式事件、使用同一套取消机制。

当然，任何扩展都不可能永远只加配置。新的模型能力可能要求扩展消息结构或调用契约。关键在于能说清楚：这次变化来自新的能力，还是旧边界没有划好。

这也是我希望独立 AI 层保持的状态。调用方能用稳定的方式完成一次请求，又能明确知道当前模型有哪些选项；适配器处理协议细节，应用保留自己的上下文、调度和工具执行策略。等到下一家供应商、下一种模型真正进入项目时，再沿着这些边界扩展。