open-sse Architecture (中文 (简体))

为什么使用单独的工作区包？

出于以下几个原因，open-sse/ 是 OmniRoute monorepo 中的一个独立工作区：

1. 可复用性 — open-sse 以 @omniroute/open-sse 的名称发布到 npm，因此其他项目可以独立使用它
2. 清晰的边界 — 流式处理引擎与 OmniRoute 特有的 UI/DB 层解耦
3. 性能 — 该引擎不依赖 Next.js，因此在 CLI/无服务器环境中能够实现更快的冷启动
4. 版本管理 — open-sse 可以按照自己的节奏发布版本

package.json

"workspaces": ["open-sse"]

---

顶层结构

open-sse/
├── index.ts              # 公共入口点
├── types.d.ts            # 公共类型导出
├── package.json          # @omniroute/open-sse
├── config/               # 提供者配置、常量、注册表
├── executors/            # 各提供者的 HTTP 执行器（67 个，加上 base.ts/index.ts）
├── handlers/             # 请求处理器（chatCore、responses 等）
├── lib/                  # 内部工具
├── mcp-server/           # 模型上下文协议服务器
├── services/             # 约 298 个服务模块
├── transformer/          # Responses API 格式转换器
├── translator/           # 格式转换（OpenAI ↔ Claude ↔ Gemini）
└── utils/                # 共享工具（日志、错误、流等）

模块数量

| 目录 | 文件数 | 用途 | | executors/ | 167 | 各提供者的 HTTP 执行器（通过 DefaultExecutor 工厂统一） | | handlers/ | 157 | 请求入口点（chatCore、responses、embeddings） | | services/ | ~536 | 路由、缓存、速率限制、刷新等 | | translator/ | 56 | 格式转换（OpenAI ↔ Claude ↔ Gemini） | | mcp-server/ | 44 | MCP 工具和传输方式 | | utils/ | ~108 | 横切工具（日志、错误、流） | | config/ | ~339 | 提供者配置、常量、注册表 |

---

请求管道

每个 LLM 请求都会流经一个五阶段管道：

                              ┌──────────────┐
   HTTP 请求                  │  1. 路由     │   组合解析、模型选择
   （Next.js 路由）            └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │  2. 转换     │   格式转换（OpenAI ↔ Claude ↔ Gemini）
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │  3. 执行     │   提供者执行器、HTTP、重试、断路器
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │  4. 流式传输 │   SSE 转换、背压
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │  5. 记录     │   用量跟踪、调用日志、错误分类
                              └──────┬───────┘
                                     │
                                     ▼
                              HTTP 响应（SSE 或 JSON）

阶段 1：路由（services/combo.ts）

入口点：services/combo.ts 中的 handleComboChat()

将请求解析为具体的 (provider, model, account, credentials) 元组：

• 按 ID 查找组合（或为 auto/* 模型构建虚拟组合）
• 应用路由策略（优先级、加权、轮询等）
• 过滤掉不健康的提供者（断路器）
• 选择下一个可用目标

对于 auto/* 模型，此阶段还会：

• 运行十六因素评分算法（services/autoCombo/）
• 根据健康状况、成本、延迟等因素选择一个 provider+model 组合

阶段 2：转换（translator/）

如果源格式（例如 OpenAI）与目标格式（例如 Claude）不同，则会转换请求：

• 系统提示词 → 系统消息
• 工具定义 → 提供者特有的工具格式
• 推理/思考参数 → 提供者特有的等效参数
• 消息角色规范化（对于非 OpenAI 提供者，将 developer → system）

translator/index.ts 对外提供：

translateRequest(body, sourceFormat, targetFormat): TranslatedRequest
needsTranslation(source, target): boolean

阶段 3：执行（executors/）

入口点：getExecutor(providerId).execute(request, options)

所有提供者都通过 getExecutor() 工厂的回退机制使用 DefaultExecutor（executors/default.ts）。执行器会：

• 构建上游 URL（buildUrl()）
• 添加提供者特有的请求头（buildHeaders()）
• 转换请求正文（transformRequest()）
• 发送 HTTP 请求，并执行重试和指数退避
• 在需要时处理身份验证刷新（OAuth 提供者）

所有执行器都扩展自 BaseExecutor（executors/base.ts，1170 行代码），后者提供：

• 通用重试逻辑
• 代理集成
• 断路器集成
• 用量记录钩子

阶段 4：流式传输（utils/stream.ts）

对于流式响应，执行器会返回一个 ReadableStream。处理器会：

• 通过 SSE 转换流传递数据（createSSETransformStreamWithLogger）
• 应用心跳 ping 以检测失效连接
• 妥善处理客户端断开连接（pipeWithDisconnect）
• 为非流式客户端将 SSE 转换为 JSON

对于非流式响应，执行器会返回一个已解析的 JSON 对象，并保持原样传递。

阶段 5：记录（services/usage.ts）

响应后（无论成功或失败），都会记录使用情况：

• 响应中的 prompt_tokens、completion_tokens、cached_tokens
• 根据定价数据计算的 cost_usd
• latency_ms、status，以及失败时的 error_class
• 持久化到 usage_history 表

调用日志制品（如果已启用）将写入 ${DATA_DIR}/call_logs/。

---

关键文件深入解析

chatCore.ts（5977 行）

主请求处理器。尽管体量庞大，但其结构清晰：

// chatCore.ts 的伪结构
export async function handleChat(request: NextRequest) {
  // 1. 身份验证 + CORS
  await authenticateRequest(request);
  applyCorsHeaders(response);

  // 2. 请求体校验
  const body = await parseRequestBody(request);

  // 3. 格式检测 + 转换
  const sourceFormat = detectFormat(request);
  const targetFormat = getTargetFormat(providerId);
  if (needsTranslation(sourceFormat, targetFormat)) {
    body = translateRequest(body, sourceFormat, targetFormat);
  }

  // 4. 组合路由
  const targets = await resolveComboTargets(comboId, body);
  for (const target of targets) {
    try {
      const result = await executeOnTarget(target, body);
      await recordUsage(result);
      return result;
    } catch (err) {
      // 继续尝试下一个目标
    }
  }

  // 5. 紧急回退
  return await emergencyFallback(body);
}

尽管它是一个巨型函数，但内部按带注释的区段组织，这些区段与五阶段流水线一一对应。

combo.ts（4456 LOC）

将组合解析为有序目标列表的路由引擎。

services/combo.ts

export async function handleComboChat(body, comboId): Promise<ChatResult> {
  const targets = await resolveComboTargets(comboId, body);
  for (const target of targets) {
    try {
      return await handleSingleModel(target, body);
    } catch (err) {
      log.warn("目标失败，正在尝试下一个目标", { target, err });
    }
  }
  throw new ComboExhaustedError("所有目标均失败");
}

支持 19 种路由策略（参见 src/shared/constants/routingStrategies.ts）：

策略	行为
priority	按列表顺序优先选择第一个目标
weighted	根据每个目标的权重进行概率选择
round-robin	按顺序循环选择目标
context-relay	在不同目标之间传递上下文
fill-first	用满配额后再转向下一个目标
p2c	两个随机选择的幂
random	均匀随机选择
least-used	选择近期使用次数最少的目标
cost-optimized	优先选择成本最低且健康的目标
reset-aware	感知提供者的重置时间窗口
reset-window	基于重置时间窗口进行路由
headroom	优先选择剩余配额余量最大的目标
strict-random	真正均匀随机（不进行质量加权）
auto	使用 16 因子评分（autoCombo/）
lkgp	优先选择最近已知可用的提供者
context-optimized	最适合长上下文请求的目标
fusion	并行分发给多个目标组成的面板，然后通过裁判进行综合（fusion.ts）

base.ts（1170 LOC）

所有 107 个执行器都继承的抽象执行器。它包含：

• buildUrl() — 默认 URL 构造逻辑（子类可针对自定义需求覆盖）
• buildHeaders() — 默认请求头（身份验证、content-type）
• transformRequest() — 默认直接透传
• execute() — 包含重试、退避和断路器机制的主要 HTTP 循环

open-sse/executors/default.ts

export class DefaultExecutor extends BaseExecutor {
  // 处理所有与 OpenAI/Anthropic 兼容的提供者
  // 提供者注册配置（URL、身份验证、请求头），但共享执行器逻辑
}

提供者特定行为（身份验证请求头、基础 URL、版本请求头）通过提供者注册表进行配置，而不是通过单独的执行器类实现。

---

## 服务（117 个模块）

服务是由处理器组合的**专注于单一用途的模块**。主要类别包括：

### 路由与组合

- `combo.ts` — 组合路由请求的入口点
- `services/autoCombo/` — 16 因子评分、8 种自动路由策略
- `wildcardRouter.ts` — 匹配通配符路由（`gpt-*`）
- `modelFamilyFallback.ts` — T5 系列内回退

### 速率限制与配额

- `rateLimitManager.ts` — 每个密钥和提供者使用一个令牌桶
- `usage.ts` — 用量记录
- `quotaCache.ts` — 内存中的配额快照

### 账户与令牌

- `tokenRefresh.ts` — 遇到 401 时刷新 OAuth
- `accountFallback.ts` — 切换到备用账户
- `sessionManager.ts` — 多轮会话状态

### 智能功能

- `intentClassifier.ts` — 对请求意图进行分类
- `taskAwareRouter.ts` — 按任务类型路由
- `thinkingBudget.ts` — 分配思考令牌
- `contextManager.ts` — 注入路由上下文

### 弹性与容错

- `resilience.ts` — 重试、退避和熔断器编排
- `emergencyFallback.ts` — 最后手段回退
- `modelDeprecation.ts` — 自动路由到后继模型

### 状态

- `signatureCache.ts` — 按请求签名去重
- `volumeDetector.ts` — 负载削减
- `contextHandoff.ts` — 会话序列化

### 压缩

- `compression/`（子目录）— 完整的压缩流水线
- 39 个文件，涵盖引擎、规则包和适配器

### 技能

- （详见 [SKILLS.md](./SKILLS.md)）

### 记忆

- （详见 [MEMORY.md](./MEMORY.md)）

---

## 执行器（75+ 个文件）

每个提供者对应一个文件。它们都扩展 `BaseExecutor`，并覆写各自不同的部分。

### 通用模式

提供者通过 `getExecutor(providerId)` 解析，该函数返回已配置的执行器。兼容 OpenAI/Anthropic 的提供者使用 `DefaultExecutor`（`executors/default.ts`）。提供者特定的行为（基础 URL、身份验证标头、API 版本）在 `open-sse/config/providers/` 中配置，而请求正文转换则在 `open-sse/translator/` 中处理。

**自定义 URL** 通过提供者配置进行设置：

```ts
// open-sse/config/providers/ 中的提供者配置
export default {
  id: "together",
  baseURL: "https://api.together.xyz/v1/chat/completions",
}

自定义身份验证通过提供者注册表的身份验证配置（API 密钥、OAuth、标头配置文件）处理。

自定义请求正文转换（例如 Anthropic 将 system 与 messages 分离）在 open-sse/translator/ 中按提供者注册。

### 执行器工厂

`executors/index.ts` 导出 `getExecutor(providerId)`：

```ts
import { getExecutor } from "@omniroute/open-sse/executors";

const executor = getExecutor("anthropic");
const result = await executor.execute({
  model: "claude-sonnet-4-5",
  messages: [...],
});

解析过程通过 ExecutorRegistry（executors/registry.ts）完成：每个专用执行器都声明在 executors/index.ts 的内置表中，并在模块加载时通过 registerExecutor(alias, instance) 注册；getExecutor() 查询注册表，对于没有专用条目的提供者，则回退到记忆化的 DefaultExecutor。完整的别名 → 执行器映射由黄金测试 tests/unit/executor-map-golden.test.ts 进行特征验证。

---

转换器

在 3 种格式之间进行转换：OpenAI、Anthropic、Gemini，以及新的 Responses API。

何时进行转换

import { needsTranslation, translateRequest } from "@omniroute/open-sse/translator";

if (needsTranslation(sourceFormat, targetFormat)) {
  body = translateRequest(body, sourceFormat, targetFormat);
}

常见转换：

• OpenAI → Anthropic：独立的 system 字段、x-api-key 请求头
• OpenAI → Gemini：使用 contents 而非 messages，并使用 systemInstruction
• OpenAI → Responses API：input 数组、previous_response_id 状态

已处理的边界情况

• developer 角色 → 对于非 OpenAI 提供者转换为 system
• system 角色 → 对于 GLM/ERNIE 合并到第一条用户消息中
• json_schema → Gemini 的 responseMimeType + responseSchema
• tools → 提供者特定的工具格式
• 思考参数（o1、Claude）→ 提供者特定的等效参数

---

MCP 服务器

open-sse/mcp-server/ 实现了 Model Context Protocol 服务器：

• 110 个工具（提供者管理、组合、内存、缓存、压缩、代理、技能、游戏化、插件、Notion、Obsidian、本地语料库）
• 3 种传输方式：stdio、SSE、Streamable HTTP
• 33 个作用域，用于细粒度授权

工具注册

工具以独立文件的形式注册在 open-sse/mcp-server/tools/ 中，每个文件都导出名称、schema、handler 和 scope：

open-sse/mcp-server/tools/getHealth.ts

import { z } from "zod";
export default {
  name: "omniroute_get_health",
  description: "Get system health snapshot",
  scope: "read:health",
  inputSchema: z.object({}),
  handler: async (_args, ctx) => {
    return await getSystemHealth();
  },
};

传输方式

// stdio（CLI 用法）
startMcpStdio(server);

// SSE（基于 HTTP 的流式传输）
startMcpSse(server, port);

// Streamable HTTP（现代 MCP）
startMcpStreamable(server, port);

授权

每次工具调用都会经过作用域检查（open-sse/mcp-server/auth/）：

if (!hasScope(apiKey, "providers:read")) {
  throw new Error("Insufficient scope");
}

---

转换器

open-sse/transformer/ 在 Chat Completions 和 Responses API 格式之间进行转换。

为什么需要独立的转换器？

Responses API 是 OpenAI 推出的新格式，支持有状态对话（previous_response_id）。当客户端发送 Responses 请求时，OmniRoute 会：

1. 在内部将 Responses 转换为 Chat Completions
2. 将请求发送给提供者（任何支持 Chat Completions 的提供者）
3. 将响应转换回 Responses 格式
4. 将转换后的响应以流式方式发送给客户端

转换器（transformer/responsesTransformer.ts）提供：

createResponsesApiTransformStream(): TransformStream

它会处理：

• response.output_item.added 事件
• response.output_text.delta 事件
• response.completed 事件
• 工具调用映射（function_call ↔ tool_calls）

---

配置

open-sse/config/ 包含配置层：

文件	用途
providerRegistry.ts	基于 352 个提供者目录的聊天模型注册表
providerModels.ts	模型别名、格式映射
constants.ts	超时时间、限制、状态码
defaultThinkingSignature.ts	默认 Claude 思考签名
modelStrip.ts（位于 services 中）	按提供者移除字段

提供者注册表 Schema

interface ProviderConfig {
  id: string;
  name: string;
  baseUrl: string;
  authType: "bearer" | "api-key" | "oauth" | "cookie";
  executorClass: string;
  defaultModel: string;
  capabilities: ProviderCapabilities;
  models: ModelDefinition[];
}

模块加载时进行 Zod 验证，以确保所有提供者配置均有效。

---

性能约束

路由引擎具有严格的性能预算：

操作	目标	测量条件
组合解析	<10ms	针对 50 个目标
速率限制检查	<1ms	内存令牌桶
模型系列回退	<5ms	已缓存的系列定义
请求路由分发	<2ms	热路径
路由热路径中无阻塞 I/O	—	全部异步

---

反模式

❌ 在 combo.ts 中进行同步数据库调用 — 预先计算并缓存 ❌ 在处理程序中实现重试逻辑 — 使用弹性服务中的 retry() ❌ 直接访问提供者配置 — 使用 providerRegistry 获取器 ❌ 硬编码回退链 — 在 modelFamilyFallback.ts 中定义 ❌ 跨并发请求修改状态 — 仅使用请求作用域上下文

---

添加新组件

添加新服务

1. 创建 open-sse/services/[serviceName].ts，并确保职责单一
2. 导出主处理函数和所有常量
3. 在 tests/unit/services/[serviceName].test.mjs 中添加单元测试
4. 在 handlers/chatCore.ts 中集成到请求管道（如果与路由相关）
5. 如果服务影响目标选择，则更新 combo.ts 中的路由逻辑
6. 在此文件中添加文档

添加新执行器

1. 创建扩展 BaseExecutor 的 open-sse/executors/[provider].ts
2. 在 config/providerRegistry.ts 中注册
3. 添加到 executors/index.ts 工厂
4. 为执行器添加单元测试
5. 在 docs/architecture/ARCHITECTURE.md 中添加文档

添加新 MCP 工具

1. 创建或更新 open-sse/mcp-server/tools/[category]Tools.ts
2. 为输入定义 Zod 模式
3. 在 mcp-server/index.ts 中注册工具
4. 添加到 mcp-server/auth/ 中的作用域矩阵
5. 添加单元测试

---

另请参阅

• ARCHITECTURE.md — 高层架构
• CODEBASE_DOCUMENTATION.md — 工程参考
• REPOSITORY_MAP.md — 逐目录说明
• AUTO-COMBO.md — 16 因子评分
• MCP-SERVER.md — MCP 服务器
• A2A-SERVER.md — A2A 服务器
• 源代码：open-sse/（400 多个文件，约 143K 行代码）

---

OmniRoute 源码 (a58000c7685f)