Nuxt3中使用Mistral AI API流式返回时EventSource重复问题

问题原因 #

EventSource默认会在连接断开后自动重连，当后端流式传输完成并关闭响应时，前端EventSource会重新发起请求，导致内容从头重复展示。此外，后端未发送明确的流结束标记，前端无法判断传输是否完成，也会加剧这个问题。

前端修复方案 #

修改EventSource逻辑，在收到流结束标记时手动关闭连接，同时处理错误情况避免自动重连：

// 重置内容，避免多次请求累加
bio.value = '';
const eventSource = new EventSource(`/api/mistral/analysis?songs=${encodeURIComponent(songs)}`);

eventSource.onmessage = (event) => {
    // 判断是否收到结束标记
    if (event.data === '[DONE]') {
        eventSource.close();
        return;
    }
    bio.value += event.data;
};

// 监听错误事件，关闭连接防止自动重连
eventSource.onerror = () => {
    eventSource.close();
};

后端优化方案 #

1. 添加X-Accel-Buffering: no响应头，防止服务器缓冲SSE流（部分部署环境如Vercel/Nginx需要）；
2. 流式传输完成后发送明确的结束标记[DONE]；
3. 确保响应正确处理，避免异常导致的连接问题。

修改后的后端代码：

import { Mistral } from '@mistralai/mistralai';
import { getQuery } from 'h3';

export default defineEventHandler(async (e) => {
    const { songs } = getQuery(e);
    const config = useRuntimeConfig();

    const messages = [{ 
        role: 'system', 
        content: `
        You are a helpful creative psychiatrist who determines peoples personalities by looking at the songs they listen to. Besides that, 
        give an in-depth analasys of who I might prefer as a romantic partner and why. Mention a few songs on which you based the analysis.
        Respond in markdown and do not ask if you can assist any further.
        `
    },{ 
        role: 'user', 
        content: `${songs}`
    }];

    try {
        const mistral = new Mistral({ apiKey: config.mistralAiApiKey });

        const stream = await mistral.chat.stream({
            model: 'mistral-small-latest',
            messages
        });

        // 设置SSE响应头，添加防缓冲头
        setHeader(e, 'Content-Type', 'text/event-stream');
        setHeader(e, 'Cache-Control', 'no-cache');
        setHeader(e, 'Connection', 'keep-alive');
        setHeader(e, 'X-Accel-Buffering', 'no');

        for await (const chunk of stream) {
            const content = chunk.data.choices[0].delta.content;
            if (content) { // 确保内容存在再发送
                e.node.res.write(`data: ${content}\n\n`);
                // 强制刷新响应，避免缓冲区积压
                await new Promise(resolve => e.node.res.flush(resolve));
            }
        }

        // 发送流结束标记
        e.node.res.write(`data: [DONE]\n\n`);
        e.node.res.end();
    } 
    catch (error) {
        console.error(error);
        // 关闭响应，避免前端等待
        e.node.res.end();
            
        throw createError({
            statusCode: 500,
            message: error.message || '服务器内部错误'
        });
    }
});

额外注意事项 #

• 前端请求时用encodeURIComponent处理songs参数，避免特殊字符导致的请求异常；
• 部分部署平台可能需要额外配置以支持SSE，比如Vercel需要在vercel.json中添加SSE支持；
• 确保后端在异常时正确关闭响应，防止前端EventSource一直等待。

内容的提问来源于stack exchange，提问作者Rik Verbeek