Vercel AI SDK Stream protocal详解

AI-SDK Stream Protocol

AI SDK UI 流式协议（Stream Protocol）摘要

参考文档：AI SDK UI - Stream Protocols

本摘要面向在自定义后端（如 FastAPI / Python）与前端（@ai-sdk/react）之间，通过 SSE 协议实现与 AI SDK UI 兼容的流式通讯。

核心概念

• 两类协议：
  • Text Stream：仅流式输出纯文本。适用于简单补全与聊天文本，不含工具调用/结构化数据。
  • Data Stream（UI Message Stream）：在 SSE 基础上定义了标准化事件类型（start/text/reasoning/data/tool/finish 等），可承载更丰富的结构化数据、工具调用、推理过程与多 step。
• 首选协议：
  • 前端 useChat/useCompletion 默认使用 Data Stream。
  • 若后端自定义实现 Data Stream，必须设置响应头：x-vercel-ai-ui-message-stream: v1。

HTTP 与 SSE 要点

• 响应类型：Content-Type: text/event-stream
• 推荐头：
  • Cache-Control: no-cache
  • Connection: keep-alive
  • x-vercel-ai-ui-message-stream: v1（Data Stream 必需）
• 传输格式：标准 SSE 行，使用 data: <JSON> 加空行分隔；流末尾必须发送 data: [DONE]。

事件类型（Data Stream 使用）

• 消息生命周期：

  • start：开始一条新的消息

    data: {"type":"start","messageId":"..."}

  • finish：当前消息完成

    data: {"type":"finish"}

  • 终止标记：

    data: [DONE]

• Step（后端一次 LLM 调用粒度）：

  • start-step / finish-step

    data: {"type":"start-step"}
    data: {"type":"finish-step"}

• 文本分段（增量拼接）：

  • text-start / text-delta / text-end

    data: {"type":"text-start","id":"msg_..."}
    data: {"type":"text-delta","id":"msg_...","delta":"Hello"}
    data: {"type":"text-end","id":"msg_..."}

• 推理分段（类似文本分段）：

  • reasoning-start / reasoning-delta / reasoning-end

• 引用与文件：

  • source-url、source-document、file

• 工具调用（可选）：

  • tool-input-start / tool-input-delta / tool-input-available
  • tool-output-available

• 自定义数据：

  • data-* 命名空间的任意结构化数据

    data: {"type":"data-weather","data":{"location":"SF","temperature":100}}

• 错误：

  • error

    data: {"type":"error","errorText":"error message"}

常见注意事项

• 始终以 [DONE] 结束流；否则前端无法正确终止流程。
• 为每一段文本匹配 text-start/text-delta/text-end，避免粘连或丢失。
• 多次 LLM 子调用时配对 start-step/finish-step，便于前端区分阶段和 Stitching。
• 错误时仍应发送 error 事件，并以 finish + [DONE] 正常完结流。

完整示例：覆盖全部关键事件

下面给出一个后端（FastAPI/Python）示例，演示如何在一次请求中按顺序输出常见与可选事件：

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


def sse(data: dict | str) -> str:
    if isinstance(data, dict):
        return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
    return f"data: {data}\n\n"  # 用于 [DONE]


@app.get('/api/chat')
def chat_stream():
    def _gen():
        # 1) 消息开始
        yield sse({"type": "start", "messageId": "msg_0001"})

        # 2) Step 1 开始：模型生成主内容
        yield sse({"type": "start-step"})

        # 2.1) 可选：推理过程（reasoning-*）
        yield sse({"type": "reasoning-start", "id": "rsn_1"})
        yield sse({"type": "reasoning-delta", "id": "rsn_1", "delta": "Analyzing user intent..."})
        yield sse({"type": "reasoning-delta", "id": "rsn_1", "delta": "Planning answer structure."})
        yield sse({"type": "reasoning-end", "id": "rsn_1"})

        # 2.2) 文本输出（text-*）
        yield sse({"type": "text-start", "id": "txt_1"})
        yield sse({"type": "text-delta", "id": "txt_1", "delta": "Hello, this is a demo. "})
        yield sse({"type": "text-delta", "id": "txt_1", "delta": "I can stream text, "})
        yield sse({"type": "text-delta", "id": "txt_1", "delta": "reasoning, tools, and sources."})
        yield sse({"type": "text-end", "id": "txt_1"})

        # 2.3) 引用/来源与文件（可选）
        yield sse({"type": "source-url", "sourceId": "https://example.com", "url": "https://example.com"})
        yield sse({
            "type": "source-document",
            "sourceId": "doc_1",
            "mediaType": "file",
            "title": "Whitepaper.pdf"
        })
        yield sse({
            "type": "file",
            "url": "https://example.com/image.png",
            "mediaType": "image/png"
        })

        # 2.4) 自定义数据（data-*）
        yield sse({"type": "data-status", "data": {"stage": "writing", "progress": 70}})

        # 2.5) Step 1 完成
        yield sse({"type": "finish-step"})

        # 3) Step 2 开始：工具调用（可选）
        yield sse({"type": "start-step"})
        yield sse({
            "type": "tool-input-start",
            "toolCallId": "call_1",
            "toolName": "getWeatherInformation"
        })
        yield sse({
            "type": "tool-input-delta",
            "toolCallId": "call_1",
            "inputTextDelta": "San Francisco"
        })
        # 输入完整可用
        yield sse({
            "type": "tool-input-available",
            "toolCallId": "call_1",
            "toolName": "getWeatherInformation",
            "input": {"city": "San Francisco"}
        })

        # 工具输出返回
        yield sse({
            "type": "tool-output-available",
            "toolCallId": "call_1",
            "output": {"city": "San Francisco", "weather": "sunny"}
        })

        # 工具结果回填到文本
        yield sse({"type": "text-start", "id": "txt_2"})
        yield sse({"type": "text-delta", "id": "txt_2", "delta": "Weather: sunny, 23℃."})
        yield sse({"type": "text-end", "id": "txt_2"})

        # 3.1) Step 2 完成
        yield sse({"type": "finish-step"})

        # 4) 错误示例（可选，在流程中任何时刻都可出现）
        # 若出现错误，仍建议最终发送 finish + [DONE]
        # yield sse({"type": "error", "errorText": "Something went wrong"})

        # 5) 消息完成与终止
        yield sse({"type": "finish"})
        yield sse("[DONE]")

    return StreamingResponse(
        _gen(),
        media_type='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'x-vercel-ai-ui-message-stream': 'v1',
        },
    )

前端（@ai-sdk/react）依然可以用默认 useChat 消费上述流；如果需要对 reasoning-*、tool-*、data-*、source-* 做更细 UI，可在客户端中按事件类型进行路由与渲染。