Skip to content
大模型接入层设计：多模型统一调用与 Provider 抽象
概述
直接调用多个模型服务时，业务代码需要同时维护各家供应商的请求结构、鉴权方式、流式事件格式和错误语义。这些细节会随着模型版本和供应商策略变化而变化，分散在调用点里的判断逻辑越多，后续维护成本越高。
本章在业务代码与模型服务之间设计一个接入层，由四个部分组成：
- 统一协议：消息、请求参数、响应结果、错误对象的固定格式；
- Provider 适配器：把统一协议翻译成各家 API 的请求，并把各家响应翻译回统一格式；
- 流式归一化：把不同的 SSE 事件形态变成同一组异步生成器事件；
- 路由与重试：按配置选择 Provider，并在可重试错误发生时切换或重试。
整个实现建立在 ES6 的 class、async/await、异步生成器和 ES Module 之上，不依赖具体厂商 SDK。目标是在保持业务代码稳定的前提下，让新增一个模型服务时只需要新增一个适配器。
直接对接多家模型 API 的问题
请求格式、响应格式与错误语义的差异
三个主流模型服务的 HTTP 接口在请求结构上并不一致。以最简单的文本对话为例：
- OpenAI Chat Completions 使用一个
messages数组，系统提示词是数组里的system消息； - Anthropic Messages 把系统提示词放在独立参数
system中，messages只允许user和assistant两种角色； - Gemini
generateContent把历史消息放在contents数组中，系统提示词放在独立参数systemInstruction中。
响应结构同样不同。OpenAI 的文本在 choices[0].message.content，Anthropic 的文本是 content 数组中的文本块，Gemini 的文本在 candidates[0].content.parts 数组中。
| 维度 | OpenAI Chat Completions | Anthropic Messages | Gemini generateContent |
|---|---|---|---|
| 系统提示词 | messages 中的 system 消息 | 独立参数 system | 独立参数 systemInstruction |
| 用户消息 | { role: 'user', content: '...' } | 同左 | { role: 'user', parts: [{ text: '...' }] } |
| 模型回复位置 | choices[0].message.content | content 文本块数组 | candidates[0].content.parts[*].text |
| 最大输出长度 | max_tokens | max_tokens（必填） | generationConfig.maxOutputTokens |
| 鉴权方式 | Authorization: Bearer | x-api-key 或 Authorization: Bearer | URL 参数 ?key= 或 Authorization: Bearer |
错误语义也不统一。401/403 表示鉴权失败，429 表示触发速率限制，5xx 表示服务端错误，这些状态码各家基本一致；但错误响应体的结构、是否携带 HTTP 标准的 Retry-After 头、限流后是否需要等待，却没有统一约定。业务代码如果直接对接三家 API，以上每一项差异都要在调用点处理。
业务代码中重复的胶水代码
假设一个业务函数需要先接入 OpenAI，再接入 Anthropic：
js
// 业务代码直接对接两家模型服务
async function chatWithOpenAI(prompt) {
const response = await fetch('https://api.openai.com/v1/chat/completions', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
},
body: JSON.stringify({
model: process.env.OPENAI_MODEL,
messages: [{ role: 'user', content: prompt }],
}),
});
const data = await response.json();
return data.choices[0].message.content;
}
async function chatWithAnthropic(prompt) {
const response = await fetch('https://api.anthropic.com/v1/messages', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'x-api-key': process.env.ANTHROPIC_API_KEY,
'anthropic-version': '2023-06-01',
},
body: JSON.stringify({
model: process.env.ANTHROPIC_MODEL,
max_tokens: 1024,
messages: [{ role: 'user', content: prompt }],
}),
});
const data = await response.json();
return data.content[0].text;
}这两个函数输入相同、输出语义相同，却没有任何共享代码。一旦把流式输出、工具调用、超时重试、错误分类加进来，每个函数都会膨胀为数百行，并且互相之间无法复用。接入层要解决的就是这个问题：把与具体服务商相关的逻辑收进适配器，让业务函数只面对一个稳定接口。
统一调用协议设计
统一协议是接入层内部使用的数据格式。它不追求覆盖所有厂商的全部能力，而是描述各家 API 的公共子集。
消息模型：system / user / assistant / tool
统一消息模型使用四种角色：
system：系统提示词；user：用户输入；assistant：模型输出，可能同时包含文本和工具调用；tool：一次工具调用的执行结果，通过toolCallId关联到某个调用。
js
// src/model.js
export class ChatMessage {
constructor({ role, content = '', toolCalls = [], toolCallId = '', name = '' }) {
this.role = role; // 'system' | 'user' | 'assistant' | 'tool'
this.content = content; // 文本内容
this.toolCalls = toolCalls; // assistant 消息携带的 ToolCall 数组
this.toolCallId = toolCallId; // tool 消息对应的调用 ID
this.name = name; // tool 消息对应的函数名
}
}
export class ToolCall {
constructor({ id, name, args = {} }) {
this.id = id;
this.name = name;
this.args = args; // 已解析为对象的参数
}
}一次带工具调用的会话，消息序列是固定模式：
js
const messages = [
new ChatMessage({ role: 'user', content: '北京今天冷吗？' }),
new ChatMessage({
role: 'assistant',
content: '我来查一下天气。',
toolCalls: [
new ToolCall({ id: 'call_1', name: 'get_weather', args: { city: '北京' } }),
],
}),
new ChatMessage({
role: 'tool',
toolCallId: 'call_1',
name: 'get_weather',
content: '晴，-2℃',
}),
];业务流程是：先发送用户消息；模型返回 assistant 消息并附带 toolCalls；应用执行函数后追加一条 tool 消息；再调用一次模型。接入层不执行工具，只负责把这条消息序列正确翻译成各家 API 的格式。
注意：统一层只依赖字段名，不要求消息必须是
ChatMessage实例。适配器代码直接读取role、content等属性，传入普通对象字面量同样可以工作。使用类是为了约束字段形状。
请求参数与模型能力映射
js
export class ChatRequest {
constructor({ model, messages, temperature, maxTokens, topP, stop, tools }) {
this.model = model; // 模型 ID，可选
this.messages = messages; // ChatMessage 数组
this.temperature = temperature;
this.maxTokens = maxTokens;
this.topP = topP;
this.stop = stop;
this.tools = tools; // 统一的工具声明数组
}
}tools 中的每个工具声明使用统一形状：
js
{
name: 'get_weather',
description: '查询指定城市的天气',
parameters: {
type: 'object',
properties: {
city: { type: 'string' },
},
required: ['city'],
},
}统一参数到各家参数的映射如下：
| 统一参数 | OpenAI | Anthropic | Gemini |
|---|---|---|---|
temperature | temperature | temperature | generationConfig.temperature |
maxTokens | max_tokens | max_tokens | generationConfig.maxOutputTokens |
topP | top_p | top_p | generationConfig.topP |
stop | stop | stop_sequences | generationConfig.stopSequences |
tools | tools（type: 'function' 包裹） | tools（扁平结构） | tools.functionDeclarations |
适配器负责把 ChatRequest 转换成上表中的请求体，因此业务代码不需要感知字段名差异。Anthropic 的 max_tokens 为必填参数 [3]。
注意：三家 API 的能力并不是完全对齐的。例如
top_k在 Anthropic 和 Gemini 原生支持，OpenAI Chat Completions 没有该参数；Anthropic 的max_tokens是必填参数，OpenAI 和 Gemini 则是可选。统一层只选取公共参数，缺失的能力要么在适配器中给出合理默认值，要么通过扩展字段透传。
统一响应结果与错误对象
js
export class ChatResult {
constructor({ id, provider, model, content = '', toolCalls = [], finishReason = 'stop', usage = {} }) {
this.id = id; // 响应 ID
this.provider = provider; // 实际返回结果的适配器名称
this.model = model; // 实际使用的模型 ID
this.content = content; // 拼接后的完整文本
this.toolCalls = toolCalls; // ToolCall 数组
this.finishReason = finishReason; // 'stop' | 'length' | 'tool_calls' | 'content_filter'
this.usage = usage; // { promptTokens, completionTokens, totalTokens }
}
}finishReason 被归一化为少量取值，适配器负责把厂商值映射过来：OpenAI 的 finish_reason、Anthropic 的 stop_reason、Gemini 的 finishReason 各不相同。
错误对象统一为 ProviderError：
js
// src/errors.js
export class ProviderError extends Error {
constructor({ type, message, provider, statusCode, retryable = false, retryAfterMs }) {
super(message);
this.name = 'ProviderError';
this.type = type; // 'auth' | 'rate_limit' | 'timeout' | 'server' | 'invalid_request' | 'network'
this.provider = provider;
this.statusCode = statusCode;
this.retryable = retryable;
this.retryAfterMs = retryAfterMs;
}
}retryable 字段是重试层判断的依据，retryAfterMs 保留服务端在 429 响应中给出的等待时间。
说明：统一消息模型与 OpenAI Chat Completions 的
messages结构基本一致。OpenAI 后来推出的 Responses API 改用Items联合类型（message、function_call、function_call_output等），并移除了并行生成参数n[1]。统一层选择messages形态，是因为它与 Anthropic、Gemini 的对话历史结构更接近，迁移成本较低。如果业务大量使用 Responses API 的Items语义，统一消息模型需要另行扩展。
Provider 适配器接口
最小适配器接口：chat、stream 与工具调用
适配器只要求两个方法：
chat(request)：发送非流式请求，返回Promise<ChatResult>；stream(request)：发送流式请求，返回异步生成器，逐块产出统一流事件。
这两个方法是接入层自定义的接口约定，不是任何厂商 SDK 的方法签名。工具调用不是一个独立方法。它通过 ChatRequest.tools 传入，并在 ChatResult.toolCalls 或流式事件的 tool_call_delta 中返回。工具调用的差异本质上仍然是请求/响应格式差异，归入适配器的消息转换职责即可，不需要增加接口数量。
使用 ES6 类实现适配器基类
基类负责三件公共事务：保存 API 配置、发送 HTTP 请求、把非 2xx 响应归类为 ProviderError。
js
// src/provider/base.js
import { ProviderError } from '../errors.js';
export class ProviderAdapter {
constructor(config = {}) {
this.name = 'base';
this.apiKey = config.apiKey;
this.baseURL = config.baseURL;
this.model = config.model;
this.timeoutMs = config.timeoutMs || 30000;
}
async chat(request) {
throw new Error(`${this.name} 适配器没有实现 chat(request)`);
}
async *stream(request) {
throw new Error(`${this.name} 适配器没有实现 stream(request)`);
}
async requestRaw(path, { method = 'POST', body, headers = {}, timeoutMs = this.timeoutMs } = {}) {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), timeoutMs);
let response;
try {
response = await fetch(`${this.baseURL}${path}`, {
method,
headers: { 'Content-Type': 'application/json', ...headers },
body: body === undefined ? undefined : JSON.stringify(body),
signal: controller.signal,
});
} catch (error) {
if (error.name === 'AbortError') {
throw new ProviderError({
type: 'timeout',
message: `请求超时（${timeoutMs}ms）`,
provider: this.name,
retryable: true,
});
}
throw new ProviderError({
type: 'network',
message: error.message,
provider: this.name,
retryable: true,
});
} finally {
clearTimeout(timer);
}
if (!response.ok) {
throw await this.classifyError(response);
}
return response;
}
async requestJSON(path, options) {
const response = await this.requestRaw(path, options);
const contentType = response.headers.get('content-type') || '';
if (contentType.includes('application/json')) {
return response.json();
}
return response.text();
}
async classifyError(response) {
let message = response.statusText;
try {
const detail = await response.json();
message = detail?.error?.message || message;
} catch {
// 响应体不是 JSON，保留状态文本
}
let type = 'invalid_request';
let retryable = false;
if (response.status === 401 || response.status === 403) {
type = 'auth';
} else if (response.status === 429) {
type = 'rate_limit';
retryable = true;
} else if (response.status >= 500) {
type = 'server';
retryable = true;
}
return new ProviderError({
type,
message,
provider: this.name,
statusCode: response.status,
retryable,
retryAfterMs: this.parseRetryAfter(response),
});
}
parseRetryAfter(response) {
const value = response.headers.get('retry-after');
if (!value) return undefined;
const seconds = Number(value);
if (Number.isFinite(seconds)) return seconds * 1000;
const time = Date.parse(value);
return Number.isFinite(time) ? Math.max(0, time - Date.now()) : undefined;
}
}requestJSON 适用于普通 JSON 请求。流式方法需要直接消费响应体，因此使用 requestRaw 拿到原始 Response 后再读取 response.body。
注意：
requestRaw中的超时只覆盖到响应头到达之前。SSE 流开始后，相邻事件之间的等待需要由流的消费方另行控制，避免长时间没有新数据却一直占用连接。
注意：基类不添加任何鉴权头。OpenAI 使用
Authorization: Bearer，Anthropic 使用x-api-key和anthropic-version，Gemini 使用 URL 查询参数key。这些差异由各适配器自己处理。
OpenAI 适配器实现
chat 与 messages 映射
OpenAI 适配器把统一消息转换成 Chat Completions 的消息格式：
js
// src/provider/openai.js
import { ProviderAdapter } from './base.js';
import { ChatResult, ToolCall } from '../model.js';
export class OpenAIAdapter extends ProviderAdapter {
constructor(config) {
super(config);
this.name = 'openai';
this.baseURL = config.baseURL || 'https://api.openai.com/v1';
}
buildHeaders() {
return { Authorization: `Bearer ${this.apiKey}` };
}
toOpenAIMessages(messages) {
return messages.map((m) => {
if (m.role === 'assistant' && m.toolCalls.length > 0) {
return {
role: 'assistant',
content: m.content || null,
tool_calls: m.toolCalls.map((call) => ({
id: call.id,
type: 'function',
function: {
name: call.name,
arguments: JSON.stringify(call.args || {}),
},
})),
};
}
if (m.role === 'tool') {
return {
role: 'tool',
tool_call_id: m.toolCallId,
content: m.content,
};
}
return { role: m.role, content: m.content };
});
}
buildRequest(request) {
const body = {
model: request.model || this.model,
messages: this.toOpenAIMessages(request.messages),
};
if (request.temperature !== undefined) body.temperature = request.temperature;
if (request.maxTokens !== undefined) body.max_tokens = request.maxTokens;
if (request.topP !== undefined) body.top_p = request.topP;
if (request.stop !== undefined) body.stop = request.stop;
if (request.tools && request.tools.length > 0) {
body.tools = request.tools.map((tool) => ({
type: 'function',
function: {
name: tool.name,
description: tool.description,
parameters: tool.parameters,
},
}));
}
return body;
}
async chat(request) {
const data = await this.requestJSON('/chat/completions', {
headers: this.buildHeaders(),
body: this.buildRequest(request),
});
const choice = data.choices?.[0];
const message = choice?.message || {};
return new ChatResult({
id: data.id,
provider: this.name,
model: data.model,
content: message.content || '',
toolCalls: (message.tool_calls || []).map(
(tc) => new ToolCall({
id: tc.id,
name: tc.function?.name,
args: JSON.parse(tc.function?.arguments || '{}'),
})
),
finishReason: this.mapFinishReason(choice?.finish_reason),
usage: {
promptTokens: data.usage?.prompt_tokens,
completionTokens: data.usage?.completion_tokens,
totalTokens: data.usage?.total_tokens,
},
});
}
mapFinishReason(reason) {
if (reason === 'function_call') return 'tool_calls';
return reason || 'stop';
}
}两个关键转换点：
- OpenAI 的
function.arguments是一个 JSON 字符串，统一层中的ToolCall.args是解析后的对象； - 工具执行结果在 OpenAI 中是独立消息
{ role: 'tool', tool_call_id }，适配器直接从统一tool消息映射。
使用示例：
js
const adapter = new OpenAIAdapter({
apiKey: process.env.OPENAI_API_KEY,
model: process.env.OPENAI_MODEL,
});
const result = await adapter.chat(new ChatRequest({
messages: [
new ChatMessage({ role: 'system', content: '你是一名前端架构师。' }),
new ChatMessage({ role: 'user', content: '解释一下 Provider 模式。' }),
],
temperature: 0.7,
}));
console.log(result.content);示例中 temperature: 0.7 只是普通参数。temperature 为 undefined 时不会出现在请求体里，由服务端使用默认值。
OpenAI 兼容服务的处理
由于 baseURL 和 apiKey 都是构造参数，同一个 OpenAIAdapter 可以直接对接所有提供 /chat/completions 的兼容服务。例如 Google Vertex AI 提供了 OpenAI 兼容端点，把 baseURL 指向该端点，apiKey 传入 Google Cloud 访问令牌即可 [2]：
js
const vertexAdapter = new OpenAIAdapter({
baseURL: `https://${location}-aiplatform.googleapis.com/v1/projects/${projectId}/locations/${location}/endpoints/openapi`,
apiKey: googleAccessToken, // 短时有效的 Google Cloud 访问令牌
model: 'google/gemini-2.0-flash-001',
});兼容服务指的是请求和响应结构都模仿 Chat Completions 的服务，包括各类模型网关和本地推理服务。它们通常会保留 OpenAI 的 SSE 事件结构，但可能在限流响应头、工具调用参数格式等细节上存在偏差。适配器能处理的是结构一致的场景，细节差异需要通过契约测试（见「测试策略」一节）来发现和锁定。
注意：
OpenAIAdapter的chat方法始终把apiKey放进Authorization: Bearer。对于需要其它鉴权方式的服务，应新增适配器，而不是在 OpenAI 适配器里堆条件分支。
Anthropic 适配器实现
消息转换与 system 指令
Anthropic Messages API 的调用方式是 POST /v1/messages，参数包括 model、messages、system、temperature、top_p、top_k、stop_sequences、stream 等；鉴权支持 x-api-key 和 Authorization: Bearer 两种方式 [3]。
Anthropic 适配器需要处理两个差异：
system是独立参数，要从统一消息中拆出来；messages中只允许user和assistant角色，工具结果必须作为tool_result块放进user消息。
js
// src/provider/anthropic.js
import { ProviderAdapter } from './base.js';
import { ChatResult, ToolCall } from '../model.js';
export class AnthropicAdapter extends ProviderAdapter {
constructor(config) {
super(config);
this.name = 'anthropic';
this.baseURL = config.baseURL || 'https://api.anthropic.com';
this.version = config.version || '2023-06-01';
}
buildHeaders() {
return {
'x-api-key': this.apiKey,
'anthropic-version': this.version,
};
}
toAnthropicMessages(messages) {
const systemParts = [];
const apiMessages = [];
for (const m of messages) {
if (m.role === 'system') {
systemParts.push(m.content);
continue;
}
if (m.role === 'assistant') {
const content = [];
if (m.content) content.push({ type: 'text', text: m.content });
for (const call of m.toolCalls) {
content.push({ type: 'tool_use', id: call.id, name: call.name, input: call.args });
}
apiMessages.push({ role: 'assistant', content });
continue;
}
if (m.role === 'tool') {
const block = { type: 'tool_result', tool_use_id: m.toolCallId, content: m.content };
const last = apiMessages[apiMessages.length - 1];
if (last && last.role === 'user') {
if (typeof last.content === 'string') {
last.content = [{ type: 'text', text: last.content }];
}
last.content.push(block);
} else {
apiMessages.push({ role: 'user', content: [block] });
}
continue;
}
apiMessages.push({ role: m.role, content: m.content });
}
return {
system: systemParts.join('\n\n'),
messages: apiMessages,
};
}
buildRequest(request) {
const { system, messages } = this.toAnthropicMessages(request.messages);
const body = {
model: request.model || this.model,
max_tokens: request.maxTokens || 1024,
messages,
};
if (system) body.system = system;
if (request.temperature !== undefined) body.temperature = request.temperature;
if (request.topP !== undefined) body.top_p = request.topP;
if (request.stop !== undefined) {
body.stop_sequences = Array.isArray(request.stop) ? request.stop : [request.stop];
}
if (request.tools && request.tools.length > 0) {
body.tools = request.tools.map((tool) => ({
name: tool.name,
description: tool.description,
input_schema: tool.parameters,
}));
}
return body;
}
async chat(request) {
const data = await this.requestJSON('/v1/messages', {
headers: this.buildHeaders(),
body: this.buildRequest(request),
});
const text = data.content
.filter((block) => block.type === 'text')
.map((block) => block.text)
.join('');
const toolCalls = data.content
.filter((block) => block.type === 'tool_use')
.map((block) => new ToolCall({
id: block.id,
name: block.name,
args: block.input || {},
}));
return new ChatResult({
id: data.id,
provider: this.name,
model: data.model,
content: text,
toolCalls,
finishReason: this.mapFinishReason(data.stop_reason),
usage: {
promptTokens: data.usage?.input_tokens,
completionTokens: data.usage?.output_tokens,
totalTokens: (data.usage?.input_tokens || 0) + (data.usage?.output_tokens || 0),
},
});
}
mapFinishReason(reason) {
switch (reason) {
case 'max_tokens':
return 'length';
case 'tool_use':
return 'tool_calls';
default:
return 'stop'; // end_turn、stop_sequence
}
}
}注意 toAnthropicMessages 中连续 tool 消息的合并逻辑：统一模型里每个工具结果单独一条消息，而 Anthropic 要求多个 tool_result 块放在同一条 user 消息中。如果上一条 user 消息是字符串，还需要先转换为 text 块再追加 tool_result 块。
注意：Anthropic 要求
messages中的 assistant 消息携带非空content数组。如果模型返回的消息既没有文本也没有工具调用（很少见），该请求可能会被服务端拒绝。业务层应避免构造这种空消息。
tool use / tool result 的归一化
OpenAI 与 Anthropic 的工具调用格式存在明显差异 [4]：
- OpenAI 用
{ type: 'function', function: {...} }包裹工具声明，参数 schema 字段名为parameters； - Anthropic 的工具是扁平结构，字段名为
input_schema； - OpenAI 的工具调用结果位于
message.tool_calls[i].function.arguments，内容是 JSON 字符串； - Anthropic 位于
content数组中type: 'tool_use'的块，input是已解析的对象； - OpenAI 的工具结果消息是
{ role: 'tool', tool_call_id, content }，Anthropic 是content中的tool_result块。
上面对应的适配器代码已经把这些差异映射到 ToolCall 和统一 tool 消息，业务层不需要感知 input_schema 还是 parameters、字符串还是对象。
Google Gemini 适配器实现
内容结构与 generationConfig 映射
Gemini 的 generateContent 把对话历史放在 contents 数组中，每条内容由 role（user 或 model）和 parts 组成；系统提示词放在 systemInstruction 中。
js
// src/provider/gemini.js
import { ProviderAdapter } from './base.js';
import { ChatResult, ToolCall } from '../model.js';
export class GeminiAdapter extends ProviderAdapter {
constructor(config) {
super(config);
this.name = 'gemini';
this.baseURL = config.baseURL || 'https://generativelanguage.googleapis.com/v1beta';
}
toGeminiRequest(request) {
const contents = [];
const systemParts = [];
for (const m of request.messages) {
if (m.role === 'system') {
systemParts.push({ text: m.content });
} else if (m.role === 'assistant') {
const parts = [];
if (m.content) parts.push({ text: m.content });
for (const call of m.toolCalls) {
parts.push({ functionCall: { name: call.name, args: call.args } });
}
contents.push({ role: 'model', parts });
} else if (m.role === 'tool') {
const part = {
functionResponse: {
name: m.name,
response: { result: m.content },
},
};
const last = contents[contents.length - 1];
if (last && last.role === 'user') {
last.parts.push(part);
} else {
contents.push({ role: 'user', parts: [part] });
}
} else {
contents.push({ role: 'user', parts: [{ text: m.content }] });
}
}
const body = { contents };
if (systemParts.length > 0) body.systemInstruction = { parts: systemParts };
const config = {};
if (request.temperature !== undefined) config.temperature = request.temperature;
if (request.maxTokens !== undefined) config.maxOutputTokens = request.maxTokens;
if (request.topP !== undefined) config.topP = request.topP;
if (request.stop !== undefined) {
config.stopSequences = Array.isArray(request.stop) ? request.stop : [request.stop];
}
if (Object.keys(config).length > 0) body.generationConfig = config;
if (request.tools && request.tools.length > 0) {
body.tools = [{
functionDeclarations: request.tools.map((tool) => ({
name: tool.name,
description: tool.description,
parameters: tool.parameters,
})),
}];
}
return body;
}
async chat(request) {
const model = request.model || this.model;
const data = await this.requestJSON(
`/models/${model}:generateContent?key=${encodeURIComponent(this.apiKey)}`,
{ body: this.toGeminiRequest(request) }
);
const candidate = data.candidates?.[0];
const parts = candidate?.content?.parts || [];
const text = parts
.filter((p) => p.text)
.map((p) => p.text)
.join('');
const toolCalls = parts
.filter((p) => p.functionCall)
.map((p, index) => new ToolCall({
id: `fc-${index}`,
name: p.functionCall.name,
args: p.functionCall.args || {},
}));
return new ChatResult({
id: data.responseId || `gemini-${candidate?.index ?? 0}`,
provider: this.name,
model: data.modelVersion || model,
content: text,
toolCalls,
finishReason: this.mapFinishReason(candidate?.finishReason),
usage: {
promptTokens: data.usageMetadata?.promptTokenCount,
completionTokens: data.usageMetadata?.candidatesTokenCount,
totalTokens: data.usageMetadata?.totalTokenCount,
},
});
}
mapFinishReason(reason) {
switch (reason) {
case 'MAX_TOKENS':
return 'length';
case 'SAFETY':
return 'content_filter';
default:
return 'stop';
}
}
}generateContent 的 REST 端点为 POST https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent [5]。functionResponse 在 REST 接口中作为一条 user 消息的 part 发送，与模型返回的 functionCall 通过函数名关联 [5]。
注意：Gemini 的
functionCall没有 OpenAI 那样的调用 ID 字段。统一消息中的id由适配器生成，主要用于维持消息序列的一致性；真正把结果关联回调用的是functionResponse.name，因此统一tool消息必须携带name。
候选响应到统一结果的转换
Gemini 的响应结构是 candidates 数组，每个候选包含一个 content，文本在 parts[*].text，工具调用在 parts[*].functionCall。上面的 chat 方法取第一个候选，把多个 parts 拼接为 content，把 functionCall 转换为 ToolCall。
三个提供商的文本提取路径各不相同，但经过适配器之后，业务代码拿到的都是 ChatResult.content。
流式输出归一化
SSE 事件解析
三家提供商都使用 SSE（Server-Sent Events）传输流式结果。SSE 的基本格式是若干 data: 行，事件之间以空行分隔。可以先写一个通用的 SSE 解析器：
js
// src/sse.js
export async function* parseSSE(response) {
const reader = response.body.getReader();
const decoder = new TextDecoder('utf-8');
let buffer = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let separatorIndex;
while ((separatorIndex = buffer.indexOf('\n\n')) !== -1) {
const rawEvent = buffer.slice(0, separatorIndex);
buffer = buffer.slice(separatorIndex + 2);
const data = parseDataLines(rawEvent);
if (data === null) continue;
if (data === '[DONE]') return; // OpenAI 兼容服务的结束标记
const parsed = JSON.parse(data);
yield parsed;
}
}
} finally {
reader.releaseLock();
}
}
function parseDataLines(rawEvent) {
const dataLines = rawEvent
.split('\n')
.filter((line) => line.startsWith('data:'))
.map((line) => line.slice(5).trimStart());
if (dataLines.length === 0) return null;
return dataLines.join('\n'); // 多行 data 按 SSE 规范用换行连接
}这个解析器只负责传输层，不关心 JSON 里的业务结构。三家提供商 SSE 事件的差异全部体现在 JSON 内部：
- OpenAI 每个事件是包含
choices[0].delta的补全块； - Anthropic 使用事件类型
message_start、content_block_start、content_block_delta、message_delta等 [3]； - Gemini 的
streamGenerateContent默认返回换行分隔的 JSON 序列，需要加alt=sse参数才能得到 SSE，每个事件是一个GenerateContentResponse。
使用 async generator 暴露流
统一流协议只定义三类事件：
| type | 字段 | 说明 |
|---|---|---|
content | text | 一段增量文本 |
tool_call_delta | index、id、name、arguments | 工具调用信息，arguments 是 JSON 字符串片段，可能跨多个事件 |
finish | reason | 流结束，携带归一化结束原因 |
消费方按 index 组装工具调用参数：
js
async function consumeStream(stream) {
let text = '';
const calls = new Map();
for await (const event of stream) {
if (event.type === 'content') {
text += event.text;
} else if (event.type === 'tool_call_delta') {
const call = calls.get(event.index) || { id: '', name: '', args: '' };
if (event.id) call.id = event.id;
if (event.name) call.name = event.name;
call.args += event.arguments || '';
calls.set(event.index, call);
} else if (event.type === 'finish') {
console.log('结束原因：', event.reason);
}
}
const toolCalls = [...calls.values()].map((call) => ({
id: call.id,
name: call.name,
args: JSON.parse(call.args || '{}'),
}));
return { text, toolCalls };
}OpenAI 适配器的 stream 实现：
js
async *stream(request) {
const body = { ...this.buildRequest(request), stream: true };
const response = await this.requestRaw('/chat/completions', {
headers: this.buildHeaders(),
body,
});
for await (const chunk of parseSSE(response)) {
const choice = chunk.choices?.[0];
if (!choice) continue;
const delta = choice.delta || {};
if (delta.content) {
yield { type: 'content', text: delta.content };
}
if (delta.tool_calls) {
for (const tc of delta.tool_calls) {
yield {
type: 'tool_call_delta',
index: tc.index ?? 0,
id: tc.id,
name: tc.function?.name,
arguments: tc.function?.arguments,
};
}
}
if (choice.finish_reason) {
yield { type: 'finish', reason: this.mapFinishReason(choice.finish_reason) };
}
}
}注意 requestRaw 在这里替代了 requestJSON：流式响应需要保留 response.body 供 parseSSE 读取，不能先把整个 body 解析成 JSON。
Anthropic 适配器的流式实现需要维护当前内容块的状态，因为 content_block_delta 里的 text_delta 和 input_json_delta 是按块到达的：
js
async *stream(request) {
const body = { ...this.buildRequest(request), stream: true };
const response = await this.requestRaw('/v1/messages', {
headers: this.buildHeaders(),
body,
});
let currentBlock = null;
for await (const event of parseSSE(response)) {
if (event.type === 'message_start') {
continue; // 可在此读取 usage 元信息
}
if (event.type === 'content_block_start') {
currentBlock = {
index: event.index,
kind: event.content_block.type,
toolUse: event.content_block.type === 'tool_use'
? { id: event.content_block.id, name: event.content_block.name, args: '' }
: null,
};
continue;
}
if (event.type === 'content_block_delta') {
if (event.delta.type === 'text_delta') {
yield { type: 'content', text: event.delta.text };
} else if (event.delta.type === 'input_json_delta') {
currentBlock.toolUse.args += event.delta.partial_json;
}
continue;
}
if (event.type === 'content_block_stop') {
if (currentBlock?.toolUse) {
yield {
type: 'tool_call_delta',
index: currentBlock.index,
id: currentBlock.toolUse.id,
name: currentBlock.toolUse.name,
arguments: currentBlock.toolUse.args,
};
}
currentBlock = null;
continue;
}
if (event.type === 'message_delta') {
if (event.delta?.stop_reason) {
yield { type: 'finish', reason: this.mapFinishReason(event.delta.stop_reason) };
}
}
}
}Gemini 的流式实现把每个 SSE 事件中的 candidates[0].content.parts 转换为统一事件：
js
async *stream(request) {
const model = request.model || this.model;
const url = `${this.baseURL}/models/${model}:streamGenerateContent?alt=sse&key=${encodeURIComponent(this.apiKey)}`;
const response = await this.requestRaw(url, {
body: this.toGeminiRequest(request),
});
for await (const chunk of parseSSE(response)) {
const candidate = chunk.candidates?.[0];
if (!candidate) continue;
for (const part of candidate.content?.parts || []) {
if (part.text) {
yield { type: 'content', text: part.text };
}
if (part.functionCall) {
yield {
type: 'tool_call_delta',
index: 0,
id: `fc-${part.functionCall.name}`,
name: part.functionCall.name,
arguments: JSON.stringify(part.functionCall.args || {}),
};
}
}
if (candidate.finishReason) {
yield { type: 'finish', reason: this.mapFinishReason(candidate.finishReason) };
}
}
}三个适配器的流式方法都返回异步生成器，因此消费方可以用同一套 for await...of 循环处理三种不同的事件协议。
注意：
parseSSE假设事件分隔符是\n\n。某些服务可能使用\r\n\r\n，需要先把\r\n归一化为\n再解析。
错误处理、超时与重试
错误分类与归一化
classifyError 已经按 HTTP 状态码把错误归入统一分类：
| HTTP 状态码 | type | retryable | 说明 |
|---|---|---|---|
| 401 / 403 | auth | 否 | API Key 无效或权限不足 |
| 429 | rate_limit | 是 | 触发速率限制，可读取 Retry-After |
| 5xx | server | 是 | 服务端错误，可能是瞬时故障 |
| 400 / 404 | invalid_request | 否 | 请求内容或路径错误 |
| 超时 / 网络中断 | timeout / network | 是 | 未收到响应或连接中断 |
三家提供商的错误响应体都包含 error.message 字段，因此 classifyError 中统一的取法可以覆盖 OpenAI、Anthropic 和 Gemini。
注意：接入层在记录日志时，只记录
provider、error.type、error.message、耗时这几个字段即可，不要把 API Key 或完整请求体写入日志。
速率限制与指数退避
重试逻辑放在统一层，而不是每个适配器里。retryable 为真的错误才重试，auth 和 invalid_request 直接抛出：
js
// src/retry.js
export async function withRetry(fn, { maxRetries = 3, baseDelayMs = 500, factor = 2 } = {}) {
let lastError;
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
lastError = error;
if (!error.retryable || attempt === maxRetries) {
throw error;
}
const delay = computeDelay(error, attempt, baseDelayMs, factor);
await sleep(delay);
}
}
throw lastError;
}
function computeDelay(error, attempt, baseDelayMs, factor) {
if (error.retryAfterMs !== undefined) {
return error.retryAfterMs;
}
const exponential = baseDelayMs * factor ** attempt;
const jitter = Math.random() * exponential * 0.1;
return Math.floor(exponential + jitter);
}
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}使用方法：
js
const result = await withRetry(
() => adapter.chat(request),
{ maxRetries: 4, baseDelayMs: 1000 }
);关于重试的几个注意点：
- 服务端在 429 响应中返回
Retry-After头时，以此为准，不要再用固定退避覆盖它； - 指数退避需要加随机抖动，避免多个请求同时重试形成新的限流；
- 重试次数和总时间要设上限。
retryAfterMs可能很大，超过业务可接受等待时间时应直接失败； - 流式请求在已产出部分内容后发生错误，不能简单重放整个请求，否则会产生重复输出。这种情况应由业务层决定是继续等待、终止还是发起新的请求。
ProviderRegistry 与模型路由
Provider 注册与工厂创建
当适配器数量增多时，需要一种方式按名称创建适配器。注册表把 provider 名称映射到适配器类：
js
// src/registry.js
export class ProviderRegistry {
constructor() {
this.factories = new Map();
}
register(name, AdapterClass) {
this.factories.set(name, AdapterClass);
return this;
}
create(name, config) {
const AdapterClass = this.factories.get(name);
if (!AdapterClass) {
throw new Error(`不支持的 provider: ${name}`);
}
return new AdapterClass(config);
}
}注册与使用：
js
import { OpenAIAdapter } from './provider/openai.js';
import { AnthropicAdapter } from './provider/anthropic.js';
import { GeminiAdapter } from './provider/gemini.js';
const registry = new ProviderRegistry();
registry.register('openai', OpenAIAdapter);
registry.register('anthropic', AnthropicAdapter);
registry.register('gemini', GeminiAdapter);
const adapter = registry.create('openai', {
apiKey: process.env.OPENAI_API_KEY,
model: process.env.OPENAI_MODEL,
});注册表保存的是类而不是实例，这样每个调用方可以传入自己的配置。测试代码也可以往同一个注册表里注册 MockAdapter，实现测试环境与真实环境的无缝切换。
多 Provider fallback 策略
fallback 指的是第一个 Provider 调用失败后，按顺序尝试下一个 Provider。路由选择与 fallback 可以组合：路由决定用哪个 Provider，fallback 决定失败后怎么切换。
js
// src/router.js
import { ProviderError } from './errors.js';
export async function chatWithFallback(adapters, request) {
const failures = [];
for (const adapter of adapters) {
try {
return await adapter.chat(request);
} catch (error) {
failures.push({ provider: adapter.name, error });
if (!error.retryable) {
throw error;
}
console.warn(`${adapter.name} 调用失败：${error.message}`);
}
}
throw new ProviderError({
type: 'fallback_exhausted',
message: `所有 provider 均失败：${failures.map((f) => `${f.provider}: ${f.error.message}`).join(' | ')}`,
provider: failures.map((f) => f.provider).join(','),
retryable: true,
});
}使用：
js
const adapters = [
registry.create('openai', configs.openai),
registry.create('anthropic', configs.anthropic),
];
const result = await chatWithFallback(adapters, request);fallback 的默认语义是：只有 retryable 错误才切换 Provider，鉴权或请求格式错误立即抛出。如果业务要求在非重试错误上也切换（例如一个 Provider 的密钥失效时检查另一个），需要把判断逻辑改成可配置的谓词函数。模型路由也可以仿照这个结构：根据 request.model 的前缀选择 Provider，或者维护一张模型到适配器的映射表。
配置管理与 API Key 安全
适配器不持有硬编码配置，所有密钥和模型 ID 都从环境变量读取：
js
// src/config.js
export function loadProviderConfig(provider) {
const env = process.env;
const table = {
openai: {
apiKey: env.OPENAI_API_KEY,
baseURL: env.OPENAI_BASE_URL,
model: env.OPENAI_MODEL,
},
anthropic: {
apiKey: env.ANTHROPIC_API_KEY,
baseURL: env.ANTHROPIC_BASE_URL,
model: env.ANTHROPIC_MODEL,
},
gemini: {
apiKey: env.GEMINI_API_KEY,
baseURL: env.GEMINI_BASE_URL,
model: env.GEMINI_MODEL,
},
};
const config = table[provider];
if (!config) throw new Error(`未配置的 provider: ${provider}`);
if (!config.apiKey) throw new Error(`缺少 ${provider} 的 API Key，请检查环境变量`);
return {
...config,
baseURL: config.baseURL || undefined,
};
}对应的 .env 文件：
bash
OPENAI_API_KEY=sk-...
OPENAI_MODEL=...
ANTHROPIC_API_KEY=...
ANTHROPIC_MODEL=...
GEMINI_API_KEY=...
GEMINI_MODEL=...API Key 的安全边界：
.env文件必须加入.gitignore，不能进入版本库；- 密钥只出现在构造适配器时的配置对象里，不进入任何请求体或日志；
- 不要把
apiKey挂到ChatRequest、ChatResult这类会被序列化的对象上； - 在服务端应用中，接入层不应暴露给浏览器端直接调用，否则密钥会随请求头发送到客户端；
- 密钥轮换时，如果从环境变量读取，重启进程即可生效；如果从密钥管理服务读取，需要让
loadProviderConfig支持运行时刷新。
测试策略：Mock Provider 与契约测试
Mock Provider
Mock Provider 是一个实现同一接口的假适配器，用于测试上层业务逻辑（fallback、重试、流式消费），不产生真实网络请求：
js
// test/mock_adapter.js
import { ProviderAdapter } from '../src/provider/base.js';
import { ChatResult } from '../src/model.js';
export class MockAdapter extends ProviderAdapter {
constructor(config = {}) {
super(config);
this.name = config.name || 'mock';
this.responses = config.responses || [];
this.chunks = config.chunks || ['你', '好'];
}
async chat(request) {
const response = this.responses.shift() || { content: 'mock 响应' };
return new ChatResult({
id: 'mock-1',
provider: this.name,
model: request.model || this.model,
content: response.content ?? '',
toolCalls: response.toolCalls || [],
finishReason: response.finishReason || 'stop',
});
}
async *stream(request) {
for (const text of this.chunks) {
yield { type: 'content', text };
}
yield { type: 'finish', reason: 'stop' };
}
}配合 Node.js 内置测试模块，可以验证 fallback 行为：
js
// test/fallback.test.js
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { chatWithFallback } from '../src/router.js';
import { ProviderError } from '../src/errors.js';
import { ChatRequest } from '../src/model.js';
import { MockAdapter } from './mock_adapter.js';
test('第一个 provider 返回 rate_limit 时切换到第二个', async () => {
const failing = new MockAdapter({ name: 'mock-a' });
failing.chat = async () => {
throw new ProviderError({ type: 'rate_limit', message: '限流', retryable: true });
};
const ok = new MockAdapter({ name: 'mock-b', responses: [{ content: 'ok' }] });
const result = await chatWithFallback([failing, ok], new ChatRequest({ model: 'x' }));
assert.equal(result.content, 'ok');
assert.equal(result.provider, 'mock-b');
});这个测试不关心具体 Provider 的协议，只验证「可重试错误触发切换」这一行为。
契约测试
契约测试锁定的是适配器的消息转换行为：给定统一消息，适配器发送的请求体必须符合预期。通过替换 globalThis.fetch 拦截请求，不需要真正的 API 密钥：
js
// test/openai.contract.test.js
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { OpenAIAdapter } from '../src/provider/openai.js';
import { ChatMessage, ChatRequest, ToolCall } from '../src/model.js';
test('OpenAI 适配器把工具调用转换为 tool_calls 消息', async () => {
const requests = [];
const originalFetch = globalThis.fetch;
globalThis.fetch = async (url, options) => {
requests.push({ url, body: JSON.parse(options.body) });
return new Response(JSON.stringify({
id: 'chatcmpl-1',
model: 'gpt-4o',
choices: [{
message: { role: 'assistant', content: '查询结果如下。' },
finish_reason: 'stop',
}],
usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 },
}), {
status: 200,
headers: { 'Content-Type': 'application/json' },
});
};
try {
const adapter = new OpenAIAdapter({ apiKey: 'test-key', model: 'gpt-4o' });
await adapter.chat(new ChatRequest({
model: 'gpt-4o',
messages: [
new ChatMessage({ role: 'user', content: '北京天气？' }),
new ChatMessage({
role: 'assistant',
content: '',
toolCalls: [new ToolCall({ id: 'call_1', name: 'get_weather', args: { city: '北京' } })],
}),
new ChatMessage({
role: 'tool',
toolCallId: 'call_1',
name: 'get_weather',
content: '晴',
}),
],
}));
const sent = requests[0].body;
assert.equal(sent.messages[1].role, 'assistant');
assert.equal(sent.messages[1].tool_calls[0].function.name, 'get_weather');
assert.deepEqual(
JSON.parse(sent.messages[1].tool_calls[0].function.arguments),
{ city: '北京' }
);
assert.equal(sent.messages[2].role, 'tool');
assert.equal(sent.messages[2].tool_call_id, 'call_1');
} finally {
globalThis.fetch = originalFetch;
}
});契约测试的价值在于：当适配器的转换逻辑被无意修改时，测试会立刻失败。它验证的是「接入层与供应商之间的契约」，而不是供应商的真实行为。真实行为验证可以在此基础上扩展「录制/回放」模式：记录一次真实调用的请求和响应，存入 fixture，再在测试中回放。
接入层的边界与适用场景
接入层不是所有场景都需要的抽象。判断依据可以看业务代码中是否出现了多处与具体厂商字段耦合的逻辑。
适用场景：
- 业务需要同时对接多个模型服务，并且希望切换成本可控；
- 需要在一个 Provider 出错时自动切换到另一个，且业务代码不需要感知；
- 团队希望把各家 API 的调用细节收敛到一个模块中统一维护。
不适用场景：
- 只使用一家模型服务，且没有更换计划，直接使用官方 SDK 更省事；
- 需要深度使用某家的专有能力，例如 Anthropic 的 extended thinking、Gemini 的多模态输入。统一抽象只能表达各家的公共子集，专有能力会被丢弃，或者需要以透传方式绕过抽象；
- 对单次调用路径性能极敏感时，适配层的对象转换会带来少量额外开销，不过通常不是瓶颈。
接入层、官方 SDK 与 AI 网关的定位不同：
- 官方 SDK 是对单一 API 的完整封装；
- 接入层位于应用代码库内，面向业务定义统一的调用边界，不依赖具体 SDK；
- 网关是独立部署的基础设施，统一协议、路由和策略，接入层可以作为网关的客户端。
MCP（Model Context Protocol）试图在模型与外部工具、数据源之间建立标准接口，解决的是「模型如何调用工具」的问题；接入层解决的是「应用如何调用模型」的问题。两者可以共存：接入层内部可以把工具调用映射为 MCP 工具，也可以把 MCP 服务端作为统一层的下游，但这两个抽象处于不同层次。
参考链接
- [1] https://developers.openai.com/api/docs/guides/migrate-to-responses
- [2] https://docs.cloud.google.com/vertex-ai/generative-ai/docs/samples/generativeaionvertexai-gemini-chat-completions-non-streaming
- [3] https://docs.fastrouter.ai/api-reference/anthropic-messages-format
- [4] https://docs.parallel.ai/integrations/anthropic-tool-calling