AI 对话页流式处理架构：Web Streams 与 Fetch API 实践

引言

在构建 AI 智能体应用时，为了降低模型搭建复杂度，通常会将数据流转与 UI 实现统一封装。但若要深入理解底层机制，采用'生产者 - 消费者模式'的流式处理架构是更优解。这种架构将网络 IO、数据解码、文本解析与 UI 渲染解耦，从而实现实时流式响应和 UI 增量渲染。

本文基于 Vue 3 + TypeScript + Vite 技术栈，结合 Ant Design Vue 及 Ant Design X Vue 组件库，探讨如何利用 Web Streams API 与 Fetch API 实现这一流程。

从请求发送到 UI 渲染的整体链路如下：

![流式响应整体链路示意图]

流式响应处理

请求管理

我们需要同时控制上游的网络请求和下游的字节流读取，以实现会话的可取消与可停止。

1. AbortController：作为上游网络中止句柄。在发起新提问前中止旧请求，或用户点击取消时立即终止当前请求。
2. ReadableStreamDefaultReader：作为下游传输层字节流的读取器句柄。它驱动上游生产者向管道入队字节块，并在用户取消时终止读取。

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

流处理管道

建立一条 Web Streams 解析管线：生产者 → 解码 → 按行拆分 → 消费者。整体处理流程包含以下三个核心环节。

生产者流

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

参考文档：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() 钩子中冲刷缓冲区，避免丢失收尾数据。

const textStream = producerStream.pipeThrough(
  new TextDecoderStream() as unknown as TransformStream<Uint8Array, string>
);

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 终止读取，防止连接残留。

const sseStream = textStream.pipeThrough(lineSplitter);
const downstreamReader = sseStream.getReader();

function consume() {
  downstreamReader.read().then(({ done, value }) => {
    if (done) {
      isResponding.value = false;
      downstreamReader.releaseLock();
      return;
    }
    const message = value.replace(/^data:\s*/, '');
    if (message === '[DONE]') {
      downstreamReader.cancel('stream finished');
      return;
    }
    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,
  });
}

中止请求

1. 重置 UI 响应状态：标记为非响应中，避免界面显示'正在响应'。
2. 终止流式读取：若存在读取器，主动取消读取并清理引用。
3. 终止网络请求：触发中止并清理引用。

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。后者专注于 Vue 生态的 AI 组件库，支持 TS 和 TSX，能简化对话式 AI 应用的开发。

消息展示

1. 功能：展示对话历史，支持多角色渲染及 Markdown 渲染。
2. 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,
      placement: rolesAsObject[roleKey].placement,
      avatar: rolesAsObject[roleKey].avatar,
      content: m.content,
    };
  })
);

5. 传入 BubbleList 组件：

<!-- ChatBubble 组件 -->
<template>
  <BubbleList :items="bubbleItems" :roles="bubbleListRoles" />
</template>

<!-- 使用方法 -->
<ChatBubble :chat-list="chatList" :md-render="false" />

消息发送

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

<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 对话页中最基础也是必不可少的部分。在实际业务场景中，还可以根据需求扩展更多交互配置，例如集成剪贴板工具实现一键复制功能等。