streaming

assistant-ui 流式传输

请始终查阅 assistant-ui.com/llms.txt，获取最新 API。

assistant-stream 是 assistant-ui 聊天运行时底层的传输层。它会将每个后端规范化为一个由 AssistantStreamChunk 值组成的流，为三种传输格式提供编码器和解码器，并在这些格式之上添加可恢复流层。如果你的后端已经支持 Vercel AI SDK，通常无需直接使用此软件包（只需使用 streamText 和 toUIMessageStream）；当你编写自定义端点、需要自行解码流或需要可恢复流时，再使用它。

参考资料

• ./references/data-stream.md —— Data Stream 协议、useDataStreamRuntime 及其传输格式
• ./references/assistant-transport.md —— Assistant Transport SSE 格式，以及采用状态快照的运行时 useAssistantTransportRuntime
• ./references/encoders.md —— 编码器和解码器目录、PlainTextEncoder、UIMessageStreamDecoder、累加器及调试
• ./references/resumable.md —— assistant-stream/resumable：上下文、存储及客户端接线

适用场景

Streaming the model call through the Vercel AI SDK?
├─ Yes → streamText + toUIMessageStream/createUIMessageStreamResponse (or result.toUIMessageStreamResponse())
│        assistant-stream is optional: only needed to decode the response yourself or add resumable streams
└─ No → build the response with assistant-stream
    ├─ Emitting message parts (text, reasoning, tool calls) → Data Stream
    └─ Streaming a full agent state snapshot with custom commands → Assistant Transport

安装

npm install assistant-stream

@assistant-ui/ai-sdk 是当前的 AI SDK 集成软件包（与框架无关）；@assistant-ui/react-ai-sdk 仍会为较早的安装方式重新导出相同的 API，但新代码应从 @assistant-ui/ai-sdk 导入。

构建自定义流式响应

createAssistantStreamResponse 会使用 AssistantStreamController 运行回调，并返回一个编码为 Data Stream 的 Response（有关其他编码器，请参阅 data-stream.md）。

import { createAssistantStreamResponse } from "assistant-stream";

export async function POST(req: Request) {
  return createAssistantStreamResponse(async (controller) => {
    controller.appendText("Hello ");
    controller.appendText("world!");

    controller.appendReasoning("Checking the forecast first.", {
      unstable_summary: "Looking up the weather",
    });

    controller.appendSource({
      type: "source",
      sourceType: "url",
      id: "s1",
      url: "https://example.com/forecast",
      title: "Forecast",
    });

    const tool = controller.addToolCallPart({ toolName: "get_weather" });
    tool.argsText.append('{"city":"NYC"}');
    tool.argsText.close();
    tool.setResponse({ result: { temperature: 22 } });

    controller.close();
  });
}

close() 会关闭所有仍处于打开状态的部分并结束流；回调中未捕获的异常会自动转换为 error 块。

AssistantStreamController

服务端上的每个流，无论最终由哪种编码器包装，都会通过此控制器写入（createAssistantStream、createAssistantStreamController 和 createAssistantStreamResponse 都会向你提供一个）。

方法	签名	说明
appendText	(textDelta: string) => void	首次调用时打开一个文本部分，后续调用时向其中追加内容
appendReasoning	(reasoningDelta: string, options?: { unstable_summary?: string }) => void	传入 options 时始终会打开一个新部分，因此摘要会单独放在一个部分中
appendSource	(part: SourcePart) => void	SourcePart 的结构为 { type: "source", sourceType: "url", id, url, title?, parentId? }
appendFile	(part: FilePart) => void	FilePart 的结构为 { type: "file", data, mimeType, parentId? }
appendData	(part: DataPart) => void	DataPart 的结构为 { type: "data", name, data, parentId? }，这是带名称的应用自定义部分
addTextPart	() => TextStreamController	显式的 { append(text), close() } 写入器，可与其他部分交错写入
addReasoningPart	(options?) => TextStreamController	与 addTextPart 的写入器结构相同
addToolCallPart	(toolName: string) => ToolCallStreamController	会生成一个 toolCallId；如需稳定 ID，请参阅下方的对象重载
addToolCallPart	(init: ToolCallPartInit) => ToolCallStreamController	{ toolCallId?, toolName, argsText?, args?, response? }
enqueue	(chunk: AssistantStreamChunk) => void	原始逃生口；建议优先使用上方的辅助方法
merge	(stream: AssistantStream) => void	将另一个 AssistantStream 的部分拼接到当前流中
withParentId	(parentId: string) => AssistantStreamController	返回一个控制器，其写入内容会附带 parentId（用于嵌套或相关部分）
close	() => void	关闭当前打开的部分，然后关闭流

addToolCallPart 会返回一个 ToolCallStreamController：{ argsText: TextStreamController, setResponse(response), close() }。setResponse 接受 { result, artifact?, isError?, modelContent?, messages? }（即 ToolResponse 返回的结构），会自动关闭该部分，并忽略第二次调用。

流事件和部分类型

无论线格式如何，每个解码器都会生成相同的规范化 AssistantStreamChunk 联合类型（{ path: number[] } & { type, ... }）：

type	额外字段
part-start	part: PartInit（见下文）
part-finish	无
tool-call-args-text-finish	无
text-delta	textDelta: string
annotations	annotations: ReadonlyJSONValue[]
data	data: ReadonlyJSONValue[]
step-start	messageId: string
step-finish	finishReason, usage: { inputTokens, outputTokens }, isContinued: boolean
message-finish	finishReason, usage
result	result, isError: boolean, artifact?, modelContent?, messages?
error	error: string, code?, severity?: "critical" | "warning" | "info"
update-state	operations: AssistantTransportStateOperation[]（见 assistant-transport.md）

PartInit（part-start 的 part 字段）是六种部分类型之一，每种类型都带有可选的 parentId：

type	额外字段
text	无
reasoning	unstable_summary?: string
tool-call	toolCallId: string, toolName: string
source	sourceType: "url", id, url, title?
file	data: string, mimeType: string
data	name: string, data: ReadonlyJSONValue

常见陷阱

appendSource、appendFile 或 appendData 会静默丢弃该部分

• 传入包含 type 字段（"source"、"file" 或 "data"）的完整部分对象；方法名不会替你推断该字段。

工具调用在 UI 中始终无法完成

• addToolCallPart 需要 toolName；除非你自行传入，否则会自动生成 ID。关闭 argsText（或调用 setResponse，它会替你关闭）以完成该部分，否则该部分将一直无法结束。使用 "use generative" 工具包注册渲染，而不是已弃用的 makeAssistantToolUI；请参阅工具。

客户端将两个独立的推理部分合并为一个

• 在 Data Stream 线上格式中，只有设置了 unstable_summary 才会发送推理部分开始帧；普通的 appendReasoning(text) 调用只会作为文本增量传输，解码器没有其他信息可判断一个新部分已开始。连续打开两个没有摘要的推理部分（例如在工具调用前后）时，客户端会将它们还原为一个连续的推理部分。为每个部分设置 unstable_summary（即使内容看起来为空），或通过单独的消息步骤处理工具调用，以保持它们彼此独立。

流未更新 UI

• 根据实际使用的编码器检查 Content-Type：DataStreamEncoder（createAssistantStreamResponse 的默认编码器）会发送 text/plain; charset=utf-8 和 x-vercel-ai-data-stream: v1，而不是 text/event-stream。AssistantTransportEncoder 和 AI SDK 的 UI 消息流会发送 text/event-stream。

解码器抛出“Stream ended abruptly without receiving [DONE] marker”

• AssistantTransportDecoder 和 UIMessageStreamDecoder 需要终止标记 [DONE]；代理、CDN 或中间件缓冲或截断响应正文都会导致问题。DataStreamDecoder 没有此类标记。

createAssistantStreamResponse 始终编码为 Data Stream

• 它将 DataStreamEncoder 硬编码为编码器。若要使用其他线上格式，请手动编码：AssistantStream.toResponse(createAssistantStream(callback), new AssistantTransportEncoder())，或使用 createAssistantStreamController 并自行编码返回的流。

相关技能

• 运行时 -- useLocalRuntime、useExternalStoreRuntime、useAssistantTransportRuntime React Hook 和状态 Hook
• 设置 -- 搭建 AI SDK 路由处理程序和 useChatRuntime
• 工具 -- "use generative" 工具包和工具调用渲染
• 云 -- 使用 assistant-cloud 持久化流式线程和消息