AI对话页的流式处理架构：基于Web Streams+Fetch API的实践

Ne0inhk

15 Mar 2026 — 8 min read

引言

当前AI浪潮下，基于各大agent平台，我们可以在几分钟内就搭建出一个具备页面交互的智能体，从问答输出到页面交互，这个过程中的数据流转、UI实现被统一封装以降低模型搭建复杂度。为了探索这个过程的底层实现，我们采用“生产者-消费者模式”的流式处理架构，将网络IO、数据解码、文本解析与UI渲染解耦，实现实时流式响应、UI增量渲染。

使用框架如下：

前端框架：Vue 3 + TypeScript + Vite
UI组件库：Ant Design Vue、Ant Design X Vue
流处理：Web Streams API + Fetch API

从请求发送到UI渲染，流程如下：

流式响应处理

请求管理

采用 AbortController、ReadableStreamDefaultReader 实现“上游网络请求中止”和“下游字节流读取控制”，共同实现一次会话的可取消、可停止的流式处理。
AbortController：管理当前请求的“上游网络中止”句柄，用于在开始新的提问前中止上一轮未完成的请求，或用户点击取消时终止本次请求。
ReadableStreamDefaultReader：管理“下游传输层字节流”的读取器句柄，用来驱动上游生产者向管道入队字节块，以及在用户点击取消时终止字节读取。

// 流控制相关句柄 let abortController: AbortController | null = null; let currentReader: ReadableStreamDefaultReader<Uint8Array> | null = null;

流处理管道

建立 Web Streams 流式解析管线：生产者 → 解码 → 按行拆分 → 消费者。整体处理流程如下所示：

生产者流

负责把上游 reader 的chunk字节块统一按流的背压节奏入队，供下游统一消费，实现“读-推送”连续循环。并将下游或外部触发的取消信号正确传播到上游，终止读取链路。当外部状态指示停止或上游耗尽时，关闭控制器并复位响应状态，保证资源释放与状态一致性。

实现原理可参考文档：https://developer.mozilla.org/zh-CN/docs/Web/API/ReadableStream

 const producerStream = new ReadableStream<Uint8Array>({ start(controller) { function pump() { if (!isResponding.value) { controller.close(); return; } currentReader?.read().then(({ done, value }) => { if (done) { isResponding.value = false; controller.close(); return; } controller.enqueue(value); // 推送字节块 pump(); }); } pump(); }, });

转换流

当上游将生产的字节块入队后，我们构建一条流式处理管道，兼容粘包/半包：先将上游的二进制字节流解码为字符串流，再定义转换流（TransformStream类型）按行拆分并过滤空行，确保下游以“完整且非空的文本行”为单位消费数据。并在流结束的 flush() 钩子中，再次冲刷缓冲区，以处理可能残留的最后一行，避免丢失收尾数据，最后得到的是由一对可读流和可写流组成的TransformStream。

const textStream = producerStream.pipeThrough(new TextDecoderStream() as unknown as TransformStream<Uint8Array, string>); let; const lineSplitter = new TransformStream<string, string>({ transform(chunk, controller) { buffer += chunk; const lines = buffer.split('\n'); buffer = lines.pop() || ''; for (const line of lines) { const trimmed = line.trim(); if (trimmed) controller.enqueue(trimmed); } }, flush(controller) { const trimmed = buffer.trim(); if (trimmed) controller.enqueue(trimmed); }, });

SSE解析流

构建SSE 消费者读取器并驱动 UI 增量渲染，读取时若 done = true，表示“传输层的流”已经真正结束（数据源已关闭，后续不会再有字节到来），这时调用 releaseLock 释放读取器的独占锁；若读取到 message === '[DONE]'，表示应用层的结束，业务上不再需要继续读取，但此时连接/流不一定已被对端关闭，后续仍可能存在空闲或遗留的字节，主动调用 cancel 终止读取，cancel() 方法返回一个 Promise，这个Promise 在流被取消时兑现，消费者在流中调用该方法发出取消流的信号。

const sseStream = textStream.pipeThrough(lineSplitter); // ReadableStream const downstreamReader = sseStream.getReader(); // ReadableStreamDefaultReader function consume() { downstreamReader.read().then(({ done, value }) => { if (done) { isResponding.value = false; downstreamReader.releaseLock(); return; } const message = value.replace(/^data:\s*/, ''); if (message === '[DONE]') { …… } try { const parsed = JSON.parse(message); const content = parsed?.choices?.[0]?.delta?.content; if (content) { streamReply += content; // 消息增量渲染 if (chatList.value.length < index + 1) { const newReply: ChatItem = { key: index, role: 'assistant', content: streamReply, }; chatList.value.push(newReply); } else { chatList.value[index]!.content = streamReply; } } } catch (e) { console.warn('SSE parse error:', e); } consume(); }); } consume();

整体数据流转如下所示：

API通信层

发送请求

以qwen-plus模型为例，其接入方式如下：

async function fetchReply(list: ChatItem[], signal?: AbortSignal) { const messages = list.map(({ role, content }) => ({ role, content })); return fetch('https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions', { method: 'POST', headers: { Authorization: `Bearer ${import.meta.env.VITE_ALIYUN_API_KEY}`, 'Content-Type': 'application/json', }, body: JSON.stringify({ model: 'qwen-plus', messages, stream: true, }), signal, }); }

中止请求

重置 UI 响应状态：将会话状态标记为非响应中，避免界面继续显示“正在响应”等状态提示
终止流式读取：如果存在当前的流读取器，则主动取消读取，防止继续从流中消费数据，并清理引用
终止网络请求：如果存在未完成的流，则触发终止，并清理引用

const handleCancel = () => { isResponding.value = false; // 终止读取 if (currentReader) { try { currentReader.cancel('user canceled'); } catch (e) { console.error(e); } currentReader = null; } // 终止请求 if (abortController) { try { abortController.abort('user cancellation'); } catch (e) { console.error(e); } abortController = null; } message.error('已取消发送'); };

基础UI交互组件封装

UI设计主要基于Ant Design Vue、Ant Design X Vue组件库，其中 Ant Design X Vue 专注于Vue生态的先进AI组件库，旨在简化对话式AI应用的开发，同时支持ts 和tsx。

消息展示

功能：展示对话历史，支持多角色渲染，支持Markdown渲染。
处理 markdown 输出渲染：

import { h } from 'vue'; import { type BubbleProps } from 'ant-design-x-vue'; import { Typography } from 'ant-design-vue'; import markdownit from 'markdown-it'; const md = new markdownit({ html: false, breaks: true, linkify: true, typographer: true }); const renderMarkdown: BubbleProps['messageRender'] = (content) => h(Typography, null, { default: () => h('div', { innerHTML: md.render(content) }), });

3. 角色映射配置：

import { UserOutlined } from '@ant-design/icons-vue'; import { h } from 'vue'; const rolesAsObject = { assistant: { placement: 'start', avatar: { icon: h(UserOutlined), style: { background: '#fde3cf' } }, typing: { step: 5, interval: 20 }, styles: { maxWidth: '600px', }, messageRender: renderMarkdown, }, user: { placement: 'end', avatar: { icon: h(UserOutlined), style: { background: '#87d068' } }, }, system: { placement: 'start', avatar: { icon: h(UserOutlined), style: { background: '#d9d9d9' } }, styles: { maxWidth: '600px', }, messageRender: renderMarkdown, }, } as const;

4. 适配 BubbleList 组件的 roles 类型处理

const bubbleListRoles = rolesAsObject as NonNullable<BubbleListProps['roles']>; const bubbleItems = computed(() => props.chatList.map((m, idx) => { type RoleKey = keyof typeof rolesAsObject; const roleKey = (m.role in rolesAsObject ? m.role : 'assistant') as RoleKey; return { key: m.key ?? idx, role: m.role, // 绑定role,对应rolesAsObject中的配置项 placement: rolesAsObject[roleKey].placement, avatar: rolesAsObject[roleKey].avatar, content: m.content, }; }) );

5. 最后将 bubbleItems 和 bubbleListRoles 传入 BubbleList 组件

// ChatBubble组件 <template> <BubbleList :items="bubbleItems" :roles="bubbleListRoles" /> </template> // 使用方法 <ChatBubble :chat-list="chatList" :md-render="false" />

消息发送

功能：支持消息输入、发送控制、状态展示

<template> <Sender :value="props.inputText" @update:value="onUpdateValue" :loading="props.isResponding" :auto-size="{ minRows: 2, maxRows: 6 }" :onSubmit="handleSubmit" :onCancel="handleCancel" /> </template> <script setup lang="ts"> import { Sender } from 'ant-design-x-vue'; const props = defineProps({ inputText: String, isResponding: Boolean, }); const emit = defineEmits(['submit', 'cancel', 'update:inputText']); const onUpdateValue = (val: string) => { emit('update:inputText', val); }; const handleSubmit = () => { emit('submit', props.inputText); }; const handleCancel = () => { emit('cancel'); }; </script>

以上是AI对话页中最基础也是必不可少的部分，基于业务背景和用户体验提升，我们还可以添加更多的交互配置，比如还可以使用 vue-clipboard3 库中的 toClipboard 方法实现一键复制功能，等等。