AI对话的逐字输出：流式返回才是幕后黑手

AI应用已经渗透到我们生活的方方面面。其中，AI对话系统因其自然、流畅的交流方式而备受瞩目。前段时间有人在交流群中问到，如何实现AI对话中那种逐字输出的效果，这背后，流式返回技术发挥了关键作用。

欢迎加入前端筱园交流群：点击加入交流群

​ 其实这背后并不是前端做了什么特效，而是采用的流式返回，而不是一次性返回完整的响应。流式返回允许服务器在一次连接中逐步发送数据，而不是一次性返回全部结果。这种方式使得前端可以在等待完整响应的过程中，逐步展示生成的内容，从而极大地提升了用户体验。

​ 那么，前端接收流式返回具体有哪些方式呢？接下来，本文将详细探讨几种常见的技术手段，帮助读者更好地理解并应用流式返回技术。

使用 Axios

大多数场景下，前端用的最多的就是axios来发送请求，但是axios 只有在在Node.js环境中支持设置 responseType: 'stream' 来接收流式响应。

const axios = require('axios');
const fs = require('fs');


axios.get('http://localhost:3000/stream', {
responseType: 'stream', // 设置响应类型为流

})

    .then((response) => {
// 将响应流写入文件

        response.data.pipe(fs.createWriteStream('output.txt'));

    })

    .catch((error) => {
console.error('Stream error:', error);

    });

特点

• 
  
• 仅限 Node.js：浏览器中的 axios 不支持 responseType: 'stream'。
  
• 适合文件下载：适合处理大文件下载。
  

使用 WebSocket

WebSocket 是一种全双工通信协议，适合需要双向实时通信的场景。

前端代码：

const socket = new WebSocket('ws://localhost:3000');


socket.onopen = () => {
console.log('WebSocket connected');

};

socket.onmessage = (event) => {
console.log('Received data:', event.data);

};

socket.onerror = (error) => {
console.error('WebSocket error:', error);

};

socket.onclose = () => {
console.log('WebSocket closed');

};

服务器代码

const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 3000 });


wss.on('connection', (ws) => {
console.log('Client connected');

let counter = 0;
const intervalId = setInterval(() => {

        counter++;

        ws.send(JSON.stringify({ message: 'Hello', counter }));

if (counter >= 5) {
clearInterval(intervalId);

            ws.close();

        }

    }, 1000);


    ws.on('close', () => {
console.log('Client disconnected');
clearInterval(intervalId);

    });

});

虽然WebSocket作为一种在单个TCP连接上进行全双工通信的协议，具有实时双向数据传输的能力，但AI对话情况下可能并不选择它进行通信。主要有以下几点原因：

• 
  
• 在AI对话场景中，通常是用户向AI模型发送消息，模型回复消息的单向通信模式，WebSocket的双向通信能力在此场景下并未被充分利用
  
• 使用WebSocket可能会引入不必要的复杂性，如处理双向数据流、管理连接状态等，也会增加额外的部署与维护工作
  

特点

• 
  
• 双向通信：适合实时双向数据传输
  
• 低延迟：基于 TCP 协议，延迟低
  
• 复杂场景：适合聊天、实时游戏等复杂场景
  

使用 XMLHttpRequest

虽然 XMLHttpRequest 不能直接支持流式返回，但可以通过监听 progress 事件模拟逐块接收数据

const xhr = new XMLHttpRequest();

xhr.open('GET', '/stream', true);


xhr.onprogress = (event) => {
const chunk = xhr.responseText; // 获取当前接收到的数据
console.log(chunk);

};


xhr.onload = () => {
console.log('Request complete');

};


xhr.send();

服务器代码（Koa 示例）：

router.get("/XMLHttpRequest", async (ctx, next) => {

    ctx.set({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",

    });
// 创建一个 PassThrough 流
const stream = new PassThrough();

    ctx.body = stream;
let counter = 0;
const intervalId = setInterval(() => {

        counter++;

        ctx.res.write(
JSON.stringify({ message: "Hello", counter })

        );

if (counter >= 5) {
clearInterval(intervalId);

            ctx.res.end();

        }

    }, 1000);


    ctx.req.on("close", () => {
clearInterval(intervalId);

        ctx.res.end();

    });

});

可以看到以下的输出结果，在onprogress中每次可以拿到当前已经接收到的数据。它并不支持真正的流式响应，用于AI对话场景中，每次都需要将以显示的内容全部替换，或者需要做一些额外的处理。

如果想提前终止请求，可以使用 xhr.abort() 方法；

setTimeout(() => {

    xhr.abort();

}, 3000);

特点

• 
  
• 兼容性好：支持所有浏览器。
  
• 非真正流式：XMLHttpRequest 仍然需要等待整个响应完成，progress 事件只是提供了部分数据的访问能力。
  
• 内存占用高：不适合处理大文件。
  

使用 Server-Sent Events (SSE)

SSE 是一种服务器向客户端推送事件的协议，基于 HTTP 长连接。它适合服务器向客户端单向推送实时数据

前端代码：

const eventSource = new EventSource('/sse');


eventSource.onmessage = (event) => {
console.log('Received data:', event.data);

};


eventSource.onerror = (event) => {
console.error('EventSource failed:', event);

};

服务器代码（Koa 示例）：

router.get('/sse', (ctx) => {

    ctx.set({
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',

    });

let counter = 0;
const intervalId = setInterval(() => {

        counter++;

        ctx.res.write(`data: ${JSON.stringify({ message: 'Hello', counter })}\n\n`);

if (counter >= 5) {
clearInterval(intervalId);

            ctx.res.end();

        }

    }, 1000);


    ctx.req.on('close', () => {
clearInterval(intervalId);

        ctx.res.end();

    });

});

EventSource 也具有主动关闭请求的能力，在结果没有完全返回前，用户可以提前终止内容的返回。

// 在需要时中止请求
setTimeout(() => {

    eventSource.close(); // 主动关闭请求

}, 3000); // 3 秒后中止请求

虽然EventSource支持流式请求，但AI对话场景不使用它有以下几点原因：

• 
  
• 单向通信
  
• 仅支持 get 请求：在 AI 对话场景中，通常需要发送用户输入（如文本、文件等），这需要使用 POST 请求
  
• 无法自定义请求头：EventSource 不支持自定义请求头（如 Authorization、Content-Type 等），在 AI 对话场景中，通常需要设置认证信息（如 API 密钥）或指定请求内容类型
  

注意点

返回给 EventSource 的值必须遵循 data: 开头并以 \n\n 结尾的格式，这是因为 Server-Sent Events (SSE) 协议规定了这种格式。SSE 是一种基于 HTTP 的轻量级协议，用于服务器向客户端推送事件。为了确保客户端能够正确解析服务器发送的数据，SSE 协议定义了一套严格的格式规范。SSE 协议规定，服务器发送的每条消息必须遵循以下格式：

field: value\n

其中 field 是字段名，value 是对应的值。常见的字段包括：

• 
  
• data:：消息的内容（必须）。
  
• event:：事件类型（可选）。
  
• id:：消息的唯一标识符（可选）。
  
• retry:：客户端重连的时间间隔（可选）。
  

每条消息必须以 两个换行符 (\n\n) 结尾，表示消息结束

以下是一个完整的 SSE 消息示例：

id: 1\n

event: update\n

data: {"message": "Hello", "counter": 1}\n\n

特点

• 
  
• 单向通信：适合服务器向客户端推送数据。
  
• 简单易用：基于 HTTP 协议，无需额外协议支持。
  
• 自动重连：EventSource 会自动处理连接断开和重连
  

使用 fetch API

fetch API 是现代浏览器提供的原生方法，支持流式响应。通过 response.body，可以获取一个 ReadableStream，然后逐块读取数据。

前端代码：

// 发送流式请求
fetch("http://localhost:3000/stream/fetch", {
method: "POST",

    signal,

})

    .then(async (response: any) => {
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
console.log(new TextDecoder().decode(value));

        }

    })

    .catch((error) => {
console.error("Fetch error:", error);

    });

服务器代码（Koa 示例）：

router.post("/fetch", async (ctx) => {

    ctx.set({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",

    });
// 创建一个 PassThrough 流
const stream = new PassThrough();

    ctx.body = stream;
let counter = 0;
const intervalId = setInterval(() => {

        counter++;

        ctx.res.write(JSON.stringify({ message: "Hello", counter }));

if (counter >= 5) {
clearInterval(intervalId);

            ctx.res.end();

        }

    }, 1000);


    ctx.req.on("close", () => {
clearInterval(intervalId);

        ctx.res.end();

    });

});

fetch也同样可以在客户端主动关闭请求。

// 创建一个 AbortController 实例
const controller = new AbortController();
const { signal } = controller;
// 发送流式请求
fetch("http://localhost:3000/stream/fetch", {
method: "POST",

    signal,

})

    .then(async (response: any) => {
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
console.log(new TextDecoder().decode(value));

        }

    })

    .catch((error) => {
console.error("Fetch error:", error);

    });

// 在需要时中止请求
setTimeout(() => {

    controller.abort(); // 主动关闭请求

}, 3000); // 3 秒后中止请求

打开控制台，可以看到在Response中可以看到返回的全部数据，在EventStream中没有任何内容。

这是由于返回的信息SSE协议规范，具体规范见上文的 Server-Sent Events 模块中有介绍到

ctx.res.write(
`data: ${JSON.stringify({ message: "Hello", counter })}\n\n`

);

但是客户端fetch请求中接收到的数据也包含了规范中的内容，需要前端对数据进一步的处理一下

特点

• 
  
• 原生支持：现代浏览器均支持 fetch 和 ReadableStream。
  
• 逐块处理：可以实时处理每个数据块，而不需要等待整个响应完成。
  
• 内存效率高：适合处理大文件或实时数据。
  

总结

综上所述，在 AI 对话场景中，fetch 请求 是主流的技术选择，而不是 XMLHttpRequest 或 EventSource。以下是原因和详细分析：

• 
  
• fetch 是现代浏览器提供的原生 API，基于 Promise，代码更简洁、易读
  
• fetch 支持 ReadableStream，可以实现流式请求和响应
  
• fetch 支持自定义请求头、请求方法（GET、POST 等）和请求体
  
• fetch 结合 AbortController 可以方便地中止请求
  
• fetch 的响应对象提供了 response.ok 和 response.status，可以更方便地处理错误
  

最后来看一个接入deekseek的完整例子：

resource.dengzhanyong.com/mp4/7823928…

服务器代码（Koa 示例）：

const openai = new OpenAI({
baseURL: "https://api.deepseek.com",
apiKey: "这里是你申请的deepseek的apiKey",

});


// 流式请求 DeepSeek 接口并流式返回

router.post("/fetchStream", async (ctx) => {
// 设置响应头

    ctx.set({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",

    });

try {
// 创建一个 PassThrough 流
const stream = new PassThrough();

        ctx.body = stream;

// 调用 OpenAI API，启用流式输出
const completion = await openai.chat.completions.create({
model: "deepseek-chat", // 或 'gpt-3.5-turbo'
messages: [{ role: "user", content: "请用 100 字介绍 OpenAI" }],
stream: true, // 启用流式输出

        });
// 逐块处理流式数据
for await (const chunk of completion) {
const content = chunk.choices[0]?.delta?.content || ""; // 获取当前块的内容

            ctx.res.write(content);

            process.stdout.write(content); // 将内容输出到控制台

        }

        ctx.res.end();

    } catch (err) {
console.error("Request failed:", err);

        ctx.status = 500;

        ctx.res.write({ error: "Failed to stream data" });

    }

});

前端代码：

const controller = new AbortController();
const { signal } = controller;
const Chat = () => {
const [text, setText] = useState<string>("");
const [message, setMessage] = useState<string>("");
const [loading, setLoading] = useState<boolean>(false);

function send() {
if (!message) return;
setText(""); // 创建一个 AbortController 实例
setLoading(true);

// 发送流式请求
fetch("http://localhost:3000/deepseek/fetchStream", {
method: "POST",
headers: {
"Content-Type": "application/json",

            },
body: JSON.stringify({

                message,

            }),

            signal,

        })

            .then(async (response: any) => {
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const data = new TextDecoder().decode(value);
console.log(data);
setText((t) => t + data);

                }

            })

            .catch((error) => {
console.error("Fetch error:", error);

            })

            .finally(() => {
setLoading(false);

            });

    }

function stop() {

        controller.abort();
setLoading(false);

    }

return (
<div>
<Input
value={message}
onChange={(e) => setMessage(e.target.value)}

            />
<Button
onClick={send}
type="primary"
loading={loading}
disabled={loading}

            >

                发送
</Button>
<Button onClick={stop} danger>

                停止回答
</Button>
<div>{text}</div>
</div>

    );

};

写在最后

欢迎加入前端筱园交流群：点击加入交流群

关注我的公众号【前端筱园】，不错过每一篇推送