流（Streams）

Streams 会保留仍在生成的回复（例如模型令牌），避免内容丢失。如果用户刷新页面，或在回复过程中断开连接，重新连接后即可从中断处继续，而无需丢失回复或从头开始。如果 Durable Object 在回复过程中重启，已写入的分块会被保留，回复也可以从那里继续。Think harness会以这种方式流式传输回复。

agents/streams 中的 Streams 会在 Durable Object 的 SQLite 数据库中，将每条流存储为有序的分块日志。读取方可以从任意位置重放分块，然后继续读取新追加的分块，直到流结束。

安装 capability

src/index.jsjs

import { DurableObject } from "cloudflare:workers";
import { streamText } from "ai";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/ai-sdk";
import { Streams, sseResponse } from "agents/streams";

export class ReportAgent extends DurableObject {
	streams = new Streams();
	lifecycle = Lifecycle.install(this).use(this.streams);

	async onRequest(request) {
		const url = new URL(request.url);
		const id = url.searchParams.get("id") ?? "report";

		if (request.method === "POST") {
			const { prompt } = await request.json();
			this.ctx.waitUntil(this.#generate(id, prompt));
			return Response.json({ id }, { status: 202 });
		}

		// Replay what is stored, then follow live chunks until the stream ends.
		return sseResponse(this.streams, id, { request });
	}

	async #generate(id, prompt) {
		const stream = await this.streams.open(id);
		try {
			const ai = createAI({ binding: this.env.AI });
			const result = streamText({
				model: ai("anthropic/claude-opus-5.5"),
				prompt,
			});
			for await (const text of result.textStream) {
				stream.append(text);
			}
			stream.close();
		} catch (error) {
			stream.error(String(error));
		}
	}
}

src/index.tsts

import { DurableObject } from "cloudflare:workers";
import { streamText } from "ai";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/ai-sdk";
import { Streams, sseResponse } from "agents/streams";

export class ReportAgent extends DurableObject<Env> {
	readonly streams = new Streams();
	readonly lifecycle = Lifecycle.install(this).use(this.streams);

	async onRequest(request: Request): Promise<Response> {
		const url = new URL(request.url);
		const id = url.searchParams.get("id") ?? "report";

		if (request.method === "POST") {
			const { prompt } = await request.json<{ prompt: string }>();
			this.ctx.waitUntil(this.#generate(id, prompt));
			return Response.json({ id }, { status: 202 });
		}

		// Replay what is stored, then follow live chunks until the stream ends.
		return sseResponse(this.streams, id, { request });
	}

	async #generate(id: string, prompt: string): Promise<void> {
		const stream = await this.streams.open(id);
		try {
			const ai = createAI({ binding: this.env.AI });
			const result = streamText({
				model: ai("anthropic/claude-opus-5.5"),
				prompt,
			});
			for await (const text of result.textStream) {
				stream.append(text);
			}
			stream.close();
		} catch (error) {
			stream.error(String(error));
		}
	}
}

写入分块

open(id) 会返回一个 StreamWriter：

成员	说明
`append(chunk)`	存储一个分块并唤醒实时读取方。写入是同步的。返回该分块的序列号。
`close(options?)`	将流标记为 `completed`。读取方会在最后一个分块后结束。
`error(reason?, options?)`	将流标记为 `errored` 并存储 `reason`。
`cursor`	目前已存储的分块数，也是下一个分块的序列号。
`streamId`	流的 ID。

分块可以是任意 JSON 值，默认最大为 1 MiB。可通过 new Streams({ maxChunkBytes }) 设置其他限制。

open() 接受 metadata（与流一起存储的 JSON 对象）和 tag。可以安全地多次调用：

再次打开正在运行的流会返回游标位于当前进度的写入器。
打开已结束的流会抛出 StreamClosedError。
关闭或报错一个已结束的流不会产生任何效果。

恢复生产者

由于游标是持久化的，生产者在对象逐出后重启时，可以从上一个生产者停止的位置继续。请从 stream.cursor 开始循环：

async function produce(streams, id, total) {
	const stream = await streams.open(id);
	// Chunks before the cursor were stored by an earlier run.
	for (let i = stream.cursor; i < total; i++) {
		stream.append({ step: i });
	}
	stream.close();
}

import type { Streams } from "agents/streams";

async function produce(streams: Streams, id: string, total: number) {
	const stream = await streams.open(id);
	// Chunks before the cursor were stored by an earlier run.
	for (let i = stream.cursor; i < total; i++) {
		stream.append({ step: i });
	}
	stream.close();
}

读取分块

read(id, options) 会重放已存储的分块，随后继续读取实时追加内容，并在流结束时返回。每项都是带有 seq 和 chunk 值的 StreamChunk：

async function collect(streams, id, from = 0) {
	const chunks = [];
	for await (const { seq, chunk } of streams.read(id, { from })) {
		chunks.push(chunk);
		console.log(`chunk ${seq}`);
	}
	return chunks;
}

import type { Streams } from "agents/streams";

async function collect(streams: Streams, id: string, from = 0) {
	const chunks: unknown[] = [];
	for await (const { seq, chunk } of streams.read(id, { from })) {
		chunks.push(chunk);
		console.log(`chunk ${seq}`);
	}
	return chunks;
}

传入 signal 可停止跟随实时流。

如果每次向下游写入都会产生开销（例如刷新 SSE 或调用 RPC），请分批读取。readBatches() 会生成数组。重放期间，每个数组最多包含 batchSize 个分块，默认为 100。跟随实时流时，每个数组会包含自上一个数组以来追加的所有内容：

async function forward(streams, id, send) {
	for await (const batch of streams.readBatches(id, {
		batchSize: 50,
		onUpToDate: () => console.log("caught up"),
	})) {
		await send(batch);
	}
}

import type { StreamChunk, Streams } from "agents/streams";

async function forward(
	streams: Streams,
	id: string,
	send: (batch: StreamChunk[]) => Promise<void>,
) {
	for await (const batch of streams.readBatches(id, {
		batchSize: 50,
		onUpToDate: () => console.log("caught up"),
	})) {
		await send(batch);
	}
}

读取方首次到达已存储分块的末尾时，会调用一次 onUpToDate。读取方跟随实时流时即处于最新状态，因此可以用它将 UI 从重放模式切换为实时模式。

通过 SSE 提供服务

sseResponse(streams, id, options) 会返回一个 Response，以 Server-Sent Events 形式提供整个流。它会重放已存储的分块，然后跟随实时追加内容：

sequenceDiagram
    accTitle: Reconnect to a durable stream
    accDescr: After receiving chunk 2, a browser disconnects while the producer stores chunks 3 and 4. It reconnects with Last-Event-ID 2, replays the missing chunks, and then receives live output.
    participant B as Browser
    participant S as Streams via SSE
    S-->>B: Chunk 2, id: 2
    Note over B,S: Connection drops
    Note right of S: Producer stores chunks 3 and 4
    B->>S: Reconnect with Last-Event-ID: 2
    S-->>B: Replay chunks 3 and 4
    S-->>B: up-to-date
    Note right of S: Producer appends chunk 5
    S-->>B: Live chunk 5
    Note right of S: Producer closes or errors the stream
    S-->>B: done or error

重放从 Last-Event-ID 之后开始；如果没有有效请求头，则从 ?from= 中的序列号开始。from 选项会覆盖两者，并指定要发送的第一个分块。重放完成后，响应会发送 up-to-date 事件，再跟随实时追加内容。心跳注释可防止空闲代理关闭连接。响应最后会发送 done 事件，或携带已存储原因的 error 事件。

每个分块的序列号就是其 SSE id。重新连接的浏览器 EventSource 会自动发送 Last-Event-ID，因此无需客户端代码即可从下一个分块继续。传入 request 后，响应会读取这些值，并在客户端断开时停止。

选项	说明
`request`	传入请求。提供 `Last-Event-ID`、`?from=` 和中止信号
`from`	要发送的第一个序列号
`signal`	停止跟随流
`batchSize`	重放期间每次读取的分块数
`heartbeatMs`	心跳注释之间的间隔

对于 WebSocket 等其他传输方式，请自行处理 read() 或 readBatches() 的输出。

查找和删除流

方法	说明
`status(id)`	返回流的 `StreamStatus`；如果不存在则返回 `null`
`list(options?)`	按最新优先列出流。可按 `state` 或 `tag` 筛选，并通过 `limit` 限制结果数量
`delete(id)`	删除已结束的流及其分块。没有匹配的流时返回 `false`

StreamStatus 包含流的 state（"streaming"、"completed" 或 "errored"）、cursor、tag、metadata、error 和时间戳。

标签可将属于同一操作的流分组，例如请求 ID。它并非唯一值。当某个操作生成连续的多条流（例如重试的回复）时，请为每条流添加标签，并使用 list({ tag, limit: 1 }) 查找最新的一条。流创建时，其标签即固定。

内容移出后删除流

流通常是临时的。内容保存到其他位置（例如对话消息）后，请删除该流。close() 和 error() 接受两个选项，可在一个 SQLite 事务中完成这些操作：

commit 会执行你自己的同步写入操作。必须避免等待。
discard: true 会删除该流的行。

function finish(stream, sql, text) {
	stream.close({
		commit: () => {
			sql.exec("INSERT INTO replies (text) VALUES (?)", text);
		},
		discard: true,
	});
}

import type { StreamWriter } from "agents/streams";

function finish(stream: StreamWriter, sql: SqlStorage, text: string) {
	stream.close({
		commit: () => {
			sql.exec("INSERT INTO replies (text) VALUES (?)", text);
		},
		discard: true,
	});
}

结果要么是回复已存储且流已删除，要么是 commit 抛出错误、未发生任何变更且流仍在运行。

存储

分块以数据块形式存储。一行会持续保存分块，直到达到 256 KB；之后追加的内容会写入新行。每次追加只写入一行，而数千个分块通常只占几行，因此删除开销很低。

此功能不会自行删除旧流。请使用 delete() 删除，或在流结束时使用 discard。

在 Agent 类上使用

AIChatAgent 和 Think 会将每条生成中的回复存储在 Streams 中，并在回复结束时将其移入对话。若要将 Streams 添加到 Agent，请将其实例作为字段创建，并在构造函数中通过 this.lifecycle.use() 安装。