写一个新的适配器

内置协议不适用时，从 @volcengine/chatkit/core 引入 createAdapter，提供协议转换与能力函数。最小实现只有 protocol.encodeInput、protocol.decodeEvent 和 run；业务方无需实现 start、底层 run，也无需保存运行时的写入入口。

如果只是地址、鉴权、字段位置不同，先使用内置适配器与扩展 API。完全不同的消息协议才需要新建适配器。

最小可运行示例

custom-adapter.tsx

import '@volcengine/chatkit/styles.css';

import { Chatkit, useChatRuntime } from '@volcengine/chatkit';
import { type AdapterProtocol, buildAcceptedChatInput, type ChatInput, chatInputText, createAdapter } from '@volcengine/chatkit/core';
import { useMemo } from 'react';

type EchoInput = ChatInput;

const sleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));

type EchoEvent = { type: 'text'; text: string } | { type: 'done' } | { type: 'error' };
const protocol: AdapterProtocol<EchoInput, EchoInput, EchoEvent> = {
  encodeInput: input => input,
  projectInput: (input, { runId }) => [buildAcceptedChatInput(runId, input)],
  decodeEvent: (event, { runId }) =>
    event.type === 'done'
      ? [{ type: 'completed' }]
      : event.type === 'error'
      ? [{ type: 'failed', error: { code: 'ECHO_REFUSED', message: '回显服务拒绝了这个请求' } }]
      : [
          {
            type: 'message',
            message: {
              id: `assistant:${runId}`,
              runId,
              role: 'assistant',
              parts: [{ id: 'text', type: 'text', text: event.text }],
            },
          },
        ],
};
function createEchoAdapter() {
  return createAdapter({
    protocol,
    run: async ({ body }) => ({
      async *[Symbol.asyncIterator]() {
        if (chatInputText(body).trim() === '失败') {
          yield { data: { type: 'error' as const } };
          return;
        }
        const text = `收到：「${chatInputText(body)}」。这段回答通过公共 createAdapter 流式显示。`;
        for (let i = 0; i < text.length; i += 2) {
          await sleep(40);
          yield { eventId: `e${i}`, data: { type: 'text' as const, text: text.slice(0, i + 2) } };
        }
        yield { data: { type: 'done' as const } };
      },
    }),
  });
}

export default function CustomAdapter() {
  const adapter = useMemo(() => createEchoAdapter(), []);
  const runtime = useChatRuntime({ sessionId: 'custom-adapter', adapter });
  if (!runtime) return null;
  return <Chatkit runtime={runtime} settings={{ welcome: '这个后端只会回显', recommendations: ['早上好', '失败'] }} />;
}

公共工厂接入自定义协议新窗口打开

createAdapter 参数

参数	必需	含义
protocol	是	下表的编码、解码函数
run	是	发送一次执行请求，返回协议流或受理响应
subscribe / resume / abort / steer / regenerate / edit / history	否	与内置适配器相同的能力函数
transformInput	否	在默认编码之后修改请求体
transformOutput	否	默认解码之后补充、过滤展示内容
acceptedFileTypes	否	文件选择器 accept 规则，如 image/*、.pdf；缺省不限制。工厂不校验附件内容
onError	否	`(capability, error) => void`，接收调用或转换错误

返回值直接交给 useChatRuntime、useChatSessions 或 headless 的 ChatCoreRuntime。内部的连接、取消、写入入口和按轮次的订阅分发由公共工厂管理。

protocol 的完整契约

函数	输入	返回	何时需要
encodeInput	`input, { sessionId, runId, messages }`	协议请求体 TBody	必填；messages 是本轮之前的上下文
decodeEvent	`event, { sessionId, runId, eventId? }`	`readonly ProtocolUpdate[]`	必填；同步解码一条事件，不发请求
projectInput	`input, { sessionId, runId }`	用户消息数组	可选；没有时由后端回显用户消息
identifyEvent	`event`	`{ runId, sessionId?, startsRun?, endsResponse? }` 或 undefined	配置 subscribe 时必填；可靠地分发交错事件
decodeHistory	`unknown, { sessionId }`	HistoryResult	配置 history 时必填
replayHistory	`unknown, { sessionId }`	`Iterable<{ event, message }>`	可选；把历史回放成与实时同形的事件

decodeEvent 的返回数组按顺序应用：

type	其余字段	含义
message	message: RuntimeMessage	创建或替换完整消息；必须自行带齐仍需保留的片段
delta	messageId, partId, delta	为已存在的文字/思考片段追加内容
phase	phase: RunPhase	后端当前执行阶段
completed	无	后端明确完成
failed	error: RunError	后端明确失败，需 code 与 message
interrupted	无	后端明确中断

没有明确结论不能返回 completed；连接异常也不能伪造 failed。工厂把解码更新关联到当前轮次，Core 会将实时消息的 runId 归一到事实的轮次；历史消息的归属则由 decodeHistory 提供。无法识别的事件可返回空数组，它仍会经过 transformOutput。事件流耗尽后尚无结论时，结果保持未知。

startsRun 只用于后端明确开始的轮次，让本页面没有发消息时也能接收主 Agent 回答。endsResponse 关闭本轮读取器，保留会话监听。identifyEvent 只由订阅分发调用，直接 run 流仍由 decodeEvent 的终态更新结束消费。startsRun/endsResponse 不会替 decodeEvent 生成完成或失败事实。消息、轮次与事件各用自己的稳定 id；不能用队列里的最后一条消息猜归属。

完整协议示例：文本、工具、历史与恢复

my-protocol.ts

import type { RuntimeMessage, RuntimePart } from '@volcengine/chatkit/core';

/**
 * 一个假想的协议，用来演示怎么写适配器。
 * 流是 SSE，每个事件一行 `id:`（事件编号）加一行 `data:`（JSON）。
 */
export type MyEvent = { run?: string } & (
  | { type: 'started'; run: string } // 后端给这一轮起的编号
  | { type: 'text'; text: string } // 一段正文
  | { type: 'tool'; id: string; name: string; args: Record<string, unknown> } // 开始调用工具
  | { type: 'tool_result'; id: string; result: unknown } // 工具有了结果
  | { type: 'done' } // 这一轮完成
  | { type: 'error'; code: string; message: string; retryable?: boolean }
); // 这一轮失败

export interface MyFrame {
  /** 事件编号；后端不支持续接时可以没有。 */
  id?: string;
  event: MyEvent;
}

/** 发起一轮、插话的请求内容。 */
export interface MyRunBody {
  text: string;
}

/** 历史的格式：每一轮的输入、结果和当时发过的事件。 */
export interface MyHistory {
  runs: MyHistoryRun[];
}

export interface MyHistoryRun {
  /** 后端的轮次编号。 */
  id: string;
  status: 'completed' | 'failed' | 'interrupted' | 'running';
  input: string;
  /** 开始时间，毫秒。 */
  created_at: number;
  frames: MyFrame[];
  error?: { code: string; message: string };
}

/** 把收到的文本切成事件；最后不完整的那一段留到下次。 */
export const parseFrames = (buffer: string): { frames: MyFrame[]; rest: string } => {
  const blocks = buffer.split('\n\n');
  const rest = blocks.pop() ?? '';
  const frames: MyFrame[] = [];
  for (const block of blocks) {
    let id: string | undefined;
    let data = '';
    for (const line of block.split('\n')) {
      if (line.startsWith('id:')) id = line.slice(3).trim();
      else if (line.startsWith('data:')) data += line.slice(5).trim();
    }
    if (data) frames.push({ ...(id ? { id } : {}), event: JSON.parse(data) as MyEvent });
  }
  return { frames, rest };
};

/**
 * 一条回答的内容。实时流和历史都用它来拼，所以同一段内容的消息编号、片段编号一定相同，
 * 刷新前后界面不会重复、不会闪。
 */
export class Answer {
  readonly parts: RuntimePart[] = [];

  /** 正文是第几段：工具调用之后的正文换新的一段。 */
  private segment = 0;

  constructor(readonly runId: string) {}

  get messageId(): string {
    return `assistant:${this.runId}`;
  }

  /** 追加一段正文，返回它落在哪个片段上。 */
  appendText(text: string): string {
    const id = `text:${this.segment}`;
    const index = this.parts.findIndex(part => part.id === id);
    const existing = this.parts[index];
    const before = existing?.type === 'text' ? existing.text : '';
    // 交给 Core 的对象会被冻结，不能原地改，只能整个换掉
    const part: RuntimePart = { id, type: 'text', text: before + text };
    if (index < 0) this.parts.push(part);
    else this.parts[index] = part;
    return id;
  }

  /** 工具调用开始、工具有了结果。 */
  applyTool(event: Extract<MyEvent, { type: 'tool' | 'tool_result' }>): void {
    const id = `tool:${event.id}`;
    if (event.type === 'tool') {
      this.parts.push({ id, type: 'tool-call', toolCallId: event.id, toolName: event.name, args: event.args });
      this.segment += 1;
      return;
    }
    const index = this.parts.findIndex(part => part.id === id);
    const part = this.parts[index];
    if (part?.type === 'tool-call') this.parts[index] = { ...part, result: event.result };
  }

  toMessage(): RuntimeMessage {
    return { id: this.messageId, role: 'assistant', runId: this.runId, parts: [...this.parts] };
  }
}

export const toUserMessage = (runId: string, text: string): RuntimeMessage => ({
  id: `user:${runId}`,
  role: 'user',
  runId,
  parts: [{ id: 'text', type: 'text', text }],
});

my-adapter.ts

import {
  type AdapterCapabilityFunctions,
  type AdapterProtocol,
  createAdapter,
  type HistoryResult,
  type HistoryRunSnapshot,
  type ProtocolUpdate,
} from '@volcengine/chatkit/core';

import { Answer, type MyEvent, type MyHistory, type MyRunBody, toUserMessage } from './my-protocol';
export interface MyInput {
  text: string;
}
export type MyAdapterOptions = AdapterCapabilityFunctions<MyRunBody, MyRunBody, undefined, undefined, MyEvent, MyInput>;

/** 编码和解码保留协议自己的规则；公共工厂负责请求、取消和订阅。 */
export function createMyAdapter(options: MyAdapterOptions) {
  const answers = new Map<string, Answer>();
  const keyOf = (sessionId: string, runId: string) => JSON.stringify([sessionId, runId]);
  const apply = (event: MyEvent, answer: Answer): ProtocolUpdate[] => {
    if (event.type === 'done') return [{ type: 'completed' }];
    if (event.type === 'error')
      return [
        {
          type: 'failed',
          error: {
            code: event.code,
            message: event.message,
            ...(event.retryable === undefined ? {} : { retryable: event.retryable }),
          },
        },
      ];
    if (event.type === 'started') return [];
    if (event.type === 'text') answer.appendText(event.text);
    else answer.applyTool(event);
    return [{ type: 'message', message: answer.toMessage() }];
  };
  const protocol: AdapterProtocol<MyInput, MyRunBody, MyEvent> = {
    encodeInput: input => ({ text: input.text }),
    projectInput: (input, { runId }) => [toUserMessage(runId, input.text)],
    identifyEvent: event =>
      event.run
        ? {
            runId: event.run,
            startsRun: event.type === 'started',
            endsResponse: event.type === 'done' || event.type === 'error',
          }
        : undefined,
    decodeEvent(event, { sessionId, runId }) {
      const key = keyOf(sessionId, runId);
      let answer = answers.get(key);
      if (!answer) {
        answer = new Answer(runId);
        answers.set(key, answer);
      }
      const updates = apply(event, answer);
      if (event.type === 'done' || event.type === 'error') answers.delete(key);
      return updates;
    },
    decodeHistory(value, { sessionId }): HistoryResult {
      const history = value as MyHistory;
      if (!Array.isArray(history.runs)) throw new Error('历史缺少 runs');
      const messages = [];
      const runSnapshots: HistoryRunSnapshot[] = [];
      let activeRun: HistoryResult['activeRun'] = null;
      for (const run of history.runs) {
        const answer = new Answer(run.id);
        for (const frame of run.frames) apply(frame.event, answer);
        messages.push(toUserMessage(run.id, run.input));
        if (answer.parts.length) messages.push(answer.toMessage());
        if (run.status === 'running') {
          answers.set(keyOf(sessionId, run.id), answer);
          const eventId = run.frames.at(-1)?.id;
          activeRun = { runId: run.id, ...(eventId ? { eventId } : {}) };
        } else
          runSnapshots.push({
            id: run.id,
            origin: 'remote',
            status: run.status,
            phase: 'idle',
            createdAt: run.created_at,
            workItems: {},
            ...(run.error ? { error: run.error } : {}),
            ...(run.status === 'interrupted' ? { interruptReason: 'protocol' as const } : {}),
          });
      }
      return {
        messages,
        runSnapshots,
        activeRun,
        hasMore: false,
        session: {
          id: sessionId,
          status: activeRun ? 'running' : history.runs.at(-1)?.status === 'failed' ? 'error' : 'idle',
        },
      };
    },
  };
  return createAdapter({ ...options, protocol });
}

use-my-adapter.ts

import { createMyAdapter } from './my-adapter';

const send = (method: string, path: string, body?: unknown) =>
  fetch(`/api/chat${path}`, {
    method,
    headers: { 'Content-Type': 'application/json' },
    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
  });

// 用法和内置适配器一样：每个能力一个函数，地址和凭据都在函数里
export const adapter = createMyAdapter({
  run: ({ sessionId, runId, body }) => send('POST', `/sessions/${sessionId}/runs`, { runId, ...body }),
  abort: ({ sessionId, runId }) => send('POST', `/sessions/${sessionId}/runs/${runId}/stop`, {}),
  resume: ({ sessionId, runId, eventId }) =>
    send('GET', `/sessions/${sessionId}/runs/${runId}/events?after=${encodeURIComponent(eventId)}`),
  history: ({ sessionId }) => send('GET', `/sessions/${sessionId}/runs`),
});

协议需要拼接工具参数或恢复文字前缀时，按 sessionId + runId 保存转换状态。历史与实时使用同一组消息和片段编号。终态后释放临时解析状态；这些内存状态不是历史存储，刷新后的内容仍来自后端。

HistoryResult 包含 messages、可选 runSnapshots、session、activeRun、cursor 和必需 hasMore。activeRun 为 null 表示已知没有在执行的轮次；缺省表示后端没有提供。状态在另一个接口时，可在 history 能力函数中一起请求并合并响应。详见会话与历史。

验证适配器

my-adapter.conformance.test.ts

import { type ConformanceScenario, runAdapterConformance } from '@volcengine/chatkit/testing';
import { describe, expect, it } from 'vitest';

import { createMyAdapter, type MyAdapterOptions, type MyInput } from './my-adapter';
import type { MyEvent, MyFrame, MyHistoryRun } from './my-protocol';

/**
 * 把自己写的适配器接进一致性套件。套件要后端表现出四种行为，这里用一个假后端按场景回放：
 * text 正常回答、tool 调一次工具、fail 后端失败、slow 一直输出直到被停止。
 */
const script = (kind: ConformanceScenario): MyEvent[] => {
  if (kind === 'fail') return [{ type: 'error', code: 'RATE_LIMITED', message: '请求太多了', retryable: true }];
  if (kind === 'slow') return [{ type: 'text', text: '处理中' }];
  if (kind === 'tool') {
    return [
      { type: 'text', text: '我查一下。' },
      { type: 'tool', id: 'call-1', name: 'lookup', args: { q: 'a' } },
      { type: 'tool_result', id: 'call-1', result: { hits: 1 } },
      { type: 'text', text: '查到 1 条。' },
      { type: 'done' },
    ];
  }
  return [{ type: 'text', text: '这是' }, { type: 'text', text: '回答' }, { type: 'done' }];
};

const encoder = new TextEncoder();
const json = (value: unknown): Response => new Response(JSON.stringify(value), { status: 200 });
const accepted = (): Response => new Response(null, { status: 204 });

/** 事件流。给了 signal 时流一直开着，直到请求被取消。 */
const stream = (frames: MyFrame[], signal?: AbortSignal): Response =>
  new Response(
    new ReadableStream<Uint8Array>({
      start(controller) {
        for (const { id, event } of frames) {
          controller.enqueue(encoder.encode(`id: ${id}\ndata: ${JSON.stringify(event)}\n\n`));
        }
        if (!signal) {
          controller.close();
          return;
        }
        signal.addEventListener('abort', () => controller.error(new DOMException('aborted', 'AbortError')), {
          once: true,
        });
      },
    }),
    { status: 200, headers: { 'Content-Type': 'text/event-stream' } }
  );

/** 假后端：轮次和会话都存在内存里。 */
const createBackend = (kind: ConformanceScenario): MyAdapterOptions => {
  let runs: MyHistoryRun[] = [];

  const run: MyAdapterOptions['run'] = async ({ body, runId }) => {
    const id = runId;
    const frames: MyFrame[] = [{ type: 'started', run: id } as MyEvent, ...script(kind)].map((event, index) => ({
      id: `e${index}`,
      event,
    }));
    runs.push({
      id,
      status: kind === 'fail' ? 'failed' : kind === 'slow' ? 'interrupted' : 'completed',
      input: body.text,
      created_at: Date.now(),
      frames,
      ...(kind === 'fail' ? { error: { code: 'RATE_LIMITED', message: '请求太多了' } } : {}),
    });
    return stream(frames, kind === 'slow' ? new AbortController().signal : undefined);
  };

  /** 重新生成、编辑后重发：被重做的那一轮和它之后的内容作废，再执行新的一轮。 */
  const rerun: MyAdapterOptions['regenerate'] = request => {
    const index = runs.findIndex(item => item.id === request.targetRunId);
    if (index >= 0) runs = runs.slice(0, index);
    return run(request);
  };

  return {
    run,
    regenerate: rerun,
    edit: rerun,
    abort: async () => accepted(),
    history: async () => json({ runs }),
  };
};

runAdapterConformance<MyInput>(
  'my-adapter',
  {
    createAdapter: kind => createMyAdapter(createBackend(kind)),
    makeInput: text => ({ text }),
    // 重新生成、会话接口的用例默认不跑，适配器支持时打开
    capabilities: { regenerate: true },
  },
  { describe, it, expect }
);

用 @volcengine/chatkit/testing 的 runAdapterConformance 跑一致性测试。接真实后端时还要验证断线重放、历史与实时同形、远端失败、停止受理与最终取消，以及订阅事件的身份。文档里的模拟后端不能代替生产后端验收。

解码与扩展的边界

Response 路径读取的是 SSE 的 data JSON，不是任意 JSON 响应、NDJSON 或二进制流。其他传输先转换为 AsyncIterable<StreamEvent<TEvent>>。HTTP event 字段会写入事件对象；解析或 decodeEvent 抛错会结束本次读取。
decodeEvent 与 decodeHistory 同步返回；工厂不提供异步 decoder，也不验证完整后端 schema。会话目录由外部处理，不提供会话 CRUD 解码器。
encodeInput 和 transformInput 位于能力调用包装之外，解码也不统一经过 onError；应同时处理 runtime 的派发、历史操作错误。不能把 onError 当作捕获所有异常的出口。
decodeHistory 先转换整个响应。给了 replayHistory 时，transformOutput 按它回放的事件逐条调用（source: history，带每步 message 的 runId 与 id 作为 messageId），接入方写一套转换就能在刷新后重建业务内容；每步的 message 是应用这条事件之后的那条消息，同一条消息最后一步必须与 decodeHistory 结果相同，不在任何一步里的消息原样保留。没给 replayHistory 时，transformOutput 以原始整份响应（运行时强制视作 TEvent）调用一次，需要兼容两种形状时声明联合类型并在转换内区分。
工厂的恢复流仍传 source: stream；枚举中存在 recovery 不表示工厂一定使用该值。
projectInput 在命令受理后投影；需要即时用户输入展示时应验证实际交互，不能假设此 hook 在 run 返回前执行。
工厂按外层 eventId 为仍保留状态的轮次去重，终态和解绑会清理记录；它不提供跨页面的持久化去重或事件排序。

#写一个新的适配器

#最小可运行示例

#createAdapter 参数

#protocol 的完整契约

#完整协议示例：文本、工具、历史与恢复

#验证适配器