让 Agent 会「开口说话」：AGUI 协议 + Data Stream Protocol 全链路实战

当 Agent 只会吐一坨完整文本时，它是个"灰盒子"：工具调没调、在干什么、结果长啥样，用户全看不见。 AGUI（Agent-User Interaction，Agent 与用户的交互协议）把这件事拆成一串结构化事件——文本分块流、工具调用的生命周期事件；Data Stream Protocol（DSP）再把这些事件通过 SSE 传到前端，前端就能把文本流式渲染、把工具调用画成组件。

本文用一个 NestJS（后端）+ React/Vite（前端）的真实项目，从协议到代码，一步步拆解这条链路。

为什么要学：从「打字机文本」到「看得见的 Agent」

先看三个常见痛点：

1. 纯文本流体验差：很多实现是 fetch 一段完整字符串再塞进 innerHTML，等半天才出现一大坨；好一点的用 SSE 逐字拼字符串，但遇到 Markdown 表格、代码块、流程图，还得自己另写解析器。
2. 工具调用是黑盒：Agent 后台明明调了搜索引擎、发了邮件，用户却只看到一句"让我查一下…"然后干等，不知道进展。
3. 消息结构不统一：文本、工具调用、错误混在一起，前端要么写死分支，要么无从下手。

AGUI 解决的是"Agent 应该如何把 UI 相关的事件告诉前端"，Data Stream Protocol 解决的是"这些事件在网络上怎么传、前端怎么解析"。这两层打通，上面的痛点就都消失了。

下面按"先概念、后代码、由浅入深"的顺序讲。

---

一、AGUI 协议：Agent 与界面之间的「通用语言」

一句话

AGUI 是 CopilotKit 发起的一个开放、事件驱动的协议（仓库：github.com/ag-ui-protocol/ag-ui），规定 Agent 后端和前端应用之间交互事件的标准格式。

它解决什么

没有协议时，每个 Agent 后端自定义一套输出格式，前端要为每家写适配。AGUI 把交互抽象成标准化事件流，让任何 Agent 框架 + 任何前端框架都能互通。

核心事件类型

AGUI 的事件围绕两个重点：文本消息和工具调用。

• 文本：TEXT_MESSAGE_START → TEXT_MESSAGE_CONTENT（可多条，即流式片段）→ TEXT_MESSAGE_END
• 工具调用：TOOL_CALL_START → TOOL_CALL_ARGS（参数可能边生成边传）→ TOOL_CALL_END
• 生命周期：RUN_STARTED / RUN_FINISHED / RUN_ERROR
• 状态同步：STATE_SNAPSHOT / STATE_DELTA / MESSAGES_SNAPSHOT

要点：

• AGUI 是事件流而非一次性 JSON，天然适配流式输出。
• 它是传输无关的：SSE、WebSocket、HTTP 二进制都行。
• 它明确区分「文本」和「工具调用」两类事件——这正是我们前端分头渲染的基础。

---

二、Data Stream Protocol：把事件搬上网络的「运输线」

一句话

Data Stream Protocol（DSP）是 Vercel AI SDK 采用的基于 SSE 的流式协议，用一行行 data: {...} 把「消息部件」（parts）推给前端。

它和 AGUI 的关系

AGUI 定义"该发什么事件"，DSP 定义"每个事件在线上长什么样"。本项目落地方式是：后端 LangChain 产出 Agent 流 → @ai-sdk/langchain 转成 DSP 流 → pipeUIMessageStreamToResponse 以 SSE 写回 → 前端 useChat 再按 DSP 解析成 UIMessage。

SSE 线上长什么样

data: {"type":"start"}

data: {"type":"text-start","id":"t1"}

data: {"type":"text-delta","id":"t1","delta":"北京今天"}

data: {"type":"tool-input-start","toolCallId":"c1","toolName":"web_search"}

data: {"type":"tool-input-available","toolCallId":"c1","toolName":"web_search","args":"{\"query\":\"北京天气\"}"}

data: {"type":"tool-output-available","toolCallId":"c1","output":"引用: 1\n标题: …"}

data: {"type":"finish","finishReason":"stop"}

常用 chunk 类型

要点：

• DSP 是**部件化（parts）**的：一条消息由多个 part 组成，每个 part 有自己的类型和状态。
• 文本用 text-delta 推增量，前端逐字累加，天然支持打字机效果。
• 工具参数支持 tool-input-delta，所以参数还没生成完就能先渲染"正在生成"的卡片。

---

三、后端：LangChain createAgent + 一行协议转换

不再手写 Agent Loop

传统写 Agent 要自己维护循环：把历史 + 工具结果拼进 messages、判断该不该再调工具……LangChain.js 1.x 的 createAgent() 把这些封装好了。我们不再手写 agent loop。

// ai.service.ts
import { ChatOpenAI } from '@langchain/openai';
import { createAgent } from 'langchain';
import { toBaseMessages, toUIMessageStream } from '@ai-sdk/langchain';
import type { AIMessageChunk, UIMessage } from 'ai';

export class AiService {
  private readonly agent: ReturnType<typeof createAgent>;

  constructor(model: ChatOpenAI, webSearchTool: any, sendMailTool: any) {
    this.agent = createAgent({
      model,
      tools: [webSearchTool, sendMailTool],
      systemPrompt:
        '你是 AI 助手，需要最新信息时请使用 web_search 搜索后再作答；发邮件用 send_mail。',
    });
  }

  async stream(messages: UIMessage[]) {
    // 1. 前端传来的 UIMessage[] → LangChain 的 BaseMessage[]
    const lcMessages = await toBaseMessages(messages);

    // 2. createAgent 自带 agent loop，按两种模式流式产出
    const lgStream = await this.agent.stream(
      { messages: lcMessages },
      { streamMode: ['messages', 'values'], recursionLimit: 30 },
    );

    // 3. LangChain 流 → AI SDK 的 UI Message 流（DSP）
    return toUIMessageStream(lgStream as AsyncIterable<AIMessageChunk>);
  }
}

streamMode 两个模式的含义：

• messages：逐 token 产出 AIMessageChunk——用来推文本增量。
• values：每个阶段结束产出完整状态——用来拿到工具调用结果、最终答案。

Controller 端用 pipeUIMessageStreamToResponse 把流直接写进 Express 响应：

// ai.controller.ts
import { pipeUIMessageStreamToResponse } from 'ai';

@Post('chat')
async postChat(@Body() body: { messages: UIMessage[] }, @Res() res: Response) {
  const stream = await this.aiService.stream(body.messages);
  pipeUIMessageStreamToResponse({ response: res, stream }); // SSE 输出 DSP 流
}

工具：zod 定义参数

工具用 @langchain/core/tools 的 tool() 定义，参数用 zod 声明——模型会自动生成合法参数。

import { tool } from '@langchain/core/tools';
import { z } from 'zod';

const webSearchTool = tool(
  async ({ query, count }) => {
    // 调 Web Search API，返回格式化文本
    return '引用: 1\n标题: ...\nURL: ...\n摘要: ...';
  },
  {
    name: 'web_search',
    description: '使用 Web Search API 搜索互联网网页',
    schema: z.object({
      query: z.string().min(1).describe('搜索关键词'),
      count: z.number().int().min(1).max(20).optional().describe('返回数量'),
    }),
  },
);

要点：

• createAgent 接管 agent loop，开发者只需给模型 + 工具 + systemPrompt。
• streamMode: ['messages','values'] 同时拿到逐 token 文本和完整状态。
• toUIMessageStream 是关键桥接：LangChain 流 → DSP 流，后端无需手写 SSE 拼装。
• 工具参数用 zod 声明，模型自动生成合法参数，前端也能反推参数结构。

---

四、前端：useChat 解析 SSE，拿到的是 parts 不是字符串

一句话

@ai-sdk/react 的 useChat 负责消费 DSP 流，把每条消息解析成一个带 parts 数组的 UIMessage。我们要做的只是"每个 part 怎么渲染"。

// App.tsx
import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport, type UIMessage } from 'ai';

const transport = useMemo(
  () => new DefaultChatTransport({ api: 'http://localhost:3000/ai/chat' }),
  [],
);

const { messages, sendMessage, status, stop, error, clearError } =
  useChat<UIMessage>({ transport });

const busy = status === 'submitted' || status === 'streaming';

每条消息由 parts 组成

一个 assistant 消息长这样（逻辑上）：

message.parts = [
  { type: 'text',                 text: '我帮你查一下…' },
  { type: 'tool-input-available', toolCallId, toolName: 'web_search', args: {...} },
  { type: 'tool-output-available', toolCallId, output: '引用: 1…' },
  { type: 'text',                 text: '北京今天 20°C…' },
]

渲染的核心就是一个 MessagePart 分发器：

function MessagePart({ part, textStreamActive }) {
  if (part.type === 'text') {
    return <StreamdownText isStreaming={textStreamActive}>{part.text}</StreamdownText>;
  }
  if (isToolUIPart(part)) {
    return <ToolMessagePart part={part} />; // 工具 → 自定义组件
  }
  return null;
}

判断"最后一段文本在流式输出"

因为文本是增量到达的，我们只让最后一段正在流的文本开动画：

const lastAssistant = messages.filter(m => m.role === 'assistant').at(-1);
// 对每个 part：是 text + 是最后一条助手消息 + 是最后一个 text part + busy
const textStreamActive =
  part.type === 'text' &&
  message.role === 'assistant' &&
  message.id === lastAssistant?.id &&
  index === lastTextPartIdx &&
  busy;

要点：

• useChat + DefaultChatTransport 自带停止（stop()）、错误（error）、状态（status）管理。
• 消息是结构化的 parts 数组，文本和工具各自成 part，渲染逻辑天然分离。
• 用 isToolUIPart() 判断工具部件，不用手写正则匹配。

---

五、文本渲染：Streamdown 让 Markdown 流式「长出来」

为什么不用 react-markdown

react-markdown 每次 token 更新都要全量重新解析，流式场景卡顿；而 Streamdown 是专门为 AI 流式渲染设计的 react-markdown 替代品，能容忍未闭合的 Markdown（代码块还没写完、表格还在生成），并用 remend 预处理器按块增量解析。

// StreamdownText.tsx
import { createCodePlugin } from '@streamdown/code';
import { mermaid } from '@streamdown/mermaid';
import { Streamdown, type ThemeInput } from 'streamdown';
import 'streamdown/styles.css';

const shikiTheme: [ThemeInput, ThemeInput] = ['github-light', 'github-dark'];
const codePlugin = createCodePlugin({ themes: shikiTheme });

export function StreamdownText({ children, isStreaming = false }) {
  return (
    <Streamdown
      mode="streaming"
      isAnimating={isStreaming}       // 流式时开动画、禁用交互
      parseIncompleteMarkdown         // 容忍未闭合的语法块
      shikiTheme={shikiTheme}
      plugins={{ mermaid, code: codePlugin }}
    >
      {children}
    </Streamdown>
  );
}

要点：

• parseIncompleteMarkdown：代码块 / 表格没写完也能正常渲染，不闪屏。
• mode="streaming" + isAnimating：流式期间的动画与静态渲染分离。
• plugins：mermaid 流程图、Shiki 高亮代码都内建，不用自己写。
• 表格、任务列表、数学公式、CJK 标点都能处理，对 AI 输出友好。

---

六、Tool Call 渲染：自定义组件 + 流式状态机

工具部件的生命周期

工具 part 有四个状态，前端据此渲染"进行中"或"已完成"的卡片：

function ToolMessagePart({ part }) {
  const name = getToolName(part);

  if (part.state === 'output-error') {
    return <ToolErrorPanel name={name} message={part.errorText} />;
  }
  if (part.state !== 'output-available') {
    // 还没执行完：参数流式生成中，或等待执行
    if (name === 'send_mail') return <SendMailToolPanel progress="input-streaming" />;
    return <ToolPendingPanel name={name} hint={/* 从部分参数里提取提示 */} />;
  }
  // 执行完成，按工具名分发到各自的"成品组件"
  switch (name) {
    case 'web_search': return <WebSearchToolPanel input={...} output={...} />;
    case 'send_mail':  return <SendMailToolPanel input={...} output={...} />;
    default:           return <DefaultToolOutput value={...} />;
  }
}

实例 1：web_search —— 把结果文本解析成「卡片列表」

后端把搜索结果拼成了多段文本（引用 / 标题 / URL / 摘要），前端按块解析，渲染成可点击的结果卡片：

function WebSearchToolPanel({ input, output }) {
  const query = input?.query;
  const items = parseWebSearchBlocks(output); // 多段文本 → 结构化数组
  return (
    <div className="tool-panel">
      <div className="tool-panel__head">联网搜索「{query}」</div>
      {items.map(it => (
        <a key={it.ref} href={it.url}>
          <b>{it.title}</b>
          <p>{it.summary}</p>
          <span>{it.siteName} · {it.publishedAt}</span>
        </a>
      ))}
    </div>
  );
}

实例 2：send_mail —— 参数流式生成时就有进度

发邮件不需要等参数全齐：subject、正文可能是流式生成的。前端在 input-streaming 阶段就渲染一张"正在生成邮件参数"的卡片，收件人 / 主题 / 正文逐字段出现，带闪烁光标。

兜底渲染

没写专属组件的工具，走 DefaultToolOutput——把 JSON 格式化后塞进 <pre>，保证任何工具都有 UI。

要点：

• 按状态机渲染：参数生成中 / 待执行 / 完成 / 出错，各有各的组件。
• 按工具名分发：每个工具有专属"成品组件"，未知工具走兜底 JSON 展示。
• 工具参数支持 tool-input-delta，所以流式生成参数时就能渲染进度卡片——这是 AGUI/DSP 相对传统实现最大的体验优势。

---

七、全链路一图流

一张图总结：这一层栈怎么分工

最后

把 Agent 从"灰盒子"变成"看得见的过程"，本质就是把文本和工具调用拆成结构化事件流：AGUI 负责定义事件，Data Stream Protocol 负责传输，createAgent + @ai-sdk/langchain 负责后端产出，useChat + Streamdown + 自定义组件负责前端消费。整条链路没有手写 agent loop，也没有手写 SSE 解析器——协议和 SDK 把最脏的活都包了。

下一步建议：给 send_mail 之外的更多工具写专属组件；把工具结果里的"引用列表"升级成可交互的侧栏；再给 Streamdown 接入 @streamdown/math 渲染公式。协议层不动，前端就能越做越丰富。