跳到主要内容

事件流

任务运行时会流式发送：生成中的回复文字、它执行的步骤、它生成的媒体，以及它的结束。有两种接收方式：

  • 跟踪会话：GET /v1/sessions/{session_id}/events 会发送会话里的每一个任务，只要你保持连接。
  • 在请求上流式接收：发送消息时带上 "stream": true，响应就是这一个任务的帧，以它的结束帧收尾，最长 10 分钟。

两种方式发送的帧完全相同，格式为服务器推送事件（SSE）。

帧​

每一帧有一个 id、一个 event 名称和 JSON 格式的 data，后面跟一个空行：

id: 1760860912345-0:0
event: task.started
data: {"task_id": "exec-5a2c…"}

id: 1760860912400-0:0
event: text.delta
data: {"task_id": "exec-5a2c…", "delta": "Here is a vegetarian version"}

每一帧都带着所属任务的 task_id。

eventdata何时发送
task.startedtask_id任务开始。
text.deltatask_id、delta新增的回复文字。一个任务的所有增量拼起来就是它的最终文字。
task.progresstask_id、category、text正在进行的一个步骤。category 为 search、image、video、music、memory、analysis 或 other；text 是给用户看的简短说明。
media.updatedtask_id、slot、kind、status、attachment、error任务生成的图片、视频、音频或文件改变了 status（见下）。
ui.updatedtask_id、ui仅 a2ui 会话：任务的完整界面数据，每次变化都会发送。详见 A2UI 回复。
suggestionstask_id、items用户接下来可能发送的消息，最多三条。紧挨着结束帧之前发送。
task.completedtask_id、session_id、status、text、ui任务完成。text 是完整回复；在 a2ui 会话里，ui 是它的界面数据。
task.stoppedtask_id、session_id、status、text、ui任务被停止。text 是到目前为止的回复。
task.failedtask_id、session_id、status、error任务失败。error 带 code 和 message。

每个任务都恰好以 task.completed、task.stopped 或 task.failed 之一结束——这个集合永远不会扩充。其他事件类型可能新增：遇到不认识的事件请忽略。

一帧可能很大：结束帧带着完整回复，在 a2ui 会话里还带着界面数据。

媒体​

media.updated 按 slot 跟踪任务生成的每一张图片、每一段视频或音频、每一个文件：

  • 生成过程中 status 为 running，之后变为最终状态之一：completed、failed 或 canceled。每种状态只发送一次。
  • kind 为 image、video、audio 或 file（尚不知道时为 null）。
  • completed 的媒体带 attachment：type、filename、url、content_type 和 expires_at。url 是会过期的下载链接；会话的消息列表里总有新的链接。
  • failed 的媒体带 error 错误码，例如 insufficient_credits 或 generation_failed。
  • 有些媒体比它的任务耗时更长：media.updated 帧可能在任务的结束帧之后才到。继续跟踪会话就能收到。

任务失败​

task.failed 用 error.code 说明原因，取值来自一个可能扩充的短列表：insufficient_credits（项目余额耗尽）、timeout、input（请求或其媒体无法使用）、upstream 或 network（任务依赖的服务出错：请稍后重试），以及 provision、config、system 或 unknown（我们这边出了问题）。error.message 是给用户看的一句话。

续传：帧 id 与 Last-Event-ID​

每一帧都有自己的 id——一个不透明的字符串：保存并原样传回，不要解析它。打开事件流时在 Last-Event-ID 请求头里传回其中一个，事件流就会从那一帧之后继续——不丢失，也不重复：

Last-Event-ID事件流发送
某一帧的 id那一帧之后的一切。断线重连时使用。
某个任务的 last_event_id这个任务，从它的第一帧开始。跟踪刚开启的任务时使用。
0整个会话，从头开始。
不带，或会话已不再保存的 id只发送从现在起发生的内容。之前的回复在消息列表里。

跟踪你开启的任务​

POST /v1/sessions（带第一条消息时）和 POST /v1/sessions/{session_id}/messages 都会返回 last_event_id：新任务在事件流上的起点。用它打开事件流，就会从第一帧开始收到这个任务——即使任务在你连上之前就已经开始输出。

curl -N "$NEWEYES_API_BASE/v1/sessions/$SESSION_ID/events" \
  -H "Authorization: Bearer $NEWEYES_USER_TOKEN" \
  -H "Last-Event-ID: $LAST_EVENT_ID"

重连​

一条连接最长持续 10 分钟，网络也会中断。记住你处理过的最后一帧的 id；连接结束后，用它作为 Last-Event-ID 重新打开事件流。

心跳​

没有内容可发时，事件流每 15 秒写一行注释：

: ping

如果 45 秒内什么都没有收到——没有帧，也没有 ping——就当作连接已断开，用最后一帧的 id 重连。

在请求上流式接收​

在消息里加上 "stream": true，响应就是 200，内容是这个任务自己的帧，从第一帧到结束帧，然后响应结束：

curl -N "$NEWEYES_API_BASE/v1/sessions/$SESSION_ID/messages" -X POST \
  -H "Authorization: Bearer $NEWEYES_USER_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"text": "Make it vegetarian.", "stream": true}'
  • 被拒绝的消息仍然在任何帧之前，以错误状态码和 JSON 错误返回。
  • 响应里只有这个任务的帧。结束帧之后才到的 media.updated 不在其中：请跟踪会话的事件流来接收。
  • 和所有事件流一样，这个响应最长持续 10 分钟，而任务可能更久，比如生成视频。
  • 响应没收到结束帧就结束了，任务也不会停——不管是连接断了还是 10 分钟到了。用你收到的最后一帧的 id 作为 Last-Event-ID，在会话事件流上继续接收；或者用同一个 Idempotency-Key 重试请求，从第一帧重新接收同一个任务。
  • 这个响应计入该用户的事件流数量。

在代码里读取事件流​

浏览器里的 EventSource 无法发送 Authorization 请求头，所以请用 fetch（或任何能流式读取响应体的 HTTP 客户端）读取。

const END = ['task.completed', 'task.failed', 'task.stopped'];
const SILENCE_MS = 45_000; // three missed pings

// Follows one task to its end, reconnecting with the last frame's id when the connection drops
// or goes silent.
async function followTask(api, token, sessionId, taskId, startAt, onFrame) {
  let lastId = startAt;
  for (;;) {
    const silence = new AbortController();
    let timer = setTimeout(() => silence.abort(), SILENCE_MS);
    try {
      const response = await fetch(`${api}/v1/sessions/${sessionId}/events`, {
        headers: { Authorization: `Bearer ${token}`, 'Last-Event-ID': lastId },
        signal: silence.signal,
      });
      if (!response.ok) throw new Error((await response.json()).error.code);
      const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
      let buffer = '';
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break; // the connection ended: reconnect from lastId
        clearTimeout(timer);
        timer = setTimeout(() => silence.abort(), SILENCE_MS);
        buffer += value;
        let end;
        while ((end = buffer.indexOf('\n\n')) !== -1) {
          const frame = parse(buffer.slice(0, end));
          buffer = buffer.slice(end + 2);
          if (!frame) continue; // a ping
          lastId = frame.id;
          if (frame.data.task_id !== taskId) continue;
          onFrame(frame);
          if (END.includes(frame.event)) {
            await reader.cancel();
            return frame;
          }
        }
      }
    } catch (error) {
      if (error.name !== 'AbortError' && error.name !== 'TypeError') throw error;
      // silent too long (AbortError) or the network dropped (TypeError): reconnect from lastId
    } finally {
      clearTimeout(timer);
    }
  }
}

function parse(block) {
  const fields = {};
  for (const line of block.split('\n')) {
    if (!line || line.startsWith(':')) continue;
    const at = line.indexOf(':');
    fields[line.slice(0, at)] = line.slice(at + 1).replace(/^ /, '');
  }
  return fields.event ? { id: fields.id, event: fields.event, data: JSON.parse(fields.data) } : null;
}

限制​

每个终端用户同时最多保持 16 条事件流，每个会话最多 4 条。再打开一条时，会结束最早的一条——先结束该会话最早的一条，否则结束该终端用户最早的一条，被结束的事件流随即关闭。只有该终端用户的所有事件流都是最近 10 秒内打开的，新的事件流才会被拒绝，返回 429 rate_limited。