第 22 章 · 实战三：全栈流式聊天应用

本章目标：
打通「服务端 streamText + UI Message Stream + 前端 useChat」的完整链路
掌握 toUIMessageStream / createUIMessageStreamResponse 的服务端用法
用 DefaultChatTransport 连接自定义后端接口
实现消息持久化与历史会话恢复的关键要点

22.1 目标与技术栈

我们要交付一个可日常使用的流式聊天应用：

层	技术	职责
前端	React + `useChat`（`@ai-sdk/react`）	消息渲染、发送、流式更新
传输	`UIMessage Stream Protocol`（SSE）	结构化消息流
服务端	Node.js + `streamText`	模型调用与流转码
存储	任意 KV/DB（示例用内存 Map）	会话持久化

核心数据结构是 UIMessage：带 parts 数组的结构化消息（文本、工具调用等都是 part），前后端统一以它交换数据。

22.2 服务端：streamText 与 UI Message Stream

先用 Node.js 原生 HTTP 服务实现聊天接口（Next.js 项目则把同样的逻辑放进 route handler）：

import {
  streamText,
  toUIMessageStream,
  convertToModelMessages,
  createUIMessageStreamResponse,
  type UIMessage,
} from 'ai';
import { getLanguageModel } from './provider'; // 内部兼容 AI Gateway 与自定义 OpenAI 兼容 Provider，见第 20 章
import { createServer } from 'node:http';

export async function handleChat(req: Request): Promise<Response> {
  // 1. 前端发来的是 UIMessage[]（含 parts 的结构化消息）
  const { messages }: { messages: UIMessage[] } = await req.json();

  // 2. UIMessage → ModelMessage（剥离 UI 专属字段）
  const modelMessages = await convertToModelMessages(messages);

  const result = streamText({
    model: getLanguageModel(),
    instructions: '你是一个乐于助人的中文助手。',
    messages: modelMessages,
  });

  // 3. 把模型输出流转码为 UI Message Stream 并包装为 Response
  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

createServer(async (req, res) => {
  if (req.method === 'POST' && req.url === '/api/chat') {
    const response = await handleChat(
      new Request('http://localhost/api/chat', {
        method: 'POST',
        body: JSON.stringify({
          messages: extractMessages(await readBody(req)),
        }),
      }),
    );
    // 将 Web Response 写回 Node 响应
    res.writeHead(response.status, Object.fromEntries(response.headers));
    res.end(Buffer.from(await response.arrayBuffer()));
    return;
  }
  res.statusCode = 404;
  res.end();
}).listen(3000, () => console.log('listening on http://localhost:3000'));

三个关键转换：

入站：convertToModelMessages(messages) 把 UIMessage[] 转成模型能理解的 ModelMessage[]；
生成：streamText 返回的原始 text-delta 流对前端不友好；
出站：toUIMessageStream({ stream }) 把它转码为 UIMessage 流（SSE 格式），createUIMessageStreamResponse 补上正确的响应头。

22.3 前端：useChat 与 DefaultChatTransport

tsx

'use client';

import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport } from 'ai';
import { useState } from 'react';

export default function Chat() {
  const [input, setInput] = useState('');

  const { messages, sendMessage, status, error, stop } = useChat({
    transport: new DefaultChatTransport({
      api: '/api/chat', // 你的后端聊天接口
    }),
  });

  const handleSubmit = (e: React.FormEvent) => {
    e.preventDefault();
    sendMessage({ text: input });
    setInput('');
  };

  return (
    <div className="chat">
      {messages.map((message) => (
        <div key={message.id} className={`msg ${message.role}`}>
          {message.parts.map((part, i) =>
            part.type === 'text' ? <span key={i}>{part.text}</span> : null,
          )}
        </div>
      ))}

      {status === 'submitted' && <p>思考中…</p>}
      {error && <p role="alert">出错了：{error.message}</p>}

      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="输入消息…"
          disabled={status !== 'ready'}
        />
        <button type="submit">发送</button>
        {status === 'streaming' && <button onClick={stop}>停止</button>}
      </form>
    </div>
  );
}

要点：

渲染时遍历的是 message.parts，只处理 type === 'text' 的部分——后续接入工具调用、文件等只需增加分支；
status 状态机：ready → submitted → streaming → ready，据此控制按钮禁用与加载提示。

22.4 消息持久化与会话恢复

持久化的原则：存储 UIMessage[]，而不是 ModelMessage——前者包含完整的 UI 信息（parts、元数据），可直接用于恢复会话。

// 示例用内存 Map；生产环境换成 Redis / Postgres 等
const chatStore = new Map<string, UIMessage[]>();

export async function saveChat(chatId: string, messages: UIMessage[]) {
  chatStore.set(chatId, messages);
}

export async function loadChat(chatId: string): Promise<UIMessage[]> {
  return chatStore.get(chatId) ?? [];
}

在服务端接口中接入存取：

export async function handleChatWithPersistence(
  chatId: string,
  messages: UIMessage[],
): Promise<Response> {
  // 先落库用户刚发来的完整消息列表
  await saveChat(chatId, messages);

  const result = streamText({
    model: getLanguageModel(),
    instructions: '你是一个乐于助人的中文助手。',
    messages: await convertToModelMessages(messages),
    onFinish: async ({ response }) => {
      // 生成结束后把助手回复追加进会话
      const history = await loadChat(chatId);
      await saveChat(chatId, [...history, ...response.messages]);
    },
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

恢复历史会话时，把存储的 UIMessage[] 直接传给前端的初始状态即可，无需任何格式转换。

22.5 运行与扩展方向

bash

npx tsx server.ts   # 启动 http://localhost:3000

验证流式效果：打开浏览器 Network 面板观察 /api/chat 响应——应能看到 SSE 分块逐段到达，而非一次性返回。

推荐扩展路线：

接工具调用：把第 21 章的客服代理挂到本应用后端，前端增加 tool part 渲染分支；
接 RAG：回答前先走第 20 章的检索流水线，注入知识库上下文；
断线续传：利用 AI SDK 的 resume streams 能力在网络中断后继续接收未完成的流。

本章小结

全栈链路：前端 useChat ⇄ UIMessage Stream Protocol（SSE）⇄ 服务端 streamText；
服务端两个转码函数是关键：入站 convertToModelMessages、出站 toUIMessageStream + createUIMessageStreamResponse；
前端渲染基于 message.parts，天然支持文本之外的工具调用等多模态内容扩展；
持久化存 UIMessage[] 原始格式，恢复零转换成本；
status 状态机驱动 UI 反馈，onFinish 回调是落库助手回复的正确时机。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 客户端发到服务端的 messages 是什么类型？

2. toUIMessageStream 在链路中的作用是？

3. 持久化聊天记录时应存储哪种格式？

4. 关于 useChat 的 status 状态机，正确的顺序是？

🛠️ 动手实践

为聊天界面添加 Markdown 渲染：将 assistant 消息的 text part 用 markdown-it 渲染后再展示。
实现「多会话切换」：左侧列出历史会话（来自 store），点击后把对应 UIMessage[] 设为 useChat 的初始消息。
在 onFinish 中记录本次调用的 usage（token 用量），并按会话累加展示成本统计。

第 22 章 · 实战三：全栈流式聊天应用 ​

22.1 目标与技术栈 ​

22.2 服务端：streamText 与 UI Message Stream ​

22.3 前端：useChat 与 DefaultChatTransport ​

22.4 消息持久化与会话恢复 ​

22.5 运行与扩展方向 ​

本章小结 ​