One post tagged with "SSE"

可以点击右上角的[Ask AI]体验，目前只是一个基础的RAG服务

其实我的需求很简单：

• 原先基于algolia的文本检索有点落后了，想给自己的知识库接入AI问答，可以同时检索多处内容并做对话式的总结，而不是简单的带文本字符串的文章列表（algolia后面推出了NeuralSearch，不过得额外开启，还要付费）
• 学习 ai agent 相关技术，本篇主要涉及 prompt 和 RAG 技术

整体设计​

怎么交互？

因为原先algolia的检索窗口在右上角，所以直接在旁边加上ai问答按钮，触发对话窗口。有点像客服助手，emm说起来，客服助手算是入门ai agent最好的项目吧，根据问题设计多种回复prompt，通过function calling调用api方法或查询数据库等等，我刚开始练手就是搞了一个客服助手，熟悉了ai agent的基本体

然后需要一个轻量的后端服务来实现RAG服务接口，对接向量数据库，这里用了 chroma db来存储知识库的文章向量，到这里就先明确技术选型如下：

• 前端：docusaurus(SSG)
• 后端：hono。hono 是一个轻量级的 web 框架，支持 nodejs，据说这个库还是一个跨运行时的轻量框架，能支持 Bun/Deno 等运行时环境。这个框架对比express 体积更小(12-15kb)，api 风格跟 express 挺接近的，其实都可以，这里主要是尝鲜，这个服务也相对不会很复杂，就用hono来实现
• 向量数据库：chroma db。这是一个轻量的向量数据库，基于 sqlite（文件型存储），并且 nodejs 也能快速接入，而且我的知识库属于个人场景，单用户访问，不需要分布式高 QPS。关于向量，可以先记住这句话：语义相近的文本，向量在空间中的距离就小
• 模型：deepseek-v3，性价比高，这种场景只是简单的总结和问答，不需要复杂的模型
• 向量模型：阿里的 text-embedding-3

关键工程问题​

文档切片​

切片策略：根据标题栏和段落，按语义切分文本

将 Docusaurus 项目中的 docs/ 和 blog/ 目录下的所有 Markdown (.md/.mdx) 文档提取出来，进行合理的切片处理，以便后续进行向量化 (Embedding)。因为 Markdown 本身就是结构化的，H1/H2/H3 天然是语义边界，切出来的每片都是一个完整的小主题，直接喂给 embedding 模型。每个切片必须保留路由信息 (sourceUrl)，以便于前端问答时回显来源链接

这里使用了几个库来协助处理：

• unified + remark-parse：用于解析 Markdown 生成 AST（抽象语法树）。如果不这么做，只能正则匹配 #，碰到代码块里的 # 就误判
• remark-frontmatter：用于识别并提取文件头部的 YAML 属性（如 title/tags）。切片丢了标题，回显时无法显示"这是哪篇文档"
• mdast-util-to-string：用于将 AST 节点转回纯文本进行分析。因为经过上述处理后拿到的还是带结构的对象，没法直接喂给 embedding

整个 pipeline：

.md 源文件
  → unified + remark-parse：解析成 AST（树状结构）
  → remark-frontmatter：把头部的 YAML 抽出来作为 metadata
  → 遍历 AST 找 H2/H3 节点作为切片锚点
  → mdast-util-to-string：把切片节点的文本内容转成纯文本
  → 输出：{ sourceUrl, title, content, ...metadata }

为什么不一步到位？Markdown 看似简单，其实头部的 YAML、代码块、链接、图片都是「结构化信息」，得先把它们都识别出来，才能精准地切、按层级地切、不破坏语境地切。

注意：只切到 H2/H3 比较合适，再细就碎了；遇到特别长的章节，可以按段落再细分

这里可以配置 package.json 的 scripts 命令，这样在 github 的 workflow 流程里就能执行相关脚本来触发文档切片，并自动上传向量数据库

"scripts": {
    "rag:parse": "tsx scripts/rag/test-parser.ts"
}

文本向量化​

借助 text-embedding-3 模型，把切好的文本转成向量表示（就是把「文字」转成「坐标」，后续在向量空间里找距离最近的那几个）。这里选了阿里的向量模型，性价比相对高一些，中文支持也 OK。

存到 ChromaDB 时几个细节：

• 向量维度：text-embedding-3 默认输出 1024 维，每片文本对应一个 1024 维的浮点数数组
• 元数据一起存：除了向量，把原文、文档路径、章节标题作为 metadata 一起入库，召回时靠 metadata 做过滤和展示

切完的格式如下：

{
    "content": "[文档路径: /docs/afreshjs/Node.js/高级核心概念 | 章节: 高级核心概念 > 高级核心概念 > 3. 模块化的底层原理 (CJS vs ESM) > 3.2 循环引用 (Circular Dependency)]\n### 3.2 循环引用 (Circular Dependency)\n\n当 `a.js` 引用 `b.js`，同时 `b.js` 又引用 `a.js` 时：\n\n- **CommonJS 的表现**：\n  CJS 在加载模块时，会优先在 `require.cache` 中创建该模块的空对象 `module.exports`。当发生循环引用时，`b.js` 会拿到 `a.js` **还没执行完的、不完整的 `exports` 对象**。这会导致运行时拿到 `undefined` 而报错。\n  _(CJS 导出的是值的拷贝/浅拷贝)_\n\n- **ESM (ECMAScript Modules) 的表现**：\n  ESM 的加载分为“解析”、“实例化”、“执行”三个阶段。ESM 导出的是**实时绑定 (Live Bindings)**，即导出的变量和原模块内部的变量指向同一块内存地址。\n  因此在处理循环引用时，只要你不立刻去读取那个还没初始化的变量，引擎就能完美处理好模块的依赖图谱。\n\n---",
    "metadata": {
      "sourceUrl": "/docs/afreshjs/Node.js/高级核心概念",
      "title": "高级核心概念",
      "h1": "高级核心概念",
      "h2": "3. 模块化的底层原理 (CJS vs ESM)",
      "h3": "3.2 循环引用 (Circular Dependency)",
      "chunkIndex": 5
    }
  },

增量灌库​

这里有个小插曲，第一次需要全量灌库，接口携带的数据太大，导致接口报了body体积过大的异常(对应http响应码413)。除了调整nginx对应配置的阈值（client_max_body_size），这里还用了多个接口分批上传，稳妥点，毕竟接口处理或写库太久了估计还要触发接口超时异常(504)

增量灌库主要原理是基于 Git Diff 仅对变更文件重新 Embedding，避免每次修改知识库后，都需要重新全量上传，而是只上传新增或修改的文章向量。并且增量灌库逻辑都是写到github actions中，每次修改知识库后，都需要触发github actions来增量灌库。

chroma 数据库是怎么存取和更新的？

ChromaDB 本质就是 SQLite（存元数据 + 原文）+ HNSW 索引（存向量，做相似匹配）。每个 collection 是一组「记录」，每条含 id / document / embedding / metadata 四个字段。

// 新增
await collection.add({ ids, documents, embeddings, metadatas });
// 有则更新、无则新增（按 id）—— 灌库最常用
await collection.upsert({ ids, documents, embeddings, metadatas });
// 更新已有（不存在会报错）
await collection.update({ ids, embeddings, metadatas });
// 删除
await collection.delete({ ids: [...] });

为啥这么简单？因为底层 HNSW 索引是自维护的——你 add / upsert / delete 的时候，索引自动更新，不用手动 rebuild。这也是个人项目适合用 Chroma 的原因：不用自己折腾向量索引的实现。

prompt设计​

针对ai问答的回复，设计了如下 prompt：

# 角色

你是一个知识库助手，只能基于「参考文档」回答用户问题，不要编造。

# 回答要求

1. 先用 1-2 句话直接回答用户问题
2. 再补充关键细节（来自参考文档）
3. 最后列出引用来源（仅列文档名 + 章节，不要贴 URL 长链接）
4. 如果参考文档不足以回答，明确说"知识库没有相关内容"

# 参考文档

${retrieved_chunks}

# 用户问题

${user_query}

流式问答​

这里就是SSE的典型应用场景，用户输入问题，后端检索数据库将最相关的数据喂给模型，然后将模型返回的结果处理成一个流式响应，前端可以实时接收并展示给用户。

开启 SSE 接口，三件事缺一不可：响应头 + 后端实现 + Nginx 反代配置

1. 响应头

content-type: text/event-stream
cache-control: no-cache          # 防止中间代理缓存
connection: keep-alive           # 保持长连接
x-accel-buffering: no            # 告诉 Nginx 不要缓冲（关键）

2. 后端实现

用 hono/streaming 的 streamSSE(把后端响应包成 SSE 格式（data: ...\n\n），上游 stream: true 让 DeepSeek 边想边吐——两端一配合才有"打字机"效果)，几行就能写出来，核心是「透传上游流」，参考代码如下：

import { Hono } from "hono";
import { streamSSE } from "hono/streaming";

const app = new Hono();

app.get("/rag-api/ask", async (c) => {
  return streamSSE(c, async (stream) => {
    // 1. 请求上游 LLM，开启流式
    const upstream = await fetch("https://api.deepseek.com/chat/completions", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.DEEPSEEK_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        model: "deepseek-chat",
        messages: [{ role: "user", content: prompt }],
        stream: true,
      }),
    });

    if (!upstream.body) return;

    // 2. 透传上游流：read → 解析 → writeSSE
    const reader = upstream.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split("\n");
      buffer = lines.pop() || ""; // 最后一行可能不完整，留到下轮

      for (const line of lines) {
        if (!line.startsWith("data: ")) continue;
        const data = line.slice(6).trim();
        if (data === "[DONE]") {
          await stream.writeSSE({ data: "[DONE]" });
          continue;
        }
        try {
          const json = JSON.parse(data);
          const content = json.choices[0]?.delta?.content || "";
          if (content) {
            await stream.writeSSE({ data: JSON.stringify({ content }) });
          }
        } catch (e) {
          console.error("SSE 解析错误:", e);
        }
      }
    }
  });
});

3. Nginx 反代必须关缓冲（最常踩的坑）

Nginx 默认会缓冲响应，SSE 会被"卡住"看不到流。必须加：

location /test-api/ {
    proxy_pass http://127.0.0.1:3000/;
    proxy_buffering off;          # 关键！禁用缓冲
    proxy_cache off;              # 禁用缓存
    proxy_http_version 1.1;       # SSE 需要 HTTP/1.1
    chunked_transfer_encoding on;
    proxy_read_timeout 60s;       # 防止超时断流
}

4. 客户端断开要清理上游（不浪费 token）

stream.onAbort(async () => {
  await reader.cancel(); // 关闭上游连接
});

易踩的坑：

总结下上述流程：

DeepSeek API (ReadableStream)
    │ raw bytes (SSE 格式: data: {...}\n\n)
    ▼
getReader() 异步迭代
    │
    ▼
TextDecoder → buffer 累积 → 按 \n 切行
    │
    ▼
JSON.parse → 提取 content
    │
    ▼
writeSSE({ data: JSON.stringify({content}) })
    │
    ▼
Hono 内部写到 response.body (writable)
    │
    ▼
Nginx 不缓冲 → 客户端立即可见

话说怎么中断ai的回复？

1. 在对话窗口点击取消按钮
2. 后端需要在收到用户中断请求后，立即停止模型调用，返回一个空的流式响应
3. 前端需要在收到空的流式响应后，立即关闭对话窗口

为啥要用 @microsoft/fetch-event-source 这个库来做SSE请求？

• 原生 EventSource：只支持 GET 请求，header 也不能自定义——传个 token 都麻烦
• fetch + getReader()：微信内置浏览器（X5/WKWebView）会拿到 null 直接抛异常

核心思路：用 fetch 发请求 + 自己解析流式响应体，各取两条路的一半优点：

• 用 fetch：支持 POST、自定义 header——EventSource 做不到的它都能做
• 自己解析：内部封装了 ReadableStream 兼容处理，遇到 X5/WKWebView 拿不到 body 时会自动 fallback 到 XHR streaming 方案

把"流式解析"这层脏活在 SDK 里干完了，业务代码只关心 onmessage 回调：

import { fetchEventSource } from "@microsoft/fetch-event-source";

await fetchEventSource("/rag-api/ask", {
  method: "POST",
  headers: { "Content-Type": "application/json", Authorization: "Bearer xxx" },
  body: JSON.stringify({ query: "什么是 RAG？" }),
  onmessage(msg) {
    console.log(msg.data); // 每个 SSE 事件触发一次
  },
  onerror(err) {
    console.error("SSE 异常", err);
  },
});

API 跟 EventSource 几乎一样：onmessage / onerror / onopen——但底层是 fetch，兼容性和灵活性兼得

评估模型回复的质量​

RAG 跑起来之后，怎么知道它"答得好不好"？靠人工抽查显然不靠谱。这里用了一个比较简单的评估流程：

一、先建立一份「黄金评测集」

说白了就是测试用例集。人工标注一批「问题-标准答案」对，覆盖知识库核心知识点。

• 规模：起步 50~100 条就够
• 来源：从真实用户问题里筛，再人工写标准答案
• 维护：知识库更新时同步补几条（其实可以让ai协助写，跟写测试用例一样，要不然很累）

评测集不是一次性产物，是跟着系统一起迭代的"活文档"。

二、给每个回答打个分

两种打法：

• 传统指标：准确率（回答对不对）、召回率（覆盖了多少关键点）、命中率（召回到的 chunks 有多少相关的）
• 省事打法：直接让 LLM 当裁判 A/B 评分——准备一份评分 prompt，让 LLM 对比「标准答案」和「模型回答」打分

LLM 当裁判简单粗暴，但要注意 prompt 写得清楚，否则它会"一团和气"全给高分。

三、把所有问答日志落库，方便回头查

// 每次问答时记录
{
  query: "怎么部署 RAG？",
  retrievedChunks: ["chunk-1", "chunk-2", "chunk-3"],
  finalAnswer: "...",
  timestamp: Date.now(),
  userId: "xxx",
  feedback: "good" | "bad" | null,  // 用户反馈
}

落库之后能干嘛：

• 抽样核对：每周抽 20 条人工看一眼质量
• 找「召回失败」案例：query 没匹配到正确 chunk 的反例
• 反哺优化：高频 bad case 优先修复

四、指标差了怎么调优——先定位再下手

判断标准很简单：

黄金思路：召回问题比生成问题更常见，先看召回。90% 的"AI 答错"案例其实是"压根没找到对的资料"。

经验法则：

• 别追求一步到位：先跑通 50 条评测集，看大致分数
• 优先修召回：换个 embedding 模型、Top-K 调大，可能比改 prompt 效果明显
• 引入「用户反馈」按钮：👍👎 标记，真实反馈比评测集更准
• 指标好 ≠ 体验好：技术指标只是参考，最终还是要看用户用得爽不爽

token消耗把控​

目前只是通过 prompt 限制输入输出 token，以及增量灌库减少向量模型token的消耗

如果后面要用上高级模型的话，肯定得做好更精细的把控，毕竟真的贵 ~

自动化部署​

最近 vibe coding 了几个 web app，基本都走的这个流程，还挺方便的

• github actions，编写 workflow deploy 脚本，push 代码后自动触发镜像构建
• docker镜像部署，写 dockerfile，上传docker hub
• 个人服务器从docker hub拉取镜像部署

注意：要先在github设置secrets，以及在服务器相关项目目录里设置环境变量，避免明文存储敏感信息

最后​

这只是一个简单的RAG服务，仅仅是向量召回和模型总结回复，后面其实还能做混合检索（可叠加 BM25 关键词）和 advanced RAG，甚至还能接入function calling，从「问答」走向「任务执行」，比如让其检索其他数据库的内容或者联网搜索外部资源（如果知识库欠缺的内容，可以靠联网搜索来弥补）