1. 背景
自从小昊助手改为真流式输出以后，接口改为使用 SSE 来接收实时消息流，而不是像以前的“伪流式”直接获取到完整的模型调用结果；为了保证回答可以实时渲染，提升交互的时效性。下面从接口调用、回答解析和实时渲染三个方面来梳理小昊改为使用的“真流式”后的逻辑。
2. 接口调用
2.1 SSE介绍
SSE(Server Sent Event)，直译为服务器发送事件，顾名思义，也就是客户端可以获取到服务器发送的事件。我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕；但是在 sse 的场景下，客户端发起请求，连接一直保持，服务端有数据就可以返回数据给客户端，这个返回可以是多次间隔的方式
2.2 请求类型选择
- 数据传输的灵活性：POST请求通过请求体（Request Body）发送数据，没有长度限制，可以轻松承载复杂的JSON对象或其它格式的数据，满足未来功能的扩展需求。而GET请求因为URL长度限制会成为传输复杂参数的瓶颈（多数浏览器对URL长度上都限制在2000字符）
- RESTful API 设计规范：在对话中，每次用户提问都意味着在服务器端“新增”一条聊天记录。使用POST请求正符合其作为“创建”操作的语义，使API设计更具一致性和规范性
基于上面两点的考虑，小昊的调用方法采用基于fetch实现sse，这需要前端在发起和接收请求的时候进行额外的处理，而不能更简单的使用 EventSource（只支持GET请求） 来处理
2.3 请求处理
2.3.1 核心策略
使用 fetch API 发起一个 POST 请求，并接收服务器返回的 text/event-stream 流式数据，然后像原生的 EventSource 一样，逐条解析并处理消息。同时，需要针对开始，接收数据，结束，错误处理等多个状态进行管理。
通过options对象实现了一套回调机制（onMessage, onError, onComplete）,并获取StreamController（在fetch请求成功后），可以终止接收响应。
2.3.2 发请求
-
创建一个AbortController实例，并把controller.signal传入fetch 的 options 中，用于后续浏览器取消请求
-
fetch构建options:
const response = await fetch(API.STREAM_TEST, { method: "POST", headers: { "Content-Type": "application/json", // 请求体body中的数据是 JSON格式 "Accept": "text/event-stream", // 客户端期望接收一个SSE格式的流式响应 }, body: JSON.stringify({ ...params, modelSupplier, }), signal: controller.signal, });
2.3.3 流式读取
- 获取读取器：通过 response.body.getReader() 拿到流的读取器 （response.body.getReader()方法返回一个遍历器。这个遍历器的read()方法每次返回一个对象，表示本次读取的内容块。）
- 循环读取：在 readStream 中使用 while (true) 循环，不断调用 await reader.read()。这个方法会返回一个对象，包含 done (布尔值，表示流是否结束) 和 value (一个 Uint8Array 二进制数据块)。
- 解码与缓冲：使用 TextDecoder 将 value 解码为UTF-8字符串，然后追加到 jsonBuffer 中。
2.3.4 解析逻辑
- 使用while循环来处理jsonBuffer
- 通过正则表达式：
/^data:([\s\S]*?)(?:\r\n\r\n|\n\n)/来匹配 SSE 消息格式 (data: …\n\n)。 - 如果匹配成功，说明找到了一个完整消息，把它交给 processDataChunk 函数处理
- 从 jsonBuffer 的头部移除掉刚刚处理过的完整消息 (match[0])。
- 循环继续，直到正则无法匹配到完整的小昊data，跳出循环。
2.3.5 回调与状态管理
- processDataChunk: 尝试 JSON.parse 解析数据块，成功后调用 options.onMessage，将解析后的数据传给父组件。
- 当流中出现 [DONE] 标志时，意味着服务端数据已全部发送。此时会最后一次调用 processBufferedData 清空缓冲区，然后触发 onComplete 回调，并主动取消 reader 和中止 controller 来清理资源。
2.3.6 错误处理与资源释放
- 整个函数被 try…catch 包裹，用于捕获 fetch 失败等初始错误。
- readStream 内部也有 try…catch，用于捕获流读取过程中的错误。
- 特别处理了 AbortError，当请求是被用户主动取消时，会静默退出，而不是当作一个异常抛给 onError。
返回的 abort 方法会同时调用 controller.abort() 和 reader.cancel()，确保请求和流读取都被彻底终止。
2.3.7 完整代码
export const getStreamText = async (
params: StreamParams,
options: {
onMessage?: (data: StreamResponse) => void;
onError?: (error: any) => void;
onComplete?: () => void;
}
): Promise<StreamController> => {
// 创建 AbortController 用于取消请求
const controller = new AbortController();
let isCompleted = false;
let jsonBuffer = ""; // 用于累积不完整的数据
const safeComplete = () => {
if(!isCompleted){
isCompleted = true;
options.onComplete?.();
jsonBuffer = "";
}
}
try {
const response = await fetch(API.STREAM_TEST, {
method: "POST",
headers: {
"Content-Type": "application/json",
"Accept": "text/event-stream",
},
body: JSON.stringify({
...params,
modelSupplier,
}),
signal: controller.signal,
});
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
const reader = response.body?.getReader();
if (!reader) {
throw new Error("Response body is null");
}
const decoder = new TextDecoder("utf-8");
// 处理单个数据块
const processDataChunk = (dataChunk: string) => {
try {
if (dataChunk && !!dataChunk.trim() && !dataChunk.includes("[DONE]")) {
const data = JSON.parse(dataChunk);
options.onMessage?.(data);
}
} catch (error) {
console.error("解析数据块失败:", error);
console.log("解析数据块失败:", dataChunk);
}
};
// 处理缓冲区中的数据块
const processBufferedData = () => {
// 使用正则表达式匹配标准SSE格式数据
// SSE格式: "data:数据\n\n" 或 "data:数据\r\n\r\n"
const regex = /^data:([\s\S]*?)(?:\r\n\r\n|\n\n)/;
while (true) {
const match = regex.exec(jsonBuffer);
if (!match) break; // 没有找到完整的数据块
const dataChunk = match[1].trim();
// 处理数据块
if (dataChunk && dataChunk.trim() !== "") {
processDataChunk(dataChunk);
}
// 从缓冲区移除已处理的数据
jsonBuffer = jsonBuffer.substring(match[0].length);
}
};
// 读取流式数据
const readStream = async () => {
try {
while (true) {
const { done, value } = await reader.read();
const streamData = decoder.decode(value, { stream: true });
jsonBuffer += streamData; // 数据块添加到缓冲区
// 流结束处理
if (value && streamData.includes("[DONE]")) {
console.log("流式数据读取完成");
processBufferedData(); // 处理所有剩余数据块
safeComplete(); // 完成回调
reader.cancel();
controller.abort();
break;
}
processBufferedData(); // 处理单个数据块
}
} catch (error: any) {
if (error.name === "AbortError") {
console.log("请求已取消");
return;
}
console.error("读取流式数据失败:", error);
jsonBuffer = "";
options.onError?.(error);
}
};
// 启动流处理
readStream();
return {
abort: () => {
controller.abort();
reader?.cancel();
jsonBuffer = "";
},
};
} catch (error) {
console.error("请求流式数据失败:", error);
options.onError?.(error);
return {
abort: () => controller.abort(),
};
}
};3. 流式答案的接收
- 使用我们先前定义好的getStreamText方法，并获取到streamController ，在三个对应的回调函数中加上与业务相关的逻辑。
- 接收数据块： 每当 SSE 接收到一个新的数据块， onMessage 的回调函数就会被触发，在回调函数中分别处理对应的数据块。
- 使用 handlers对象处理不同类型的数据，针对data中的不同字段来进行不同的逻辑处理
4. 流式回答的渲染
4.1 渲染原理
使用marked库，把 markdown 格式的文本转化成 HTML ，并配合katex 插件，mermaid库等对公式、图表的渲染提供支持。
4.2 实时渲染
在项目早期，使用 v-html 指令来渲染流式返回的 Markdown 内容。然而，在引入图片实时渲染后，出现了明显的图片闪烁问题。其根源在于 v-html 的工作机制：每当流式内容更新时，它会销毁并完整地重建整个 DOM 子树。这种全量渲染导致 <img> 元素被反复创建和加载，从而引发视觉上的闪烁。
为解决这个问题，我们放弃了 v-html，转而实现了一套精细化的增量更新策略（updateDomIncrementally 函数），以确保在内容流式更新时，已存在的图片等 DOM 元素能够保持稳定。
Markdown
-> newHtml
-> newNodes
-> newNodeHtml
-> 与 previousNodeHtml 比对其核心逻辑如下：
- 状态存储：在每次渲染后，将生成的 HTML 字符串存储在 previousHtml 变量中，作为下一次比对的基准。
- DOM 节点化：当新的 Markdown 内容 (message) 传入时，首先将其转换为 newHtml。接着，我们创建临时的 div 容器，利用 innerHTML 将 previousHtml 和 newHtml 分别解析成可供比对的 DOM 节点数组（previousNodes 和 newNodes）。
- 差异化比对与更新：
- 我们遍历 newNodes 数组，并与 previousNodes 中对应位置的节点进行 outerHTML 的比对。
- 新增节点：如果 previousNodes 中不存在对应的旧节点，则判定为新增内容，直接将新节点追加到容器末尾。
- 无变化节点：如果新旧节点的 outerHTML 完全相同，则跳过，不进行任何操作。
- 内容变化节点：如果 outerHTML 不同，则执行智能更新：
- 图片智能替换 (smartReplaceWithImages)：首先检查新旧节点中是否包含
<img>标签。如果包含，我们会进一步比对图片的数量和 src 属性。若图片完全一致，则会执行 handleSameImages 函数，它通过在新节点的克隆中替换上旧的<img>元素，从而在更新节点包裹内容的同时，保留了原始的<img>DOM 对象，图片不会重复加载并闪烁。若图片不一致，则直接用新节点替换旧节点。 - 常规替换：如果不涉及图片，则直接使用 replaceWith 方法替换整个节点。
- 图片智能替换 (smartReplaceWithImages)：首先检查新旧节点中是否包含
- 后续处理：在每次增量更新完成后，会调用 handleImages 函数，为新增的图片统一添加加载失败 (onerror) 的处理逻辑和点击预览 (onclick) 事件。
注意：
-
在util.markdownToHtml方法中，我们进行了过滤掉markdown格式的
```,和给表格加一个父包装容器用于修改样式的处理。 -
在将流式响应内容通过 markedToHtml 转换为最终渲染的 HTML 之前，我们必须先进行一次预处理。此步骤的核心是利用 filterRegex 正则表达式，移除模型在调用 MCP 工具时嵌入回答中的各类 XML 标签（如
<thinking>,<use_mcp_tool>等）。const filterRegex = new RegExp( //移除 <thinking> 和 <use_mcp_tool> <ask_follow_question> <tool_call_result> 的完整或不完整块 "<(thinking|use_mcp_tool|ask_followup_question|tool_call_result)[^>]*>[\\s\\S]*?(?:<\\/\\1>|$)" + "|" + //移除 <attempt_completion> 标签自身 "<\\/?attempt_completion>" + "|" + // - `...[^>]*`: 匹配 <att, <thi, <use, <ask等片段 // - `|<`: 或 ( | ) 只匹配一个孤立的 < 符号 // - `$`: 强制以上规则只在字符串的最末尾生效 "(<\\/?(?:at|th|us|as)[^>]*|<)$", "g" );
4.3 历史记录
回答的历史记录可以直接获取到完整的小昊内容，所以只需要针对字段分别进行处理即可，唯一需要注意的一点就是部分回答内容需要前端解析content中的标签获取（如：反问，调用的mcp工具等）
5. 测试流式输出的问题
请写一份约 3000～4000 字的《智慧水务系统供水管网漏损分析报告》，直接输出完整 Markdown，不要省略任何章节，不要调用工具，不要使用 HTML。
请严格按照以下顺序组织内容：
1. 报告标题和摘要
- 摘要不少于 300 字
- 包含加粗、斜体、引用和行内代码
2. 系统示意图片
- 使用以下 Markdown 图片，并在图片后继续输出较长文字：
3. 背景和问题分析
- 至少包含 5 个二级标题
- 每个标题下至少 3 个自然段
- 同时包含有序列表和无序列表
4. 数据指标表格
- 输出一个至少 12 行、6 列的 Markdown 表格
- 字段包括：区域、供水量、售水量、漏损量、漏损率、告警数量
- 数据需要合理且各行不同
5. 数学模型
- 使用行内公式，例如：
\(R = \frac{Q_s - Q_b}{Q_s} \times 100\%\)
- 使用至少三个独立的块级公式，格式必须为：
\[
Q_l = Q_s - Q_b
\]
- 对每个公式中的变量进行详细解释
6. Mermaid 流程图
- 使用一个完整的 mermaid 代码块
- 绘制“数据采集、数据清洗、异常识别、漏点定位、生成工单、维修反馈”的处理流程
- Mermaid 节点名称使用中文
- 在 Mermaid 代码块结束后继续输出至少 1000 字内容，不要把 Mermaid 放在回答结尾
7. 分区对比和治理建议
- 比较至少 5 个供水分区
- 给出短期、中期、长期治理方案
- 包含风险等级、优先级和预期收益
8. 最终总结
- 使用任务清单形式总结
- 添加一个普通 Markdown 链接：
[智慧水务参考资料](https://example.com/water)
请确保图片、表格、公式和 Mermaid 都实际输出，不能只描述它们。整个回答必须一次连续生成，不要询问我是否继续。