不同 LLM SDK 的事件名称和字段结构并不一致，但业务流程不应该知道这些差异。本文用 TypeScript 构建一个统一的 Provider 适配层，并通过能力矩阵和契约测试约束实现。 示例重点不是绑定某一家供应商，而是展示如何把协议差异收敛在边界内。
先定义问题边界
在一个典型的 LLM 应用中，上层流程通常只关心几件事：模型产生了哪些文本、是否请求调用工具、消耗了多少 Token，以及这次生成为什么结束。它不应该关心某个 SDK 把文本增量叫作 delta、把工具参数拆成多段，还是把结束原因表示为 stop_reason。
供应商之间的差异主要集中在四个方面：
| 能力 | 常见差异 | 业务层真正需要的结果 |
|---|---|---|
| 流式事件 | 文本、工具调用、完成事件名称不同 | 可按顺序消费的统一事件流 |
| 工具调用 | 参数可能分片，ID 和名称出现时机不同 | 一个完整、可校验的工具调用 |
| Token 用量 | 字段名、统计时机和是否提供缓存 Token 不同 | 输入、输出及总量的统一快照 |
| 结束原因 | stop、length、工具调用等命名不同 | 可用于流程判断的有限枚举 |
适配层的目标不是抹平所有差异，而是定义业务确实需要的最小公共模型。供应商特有能力可以保留在能力矩阵或扩展字段中，但不能悄悄进入订单、客服、工作流等核心代码。
设计稳定的内部事件模型
先从内部接口开始，而不是从某个 SDK 的响应对象开始。下面的模型使用判别联合，调用方可以通过 type 安全地缩小类型范围：
export type FinishReason =
| "stop"
| "length"
| "tool_call"
| "content_filter"
| "unknown";
export type Usage = {
inputTokens: number;
outputTokens: number;
totalTokens: number;
};
export type ToolCall = {
id: string;
name: string;
arguments: string;
};
export type LlmEvent =
| { type: "text_delta"; text: string }
| { type: "tool_call"; call: ToolCall }
| { type: "usage"; usage: Usage }
| { type: "done"; reason: FinishReason };
export type ChatInput = {
messages: Array<{ role: "system" | "user" | "assistant"; content: string }>;
tools?: Array<{ name: string; description?: string; inputSchema: unknown }>;
};
export type ProviderCapabilities = {
streaming: boolean;
tools: boolean;
usageInStream: boolean;
structuredOutput: boolean;
};
export interface LlmProvider {
readonly name: string;
readonly capabilities: ProviderCapabilities;
stream(input: ChatInput): AsyncIterable<LlmEvent>;
}这里没有暴露 SDK 类型，也没有把 finish_reason 之类的供应商字段带到上层。tool_call 事件中的 arguments 暂时使用 JSON 字符串，原因是不同模型可能分片返回参数；适配器负责在事件边界完成拼接和校验，业务层只处理完整调用。
实现一个可运行的适配器骨架
为了让示例可以脱离网络直接运行，下面实现一个内存 Provider。真实项目中，只需要把 stream 内部替换为对应 SDK 的流式请求，并将 SDK 事件映射为 LlmEvent；上层消费代码不需要变化。
type FinishReason = "stop" | "length" | "tool_call" | "content_filter" | "unknown";
type Usage = { inputTokens: number; outputTokens: number; totalTokens: number };
type ToolCall = { id: string; name: string; arguments: string };
type LlmEvent =
| { type: "text_delta"; text: string }
| { type: "tool_call"; call: ToolCall }
| { type: "usage"; usage: Usage }
| { type: "done"; reason: FinishReason };
type ChatInput = {
messages: Array<{ role: "system" | "user" | "assistant"; content: string }>;
tools?: Array<{ name: string; description?: string; inputSchema: unknown }>;
};
type ProviderCapabilities = {
streaming: boolean;
tools: boolean;
usageInStream: boolean;
structuredOutput: boolean;
};
interface LlmProvider {
readonly name: string;
readonly capabilities: ProviderCapabilities;
stream(input: ChatInput): AsyncIterable<LlmEvent>;
}
class DemoProvider implements LlmProvider {
readonly name = "demo";
readonly capabilities: ProviderCapabilities = {
streaming: true,
tools: true,
usageInStream: true,
structuredOutput: false,
};
async *stream(input: ChatInput): AsyncIterable<LlmEvent> {
const lastMessage = input.messages.at(-1)?.content ?? "";
for (const text of [`收到问题：${lastMessage}`, "。这是流式响应。`]) {
yield { type: "text_delta", text };
}
if (input.tools?.some((tool) => tool.name === "get_weather")) {
yield {
type: "tool_call",
call: { id: "call-1", name: "get_weather", arguments: '{"city":"Shanghai"}' },
};
yield { type: "done", reason: "tool_call" };
return;
}
yield { type: "usage", usage: { inputTokens: 12, outputTokens: 9, totalTokens: 21 } };
yield { type: "done", reason: "stop" };
}
}
async function consume(provider: LlmProvider, input: ChatInput): Promise<void> {
let text = "";
for await (const event of provider.stream(input)) {
switch (event.type) {
case "text_delta":
text += event.text;
process.stdout.write(event.text);
break;
case "tool_call":
console.log(`\n调用工具：${event.call.name} ${event.call.arguments}`);
break;
case "usage":
console.log(`\n用量：${event.usage.totalTokens}`);
break;
case "done":
console.log(`\n结束：${event.reason}`);
break;
}
}
if (!text) throw new Error("provider returned no text");
}
await consume(new DemoProvider(), {
messages: [{ role: "user", content: "介绍适配层" }],
});保存为 provider.ts 后，可以使用 npx tsx provider.ts 运行。示例中的 Token 数字只是演示数据，生产实现必须使用供应商返回的用量字段，不能通过字符数或估算值冒充精确统计。
用能力矩阵隔离供应商特性
并不是每个模型都支持全部能力。例如，某些模型支持工具调用但不会在流式事件中返回用量；另一些模型支持结构化输出，却要求单独的请求参数。适配层可以暴露静态能力描述，让路由或业务编排在调用前做明确判断：
| Provider | 流式输出 | 工具调用 | 流中用量 | 结构化输出 |
|---|---|---|---|---|
openai | 支持 | 取决于模型 | 取决于请求配置 | 取决于模型 |
anthropic | 支持 | 支持 | 取决于事件与配置 | 取决于模型 |
demo | 支持 | 支持 | 支持 | 不支持 |
表格不是运行时真相，具体能力应以适配器实现和供应商文档为准。关键是让能力检查显式存在，例如：
function requireCapability(
provider: LlmProvider,
capability: keyof ProviderCapabilities,
): void {
if (!provider.capabilities[capability]) {
throw new Error(`${provider.name} does not support ${capability}`);
}
}
function chooseProvider(provider: LlmProvider, needsTools: boolean): LlmProvider {
if (needsTools) requireCapability(provider, "tools");
requireCapability(provider, "streaming");
return provider;
}这样做比在业务代码中散落 if (provider === "某模型") 更容易维护。供应商名称用于日志和观测，能力判断才用于流程决策。
用契约测试约束所有实现
适配器最容易出现的问题不是类型错误，而是行为不一致：有的实现永远不发 done，有的实现漏掉工具参数，有的实现把总 Token 留空。可以为每个 Provider 运行同一组契约测试：
async function collect(provider: LlmProvider): Promise<LlmEvent[]> {
const events: LlmEvent[] = [];
for await (const event of provider.stream({
messages: [{ role: "user", content: "hello" }],
})) events.push(event);
return events;
}
async function contractTest(provider: LlmProvider): Promise<void> {
const events = await collect(provider);
if (!provider.capabilities.streaming) throw new Error("streaming must be enabled");
if (events.length === 0) throw new Error("event stream must not be empty");
if (events.at(-1)?.type !== "done") throw new Error("done must be last");
for (const event of events) {
if (event.type === "usage") {
if (event.usage.totalTokens !== event.usage.inputTokens + event.usage.outputTokens) {
throw new Error("usage total is inconsistent");
}
}
if (event.type === "tool_call") JSON.parse(event.call.arguments);
}
console.log(`${provider.name}: contract passed`);
}
await contractTest(new DemoProvider());在真实项目中，还应为工具场景增加测试：确认参数分片可以合并、工具调用结束原因为 tool_call，以及不支持工具时能够在请求前失败。对于网络适配器，可以将 SDK 客户端注入构造函数，再用假的 SDK 事件测试映射逻辑；不要让契约测试依赖真实网络和模型输出。
总结
稳定的 Provider 层有三个边界：
- 用内部事件模型统一文本增量、工具调用、用量和结束原因。
- 用能力矩阵描述“能不能做”，不要把供应商名称写进业务分支。
- 用契约测试验证所有 Provider 的事件顺序、字段完整性和错误行为。
供应商 SDK 仍然会变化，模型能力也会增加，但变化应该停留在适配器内部。上层业务只依赖 LlmProvider 和稳定的 LlmEvent，因此新增模型时是增加一个实现，而不是重写整条业务流程。
评论