一个适配器注册表 + 一个可经 waterfall 拦截的流式调用 API,定义 agent loop、会话日志与所有插件共同使用的规范语言: 消息、内容块、流块、失败事实、重试策略、调用配置、归属头。
ctx.llm
13 个 src 文件 · 2726 行
Service 类插件(默认导出 LlmRuntime)+ llm-invariant
适配器实现:dsh-llm-deepseek · dsh-llm-pi-ai
| 项 | 内容 |
|---|---|
| Service | LlmRuntime extends Service,super(ctx, 'llm');无 Config |
| 事件 | llm/stream(waterfall,包住每次流式调用)+ llm/adapters-updated(emit,拓扑变更提交点;监听器失败被包含,只有 INVARIANT 码失败在扇出后重抛) |
| 子路径导出 | . / ./invariant / ./types / ./brand / ./message(message 用于浏览器端最小依赖导入) |
| 关键设计 | loop 构建的请求经 markAgentLoopRequest 打进程内标记且深冻结(突变抛错);监听者只读不改写 |
registerAdapter(providers: string[], adapter: LlmAdapter): AdapterRegistrationHandle registerConfigurableProviders(entries): DirectoryRegistrationHandle registerModelDiscovery(settingsNs, discover): () => void listProviders() / listModels(provider) / listConfigurableProviders() providerRetryPolicy(provider): ResolvedRetryPolicy resolveModelInfo(provider, model, signal?): Promise<LlmResolvedModelInfo> resolveCallConfig(config, signal?): Promise<LlmCallConfig> // 物化适配器默认;不 clamp、不 alias prepareCall(config, signal?): Promise<PreparedLlmCall> // 捕获注册+策略;stream 一次性 stream(options): AsyncIterable<StreamChunk> // waterfall 包裹
| 类型 | 要点 |
|---|---|
StreamChunk | 8 变体联合:block-start / text-delta / reasoning-delta / tool-call-delta / block-end / usage / finish;block index 关联交错 delta;tool arguments 保持原始 JSON 字符串 |
FinishReason | stop / tool-calls / max-tokens / aborted(+failure) / error(+failure);merge-extensible |
LlmFailure | { message, code, status?, providerRetryAfterMs?, requestId? } —— 可序列化失败事实;策略在 policy,不在 error |
TokenUsage | input/output 为不相交计数;缓存输入单独报 cacheRead/cacheWrite |
Message | 单一不可变表示:{ id: MessageId, role, content, source };交付/持久历史/模型请求共用;MessageSourceMap merge-extensible(user | plugin(+ContextFormed) | model | tool) |
ContentBlockMap | text / reasoning / image / tool-call / tool-result;merge-extensible(新核心块必须带 adapter/UI/compaction 支持) |
LlmCallConfig | { provider, model, reasoningEffort?, temperature?, maxTokens?, stop? } —— 会话请求 header 状态,影响缓存复用 |
PreparedLlmCall | { config(深冻结,含已物化默认), retryPolicy, adapterDefaults, stream } —— 重复 dispatch 或 config 不匹配 ⇒ INVALID_PREPARED_CALL |
registerAdapter 经 ctx.effect 生成器:prepareRoutes(index.ts:374)先完整校验(空名/冲突/重复 → 抛错;
providerInfo 元数据 id 必须等于 provider),然后 commitRoutes(index.ts:405)在同一同步节内删旧 route、装新 route、emitAdaptersUpdated —— 无观测间隙。
handle.replace(next) 先完整 prepare 再替换,失败不动现有 routes;replace([]) 合法。
目录(registerConfigurableProviders)同构:commit 先完整校验,replace 是 swap 而非 delete-then-add。
listModels 是 advisory 目录:适配器可接受未列出的 model,消费者不得因未列出而拒绝请求resolveCallConfig:maxTokens 省略且适配器给 defaultMaxTokens → 物化;explicit reasoningEffort 不在能力集 → UNSUPPORTED_REASONING_EFFORT(在 provider I/O 之前拒绝)prepareCall:物化的适配器默认值打标(adapterDefaults)后,下一次提案省略它们 —— 让当前精确路由重新物化自己的默认;HMR 不能把一个适配器的能力结果配另一个适配器的请求normalizeLlmFailure(adapter-failure.ts):非 Error → 包成 HarnessError(UNKNOWN);第三方 code 不进词表(只有真实 instanceof 或可信快照才信)。
llm/stream 不执行重试(单次尝试包装)。策略被捕获在注册上(resolveRetryPolicy:normal 默认 maxRetries 2、retryableCodes 默认
[EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]、backoff 500ms→10s + 0.1 jitter;always 模式重试所有失败直到成功/取消)。
由可选的 dsh-llm-retry 插件在 agent/request-error 执行 —— 因为发出 chunk 后重试没有持久尝试边界。
normalizeApiKey:trim 后可打印 ASCII 判定;错误消息绝不回显密钥,只给 ref;INVALID_CREDENTIAL(已提供但不可用)≠ MISSING_CREDENTIAL(未提供);刻意排除在默认可重试集外attributionHeaders(identity):user-agent: product/version (+url)(RFC 9110);每个 provider HTTP 请求必须带(白标可替换、不可抑制);测试要求证明 wire 请求中确实加了头
agent loop 边喂边记录原始 chunk。push 按 type 分发:已被 block-end 关闭的 index 的迟到 delta 被忽略(防坏适配器撑内存/污染);
usage/finish/replayState 覆盖式记录;未知 chunk 走 assertNever;容忍 delta-only 协议(无 start/end);
blocks() 在 finish.kind === 'max-tokens' 时丢弃无法安全执行的 tool-call 块;未知块类型永不 close 时 blocks() 抛错。
挂 llm/stream { global: true, prepend: true }:block index 非负安全整数;delta 必须落在同类型已开块;block-start 不得重复开同 index;block-end 必须关开着的块且类型匹配;usage 至多一次;finish 时仍有开块 → fail(除非 reason 是 error/aborted);finish 之后任何 chunk → fail;流必须终结于 finish 块。
LlmAdapter(index.ts:180-233),实现唯一必需方法 async * stream(options: GenerateOptions):
options.signal;usage 先于终结 finish;tool arguments 保持原始 JSON 字符串;索引关联交错 delta;失败直接 throw(service 会归一)或 emit error/aborted finishattributionHeaders()dsh-llm-deepseek(fetch + eventsource-parser;空闲看门狗映射 TIMEOUT;abort ⇒ ABORTED;429 ⇒ RATE_LIMIT;400+context ⇒ CONTEXT_WINDOW_EXCEEDED;≥500 ⇒ SERVER);dsh-llm-pi-ai(库后端 + replay 状态)providerInfo(显示名,id 必须等于 route)/ providerRetryPolicy(缺省 normal 默认)/ listModels(advisory)/ resolveModel(context 容量、defaultMaxTokens、reasoning efforts;能力字段省略是"未知",显式省略是负能力)ctx.llm.registerAdapter([route1,...], adapter)(all-or-nothing;handle 用于 replace);配置可激活的路由用 registerConfigurableProviders;自定义端点用 registerModelDiscovery(请求是 draft,直接带凭据、不读写 settings/credentials)dsh-llm-retry 在 agent/request-error 执行);失败码用共享分类器并带可验证事实(status/Retry-After/requestId)