chatkit

发送和接收：单接口、分离接口与 WebSocket

run 表示“发起一轮”；subscribe 表示“建立一个会话的持续接收通道”。订阅不依赖本页面先发送消息，也可以接收其他客户端或后端主动发起的轮次。函数个数由交互方式决定，与 HTTP 或 WebSocket 的名称无关。

run 发起一轮，subscribe 监听会话ChatKit能力函数后端建立接收通道：subscribe({ sessionId })建立监听并缓冲事件监听已就绪用户发送：run({ sessionId, runId, body })发送命令受理响应；回答从订阅进入其他端触发的新轮次：仍通过订阅接收
接收可以独立于本页面的发送发生；停止一轮不关闭会话监听。
后端形态接入方式输出从哪来
POST chat 同时返回流只写 runrun 返回的 Response 或异步事件序列
先 send，再订阅该轮 eventsrun 内先发送，再返回这一轮的流run 返回的流；无法接收本页面未发起的其他轮次
独立会话事件通道run + subscriberun 只返回受理，所有回答统一从 subscribe 接收
共享 WebSocketrun + subscribe，接入方管理共享连接subscribe 返回本会话的异步事件序列
send-and-subscribe.ts
import { createOpenAiAdapter } from '@volcengine/chatkit/adapters/openai';

// 路径为接入方代理的示意约定。仅收到订阅已就绪的响应后才允许 run。
export const adapter = createOpenAiAdapter({
  subscribe: ({ sessionId }) => fetch(`/api/chat/events?${new URLSearchParams({ sessionId })}`, {}),
  run: ({ sessionId, runId, body }) =>
    fetch('/api/chat/send', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ sessionId, runId, ...body }),
    }),
});

生命周期

  1. ChatKit 需要建立会话接收通道时调用 subscribe；已有有效订阅则复用。函数返回时监听必须已经建立，并能缓冲事件。底层可以复用已有 WebSocket，不要求每次新建连接。
  2. 本地发送先等待订阅就绪再调用 run；订阅已缓冲且身份匹配的事件可在受理响应之前到达。OpenAI 工具循环也可能继续调用 run。
  3. 每条事件带可靠 runId；后端开始事件可登记本地没有发起的轮次，交错事件分别进入对应轮次。
  4. 回答结束、用户停止一轮，都不会关闭会话订阅。切换会话保留后台监听。
  5. 释放运行时，SDK 取消 Response reader 或调用迭代器 return()；清理该会话监听，不能顺手关掉其他会话共用的连接。

重新打开页面时，history 提供基线，resume 补收游标之后的输出。subscribe 负责持续接收，不自动保证重放；可靠恢复需要后端提供重放或完整历史。

使用 useChatSessions 或 useChatRuntime 时，Hook 在初始化真实会话的运行时时自动建立订阅。“打开会话”是这套默认管理方式的调用时机，不是 subscribe 只在打开页面时才能使用的限制。只使用 Core 时，runtime.connect() 可显式建立监听，不必先发送；首次 send 也会等待连接就绪。断线后重新连接会重新调用 subscribe；这不代表 SDK 提供无限自动重连。

自定义事件流在迭代器 return() 中移除监听并唤醒 pending next()；具体用法与停止一轮的区别见能力函数 API。

可运行的独立订阅

先点按钮、不发送用户消息，也会看到主 Agent 回答；随后可以在输入框正常发送。

subscription.tsx
import { Chatkit, useChatRuntime } from '@volcengine/chatkit';
import { createOpenAiAdapter, type OpenAiChunk } from '@volcengine/chatkit/adapters/openai';
import type { StreamEvent } from '@volcengine/chatkit/core';
import { useMemo } from 'react';

/** 示例后端的会话事件总线；真实项目换成 SSE/WS 客户端即可。 */
function createDemoConnection() {
  const listeners = new Set<(event: StreamEvent<OpenAiChunk>) => void>();
  let sequence = 0;
  const publish = (data: OpenAiChunk) => {
    const event = { eventId: `event-${++sequence}`, data };
    for (const listener of listeners) listener(event);
  };
  const answer = (runId: string, text: string) => {
    publish({ run_id: runId, type: 'run.started' });
    publish({ run_id: runId, choices: [{ delta: { content: text }, finish_reason: 'stop' }] });
  };
  const adapter = createOpenAiAdapter({
    async subscribe() {
      const queue: StreamEvent<OpenAiChunk>[] = [];
      let wake: (() => void) | undefined;
      const listener = (event: StreamEvent<OpenAiChunk>) => {
        queue.push(event);
        wake?.();
      };
      // 在返回之前登记监听；run 的快速响应也会先缓存在 queue 中。
      listeners.add(listener);
      const close = () => {
        listeners.delete(listener);
        wake?.();
      };
      let closed = false;
      return {
        [Symbol.asyncIterator]() {
          return {
            async next() {
              while (!closed && !queue.length)
                await new Promise<void>(resolve => {
                  wake = resolve;
                });
              return closed
                ? { done: true as const, value: undefined }
                : { done: false as const, value: queue.shift()! };
            },
            async return() {
              closed = true;
              close();
              return { done: true as const, value: undefined };
            },
          };
        },
      };
    },
    async run({ runId, body }) {
      answer(runId, `已收到：${String(body.messages.at(-1)?.content ?? '')}`);
      return new Response(null, { status: 202 });
    },
  });
  return {
    adapter,
    push: () => answer(`remote-${++sequence}`, '这条主 Agent 回答来自会话订阅，没有在本页面发送用户消息。'),
  };
}
export default function Subscription() {
  const connection = useMemo(createDemoConnection, []);
  const runtime = useChatRuntime({ sessionId: 'subscription-demo', adapter: connection.adapter });
  return (
    <div style={{ height: '100%', display: 'flex', flexDirection: 'column' }}>
      <div style={{ padding: 12 }}>
        <button onClick={connection.push}>模拟其他端触发主 Agent 回答</button>
      </div>
      <div style={{ flex: 1, minHeight: 0 }}>
        {runtime && (
          <Chatkit
            runtime={runtime}
            settings={{ welcome: '可以直接发送，也可以先点上方按钮', recommendations: ['你好'] }}
          />
        )}
      </div>
    </div>
  );
}
发送与被动接收共用会话新窗口打开

示例的内存事件总线用于演示，不承担持久化。真实 WS 客户端将监听到的事件放进同样的异步序列，数据仍按所选协议解析。连接重连、鉴权、消息分发由接入方负责。

订阅事件身份

协议轮次身份被动开始证据结束证据
OpenAI 兼容扩展run_idtype: run.started任一 choice 的 finish_reason
ADK 扩展invocationIdtype: run.startedturnComplete / type: run.completed / errorCode
A2AtaskId，任务对象用 idstatus.state 为 submitted / workingfinal 或 completed / failed / canceled 等终态
Hermesrun_idevent: run.startedrun.completed / run.failed / run.cancelled / run.interrupted / done

表中“结束”指关闭本轮读取器，不一定是执行完成：A2A 的 final 与 Hermes 的 done 本身不能替代执行结论。OpenAI 的 finish_reason 还可能要求进入工具循环。

这些独立会话事件约定不代表原始协议天然提供广播。没有身份或开始标记时，在接入方代理中补齐；不能按最近发送的一轮猜测。自定义协议通过 identifyEvent 定义自己的身份规则。只讨论主 Agent 的消息，子任务状态由业务组件自行查询或订阅。

独立订阅恢复时，resume 必须让同一会话通道从 eventId 之后完整、有序地重放，再接入实时输出。不能让重放与实时增量乱序混入同一轮；SDK 不比较不透明游标的大小。历史活动轮次在恢复前收到的尾帧不作为重放替代。

当前接入边界

  • 本地命令的订阅事件必须使用与命令相同的路由 runId。ADK 的 invocationId、A2A 的 taskId/id、Hermes 的 run_id 若另行生成，需要代理统一编号；受理响应中的新编号不会自动绑定到订阅分发器。
  • A2A 的本地发送和被动接收使用同一个事件解码入口：HTTP SSE 通常携带 JSON-RPC 信封；异步序列可直接给 result 内的事件，也可保留已有信封，不会被二次包装。事件带 contextId 时必须与当前 sessionId 相同，错会话事件会被拒绝。
  • HTTP subscribe 由公共 SSE 读取器解析 JSON；event 行会写入对象的 event 字段。它不复用各协议直接 run 的全部解析规则。ADK 的 [DONE]、OpenAI 的无身份结束标记不能用来路由订阅；结束和错误事件也须带轮次身份。
  • 订阅层只按外层 StreamEvent.eventId 去重，HTTP 对应 SSE id。ADK 仅给 data.id 或 Hermes 仅给 data.seq 不会启用这一层去重；必要时同时提供外层编号。
  • OpenAI 被动轮次没有本地发起时的工具循环上下文，不能据此承诺执行浏览器工具。
  • 断线后再次 connect 可以建连接，但空闲会话订阅没有自动重连定时器；应用需要管理重连。已有断线轮次的有限恢复见恢复页面。