Core 3: LLM Adapter Seam

Official counterpart: llm/llm (vocabulary & seam) + llm-deepseek (DeepSeek implementation)

The Seam Concept

The official abstracts a "swappable capability" as a trio:

Role	Responsibility	Official example
Service Definition	Declares the interface and stream vocabulary	`llm/llm` defines the `stream()` contract
Provider	Implements the interface	`llm-deepseek` calls the DeepSeek API
Consumer	Consumes the interface, usually a model-facing tool	The agent loop consumes `stream()`

The payoff: swap the provider, keep the product. DeepSeek, OpenAI, a local vLLM, or a compatible gateway — as long as they implement the same stream() interface, the agent loop doesn't change a single line.

The Provider Seam: Stream Event Vocabulary

/** Provider seam: stream() produces stream events */
export interface LLMProvider {
  stream(messages: ChatMessage[], tools?: unknown[]): AsyncGenerator<StreamEvent>;
}

export type StreamEvent =
  | { type: "chunk"; delta: Record<string, unknown> }
  | { type: "message"; message: ChatMessage };

Only two events:

chunk — streaming delta (content / reasoning_content / tool-call fragments)
message — the complete message (authoritative result, including tool_calls)

complete() is a convenience wrapper: consume the whole stream, return the final message:

export async function complete(
  provider: LLMProvider,
  messages: ChatMessage[],
  tools: unknown[] = [],
): Promise<ChatMessage> {
  let message: ChatMessage = { role: "assistant", content: "" };
  for await (const ev of provider.stream(messages, tools)) {
    if (ev.type === "message") message = ev.message;
  }
  return message;
}

DeepSeekProvider: OpenAI-Compatible Client

DeepSeek's API is OpenAI-compatible: POST {base_url}/chat/completions. So our implementation needs zero SDK dependencies — just Node's built-in fetch plus an SSE parser:

/**
 * LLM 适配器缝隙:DeepSeek(OpenAI 兼容)流式客户端 + 测试用脚本化 Provider。
 *
 * 官方概念:llm 是能力缝隙(seam),由 Service Definition(流词汇)+ Provider(实现)
 * 组成。换 Provider 不换产品:baseURL 指向任意 OpenAI 兼容端点即可。
 */

import type { ChatMessage, ToolCall } from "./session.ts";

const DEFAULT_BASE_URL = "https://api.deepseek.com";
const DEFAULT_MODEL = "deepseek-chat";

export class LLMError extends Error {}

/** Provider 缝隙:stream() 产出流事件 */
export interface LLMProvider {
  stream(messages: ChatMessage[], tools?: unknown[]): AsyncGenerator<StreamEvent>;
}

export type StreamEvent =
  | { type: "chunk"; delta: Record<string, unknown> }
  | { type: "message"; message: ChatMessage };

/** 便捷封装:消费完整流,返回最终消息 */
export async function complete(
  provider: LLMProvider,
  messages: ChatMessage[],
  tools: unknown[] = [],
): Promise<ChatMessage> {
  let message: ChatMessage = { role: "assistant", content: "" };
  for await (const ev of provider.stream(messages, tools)) {
    if (ev.type === "message") message = ev.message;
  }
  return message;
}

export interface DeepSeekOptions {
  baseURL?: string;
  apiKey?: string;
  model?: string;
  timeoutMs?: number;
}

/** DeepSeek 官方适配器:OpenAI 兼容 chat/completions,SSE 流式 */
export class DeepSeekProvider implements LLMProvider {
  readonly baseURL: string;
  readonly apiKey: string;
  readonly model: string;
  private timeoutMs: number;

  constructor(opts: DeepSeekOptions = {}) {
    this.baseURL = (opts.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? DEFAULT_BASE_URL).replace(/\/$/, "");
    this.apiKey = opts.apiKey ?? process.env.DEEPSEEK_API_KEY ?? "";
    this.model = opts.model ?? process.env.DEEPSEEK_MODEL ?? DEFAULT_MODEL;
    this.timeoutMs = opts.timeoutMs ?? 180_000;
    // 注意:key 检查延迟到首次 stream(),让 dsh web 等形态可以无 key 启动
  }

  async *stream(messages: ChatMessage[], tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    if (!this.apiKey) throw new LLMError("缺少 DEEPSEEK_API_KEY(环境变量或 opts.apiKey)");
    const ctrl = new AbortController();
    const timer = setTimeout(() => ctrl.abort(), this.timeoutMs);
    try {
      const resp = await fetch(`${this.baseURL}/chat/completions`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${this.apiKey}`,
        },
        body: JSON.stringify({ model: this.model, messages, tools, stream: true }),
        signal: ctrl.signal,
      });
      if (!resp.ok) {
        throw new LLMError(`DeepSeek API ${resp.status}: ${(await resp.text()).slice(0, 300)}`);
      }
      if (!resp.body) throw new LLMError("响应没有 body");
      yield* parseSse(resp.body, resp.body.getReader(), new TextDecoder());
    } finally {
      clearTimeout(timer);
    }
  }
}

/** 从 SSE 字节流解析 chat/completions 增量,重组为 chunk/message 流事件 */
async function* parseSse(
  _body: ReadableStream<Uint8Array>,
  reader: ReadableStreamDefaultReader<Uint8Array>,
  decoder: TextDecoder,
): AsyncGenerator<StreamEvent> {
  let buffer = "";
  const toolAcc: ToolCall[] = [];
  let contentAcc = "";
  let done = false;

  while (!done) {
    const { done: streamDone, value } = await reader.read();
    if (streamDone) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? ""; // 末尾不完整行留到下一轮

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed.startsWith("data:")) continue;
      const data = trimmed.slice(5).trim();
      if (data === "[DONE]") {
        done = true; // 结束流,但还要产出最终 message 事件
        break;
      }
      let chunk: any;
      try {
        chunk = JSON.parse(data);
      } catch {
        continue;
      }
      const delta = chunk?.choices?.[0]?.delta ?? {};
      if (typeof delta.content === "string") {
        contentAcc += delta.content;
        yield { type: "chunk", delta: { content: delta.content } };
      }
      if (delta.reasoning_content) {
        yield { type: "chunk", delta: { reasoning_content: delta.reasoning_content } };
      }
      for (const tc of delta.tool_calls ?? []) {
        const idx: number = tc.index ?? 0;
        while (toolAcc.length <= idx) {
          toolAcc.push({ id: "", type: "function", function: { name: "", arguments: "" } });
        }
        toolAcc[idx].id += tc.id ?? "";
        toolAcc[idx].function.name += tc.function?.name ?? "";
        toolAcc[idx].function.arguments += tc.function?.arguments ?? "";
      }
    }
  }

  yield { type: "message", message: { role: "assistant", content: contentAcc, ...(toolAcc.length ? { tool_calls: toolAcc } : {}) } };
}

/** 脚本化 Provider:按脚本依次返回消息,测试与演示用,不发网络请求 */
export class ScriptedProvider implements LLMProvider {
  responses: ChatMessage[];
  constructor(responses: ChatMessage[]) {
    this.responses = [...responses];
  }

  async *stream(messages: ChatMessage[], _tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    const message = this.responses.shift();
    if (!message) throw new LLMError("ScriptedProvider 脚本用尽");
    const content = message.content ?? "";
    for (let i = 0; i < content.length; i += 4) {
      yield { type: "chunk", delta: { content: content.slice(i, i + 4) } };
    }
    yield { type: "message", message };
  }
}

Key design choices:

Configurable baseURL — default https://api.deepseek.com, overridable via DEEPSEEK_BASE_URL; point it at any OpenAI-compatible endpoint (local vLLM, gateway, proxy)
Lazy key validation — the constructor doesn't throw; the check happens on the first stream() call, so dsh web can boot without a key
Timeout via AbortController — aborts after 180s of silence

SSE Parsing: the hardest, most valuable 40 lines

The streaming response of chat/completions is SSE (Server-Sent Events), one data: {...} per line:

text

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

The parser must do three things: split lines, reassemble content deltas, and stitch streaming tool calls by index:

/**
 * LLM 适配器缝隙:DeepSeek(OpenAI 兼容)流式客户端 + 测试用脚本化 Provider。
 *
 * 官方概念:llm 是能力缝隙(seam),由 Service Definition(流词汇)+ Provider(实现)
 * 组成。换 Provider 不换产品:baseURL 指向任意 OpenAI 兼容端点即可。
 */

import type { ChatMessage, ToolCall } from "./session.ts";

const DEFAULT_BASE_URL = "https://api.deepseek.com";
const DEFAULT_MODEL = "deepseek-chat";

export class LLMError extends Error {}

/** Provider 缝隙:stream() 产出流事件 */
export interface LLMProvider {
  stream(messages: ChatMessage[], tools?: unknown[]): AsyncGenerator<StreamEvent>;
}

export type StreamEvent =
  | { type: "chunk"; delta: Record<string, unknown> }
  | { type: "message"; message: ChatMessage };

/** 便捷封装:消费完整流,返回最终消息 */
export async function complete(
  provider: LLMProvider,
  messages: ChatMessage[],
  tools: unknown[] = [],
): Promise<ChatMessage> {
  let message: ChatMessage = { role: "assistant", content: "" };
  for await (const ev of provider.stream(messages, tools)) {
    if (ev.type === "message") message = ev.message;
  }
  return message;
}

export interface DeepSeekOptions {
  baseURL?: string;
  apiKey?: string;
  model?: string;
  timeoutMs?: number;
}

/** DeepSeek 官方适配器:OpenAI 兼容 chat/completions,SSE 流式 */
export class DeepSeekProvider implements LLMProvider {
  readonly baseURL: string;
  readonly apiKey: string;
  readonly model: string;
  private timeoutMs: number;

  constructor(opts: DeepSeekOptions = {}) {
    this.baseURL = (opts.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? DEFAULT_BASE_URL).replace(/\/$/, "");
    this.apiKey = opts.apiKey ?? process.env.DEEPSEEK_API_KEY ?? "";
    this.model = opts.model ?? process.env.DEEPSEEK_MODEL ?? DEFAULT_MODEL;
    this.timeoutMs = opts.timeoutMs ?? 180_000;
    // 注意:key 检查延迟到首次 stream(),让 dsh web 等形态可以无 key 启动
  }

  async *stream(messages: ChatMessage[], tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    if (!this.apiKey) throw new LLMError("缺少 DEEPSEEK_API_KEY(环境变量或 opts.apiKey)");
    const ctrl = new AbortController();
    const timer = setTimeout(() => ctrl.abort(), this.timeoutMs);
    try {
      const resp = await fetch(`${this.baseURL}/chat/completions`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${this.apiKey}`,
        },
        body: JSON.stringify({ model: this.model, messages, tools, stream: true }),
        signal: ctrl.signal,
      });
      if (!resp.ok) {
        throw new LLMError(`DeepSeek API ${resp.status}: ${(await resp.text()).slice(0, 300)}`);
      }
      if (!resp.body) throw new LLMError("响应没有 body");
      yield* parseSse(resp.body, resp.body.getReader(), new TextDecoder());
    } finally {
      clearTimeout(timer);
    }
  }
}

/** 从 SSE 字节流解析 chat/completions 增量,重组为 chunk/message 流事件 */
async function* parseSse(
  _body: ReadableStream<Uint8Array>,
  reader: ReadableStreamDefaultReader<Uint8Array>,
  decoder: TextDecoder,
): AsyncGenerator<StreamEvent> {
  let buffer = "";
  const toolAcc: ToolCall[] = [];
  let contentAcc = "";
  let done = false;

  while (!done) {
    const { done: streamDone, value } = await reader.read();
    if (streamDone) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? ""; // 末尾不完整行留到下一轮

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed.startsWith("data:")) continue;
      const data = trimmed.slice(5).trim();
      if (data === "[DONE]") {
        done = true; // 结束流,但还要产出最终 message 事件
        break;
      }
      let chunk: any;
      try {
        chunk = JSON.parse(data);
      } catch {
        continue;
      }
      const delta = chunk?.choices?.[0]?.delta ?? {};
      if (typeof delta.content === "string") {
        contentAcc += delta.content;
        yield { type: "chunk", delta: { content: delta.content } };
      }
      if (delta.reasoning_content) {
        yield { type: "chunk", delta: { reasoning_content: delta.reasoning_content } };
      }
      for (const tc of delta.tool_calls ?? []) {
        const idx: number = tc.index ?? 0;
        while (toolAcc.length <= idx) {
          toolAcc.push({ id: "", type: "function", function: { name: "", arguments: "" } });
        }
        toolAcc[idx].id += tc.id ?? "";
        toolAcc[idx].function.name += tc.function?.name ?? "";
        toolAcc[idx].function.arguments += tc.function?.arguments ?? "";
      }
    }
  }

  yield { type: "message", message: { role: "assistant", content: contentAcc, ...(toolAcc.length ? { tool_calls: toolAcc } : {}) } };
}

/** 脚本化 Provider:按脚本依次返回消息,测试与演示用,不发网络请求 */
export class ScriptedProvider implements LLMProvider {
  responses: ChatMessage[];
  constructor(responses: ChatMessage[]) {
    this.responses = [...responses];
  }

  async *stream(messages: ChatMessage[], _tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    const message = this.responses.shift();
    if (!message) throw new LLMError("ScriptedProvider 脚本用尽");
    const content = message.content ?? "";
    for (let i = 0; i < content.length; i += 4) {
      yield { type: "chunk", delta: { content: content.slice(i, i + 4) } };
    }
    yield { type: "message", message };
  }
}

Streaming tool calls are the classic pitfall — the model emits tool_calls fragments across multiple deltas:

text

delta 1: {"index":0,"id":"call_1","function":{"name":"add","arguments":"{\"a\":"}}
delta 2: {"index":0,"function":{"arguments":"1,\"b\":2}"}}
final:   {"id":"call_1","function":{"name":"add","arguments":"{\"a\":1,\"b\":2}"}}

So we maintain accumulator slots indexed by index, appending id/name/arguments piece by piece.

Don't forget: the message event must still be emitted after [DONE]

[DONE] only means the stream is over — the final message event must still be produced, or the agent loop will wait forever. This is a real bug we hit while writing tests.

ScriptedProvider: The Testing Bedrock

How do you develop without a key? A scripted provider returns messages from a pre-written script, with zero network:

export class ScriptedProvider implements LLMProvider {
  responses: ChatMessage[];
  constructor(responses: ChatMessage[]) { this.responses = [...responses]; }

  async *stream(messages: ChatMessage[], _tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    const message = this.responses.shift();
    if (!message) throw new LLMError("ScriptedProvider 脚本用尽");
    const content = message.content ?? "";
    for (let i = 0; i < content.length; i += 4) {
      yield { type: "chunk", delta: { content: content.slice(i, i + 4) } }; // simulate streaming
    }
    yield { type: "message", message };
  }
}

Its value: the agent loop, tool execution, and session log can be tested deterministically in CI with no network and no key.

Verifying the Real HTTP Path

Mock alone isn't enough — we also start a local mock server with node:http and push real SSE bytes through the whole path:

it("streaming text: chunks emitted piece by piece, final message complete", async () => {
  responses = [sse(
    '{"id":"x","choices":[{"delta":{"role":"assistant","content":"你"}}]}',
    '{"id":"x","choices":[{"delta":{"content":"好"}}]}',
    "[DONE]",
  )];
  const provider = new DeepSeekProvider({ baseURL: `http://127.0.0.1:${port}`, apiKey: "test" });
  const chunks: string[] = [];
  let finalContent = "";
  for await (const ev of provider.stream([])) {
    if (ev.type === "chunk") chunks.push(String(ev.delta.content));
    else finalContent = ev.message.content;
  }
  expect(chunks).toEqual(["你", "好"]);
  expect(finalContent).toBe("你好");
});

Hooking up a real API

bash

export DEEPSEEK_API_KEY=sk-xxx
pnpm run run "write a bubble sort for me"

base_url can point at any OpenAI-compatible endpoint — that's the power of the seam.

Full Source

/**
 * LLM 适配器缝隙:DeepSeek(OpenAI 兼容)流式客户端 + 测试用脚本化 Provider。
 *
 * 官方概念:llm 是能力缝隙(seam),由 Service Definition(流词汇)+ Provider(实现)
 * 组成。换 Provider 不换产品:baseURL 指向任意 OpenAI 兼容端点即可。
 */

import type { ChatMessage, ToolCall } from "./session.ts";

const DEFAULT_BASE_URL = "https://api.deepseek.com";
const DEFAULT_MODEL = "deepseek-chat";

export class LLMError extends Error {}

/** Provider 缝隙:stream() 产出流事件 */
export interface LLMProvider {
  stream(messages: ChatMessage[], tools?: unknown[]): AsyncGenerator<StreamEvent>;
}

export type StreamEvent =
  | { type: "chunk"; delta: Record<string, unknown> }
  | { type: "message"; message: ChatMessage };

/** 便捷封装:消费完整流,返回最终消息 */
export async function complete(
  provider: LLMProvider,
  messages: ChatMessage[],
  tools: unknown[] = [],
): Promise<ChatMessage> {
  let message: ChatMessage = { role: "assistant", content: "" };
  for await (const ev of provider.stream(messages, tools)) {
    if (ev.type === "message") message = ev.message;
  }
  return message;
}

export interface DeepSeekOptions {
  baseURL?: string;
  apiKey?: string;
  model?: string;
  timeoutMs?: number;
}

/** DeepSeek 官方适配器:OpenAI 兼容 chat/completions,SSE 流式 */
export class DeepSeekProvider implements LLMProvider {
  readonly baseURL: string;
  readonly apiKey: string;
  readonly model: string;
  private timeoutMs: number;

  constructor(opts: DeepSeekOptions = {}) {
    this.baseURL = (opts.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? DEFAULT_BASE_URL).replace(/\/$/, "");
    this.apiKey = opts.apiKey ?? process.env.DEEPSEEK_API_KEY ?? "";
    this.model = opts.model ?? process.env.DEEPSEEK_MODEL ?? DEFAULT_MODEL;
    this.timeoutMs = opts.timeoutMs ?? 180_000;
    // 注意:key 检查延迟到首次 stream(),让 dsh web 等形态可以无 key 启动
  }

  async *stream(messages: ChatMessage[], tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    if (!this.apiKey) throw new LLMError("缺少 DEEPSEEK_API_KEY(环境变量或 opts.apiKey)");
    const ctrl = new AbortController();
    const timer = setTimeout(() => ctrl.abort(), this.timeoutMs);
    try {
      const resp = await fetch(`${this.baseURL}/chat/completions`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${this.apiKey}`,
        },
        body: JSON.stringify({ model: this.model, messages, tools, stream: true }),
        signal: ctrl.signal,
      });
      if (!resp.ok) {
        throw new LLMError(`DeepSeek API ${resp.status}: ${(await resp.text()).slice(0, 300)}`);
      }
      if (!resp.body) throw new LLMError("响应没有 body");
      yield* parseSse(resp.body, resp.body.getReader(), new TextDecoder());
    } finally {
      clearTimeout(timer);
    }
  }
}

/** 从 SSE 字节流解析 chat/completions 增量,重组为 chunk/message 流事件 */
async function* parseSse(
  _body: ReadableStream<Uint8Array>,
  reader: ReadableStreamDefaultReader<Uint8Array>,
  decoder: TextDecoder,
): AsyncGenerator<StreamEvent> {
  let buffer = "";
  const toolAcc: ToolCall[] = [];
  let contentAcc = "";
  let done = false;

  while (!done) {
    const { done: streamDone, value } = await reader.read();
    if (streamDone) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() ?? ""; // 末尾不完整行留到下一轮

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed.startsWith("data:")) continue;
      const data = trimmed.slice(5).trim();
      if (data === "[DONE]") {
        done = true; // 结束流,但还要产出最终 message 事件
        break;
      }
      let chunk: any;
      try {
        chunk = JSON.parse(data);
      } catch {
        continue;
      }
      const delta = chunk?.choices?.[0]?.delta ?? {};
      if (typeof delta.content === "string") {
        contentAcc += delta.content;
        yield { type: "chunk", delta: { content: delta.content } };
      }
      if (delta.reasoning_content) {
        yield { type: "chunk", delta: { reasoning_content: delta.reasoning_content } };
      }
      for (const tc of delta.tool_calls ?? []) {
        const idx: number = tc.index ?? 0;
        while (toolAcc.length <= idx) {
          toolAcc.push({ id: "", type: "function", function: { name: "", arguments: "" } });
        }
        toolAcc[idx].id += tc.id ?? "";
        toolAcc[idx].function.name += tc.function?.name ?? "";
        toolAcc[idx].function.arguments += tc.function?.arguments ?? "";
      }
    }
  }

  yield { type: "message", message: { role: "assistant", content: contentAcc, ...(toolAcc.length ? { tool_calls: toolAcc } : {}) } };
}

/** 脚本化 Provider:按脚本依次返回消息,测试与演示用,不发网络请求 */
export class ScriptedProvider implements LLMProvider {
  responses: ChatMessage[];
  constructor(responses: ChatMessage[]) {
    this.responses = [...responses];
  }

  async *stream(messages: ChatMessage[], _tools: unknown[] = []): AsyncGenerator<StreamEvent> {
    const message = this.responses.shift();
    if (!message) throw new LLMError("ScriptedProvider 脚本用尽");
    const content = message.content ?? "";
    for (let i = 0; i < content.length; i += 4) {
      yield { type: "chunk", delta: { content: content.slice(i, i + 4) } };
    }
    yield { type: "message", message };
  }
}

Recap

The provider seam is one stream() interface; swap implementations without touching the product
SSE parsing does three things: split lines, reassemble content, stitch tool calls by index
The message event must still be emitted after [DONE]
ScriptedProvider + local mock server = full-path testing without a key

Next: Tool System →

Core 3: LLM Adapter Seam ​

The Seam Concept ​

The Provider Seam: Stream Event Vocabulary ​

DeepSeekProvider: OpenAI-Compatible Client ​

SSE Parsing: the hardest, most valuable 40 lines ​

ScriptedProvider: The Testing Bedrock ​

Verifying the Real HTTP Path ​

Full Source ​