06-LLM 多协议抽象 Protocol 管道
作者：互联网 时间： 2026-07-15 19:27:52
06-LLM 多协议抽象的 Protocol 管道
一、Protocol：一个模型服务族的语义 API 契约
@opencode-ai/llm 包的核心设计目标是：将"这个 API 长什么样"（语义契约）与"请求发到哪里、用什么认证、怎么传输"（部署契约）正交分离。Protocol 是前者的载体——它只 owns 三件事：公共 LLMRequest 如何变成 provider 原生请求体、请求体必须满足什么 schema、流式响应如何解码回公共 LLMEvent。
1.1 四类型参数的定义
// 36:43:opencode/packages/llm/src/route/protocol.tsexport interface Protocol<Body, Frame, Event, State> {/** Stable id for the wire protocol implementation. */readonly id: ProtocolID/** Request side: schema for the provider-native body and how to build it. */readonly body: ProtocolBody<Body>/** Response side: streaming state machine. */readonly stream: ProtocolStream<Frame, Event, State>}
四个类型参数精确对应管道的四个阶段：
| 类型参数 | 管道阶段 | 语义 | 生命周期 |
|---|---|---|---|
Body | 请求体 | provider 原生请求体候选。Route.make 用 body.schema 校验后 JSON 编码发出 | 每次 compile 构建 |
Frame | 传输帧 | 流式响应的一个分帧单元。SSE 是一段 JSON data: 字符串；AWS event stream 是一个解析后的二进制帧 | 每个分帧周期 |
Event | 原生事件 | 从一个 Frame 经 stream.event schema 解码后的 provider 原生事件 | 每帧一个 |
State | 解析状态 | 贯穿 stream.step 的累加器，将事件序列翻译为 LLMEvent 序列 | 每次响应一个 |
1.2 请求侧：ProtocolBody
// 45:50:opencode/packages/llm/src/route/protocol.tsexport interface ProtocolBody<Body> {/** Schema for the validated provider-native body sent as the JSON request. */readonly schema: Schema.Codec<Body, unknown>/** Build the provider-native body from a common `LLMRequest`. */readonly from: (request: LLMRequest) => Effect.Effect<Body, LLMError>}
ProtocolBody 有两个字段：
schema：Body类型的 JSON Codec——它既是类型来源（TypeScript 通过 schema 推断Body），也是运行时校验器（compile用它decodeUnknown确保构建的 body 满足契约）from：把公共LLMRequest降级（lowering）为 provider 原生 body 的 Effect 函数。这是唯一知道公共消息如何映射到 provider 线格式的地方——provider 的 quirks 被锁在这里，不泄漏到LLMRequest
1.3 响应侧：ProtocolStream
// 52:63:opencode/packages/llm/src/route/protocol.tsexport interface ProtocolStream<Frame, Event, State> {/** Schema for one decoded streaming event, decoded from a transport frame. */readonly event: Schema.Codec<Event, Frame>/** Initial parser state. Called once per response with the resolved request. */readonly initial: (request: LLMRequest) => State/** Translate one event into emitted `LLMEvent`s plus the next state. */readonly step: (state: State, event: Event) => Effect.Effect<readonly [State, ReadonlyArray<LLMEvent>], LLMError>/** Optional request-completion signal for transports that do not end naturally. */readonly terminal?: (event: Event) => boolean/** Optional flush emitted when the framed stream ends. */readonly onHalt?: (state: State) => ReadonlyArray<LLMEvent>}
ProtocolStream 是一个流式状态机，五个字段对应五个阶段：
event：Schema.Codec<Event, Frame>——从一个传输帧解码出原生事件的 schema。类型上Frame → Event，是ProtocolBody.schema（Body ↔ unknown）在响应侧的镜像initial：每个响应调用一次，传入 resolved request，返回初始State。这通常是ToolStream.empty()+Lifecycle.initial()的组合step：核心翻译器——输入(当前 State, 一个 Event)，输出[下一个 State, 要发射的 LLMEvent 数组]。这是把 provider 的流式语义（OpenAI 的 delta、Anthropic 的 content_block_delta）翻译为公共事件（text-delta、tool-input-delta）的唯一位置terminal：可选——对于不自然结束的传输（如 WebSocket），标记哪个事件表示完成onHalt：可选——当分帧流结束时（如 OpenAI 发完所有 chunk），用最终 State 冲刷出剩余事件（如 Anthropic 没有 onHalt，而 OpenAI Chat 用它发射 finish 事件）
1.4 完整流水线：request → body → frames → events → LLMEvent
将四个类型参数和两块（Body / Stream）组合，完整的请求-响应流水线如下：
┌─────────────────────────────────────────────────────────────┐│compile() 编译阶段 │││LLMRequest ──► applyCachePolicy() ──► resolveRequestOptions() │││││ route.body.from(resolved)│││││Body (候选)│││││ Schema.decodeUnknown(body.schema)│││││Body (已校验)│││││route.prepareTransport(body)│││││Prepared (HTTP 预备) │└────────────────────────────────────┼────────────────────────┘ │┌────────────────────────────────────▼────────────────────────┐│streamPrepared() 流式执行阶段│││ Prepared ──► transport.frames(prepared, runtime) │││ Stream<Frame>◄── Framing.sse / bedrock-event │││Stream.mapAccumEffect│ ┌────┴────────────────────────┐ │ │decodeEvent(frame)│← stream.event │ │→ Event │ schema 解码 │ └────┬────────────────────────┘ │││ stream.step(state, event) │││ [nextState, LLMEvent[]] │││ Stream<LLMEvent>──► 公共事件流│└─────────────────────────────────────────────────────────────┘
类型流向不变量：LLMRequest → Body（经 body.from），Body → unknown（经 body.schema 编码），HTTP 传输后 Uint8Array → Frame（经 Framing.frame），Frame → Event（经 stream.event 解码），Event + State → LLMEvent[]（经 stream.step）。每一步的类型转换都有对应的 Schema.Codec 站岗——schema 既是编译期的类型推断来源，也是运行时的校验闸门。
二、Route.make()：四轴正交组合
Route.make 是协议层的核心构造器。它把四个正交的部署维度组合成一个可执行的 Route：一个 Protocol + 一个 Endpoint + 一个 Auth + 一个 Framing（+ 可选 headers/defaults）。这个分离让 DeepSeek、TogetherAI 等只需复用 OpenAIChat.protocol，零代码即可接入。
2.1 Route 的完整结构
// 36:53:opencode/packages/llm/src/route/client.tsexport interface Route<Body, Prepared = unknown> {readonly id: stringreadonly provider?: ProviderIDreadonly protocol: ProtocolIDreadonly endpoint: Endpoint<Body>readonly auth: AuthDefreadonly transport: Transport<Body, Prepared, unknown>readonly defaults: RouteDefaultsreadonly body: RouteBody<Body>readonly with: (patch: RoutePatch<Body, Prepared>) => Route<Body, Prepared>readonly model: (input: RouteMappedModelInput) => Modelreadonly prepareTransport: (body: Body, request: LLMRequest) => Effect.Effect<Prepared, LLMError>readonly streamPrepared: (prepared: Prepared,request: LLMRequest,runtime: TransportRuntime,) => Stream.Stream<LLMEvent, LLMError>}
Route 持有四轴的绑定结果（protocol/endpoint/auth/transport），加上两个运行时方法（prepareTransport + streamPrepared）和一个可变性方法（with 用于 patch）。
2.2 四轴正交模型
┌───────────────────────────────────────────────────────┐ │Route.make(input)│ ││ │ ┌─────────────┐┌──────────────┐│ │ │Protocol ││ Endpoint ││ │ │ (语义契约)││ (URL 在哪) ││ │ │ Body/Frame/ ││ baseURL+path ││ │ │ Event/State ││ +query ││ │ └──────┬──────┘└──────┬───────┘│ ││││ │ ┌──────▼──────┐┌──────▼───────┐│ │ │Auth ││ Framing││ │ │ (怎么认证)││ (怎么分帧) ││ │ │ bearer/none ││ sse/binary ││ │ │ /header/... ││││ │ └──────┬──────┘└──────┬───────┘│ ││││ │└───────┬────────┘│ │▼│ │HttpTransport.httpJson│ │(默认 transport = httpJson) │ │││ │▼│ │ Route<Body, Prepared> │ └───────────────────────────────────────────────────────┘
四轴的职责边界：
| 轴 | 接口 | 回答的问题 | 示例 |
|---|---|---|---|
| Protocol | Protocol<Body,Frame,Event,State> | 我说的是什么 API？ | OpenAI Chat / Anthropic Messages / Gemini / Bedrock Converse |
| Endpoint | Endpoint<Body> | 请求发到哪个 URL？ | baseURL + path（可为函数，URL 可嵌入 model id/region） |
| Auth | Auth | 怎么认证？ | bearer / header / none / custom，支持 orElse/andThen 组合 |
| Framing | Framing<Frame> | 响应字节流怎么切成帧？ | sse（SSE data: 行）/ AWS event stream（二进制长度前缀帧） |
2.3 Route.make 的两个重载
// 303:339:opencode/packages/llm/src/route/client.tsexport function make<Body, Prepared, Frame, Event, State>(input: MakeTransportInput<Body, Prepared, Frame, Event, State>,): Route<Body, Prepared>// ...export function make<Body, Frame, Event, State>(input: MakeInput<Body, Frame, Event, State>,): Route<Body, HttpTransport.HttpPrepared<Frame>>export function make<Body, Prepared, Frame, Event, State>(input: MakeInput<Body, Frame, Event, State> | MakeTransportInput<Body, Prepared, Frame, Event, State>,): Route<Body, Prepared> | Route<Body, HttpTransport.HttpPrepared<Frame>> {if ("transport" in input) return makeFromTransport(input)const protocol = input.protocolreturn makeFromTransport({id: input.id,provider: input.provider,protocol,endpoint: input.endpoint,auth: input.auth,headers: input.headers,transport: HttpTransport.httpJson({ framing: input.framing }),defaults: input.defaults,})}
两个重载的区别在于第四轴的提供方式：
MakeInput（常用）：调用方只提供framing，make自动用HttpTransport.httpJson({ framing })构造默认 HTTP transport。适用于绝大多数 HTTP+SSE 的 providerMakeTransportInput（高级）：调用方直接提供完整Transport<Body, Prepared, Frame>。适用于 WebSocket（如 OpenAI Responses WebSocket Route）或自定义传输
无论哪个重载，最终都走 makeFromTransport，保证两条路径的 build() 逻辑一致。
2.4 makeFromTransport：Route 的构建工厂
makeFromTransport 内部的 build() 函数完成了四轴到运行时方法的绑定：
// 247:301:opencode/packages/llm/src/route/client.tsconst build = (routeInput: BuiltRouteInput): Route<Body, Prepared> => {const route: Route<Body, Prepared> = {id: routeInput.id,provider: routeInput.provider === undefined ? undefined : ProviderID.make(routeInput.provider),protocol: protocol.id,endpoint: routeInput.endpoint,auth: routeInput.auth ?? Auth.none,transport: routeInput.transport,defaults: routeInput.defaults ?? {},body: protocol.body,with: (patch: RoutePatch<Body, Prepared>) => { /* ... 深拷贝+合并 ... */ },model: (input) => makeRouteModel(route, input),prepareTransport: (body, request) =>routeInput.transport.prepare({ /* ... endpoint/auth/encodeBody/headers ... */ }),streamPrepared: (prepared, request, runtime) => {// frames → decode → step → LLMEvent},} satisfies Route<Body, Prepared>return route}
两个运行时方法的关键实现：
prepareTransport——把 Body + LLMRequest 交给 transport 的 prepare，后者组装 URL（Endpoint.render）、叠加 headers、应用 Auth、JSON 编码 body，产出 Prepared（HTTP 场景是 HttpClientRequest + Framing）。
streamPrepared——完整的帧→事件翻译管道：
// 279:295:opencode/packages/llm/src/route/client.tsstreamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {const route = `${request.model.provider}/${request.model.route.id}`const events = routeInput.transport.frames(prepared, request, runtime).pipe(Stream.mapEffect(decodeEvent(route)),protocol.stream.terminal ? Stream.takeUntil(protocol.stream.terminal) : (stream) => stream,)return events.pipe(Stream.mapAccumEffect(() => protocol.stream.initial(request),protocol.stream.step,protocol.stream.onHalt ? { onHalt: protocol.stream.onHalt } : undefined,),Stream.catchCause((cause) => Stream.fail(streamError(route, `Failed to read ${route} stream`, cause))),)},
管道三步：
transport.frames()——执行 HTTP 请求，把响应字节流经Framing.frame切成Stream<Frame>Stream.mapEffect(decodeEvent)——每帧用protocol.stream.eventschema 解码成EventStream.mapAccumEffect(initial, step)——以initial为初始 State，逐个 Event 调step，产出Stream<LLMEvent>
decodeEvent 在 makeFromTransport 顶部通过 Schema.decodeUnknownEffect(protocol.stream.event) 预编译一次，避免每个帧重复编译 schema。
2.5 DeepSeek/TogetherAI 如何复用 OpenAIChat.protocol
正交分离的直接收益是：provider 只需覆盖 Endpoint + Auth 两轴，Protocol 和 Framing 原样复用。
第一步：OpenAIChat.protocol 定义了 chat completions 的语义契约（body schema + streaming state machine）：
// 481:493:opencode/packages/llm/src/protocols/openai-chat.tsexport const protocol = Protocol.make({id: ADAPTER,// "openai-chat"body: {schema: OpenAIChatBody,from: fromRequest,},stream: {event: Protocol.jsonEvent(OpenAIChatEvent),initial: () => ({ tools: ToolStream.empty(), toolCallEvents: [], lifecycle: Lifecycle.initial() }),step,onHalt: finishEvents,},})
第二步：OpenAICompatibleChat.route 原样复用这个 protocol，只改 route id 防止与原生 OpenAI 冲突：
// 17:22:opencode/packages/llm/src/protocols/openai-compatible-chat.tsexport const route = Route.make({id: ADAPTER,// "openai-compatible-chat"protocol: OpenAIChat.protocol,// ← 原样复用，零改动endpoint: Endpoint.path("/chat/completions"),framing: Framing.sse,})
第三步：openai-compatible.ts 的 define(profile) 工厂为每个 provider 覆盖 Endpoint（baseURL）+ Auth（bearer），生成专属 Route：
// 38:52:opencode/packages/llm/src/providers/openai-compatible.tsconst define = (profile: OpenAICompatibleProfile) => {const configureProfile = (input: FamilyModelOptions = {}) => {const facade = configure({...input,baseURL: input.baseURL ?? profile.baseURL, // ← 只改 baseURLprovider: profile.provider,})return {id: ProviderID.make(profile.provider),model: facade.model,configure: configureProfile,}}return configureProfile()}export const deepseek = define(profiles.deepseek) // baseURL = "https://api.deepseek.com/v1"export const togetherai = define(profiles.togetherai)// baseURL = "https://api.together.xyz/v1"export const cerebras = define(profiles.cerebras) // baseURL = "https://api.cerebras.ai/v1"
configure 内部通过 route.with({ endpoint: { baseURL }, auth: AuthOptions.bearer(...) }) patch 一个新 Route——因为 with 是不可变拷贝，原 OpenAICompatibleChat.route 不受影响，每个 provider 拿到自己绑定的 Route 实例。
复用度量：DeepSeek/TogetherAI/Cerebras/Groq/Fireworks/DeepInfra/Baseten 七个 provider 共享同一份 OpenAIChat.protocol（~500 行 body schema + lowering + stream parser），每个 provider 只需 3 行 define(profile)。如果没有四轴正交，每个 provider 要复制全部 500 行——这正是 protocol.ts 注释所说的"This separation is what lets DeepSeek, TogetherAI, Cerebras, etc. all reuse OpenAIChat.protocol without forking 300 lines per provider"。
三、compile()：编译边界
compile 是 LLMClient 的核心内部函数，是公共 LLMRequest 与 provider 原生世界之间的编译边界——它把请求编译成"已校验的 body + 已预备的 transport 数据"，但不执行传输。
3.1 compile 的四步流水线
// 344:359:opencode/packages/llm/src/route/client.tsconst compile = Effect.fn("LLM.compile")(function* (request: LLMRequest) {const resolved = applyCachePolicy(resolveRequestOptions(request))const route = resolved.model.routeconst body = yield* route.body.from(resolved).pipe(Effect.flatMap(ProviderShared.validateWith(Schema.decodeUnknownEffect(route.body.schema))))const prepared = yield* route.prepareTransport(body, resolved)return {request: resolved,route,body,prepared,}})
四步：
resolveRequestOptions(request)——合并三层默认值（route defaults → model defaults → request 字段），产出 resolved requestapplyCachePolicy(resolved)——根据 cache policy 注入 inline cache hints（详见第四节）route.body.from(resolved)→Schema.decodeUnknown(body.schema)——调用 protocol 的 lowering 函数构建 provider body，然后用 body schema 校验它。ProviderShared.validateWith包装了 decode，失败时产出InvalidProviderOutput类型化错误route.prepareTransport(body, resolved)——把已校验的 body + request 交给 transport 的prepare，产出 transport 私有的Prepared数据（HTTP 场景是组装好的HttpClientRequest+Framing）
3.2 resolveRequestOptions：三层默认值合并
// 167:180:opencode/packages/llm/src/route/client.tsconst resolveRequestOptions = (request: LLMRequest) => {const routeDefaults = request.model.route.defaultsconst modelDefaults = request.model.defaultsconst generation = mergeGenerationOptions(routeDefaults.generation, modelDefaults?.generation, request.generation)return LLMRequest.update(request, {generation: generation ?? new GenerationOptions({}),providerOptions: mergeProviderOptions(routeDefaults.providerOptions,modelDefaults?.providerOptions,request.providerOptions,),http: mergeHttpOptions(routeDefaults.http, modelDefaults?.http, request.http),})}
三层优先级从低到高：route.defaults（协议级默认）→ model.defaults（模型级默认）→ request.*（请求级显式）。mergeGenerationOptions 用 findLast 取最后一个非 undefined 值（后者覆盖前者），mergeProviderOptions/mergeHttpOptions 做深合并（JSON object 递归合并）。
3.3 编译边界的设计含义
compile 是"编译"而非"执行"，它有三个关键设计约束：
约束一：编译不执行传输。compile 产出 { request, route, body, prepared } 四元组，但 prepared 只是 transport 的预备数据（HTTP 是 HttpClientRequest 对象），网络请求尚未发出。这让 LLMClient.prepare() 可以安全地暴露给调试 UI 和请求预览——用户可以看到"即将发什么"而不实际发送。
约束二：body schema 是编译期闸门。body.from 产出候选 body 后，Schema.decodeUnknown(body.schema) 立即校验。这意味着 protocol 实现者的 lowering 逻辑如果有 bug（如遗漏 required 字段、类型不匹配），会在 compile 阶段就失败，而不是等请求发出去后才在 provider 端报 400。错误类型是 InvalidProviderOutputReason——"provider 输出无效"，因为这是 protocol 自己产出的 body 不合法。
约束三：cache policy 在 body 构建前注入。applyCachePolicy 在 resolveRequestOptions 之后、body.from 之前执行，这样 lowering 函数拿到的 resolved request 已经带了 cache hints，可以在构建 body 时自然地把 hints 翻译为 wire markers（如 Anthropic 的 cache_control）。这是一个"在正确位置注入，让后续自然处理"的设计模式。
3.4 prepare 与 stream 如何消费 compile 的产出
compile 的产出被三个公共方法消费：
// 361:408:opencode/packages/llm/src/route/client.tsconst prepareWith = Effect.fn("LLMClient.prepare")(function* (request: LLMRequest) {const compiled = yield* compile(request)return new PreparedRequest({id: compiled.request.id ?? "request",route: compiled.route.id,protocol: compiled.route.protocol,model: compiled.request.model,body: compiled.body,metadata: { transport: compiled.route.transport.id },})})const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest) =>Stream.unwrap(Effect.gen(function* () {const compiled = yield* compile(request)return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)}),)const generateWith = (stream: Interface["stream"]) =>Effect.fn("LLM.generate")(function* (request: LLMRequest) {const state = yield* stream(request).pipe(Stream.runFold(LLMResponse.empty, LLMResponse.reduce))const response = LLMResponse.complete(state)if (response) return responsereturn yield* ProviderShared.eventError(/* ... "stream ended without a terminal finish event" */)})
prepare：只调compile，不执行。返回PreparedRequest（包含 body + route/protocol/model 元数据）stream：调compile后用Stream.unwrap把 Effect 转为 Stream，然后执行streamPrepared（含传输 + 分帧 + 解码 + 状态机）。Stream.unwrap让 compile 的 Effect 在 Stream 首次拉取时执行，实现 lazy 语义generate：在stream基础上用Stream.runFold归约成LLMResponse。归约器是LLMResponse.reduce（定义在schema/events.ts），它把事件流组装成Message+usage+finishReason
四、cache-policy.ts：inline hint 注入
applyCachePolicy 是 compile 的第一步，负责根据请求的 cache 策略自动在正确的 part 上注入 CacheHint，让后续的 lowering 函数自然处理。
4.1 为什么只对 anthropic-messages 和 bedrock-converse 注入
// 39:42:opencode/packages/llm/src/cache-policy.ts// Protocols whose wire format ignores inline cache markers (OpenAI's implicit// prefix caching, Gemini's implicit + out-of-band CachedContent). Skip the// whole policy pass for these — emitting hints would be harmless but pointless.const RESPECTS_INLINE_HINTS = new Set(["anthropic-messages", "bedrock-converse"])
// 99:101:opencode/packages/llm/src/cache-policy.tsexport const applyCachePolicy = (request: LLMRequest): LLMRequest => {if (!RESPECTS_INLINE_HINTS.has(request.model.route.id)) return request// ...}
原因在于不同 provider 的缓存机制不同：
| Provider | 缓存机制 | 是否尊重 inline hint |
|---|---|---|
| Anthropic Messages | 显式 cache_control: { type: "ephemeral" } 标记在 content block 上 | ✅ 是——wire format 有 inline marker 字段 |
| Bedrock Converse | 显式 cachePoint 标记在 body 中 | ✅ 是——wire format 有 inline marker |
| OpenAI Chat/Responses | 隐式 prefix caching——自动缓存请求前缀，无需 marker | ❌ 否——wire format 无 marker 字段 |
| Gemini | 隐式 + out-of-band CachedContent API | ❌ 否——用独立 API 管理缓存 |
对 OpenAI/Gemini 注入 hints 是"无害但无意义的"（harmless but pointless）——lowering 函数会忽略 CacheHint 字段。因此 applyCachePolicy 在第一行就短路返回，跳过整个策略 pass，避免对消息数组做无谓的遍历和拷贝。
4.2 auto 策略的三个断点
// 18:23:opencode/packages/llm/src/cache-policy.tsconst AUTO: CachePolicyObject = {tools: true,system: true,messages: "latest-user-message",}
默认 "auto" 策略在三个位置放置 cache 断点：
tools: true——最后一个 tool definition 上。tools 在缓存层级中最高（tool 列表变化频率最低）system: true——最后一个 system part 上。system prompt 在一次会话中通常不变messages: "latest-user-message"——最近一条 user message 的最后一个 text part 上
这个设计的经济学依据在注释中写明：Anthropic 5 分钟缓存的写入成本是 1.25x base，读取成本是 0.1x base——单次复用即已盈利。在 Agent 循环中，一次 user message 会触发多轮 assistant/tool 来回，这些来回都共享同一前缀，在 "latest user message" 处设断点让每一轮的 intra-turn API call 都命中缓存。
4.3 策略解析与断点注入
// 33:37:opencode/packages/llm/src/cache-policy.tsconst resolve = (policy: CachePolicy | undefined): CachePolicyObject => {if (policy === undefined || policy === "auto") return AUTOif (policy === "none") return NONEreturn policy}
// 104:111:opencode/packages/llm/src/cache-policy.tsconst hint = makeHint(policy.ttlSeconds)const tools = policy.tools ? markLastTool(request.tools, hint) : request.toolsconst system = policy.system ? markLastSystem(request.system, hint) : request.systemconst messages = policy.messages ? markMessages(request.messages, policy.messages, hint) : request.messagesif (tools === request.tools && system === request.system && messages === request.messages) return requestreturn LLMRequest.update(request, { tools, system, messages })
解析规则：
undefined→"auto"（默认开启缓存，数学上划算）"auto"→ tools + system + latest-user-message"none"→ 不做自动放置（但手动CacheHint仍然流转）- 对象形式 → 精确按调用方指定的位置
注入逻辑尊重手动放置——markLastTool/markLastSystem/markMessageAt 都检查目标 part 是否已有 cache 字段，有则跳过。这保证用户的显式控制不被覆盖。
消息策略的三种模式：
// 85:97:opencode/packages/llm/src/cache-policy.tsconst markMessages = (messages: ReadonlyArray<Message>,strategy: NonNullable<CachePolicyObject["messages"]>,hint: CacheHint,): ReadonlyArray<Message> => {if (messages.length === 0) return messagesif (strategy === "latest-user-message") return markMessageAt(messages, lastIndexOfRole(messages, "user"), hint)if (strategy === "latest-assistant") return markMessageAt(messages, lastIndexOfRole(messages, "assistant"), hint)const start = Math.max(0, messages.length - strategy.tail)let next = messagesfor (let i = start; i < messages.length; i++) next = markMessageAt(next, i, hint)return next}
"latest-user-message"：在最后一条 user message 的最后一个 text part 上标记"latest-assistant"：在最后一条 assistant message 上标记{ tail: N }：在最后 N 条 message 上各标记一个断点
4.4 注入后的流转路径
applyCachePolicy 注入 CacheHint 后，hints 通过 LLMRequest 的 part 字段流转到 protocol 的 lowering 函数。以 Anthropic 为例：
// 243:251:opencode/packages/llm/src/protocols/anthropic-messages.tsconst cacheControl = (breakpoints: Cache.Breakpoints, cache: CacheHint | undefined) => {if (cache?.type !== "ephemeral" && cache?.type !== "persistent") return undefinedif (breakpoints.remaining <= 0) {breakpoints.dropped += 1return undefined}breakpoints.remaining -= 1return Cache.ttlBucket(cache.ttlSeconds) === "1h" ? EPHEMERAL_1H : EPHEMERAL_5M}
Anthropic 的 lowering 函数读取每个 part 的 cache 字段，通过 cacheControl 翻译为 wire 格式的 cache_control: { type: "ephemeral", ttl: "5m"|"1h" }。同时管理 Anthropic 的 4 断点上限——ANTHROPIC_BREAKPOINT_CAP = 4，超过时按 tools → system → messages 的失效顺序丢弃（tools 层级最高优先保留），并 log warning。
这样 applyCachePolicy（protocol 无关）+ cacheControl（protocol 特定）形成两层：第一层在编译边界注入语义 hints，第二层在 lowering 时翻译为 wire markers 并管理 provider 限制。
五、executor.ts：类型化错误映射与预输出重试
RequestExecutor 是 LLM 网关的 HTTP 传输层。它把 Effect 的 HttpClient 包裹成类型化的 LLMError 产出器，并实现"仅预输出重试"（pre-output retry）——只在流开始前重试，流开始后不再重试。
5.1 类型化错误体系
executor.ts 不自己定义错误类型，而是消费 schema/errors.ts 中定义的 10 种 LLMErrorReason。每种 Reason 都有 retryable getter 标记是否可重试：
// 160:172:opencode/packages/llm/src/schema/errors.tsexport const LLMErrorReason = Schema.Union([InvalidRequestReason,// 400/404/409/413/422 — retryable: falseNoRouteReason, // 无路由 — retryable: falseAuthenticationReason, // 401/403 — retryable: falseRateLimitReason,// 429 — retryable: trueQuotaExceededReason,// 429 + quota — retryable: falseContentPolicyReason,// content_filter — retryable: falseProviderInternalReason,// 5xx — retryable: trueTransportReason,// 网络层 — retryable: falseInvalidProviderOutputReason, // provider 输出无效 — retryable: falseUnknownProviderReason,// 未知状态 — retryable: false]).pipe(Schema.toTaggedUnion("_tag"))
// 174:192:opencode/packages/llm/src/schema/errors.tsexport class LLMError extends Schema.TaggedErrorClass<LLMError>()("LLM.Error", {module: Schema.String,method: Schema.String,reason: LLMErrorReason,}) {override readonly cause = this.reasonget retryable() {return this.reason.retryable}get retryAfterMs() {return "retryAfterMs" in this.reason ? this.reason.retryAfterMs : undefined}// ...}
LLMError 是 TaggedErrorClass，其 retryable 和 retryAfterMs 袋里到 reason。这让 executor 的重试逻辑可以统一查询 error.retryable 而不需要 switch reason 类型。
5.2 HTTP 状态码到 Reason 的映射
statusReason 是映射的核心——输入 HTTP 状态码 + 响应体，输出对应的 LLMErrorReason：
// 225:275:opencode/packages/llm/src/route/executor.tsconst statusReason = (input: {readonly status: numberreadonly message: stringreadonly retryAfterMs?: number | undefinedreadonly rateLimit?: HttpRateLimitDetails | undefinedreadonly http: HttpContext}) => {const body = input.http.body ?? ""if (/content[-_s]?policy|content_filter|safety/i.test(body)) {return new ContentPolicyReason({ message: input.message, http: input.http })}if (input.status === 401) {return new AuthenticationReason({ message: input.message, kind: "invalid", http: input.http })}if (input.status === 403) {return new AuthenticationReason({ message: input.message, kind: "insufficient-permissions", http: input.http })}if (input.status === 429) {if (/insufficient[-_s]?quota|quota[-_s]?exceeded/i.test(body)) {return new QuotaExceededReason({ message: input.message, http: input.http })}return new RateLimitReason({message: input.message,retryAfterMs: input.retryAfterMs,rateLimit: input.rateLimit,http: input.http,})}if (input.status === 400 ||input.status === 404 ||input.status === 409 ||input.status === 413 ||input.status === 422) {return new InvalidRequestReason({message: input.message,classification: isContextOverflow(body) ? "context-overflow" : undefined,http: input.http,})}if (input.status >= 500 || retryableStatus(input.status)) {return new ProviderInternalReason({message: input.message,status: input.status,retryAfterMs: input.retryAfterMs,http: input.http,})}return new UnknownProviderReason({ message: input.message, status: input.status, http: input.http })}
映射逻辑按优先级排列：
| HTTP 状态码 | Reason | retryable | 特殊逻辑 |
|---|---|---|---|
任何 + body 含 content_policy/safety | ContentPolicyReason | false | body 正则匹配优先于状态码 |
| 401 | AuthenticationReason (kind="invalid") | false | — |
| 403 | AuthenticationReason (kind="insufficient-permissions") | false | — |
429 + body 含 quota_exceeded | QuotaExceededReason | false | 配额耗尽不可重试 |
| 429 (其他) | RateLimitReason | true | 携带 retryAfterMs + rateLimit 详情 |
| 400/404/409/413/422 | InvalidRequestReason | false | 检测 isContextOverflow(body) → classification="context-overflow" |
| 500+ 或 429/503/504/529 | ProviderInternalReason | true | 携带 retryAfterMs |
| 其他 | UnknownProviderReason | false | 兜底 |
两个值得注意的设计：
- ContentPolicy 优先于状态码：即使状态码是 400，如果 body 含
content_filter/safety，优先归类为ContentPolicyReason。这让上层可以根据"内容政策"做差异化处理（如提示用户修改 prompt 而非重试） - Context overflow 检测：400/413 等错误会检查 body 是否匹配 context overflow 的正则模式（
isContextOverflow）。匹配则给InvalidRequestReason加classification: "context-overflow"，这让上层（如core的 compaction 逻辑）可以精确检测"上下文溢出"而非通用"无效请求"
5.3 context overflow 的模式匹配
provider-error.ts 定义了 22 个正则模式匹配各 provider 的 context overflow 消息：
// 4:28:opencode/packages/llm/src/provider-error.tsconst patterns = [/prompt is too long/i,/input is too long for requested model/i,/exceeds the context window/i,/input token count.*exceeds the maximum/i,/tokens in request more than max tokens allowed/i,// ... 共 22 个模式 .../model_context_window_exceeded/i,]export const isContextOverflow = (message: string) =>patterns.some((pattern) => pattern.test(message)) || /^4(00|13)s*(status code)?s*(no body)/i.test(message)export const isContextOverflowFailure = (failure: unknown) =>failure instanceof LLMError? failure.reason._tag === "InvalidRequest" && failure.reason.classification === "context-overflow": Schema.is(ProviderErrorEvent)(failure) && failure.classification === "context-overflow"
isContextOverflow 覆盖了 Anthropic（prompt is too long）、OpenAI（maximum context length）、Gemini、Bedrock 等各 provider 的不同措辞。isContextOverflowFailure 则用于在流式事件中检测 ProviderErrorEvent——因为 context overflow 也可能在流中途发生（provider 先开始流再发现超长）。
5.4 传输层错误的归一化
HTTP 传输层（DNS/连接/超时）的错误也需要归一化为 LLMError。toHttpError 把 Effect HttpClientError 转为 TransportReason：
// 307:343:opencode/packages/llm/src/route/executor.tsconst toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {const transportError = (input: { /* ... */ }) =>new LLMError({module: "RequestExecutor",method: "execute",reason: new TransportReason({message: input.message,kind: input.kind,url: input.request ? redactUrl(input.request.url) : undefined,http: input.request ? new HttpContext({ request: requestDetails(input.request, redactedNames) }) : undefined,}),})if (Cause.isTimeoutError(error)) {return transportError({ message: error.message, kind: "Timeout" })}if (!HttpClientError.isHttpClientError(error)) {return transportError({ message: "HTTP transport failed" })}const request = "request" in error ? error.request : undefinedif (error.reason._tag === "TransportError") {return transportError({message: error.reason.description ?? "HTTP transport failed",kind: error.reason._tag,request,})}return transportError({message: `HTTP transport failed: ${error.reason._tag}`,kind: error.reason._tag,request,})}
TransportReason 的 retryable 为 false——传输层错误不自动重试（因为可能是 DNS/证书等持久性故障）。但超时（Cause.isTimeoutError）有专门的 kind: "Timeout" 标记，上层可以据此判断是否值得重试。
5.5 预输出重试策略
// 345:364:opencode/packages/llm/src/route/executor.tsconst retryDelay = (error: LLMError, attempt: number) => {if (error.retryAfterMs !== undefined) return Effect.succeed(Math.min(error.retryAfterMs, MAX_DELAY_MS))return Random.nextBetween(Math.min(BASE_DELAY_MS * 2 ** attempt * 0.8, MAX_DELAY_MS),Math.min(BASE_DELAY_MS * 2 ** attempt * 1.2, MAX_DELAY_MS),).pipe(Effect.map((delay) => Math.round(delay)))}const retryStatusFailures = <A, R>(effect: Effect.Effect<A, LLMError, R>,retries = MAX_RETRIES,attempt = 0,): Effect.Effect<A, LLMError, R> =>Effect.catchTag(effect, "LLM.Error", (error): Effect.Effect<A, LLMError, R> => {if (!error.retryable || retries <= 0) return Effect.fail(error)return retryDelay(error, attempt).pipe(Effect.flatMap((delay) => Effect.sleep(delay)),Effect.flatMap(() => retryStatusFailures(effect, retries - 1, attempt + 1)),)})
重试参数：MAX_RETRIES = 2（最多重试 2 次，共 3 次请求）、BASE_DELAY_MS = 500、MAX_DELAY_MS = 10_000。
退避策略：
- 如果错误携带
retryAfterMs（如RateLimitReason从Retry-Afterheader 解析），优先使用它，但封顶MAX_DELAY_MS - 否则用指数退避 + 抖动：
BASE_DELAY_MS * 2^attempt，乘以 0.8~1.2 的随机抖动，封顶MAX_DELAY_MS。抖动避免多个客户端同步重试（thundering herd）
"仅预输出重试"的语义：
// 366:380:opencode/packages/llm/src/route/executor.tsexport const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.effect(Service,Effect.gen(function* () {const http = yield* HttpClient.HttpClientconst executeOnce = (request: HttpClientRequest.HttpClientRequest) =>Effect.gen(function* () {const redactedNames = yield* Headers.CurrentRedactedNamesreturn yield* http.execute(request).pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames)))})return Service.of({execute: (request) => retryStatusFailures(executeOnce(request)),})}),)
retryStatusFailures 包裹的是 executeOnce——即一次完整的 HTTP 请求（发请求 + 读状态码 + 解析 body + 映射为 LLMError 或返回 response）。重试发生在这个层面：如果 executeOnce 失败（返回 LLMError），且 error.retryable 为 true，则延迟后重试整个请求。
关键设计：重试只在"获取 HTTP Response"阶段发生。一旦 executeOnce 成功返回 response（status < 400），retryStatusFailures 就结束了——后续的 transport.frames() 读取响应流不再被重试包裹。这意味着：
- ✅ 请求被 rate limit → 自动重试（因为
executeOnce失败） - ✅ 请求 503 → 自动重试（因为
executeOnce失败） - ❌ 流开始后 provider 中途断开 → 不重试（因为已经过了
executeOnce，进入transport.frames()） - ❌ 流中途的
provider-error事件 → 不重试（因为这是 stream 层面的 LLMEvent，不是 LLMError）
这个设计是正确的——流已经开始后重试会导致重复输出（用户已经看到部分文本），而预输出阶段的重试对用户透明（什么都没输出过）。
5.6 敏感信息脱敏
executor 还负责错误中的敏感信息脱敏，确保 LLMError 的 HttpContext 不会泄漏 API key：
// 48:54:opencode/packages/llm/src/route/executor.tsconst SENSITIVE_NAME_SOURCE ="authorization|api[-_]?key|access[-_]?token|refresh[-_]?token|id[-_]?token|token|secret|credential|signature|x-amz-signature"const SENSITIVE_NAME = new RegExp(SENSITIVE_NAME_SOURCE, "i")const SHORT_QUERY_NAME = /^(key|sig)$/iconst SENSITIVE_BODY_FIELD = new RegExp(`(?:${SENSITIVE_NAME_SOURCE}|key)`, "i")const REDACT_JSON_FIELD = new RegExp(`("(?:${SENSITIVE_BODY_FIELD.source})"s*:s*)"[^"]*"`, "gi")const REDACT_QUERY_FIELD = new RegExp(`((?:${SENSITIVE_BODY_FIELD.source})=)[^&s"]+`, "gi")
脱敏分两层：
- 结构脱敏：用正则匹配 body/query 中所有看起来像敏感字段名（
authorization/token/secret/key等）的 JSON 字段和 query 参数，替换为<redacted> - 字面脱敏：收集请求中实际发送的 secret 值（header 里的 Bearer token、query 里的
?key=...），在响应 body 中搜索替换——防止 provider 把 secret 回显到错误消息中
URL 中的敏感 query 参数也被脱敏（redactUrl），request-id 提取支持 5 种 header 变体（x-request-id/request-id/x-amzn-requestid/x-amz-request-id/x-goog-request-id/cf-ray），兼容各 provider 的 request id 惯例。
5.7 Rate Limit 详情的解析
// 112:148:opencode/packages/llm/src/route/executor.tsconst rateLimitDetails = (headers: Record<string, string>, retryAfter: number | undefined) => {const limit: Record<string, string> = {}const remaining: Record<string, string> = {}const reset: Record<string, string> = {}Object.entries(headers).forEach(([name, value]) => {const openaiLimit = /^x-ratelimit-limit-(.+)$/.exec(name)?.[1]if (openaiLimit) return addRateLimitValue(limit, openaiLimit, value)const openaiRemaining = /^x-ratelimit-remaining-(.+)$/.exec(name)?.[1]if (openaiRemaining) return addRateLimitValue(remaining, openaiRemaining, value)const openaiReset = /^x-ratelimit-reset-(.+)$/.exec(name)?.[1]if (openaiReset) return addRateLimitValue(reset, openaiReset, value)const anthropic = /^anthropic-ratelimit-(.+)-(limit|remaining|reset)$/.exec(name)if (!anthropic) returnif (anthropic[2] === "limit") return addRateLimitValue(limit, anthropic[1], value)if (anthropic[2] === "remaining") return addRateLimitValue(remaining, anthropic[1], value)return addRateLimitValue(reset, anthropic[1], value)})// ...return new HttpRateLimitDetails({ retryAfterMs: retryAfter, limit, remaining, reset })}
rateLimitDetails 解析两种 rate limit header 风格：
- OpenAI 风格：
x-ratelimit-limit-requests、x-ratelimit-remaining-tokens、x-ratelimit-reset-requests - Anthropic 风格：
anthropic-ratelimit-requests-limit、anthropic-ratelimit-requests-remaining、anthropic-ratelimit-requests-reset
两种风格的 limit/remaining/reset 按 dimension（requests/tokens）分组存入 HttpRateLimitDetails，附带 retryAfterMs（从 Retry-After/retry-after-ms header 解析）。这些详情附加到 RateLimitReason 上，让上层可以精确展示"还剩多少 quota、何时重置"。
六、LLM 网关的类型化设计范式
6.1 设计范式总结
OpenCode 的 LLM 网关体现了一个完整的类型化设计范式，可以概括为五层：
┌──────────────────────────────────────────────────────────────┐│第五层：类型化错误（10 种 LLMErrorReason + retryable getter）││schema/errors.ts — 每种错误自带可重试性 + HTTP 上下文 │└───────────────────────────┬──────────────────────────────────┘│ statusReason() 映射┌───────────────────────────▼──────────────────────────────────┐│第四层：传输执行（executor.ts — HTTP + 重试 + 脱敏）││预输出重试（MAX_RETRIES=2）+ 敏感信息脱敏 + rate limit 解析│└───────────────────────────┬──────────────────────────────────┘│ transport.frames() + streamPrepared()┌───────────────────────────▼──────────────────────────────────┐│第三层：编译边界（compile — cache policy + body + schema）││resolveRequestOptions → applyCachePolicy → body.from → ││schema.decode → prepareTransport │└───────────────────────────┬──────────────────────────────────┘│ Route.make() 四轴组合┌───────────────────────────▼──────────────────────────────────┐│第二层：部署组合（Route = Protocol + Endpoint + Auth +││Framing/Transport）— 四轴正交，DeepSeek 只改 Endpoint+Auth│└───────────────────────────┬──────────────────────────────────┘│ Protocol.make()┌───────────────────────────▼──────────────────────────────────┐│第一层：语义契约（Protocol<Body,Frame,Event,State>）││body.schema + body.from + stream.event + stream.step │└──────────────────────────────────────────────────────────────┘
6.2 关键设计不变量
| 不变量 | 位置 | 约束 |
|---|---|---|
| Protocol ≠ 部署 | protocol.ts:20-24 | Protocol 不知道 URL、header、auth scheme——这些是 Route.make 的部署关切 |
| 四轴正交 | client.ts:321-339 | Protocol/Endpoint/Auth/Framing 独立可替换，DeepSeek/TogetherAI 只覆盖 Endpoint+Auth |
| 编译不执行 | client.ts:344-359 | compile 产出 { body, prepared } 但不发送网络请求——prepare() 安全暴露给调试 |
| body schema 是闸门 | client.ts:348-350 | body.from 后立即 schema.decodeUnknown 校验，lowering bug 在 compile 阶段就失败 |
| cache policy 先于 lowering | client.ts:345 | applyCachePolicy 在 body.from 之前注入 hints，让 lowering 自然翻译为 wire markers |
| inline hints 仅 Anthropic/Bedrock | cache-policy.ts:42 | OpenAI/Gemini 隐式缓存，注入 hints 无害但无意义，短路跳过 |
| 错误类型化 | errors.ts:160-192 | 10 种 Reason 各有 retryable getter，executor 统一查询不 switch |
| 预输出重试 | executor.ts:353-364 | 重试只包裹 executeOnce（获取 response），流开始后不重试 |
| 敏感信息脱敏 | executor.ts:48-55 | 结构脱敏 + 字面脱敏双层，防 provider 回显 secret |
| context overflow 检测 | provider-error.ts:4-28 | 22 个正则覆盖各 provider 措辞，classification="context-overflow" |
6.3 与 AI SDK 的对比
OpenCode 的 LLM 网关设计相比 Vercel AI SDK（@ai-sdk/*，core 依赖 18 个 @ai-sdk/* provider）有几个关键差异：
| 维度 | OpenCode @opencode-ai/llm | Vercel AI SDK |
|---|---|---|
| 协议抽象 | Protocol<Body,Frame,Event,State> 四类型参数管道 | 无显式 Protocol 类型，每个 provider 自己实现 |
| 部署正交 | Route.make 四轴正交，DeepSeek 3 行接入 | 每个 @ai-sdk/* 包独立实现，无共享 protocol |
| 请求编译 | compile() 显式编译边界（schema 校验 + transport 预备） | 无编译概念，直接 streamText |
| 缓存策略 | cache-policy.ts 协议无关 auto 注入 | 各 provider 各自处理，无统一策略 |
| 错误类型化 | 10 种 LLMErrorReason + retryable getter | ProviderError 无细分 reason |
| 重试策略 | 预输出重试（只在获取 response 阶段） | 通常无内置重试或各 provider 自行实现 |
OpenCode 的设计更重、更显式，但换来的是：新增 provider 的边际成本极低（只需 Endpoint+Auth patch）、错误处理可类型化决策（上层可以 if (error.reason._tag === "RateLimit") 做不同策略）、缓存策略统一管理（agent 循环自动获得 prompt caching 收益）。
6.4 设计哲学
- Protocol 是"API 语义"的单一来源：一个 protocol 定义了"这个 API 长什么样"的全部知识——body 怎么构建、stream 怎么解析。部署（URL/auth/transport）是正交的、可替换的维度。这让 protocol 成为可复用的语义资产，而非一次性的 provider 实现
- compile 是"类型安全"的编译边界：在 body.from 和 transport 之间插入 schema 校验，让 protocol 实现者的 lowering bug 在编译期（而非运行时 400 错误）就被捕获。compile 不执行的设计让请求预览成为免费的副产品
- cache policy 是"协议无关"的策略层：auto 策略在编译边界注入语义 hints，protocol 的 lowering 自然翻译为 wire markers。两层分离让缓存策略可以统一演进而不动每个 protocol
- 错误是"类型化决策"的输入：10 种 Reason + retryable getter 让上层可以用类型 switch 做不同策略（context-overflow → compaction，rate-limit → 退避重试，authentication → 提示用户）。executor 的预输出重试是这一层的默认消费者
- 正交组合优于继承：DeepSeek 不继承 OpenAIChat，而是
OpenAIChat.protocol + DeepSeek.endpoint + bearer.auth。这避免了继承层次的爆炸（M×N 个组合），用 M+N 个正交维度覆盖所有组合
相关资讯
-
- 元气骑士前传洗点商人在哪 元气骑士前传洗点商人位置说明
- 2026-07-15
-
- 红色沙漠回音之森古代遗迹解密指南
- 2026-07-15
-
- 元气骑士前传诺伦三女神宝箱如何开启 诺伦三女神宝箱开启指南
- 2026-07-15
-
- 古代遗迹红色沙漠哀嚎山丘解密攻略
- 2026-07-15
-
- 德莱克福尔峡谷红色沙漠古代遗迹解密攻略
- 2026-07-15
-
- 元气骑士博士上线时间 元气骑士博士全技能介绍
- 2026-07-15