统一多提供商 LLM API 设计详解#

基于 packages/ai/src/providers/ 源码分析

真实调用关系全景图：

上图展示了从应用层到 LLM API 层的完整调用链路，贯穿注册表、适配器、事件流三大核心组件。理解这张图，是掌握统一多提供商 API 设计的关键起点。

概述：什么是统一多提供商 LLM API？#

本质问题抽象#

表面痛点只是症状,真正需要解决的是三个本质问题：

1
┌────────────────────────────────────────────────────────────────┐
2
│                     本质问题一：异构性                          │
3
├────────────────────────────────────────────────────────────────┤
4
│  问题本质：                                                     │
5
│  Anthropic、OpenAI、Google 是三个完全不同的"世界"               │
6
│  - 协议不同：SSE 格式、事件命名、数据结构                        │
7
│  - 能力不同：图片、thinking、工具调用支持度                      │
8
│  - 约束不同：ID 格式限制、token 限制、参数格式                   │
9
│                                                                 │
10
│  表面症状：                                                     │
11
│  ❌ API 格式完全不同                                            │
12
│  ❌ 消息结构、字段名、事件类型都不统一                           │
13
│  ❌ 某些模型不支持图片、某些不支持 thinking                      │
14
│                                                                 │
15
│  核心挑战：如何让上层代码不感知这些差异？                        │
16
└────────────────────────────────────────────────────────────────┘
17

18
┌────────────────────────────────────────────────────────────────┐
19
│                     本质问题二：流式性                          │
20
├────────────────────────────────────────────────────────────────┤
21
│  问题本质：                                                     │
22
│  LLM 生成是"涌现"过程，不是一次性产出                           │
23
│  - 内容逐 token 生成，无法预知完整结果                          │
24
│  - 生成过程可能被中断（错误、超时、用户取消）                    │
25
│  - 需要实时反馈，不能阻塞等待                                    │
26
│                                                                 │
27
│  表面症状：                                                     │
28
│  ❌ 每个 API 的 SSE 格式各异                                    │
29
│  ❌ 流式调用中途失败，如何处理？                                 │
30
│  ❌ 用户要等到全部内容生成完才能看到第一个字                     │
31
│                                                                 │
32
│  核心挑战：如何建模"涌现过程"，让数据实时流动？                  │
33
└────────────────────────────────────────────────────────────────┘
34

35
┌────────────────────────────────────────────────────────────────┐
36
│                     本质问题三：迁移性                          │
37
├────────────────────────────────────────────────────────────────┤
38
│  问题本质：                                                     │
39
│  对话历史是"状态"，但不同提供商状态格式不互通                    │
40
│  - 加密内容（thinkingSignature）只对原模型有效                  │
41
│  - ID 格式限制不同（Anthropic ≤64字符，OpenAI 450+字符）        │
42
│  - 能力差异（从有图片模型切换到无图片模型）                      │
43
│                                                                 │
44
│  表面症状：                                                     │
45
│  ❌ 用户切换模型时，历史消息可能不兼容                           │
46
│  ❌ Anthropic 的 thinkingSignature 对 OpenAI 无效               │
47
│  ❌ 工具调用 ID 格式跨模型不兼容                                 │
48
│                                                                 │
49
│  核心挑战：如何让"状态"在不同"世界"之间迁移？                    │
50
└────────────────────────────────────────────────────────────────┘

设计核心：三个抽象原则#

对应三个本质问题，设计遵循三个核心原则：

1
┌────────────────────────────────────────────────────────────────┐
2
│                 原则一：统一抽象层（解决异构性）                 │
3
├────────────────────────────────────────────────────────────────┤
4
│                                                                 │
5
│  核心思想：                                                     │
6
│  "在差异之上建立统一，让上层只看到统一接口"                      │
7
│                                                                 │
8
│  实现：                                                         │
9
│  ┌─────────────┐                                                │
10
│  │ Anthropic   │──►                                             │
11
│  └─────────────┘    ┌─────────────────┐    ┌─────────────────┐  │
12
│  ┌─────────────┐──► │   适配器层       │──► │ 统一内部格式    │  │
13
│  │ OpenAI      │──► │ (anthropic.ts)  │    │ AssistantMessage│  │
14
│  └─────────────┘    │ (openai.ts)     │    │ - content[]     │  │
15
│  ┌─────────────┐──► │ (google.ts)     │    │ - usage         │  │
16
│  │ Google      │──► └─────────────────┘    │ - stopReason    │  │
17
│  └─────────────┘                           └─────────────────┘  │
18
│                                                                 │
19
│  关键技术：                                                     │
20
│  - 适配器模式：每个提供商一个适配器                              │
21
│  - 注册表模式：统一入口，Map<api, adapter>                       │
22
│  - 懒加载：按需加载适配器，减少资源消耗                          │
23
│                                                                 │
24
└────────────────────────────────────────────────────────────────┘
25

26
┌────────────────────────────────────────────────────────────────┐
27
│                 原则二：事件流模型（解决流式性）                 │
28
├────────────────────────────────────────────────────────────────┤
29
│                                                                 │
30
│  核心思想：                                                     │
31
│  "把涌现过程建模为事件流，而不是一次性结果"                      │
32
│                                                                 │
33
│  传统模型（结果导向）：                                         │
34
│  request ──────► [等待] ──────► response                        │
35
│                   (阻塞)                                        │
36
│                                                                 │
37
│  事件流模型（过程导向）：                                       │
38
│  request ──────► event1 ─► event2 ─► event3 ─► done             │
39
│                   (实时流动)                                    │
40
│                                                                 │
41
│  关键技术：                                                     │
42
│  - EventStream：实现 AsyncIterable，支持 for await...of         │
43
│  - delta + partial：增量 + 累积，满足不同消费场景               │
44
│  - 错误即事件：错误不打断流，作为事件推送                        │
45
│                                                                 │
46
│  为什么重要？                                                   │
47
│  - 实时反馈：用户立即看到第一个字                                │
48
│  - 可中断：用户可以随时取消，已生成内容不丢失                    │
49
│  - 可组合：多个流可以 merge、filter、transform                   │
50
│                                                                 │
51
└────────────────────────────────────────────────────────────────┘
52

53
┌────────────────────────────────────────────────────────────────┐
54
│                 原则三：状态转换层（解决迁移性）                 │
55
├────────────────────────────────────────────────────────────────┤
56
│                                                                 │
57
│  核心思想：                                                     │
58
│  "状态迁移时，转换不兼容部分，保留兼容部分"                      │
59
│                                                                 │
60
│  转换规则：                                                     │
61
│  ┌───────────────────────────────────────────────────────────┐ │
62
│  │ 内容类型      │ 同模型   │ 跨模型                          │ │
63
│  ├───────────────────────────────────────────────────────────┤ │
64
│  │ text         │ 保留     │ 保留                             │ │
65
│  │ thinking     │ 保留     │ 转为 text（丢弃加密签名）        │ │
66
│  │ redacted     │ 保留     │ 丢弃                             │ │
67
│  │ toolCall     │ 保留     │ ID 格式标准化                    │ │
68
│  │ image        │ 保留     │ 目标不支持时降级为占位符文本     │ │
69
│  └───────────────────────────────────────────────────────────┘ │
70
│                                                                 │
71
│  关键技术：                                                     │
72
│  - transformMessages：统一转换入口                              │
73
│  - isSameModel 判断：同模型保留原样，跨模型转换                  │
74
│  - normalizeToolCallId：ID 格式标准化                           │
75
│  - replaceImagesWithPlaceholder：图片降级                       │
76
│                                                                 │
77
└────────────────────────────────────────────────────────────────┘

设计目标映射#

将表面目标映射到核心原则：

表面目标	对应原则	原理
应用层只调用 `stream()`	统一抽象层	适配器隐藏差异
不关心底层是谁	统一抽象层	注册表统一入口
流式数据实时传递	事件流模型	AsyncIterable + delta
错误不打断流	事件流模型	错误即事件
按需加载	统一抽象层	懒加载减少资源
跨提供商兼容	状态转换层	transformMessages

一、整体架构图#

1
┌─────────────────────────────────────────────────────────────┐
2
│                      应用层                                  │
3
│         stream(model, context, options)                     │
4
└─────────────────────────────────────────────────────────────┘
5
                            │
6
                            ▼
7
┌─────────────────────────────────────────────────────────────┐
8
│                      注册层                                  │
9
│    api: "anthropic-messages" → lazyAnthropicWrapper         │
10
│    api: "openai-responses"   → lazyOpenAIWrapper            │
11
└─────────────────────────────────────────────────────────────┘
12
                            │
13
                            ▼ 动态 import
14
┌─────────────────────────────────────────────────────────────┐
15
│                      提供商层                                │
16
│    anthropic.ts / openai.ts / google.ts / ...               │
17
│    - SSE 解析                                                │
18
│    - 消息格式转换                                            │
19
│    - 统一事件输出                                            │
20
└─────────────────────────────────────────────────────────────┘
21
                            │
22
                            ▼ HTTP SSE
23
┌─────────────────────────────────────────────────────────────┐
24
│                      LLM API 层                              │
25
│    Anthropic API / OpenAI API / Google API                  │
26
└─────────────────────────────────────────────────────────────┘

二、逐层解析：问题 → 原则 → 实现#

每一层都对应一个本质问题，实现一个核心原则

第 1 层：注册表 + 懒加载#

对应本质问题：异构性#

1
问题：Anthropic、OpenAI、Google 是三个完全不同的"世界"
2
挑战：如何让上层代码不感知这些差异？

实现核心原则：统一抽象层#

1
原则："在差异之上建立统一，让上层只看到统一接口"
2
实现：注册表模式 + 懒加载 + 类型包装

具体实现#

问题分解：

问题 1: 如何让应用层不关心具体是哪个提供商？ → 注册表统一入口
问题 2: 用户只用 Anthropic，为什么要加载 OpenAI、Google 的代码？ → 懒加载
问题 3: 如何保证类型安全？ → 类型包装

代码实现：

1
// 1. 注册表：统一入口，Map<api, adapter>
2
const registry = new Map<string, Provider>();
3

4
export function registerApiProvider(api: string, stream: StreamFunction) {
5
    registry.set(api, stream);
6
}
7

8
export function getApiProvider(api: string): StreamFunction | undefined {
9
    return registry.get(api);  // 👈 上层只调用这个，不关心具体是谁
10
}
11

12
// 2. 懒加载：首次使用才 import
13
function createLazyStream(loadModule: () => Promise<Module>) {
14
    return (model, context, options) => {
15
        const stream = new EventStream();
16

17
        loadModule()  // 👈 首次调用才 import("./anthropic.ts")
18
            .then(module => module.stream(model, context, options))
19
            .catch(error => stream.push({ type: "error", error }));
20

21
        return stream;  // 👈 立即返回，不阻塞
22
    };
23
}
24

25
// 3. 类型包装：运行时校验 + 泛型
26
function wrapStream<TApi extends Api>(api: TApi, stream: StreamFunction<TApi>) {
27
    return (model, context, options) => {
28
        if (model.api !== api) {
29
            throw new Error(`Mismatched api: ${model.api} expected ${api}`);
30
        }
31
        return stream(model as Model<TApi>, context, options);
32
    };
33
}
34

35
// 注册
36
registerApiProvider({
37
    api: "anthropic-messages",
38
    stream: createLazyStream(() => import("./anthropic.ts")),  // 👈 按需加载
39
});

设计呼应#

本质问题	核心原则	关键技术
异构性：不同提供商是不同”世界”	统一抽象层：在差异之上建立统一	注册表统一入口 + 懒加载 + 类型包装

第 2 层：SSE 解析#

对应本质问题：流式性#

1
问题：LLM 生成是"涌现"过程，不是一次性产出
2
挑战：如何建模"涌现过程"，让数据实时流动？

实现核心原则：事件流模型#

1
原则："把涌现过程建模为事件流，而不是一次性结果"
2
实现：SSE 解析器 + 流式 JSON 解析

具体实现#

问题分解：

问题 1: 不同 API 的 SSE 格式完全不同 → 每个提供商独立解析器
问题 2: 如何处理连接中断、解析失败？ → try-catch + 错误事件
问题 3: 如何增量解析 JSON（工具调用参数）？ → partial-json 库

不同 API 的 SSE 格式对比：

API	SSE 格式	特点
Anthropic	`event: content_block_delta` `data: {"delta":{"text":"Hello"}}`	事件类型明确
OpenAI	`data: {"choices":[{"delta":{"content":"Hello"}}]}`	无事件类型，只有 data
Google	完全不同的 JSON 结构	需要独立解析

代码实现：

1
// SSE 解析器：逐行读取，解析事件
2
async function* parseSSE(body: ReadableStream) {
3
    const reader = body.getReader();
4
    const decoder = new TextDecoder();
5
    let buffer = "";
6

7
    while (true) {
8
        const { value, done } = await reader.read();
9
        if (done) break;
10

11
        buffer += decoder.decode(value);
12

13
        // 逐行解析
14
        for (const line of buffer.split("\n")) {
15
            if (line.startsWith("data: ")) {
16
                yield JSON.parse(line.slice(6));  // 👈 产出事件
17
            }
18
        }
19
    }
20
}
21

22
// 流式 JSON 解析：解析不完整的 JSON
23
export function parseStreamingJson<T>(partialJson: string): T {
24
    try {
25
        return JSON.parse(partialJson);  // 尝试直接解析
26
    } catch {
27
        return partialParse(partialJson);  // 👈 partial-json 库解析不完整 JSON
28
    }
29
}
30

31
// 使用：边解析边推送事件
32
for await (const event of parseSSE(response.body)) {
33
    if (event.type === "content_block_delta") {
34
        stream.push({ type: "text_delta", delta: event.delta.text });
35
    }
36
}

设计呼应#

本质问题	核心原则	关键技术
流式性：LLM 生成是涌现过程	事件流模型：建模涌现过程	SSE 解析器 + partial-json

第 3 层：消息格式统一#

对应本质问题：异构性#

1
问题：不同提供商消息结构、字段名完全不同
2
挑战：如何让上层只看到统一格式？

实现核心原则：统一抽象层#

1
原则："在差异之上建立统一"
2
实现：统一内部格式 + 转换函数

具体实现#

问题分解：

问题 1: Anthropic 用 content blocks 数组 → 统一用数组
问题 2: OpenAI 用 content 字符串或数组 → 统一用数组
问题 3: 工具调用的字段名不同（input vs arguments） → 统一为 arguments
问题 4: thinking block 格式不同 → 统一 thinking + thinkingSignature

格式对比：

内容类型	Anthropic	OpenAI	统一内部格式
文本	`{type: "text", text: "..."}`	`delta.content`	`{type: "text", text: "..."}`
思考	`{type: "thinking", thinking: "...", signature: "..."}`	无	`{type: "thinking", thinking: "...", thinkingSignature: "..."}`
工具调用	`{type: "tool_use", id, name, input}`	`{id, function.name, function.arguments}`	`{type: "toolCall", id, name, arguments}`

代码实现：

1
// 统一内部格式
2
interface AssistantMessage {
3
    role: "assistant";
4
    content: ContentBlock[];  // 👈 统一用数组
5
    usage: Usage;
6
    stopReason: StopReason;
7
}
8

9
// Anthropic → 内部
10
function fromAnthropic(block) {
11
    if (block.type === "text") return { type: "text", text: block.text };
12
    if (block.type === "thinking") return { type: "thinking", thinking: block.thinking, thinkingSignature: block.signature };
13
    if (block.type === "tool_use") return { type: "toolCall", id: block.id, name: block.name, arguments: block.input };  // 👈 input → arguments
14
}
15

16
// OpenAI → 内部
17
function fromOpenAI(chunk) {
18
    if (chunk.choices[0].delta.content) {
19
        return { type: "text", text: chunk.choices[0].delta.content };
20
    }
21
    if (chunk.choices[0].delta.tool_calls) {
22
        return { type: "toolCall", id: chunk.id, name: chunk.function.name, arguments: chunk.function.arguments };
23
    }
24
}

设计呼应#

本质问题	核心原则	关键技术
异构性：字段名、结构不同	统一抽象层：在差异之上建立统一	统一内部格式 + 转换函数

第 4 层：事件驱动 + 统一抽象#

对应本质问题：流式性#

1
问题：LLM 生成是流式的，如何让应用层实时收到？
2
挑战：如何解耦底层与应用层？

实现核心原则：事件流模型#

1
原则："把涌现过程建模为事件流"
2
实现：EventStream + delta + partial

具体实现#

问题分解：

问题 1: LLM 生成是流式的，如何让应用层实时收到？ → EventStream 实现 AsyncIterable
问题 2: 不同事件类型，如何统一处理？ → 统一事件类型定义
问题 3: 如何让应用层不关心底层细节？ → 只暴露事件，隐藏内部

事件类型设计：

1
type AssistantMessageEvent =
2
    | { type: "start"; partial: AssistantMessage }          // 开始
3
    | { type: "text_delta"; delta: string; partial: AssistantMessage }     // 文本增量
4
    | { type: "thinking_delta"; delta: string; partial: AssistantMessage } // 思考增量
5
    | { type: "toolcall_start"; toolCall: ToolCall; partial: AssistantMessage }  // 工具调用开始
6
    | { type: "toolcall_delta"; delta: string; partial: AssistantMessage }      // 工具调用增量
7
    | { type: "done"; message: AssistantMessage }           // 完成
8
    | { type: "error"; error: AssistantMessage };           // 错误

delta vs partial 双模式：

字段	含义	用途
`delta`	本次新增的一小段	追加显示（打字机效果）
`partial`	当前完整状态	整体替换、访问元数据

EventStream 实现：

1
class EventStream implements AsyncIterable<Event> {
2
    private queue: Event[] = [];
3
    private resolvers: ((value: Event) => void)[] = [];
4
    private ended = false;
5

6
    push(event: Event) {
7
        if (this.resolvers.length > 0) {
8
            this.resolvers.shift()!(event);  // 👈 有人等待，直接交付
9
        } else {
10
            this.queue.push(event);          // 👈 暂存队列
11
        }
12
    }
13

14
    end() { this.ended = true; }
15

16
    async *[Symbol.asyncIterator]() {  // 👈 支持 for await...of
17
        while (!this.ended || this.queue.length > 0) {
18
            if (this.queue.length > 0) {
19
                yield this.queue.shift()!;
20
            } else {
21
                yield new Promise<Event>(resolve => this.resolvers.push(resolve));  // 👈 等待新事件
22
            }
23
        }
24
    }
25
}
26

27
// 使用：消费事件流
28
for await (const event of stream) {
29
    if (event.type === "text_delta") {
30
        process.stdout.write(event.delta);  // 👈 打字机效果
31
    }
32
}

设计呼应#

本质问题	核心原则	关键技术
流式性：如何实时传递	事件流模型：建模涌现过程	EventStream + delta/partial + AsyncIterable

第 5 层：错误处理#

对应本质问题：流式性#

1
问题：流式调用中途失败，如何处理？
2
挑战：不能用 throw 打断流，需要保留已生成内容

实现核心原则：事件流模型#

1
原则："错误也是事件，不打断流"
2
实现：错误即事件 + 保留部分结果

具体实现#

问题分解：

问题 1: 流式调用中途失败，如何通知应用层？ → 推送 error 事件
问题 2: 不能用 throw 打断流 → 错误作为事件，不抛异常
问题 3: 需要保留已生成的部分内容 → output 包含已生成内容

传统方式 vs 事件驱动方式：

1
// ❌ 传统方式：throw 打断流
2
try {
3
    for await (const chunk of stream) { ... }
4
} catch (e) {
5
    // 流被打断，已生成内容丢失
6
}
7

8
// ✅ 事件驱动：错误也是事件
9
for await (const event of stream) {
10
    if (event.type === "error") {
11
        // 处理错误，已生成内容在 event.error 中
12
        console.log("部分内容:", event.error.content);
13
    }
14
}

代码实现：

1
export const streamAnthropic = (model, context, options) => {
2
    const stream = new EventStream();
3

4
    (async () => {
5
        const output = createEmptyMessage();  // 👈 初始化输出
6

7
        try {
8
            stream.push({ type: "start", partial: output });
9

10
            for await (const event of parseSSE(response.body)) {
11
                // 处理事件，更新 output
12
                output.content.push(...);
13
                stream.push({ type: "text_delta", delta: ..., partial: output });
14
            }
15

16
            stream.push({ type: "done", message: output });
17
            stream.end();
18
        } catch (error) {
19
            // 👈 错误处理：不 throw，推送事件
20
            output.stopReason = "error";
21
            output.errorMessage = error.message;
22
            stream.push({ type: "error", error: output });  // 👈 output 包含已生成内容
23
            stream.end();  // 👈 正常结束流
24
        }
25
    })();
26

27
    return stream;
28
};

设计呼应#

本质问题	核心原则	关键技术
流式性：中途失败如何处理	事件流模型：错误即事件	error 事件 + 保留部分结果

第 6 层：跨供应商消息转换#

对应本质问题：迁移性#

1
问题：对话历史是"状态"，但不同提供商状态格式不互通
2
挑战：如何让"状态"在不同"世界"之间迁移？

实现核心原则：状态转换层#

1
原则："状态迁移时，转换不兼容部分，保留兼容部分"
2
实现：transformMessages + isSameModel + normalizeToolCallId

具体实现#

问题分解：

问题 1: Anthropic 的 thinkingSignature 对 OpenAI 无效 → 跨模型丢弃加密签名
问题 2: 工具调用 ID 格式不同 → normalizeToolCallId 标准化
问题 3: 某些模型不支持图片 → 图片降级为占位符
问题 4: 加密的 thinking block 跨模型无效 → 丢弃或转文本

转换规则表：

内容类型	同模型	跨模型	原因
text	保留	保留	无差异
thinking	保留	转为 text	加密签名无效
redacted thinking	保留	丢弃	只对原模型有效
toolCall	保留	ID 标准化	格式限制不同
image	保留	降级为占位符	能力差异

代码实现：

1
function transformMessages(messages: Message[], targetModel: Model): Message[] {
2
    return messages.map(msg => {
3
        if (msg.role === "assistant") {
4
            const isSameModel = msg.model === targetModel.id;  // 👈 判断是否跨模型
5

6
            return {
7
                ...msg,
8
                content: msg.content.map(block => {
9
                    // 加密 thinking：跨模型丢弃
10
                    if (block.type === "thinking" && block.redacted && !isSameModel) {
11
                        return [];  // 👈 丢弃
12
                    }
13

14
                    // 普通 thinking：跨模型转文本
15
                    if (block.type === "thinking" && !isSameModel) {
16
                        return { type: "text", text: block.thinking };  // 👈 转文本
17
                    }
18

19
                    // 工具调用：跨模型标准化 ID
20
                    if (block.type === "toolCall" && !isSameModel) {
21
                        return { ...block, id: normalizeToolCallId(block.id) };  // 👈 ID 标准化
22
                    }
23

24
                    return block;  // 👈 同模型保留原样
25
                }).filter(Boolean),
26
            };
27
        }
28

29
        // 用户消息：检查图片支持
30
        if (msg.role === "user" && !targetModel.input.includes("image")) {
31
            return {
32
                ...msg,
33
                content: msg.content.map(block =>
34
                    block.type === "image"
35
                        ? { type: "text", text: "(image omitted)" }  // 👈 图片降级
36
                        : block
37
                ),
38
            };
39
        }
40

41
        return msg;
42
    });
43
}
44

45
// ID 标准化：Anthropic 要求 ≤64 字符，只允许 [a-zA-Z0-9_-]
46
function normalizeToolCallId(id: string): string {
47
    return id.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 64);
48
}

设计呼应#

本质问题	核心原则	关键技术
迁移性：状态格式不互通	状态转换层：转换不兼容部分	transformMessages + isSameModel + normalizeToolCallId

层次与原则对应总表#

层次	对应本质问题	实现核心原则	关键技术
第 1 层：注册表+懒加载	异构性	统一抽象层	注册表 + 懒加载 + 类型包装
第 2 层：SSE解析	流式性	事件流模型	SSE解析器 + partial-json
第 3 层：消息格式统一	异构性	统一抽象层	统一内部格式 + 转换函数
第 4 层：事件驱动	流式性	事件流模型	EventStream + delta/partial
第 5 层：错误处理	流式性	事件流模型	错误即事件 + 保留部分结果
第 6 层：跨供应商转换	迁移性	状态转换层	transformMessages + ID标准化

三条主线：

1
异构性 ──► 统一抽象层 ──► 第1层（注册表） + 第3层（消息格式）
2

3
流式性 ──► 事件流模型 ──► 第2层（SSE解析） + 第4层（事件驱动） + 第5层（错误处理）
4

5
迁移性 ──► 状态转换层 ──► 第6层（跨供应商转换）

三、完整最小实现示例#

1
// ==================== 1. 统一类型 ====================
2
interface Message {
3
    role: "user" | "assistant" | "toolResult";
4
    content: ContentBlock[];
5
    model?: string;
6
}
7

8
interface ContentBlock {
9
    type: "text" | "thinking" | "toolCall";
10
    text?: string;
11
    thinking?: string;
12
    id?: string;
13
    name?: string;
14
    arguments?: object;
15
}
16

17
type Event =
18
    | { type: "start" }
19
    | { type: "text_delta"; delta: string }
20
    | { type: "done" }
21
    | { type: "error"; message: string };
22

23
// ==================== 2. 事件流 ====================
24
// event-stream.ts
25
class EventStream implements AsyncIterable<Event> {
26
    private queue: Event[] = [];
27
    private ended = false;
28

29
    push(event: Event) { this.queue.push(event); }
30
    end() { this.ended = true; }
31

32
    async *[Symbol.asyncIterator]() {
33
        while (!this.ended || this.queue.length > 0) {
34
            while (this.queue.length > 0) yield this.queue.shift()!;
35
            if (!this.ended) await new Promise(r => setTimeout(r, 10));
36
        }
37
    }
38
}
39

40
// ==================== 3. 注册表 ====================
41
// registry.ts
42
const registry = new Map<string, StreamFn>();
43

44
function register(api: string, stream: StreamFn) {
45
    registry.set(api, stream);
46
}
47

48
function getProvider(api: string) {
49
    return registry.get(api);
50
}
51

52
// ==================== 4. 懒加载 ====================
53
// lazy.ts
54
function lazyStream(load: () => Promise<Module>): StreamFn {
55
    return (model, ctx, opts) => {
56
        const stream = new EventStream();
57
        load().then(m => m.stream(model, ctx, opts, stream)).catch(e => {
58
            stream.push({ type: "error", message: e.message });
59
            stream.end();
60
        });
61
        return stream;
62
    };
63
}
64

65
// ==================== 5. Anthropic 提供商 ====================
66
// anthropic.ts
67
function streamAnthropic(model, ctx, opts, stream: EventStream) {
68
    stream.push({ type: "start" });
69

70
    fetch("https://api.anthropic.com/v1/messages", {
71
        method: "POST",
72
        headers: { "x-api-key": opts.apiKey },
73
        body: JSON.stringify({
74
            model: model.id,
75
            messages: convertMessages(ctx.messages),
76
            stream: true,
77
        }),
78
    }).then(async res => {
79
        for await (const line of parseSSE(res.body!)) {
80
            if (line.type === "content_block_delta") {
81
                stream.push({ type: "text_delta", delta: line.delta.text });
82
            }
83
        }
84
        stream.push({ type: "done" });
85
        stream.end();
86
    }).catch(e => {
87
        stream.push({ type: "error", message: e.message });
88
        stream.end();
89
    });
90
}
91

92
// ==================== 6. SSE 解析 ====================
93
// sse.ts
94
async function* parseSSE(body: ReadableStream) {
95
    const reader = body.getReader();
96
    const decoder = new TextDecoder();
97
    let buffer = "";
98

99
    while (true) {
100
        const { value, done } = await reader.read();
101
        if (done) break;
102
        buffer += decoder.decode(value);
103

104
        for (const line of buffer.split("\n")) {
105
            if (line.startsWith("data: ")) {
106
                yield JSON.parse(line.slice(6));
107
            }
108
        }
109
        buffer = "";
110
    }
111
}
112

113
// ==================== 7. 消息转换 ====================
114
// transform.ts
115
function convertMessages(messages: Message[]): any[] {
116
    return messages.map(m => ({
117
        role: m.role,
118
        content: m.content.map(b => {
119
            if (b.type === "text") return { type: "text", text: b.text };
120
            if (b.type === "toolCall") return { type: "tool_use", id: b.id, name: b.name, input: b.arguments };
121
            return b;
122
        }),
123
    }));
124
}
125

126
function transformForTarget(messages: Message[], targetModel: string): Message[] {
127
    return messages.map(m => m.model !== targetModel ? {
128
        ...m,
129
        content: m.content.map(b => b.type === "thinking" ? { type: "text", text: b.thinking } : b),
130
    } : m);
131
}
132

133
// ==================== 8. 注册内置提供商 ====================
134
// register-builtins.ts
135
register("anthropic-messages", lazyStream(() => import("./anthropic.ts")));
136
register("openai-responses", lazyStream(() => import("./openai.ts")));
137

138
// ==================== 9. 应用层使用 ====================
139
// app.ts
140
const provider = getProvider("anthropic-messages");
141
const stream = provider(
142
    { id: "claude-3-opus", api: "anthropic-messages" },
143
    { messages: transformForTarget(historyMessages, "claude-3-opus") },
144
    { apiKey: process.env.ANTHROPIC_API_KEY }
145
);
146

147
for await (const event of stream) {
148
    if (event.type === "text_delta") {
149
        process.stdout.write(event.delta);  // 打字机效果
150
    }
151
    if (event.type === "done") {
152
        console.log("\n完成");
153
    }
154
    if (event.type === "error") {
155
        console.error("错误:", event.message);
156
    }
157
}

四、额外技巧#

1. Unicode 代理对清理#

1
// 防止不完整的 Unicode 字符导致 API 失败
2
export function sanitizeSurrogates(text: string): string {
3
    return text.replace(
4
        /[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g,
5
        ""
6
    );
7
}

2. JSON 修复#

1
// LLM 可能生成非法 JSON，修复后解析
2
export function repairJson(json: string): string {
3
    // 转义字符串内的控制字符
4
    // 修复非法转义序列
5
}

3. 缓存控制#

1
// Anthropic 支持 prompt caching
2
params.system = [
3
    {
4
        type: "text",
5
        text: systemPrompt,
6
        cache_control: { type: "ephemeral" },  // 👈 缓存标记
7
    },
8
];

五、总结：问题 → 原则 → 实现#

三条主线#

1
┌─────────────────────────────────────────────────────────────────┐
2
│  本质问题         核心原则              实现层次                │
3
├─────────────────────────────────────────────────────────────────┤
4
│                                                                 │
5
│  异构性    ───►   统一抽象层   ───►   第1层 + 第3层             │
6
│  (不同世界)       (差异之上统一)        (注册表 + 消息格式)      │
7
│                                                                 │
8
│  流式性    ───►   事件流模型   ───►   第2层 + 第4层 + 第5层     │
9
│  (涌现过程)       (建模涌现)            (SSE + Event + 错误)    │
10
│                                                                 │
11
│  迁移性    ───►   状态转换层   ───►   第6层                     │
12
│  (状态不互通)     (转换不兼容)          (transformMessages)      │
13
│                                                                 │
14
└─────────────────────────────────────────────────────────────────┘

每层对应关系#

层次	本质问题	核心原则	关键技术
第 1 层：注册表+懒加载	异构性：不同提供商是不同”世界”	统一抽象层：在差异之上建立统一	注册表 + 懒加载 + 类型包装
第 2 层：SSE解析	流式性：LLM 生成是涌现过程	事件流模型：建模涌现过程	SSE解析器 + partial-json
第 3 层：消息格式统一	异构性：字段名、结构不同	统一抽象层：在差异之上建立统一	统一内部格式 + 转换函数
第 4 层：事件驱动	流式性：如何实时传递	事件流模型：建模涌现过程	EventStream + delta/partial
第 5 层：错误处理	流式性：中途失败如何处理	事件流模型：错误即事件	error事件 + 保留部分结果
第 6 层：跨供应商转换	迁移性：状态格式不互通	状态转换层：转换不兼容部分	transformMessages + ID标准化

六、设计模式总结#

模式	对应原则	应用场景
适配器模式	统一抽象层	不同 API 格式 → 统一内部格式
注册表模式	统一抽象层	统一管理多个提供商
懒加载模式	统一抽象层	按需加载，减少资源消耗
事件驱动模式	事件流模型	流式数据实时传递
状态转换模式	状态转换层	跨提供商兼容转换
防御性编程	所有原则	Unicode/JSON 修复，防止失败

七、核心启示#

1. 抽象比列举更重要#

1
表面痛点：
2
- API 格式不同
3
- SSE 格式各异
4
- 消息结构不同
5

6
本质问题：
7
- 异构性：不同提供商是不同"世界"

启示：不要只解决表面症状，找到本质问题，抽象出核心原则。

2. 原则指导实现#

1
原则："在差异之上建立统一"
2

3
→ 注册表：统一入口
4
→ 懒加载：按需加载
5
→ 类型包装：运行时校验
6
→ 消息格式：统一内部结构

启示：一个原则可以指导多个层次的实现。

3. 层次相互呼应#

1
第1层（注册表）解决异构性 ──► 第3层（消息格式）也解决异构性
2

3
两者共同实现"统一抽象层"原则

启示：不同层次可能解决同一个问题，需要整体设计。

音乐

音乐

统一多提供商 LLM API 设计详解

统一多提供商 LLM API 设计详解#

概述：什么是统一多提供商 LLM API？#

本质问题抽象#

设计核心：三个抽象原则#

设计目标映射#

一、整体架构图#

二、逐层解析：问题 → 原则 → 实现#

第 1 层：注册表 + 懒加载#

对应本质问题：异构性#

实现核心原则：统一抽象层#

具体实现#

设计呼应#

第 2 层：SSE 解析#

对应本质问题：流式性#

实现核心原则：事件流模型#

具体实现#

设计呼应#

第 3 层：消息格式统一#

对应本质问题：异构性#

实现核心原则：统一抽象层#

具体实现#

设计呼应#

第 4 层：事件驱动 + 统一抽象#

对应本质问题：流式性#

实现核心原则：事件流模型#

具体实现#

设计呼应#

第 5 层：错误处理#

对应本质问题：流式性#

实现核心原则：事件流模型#

具体实现#

设计呼应#

第 6 层：跨供应商消息转换#

对应本质问题：迁移性#

实现核心原则：状态转换层#

具体实现#

设计呼应#

层次与原则对应总表#

三、完整最小实现示例#

四、额外技巧#

1. Unicode 代理对清理#

2. JSON 修复#

3. 缓存控制#

五、总结：问题 → 原则 → 实现#

三条主线#

每层对应关系#

六、设计模式总结#

七、核心启示#

1. 抽象比列举更重要#

2. 原则指导实现#

3. 层次相互呼应#

支持与分享

音乐

文章目录