LobeHub 前端 E2E 中 LLM Mock 的实现原理与 SSE 流式响应模拟实战

导读

LobeHub 的端到端（E2E）测试基于 Cucumber（BDD）与 Playwright 构建，整套测试链路依赖真实对话能力：从发送用户消息到渲染助手流式回复。而 E2E 场景显然不能在 CI 中触发真实 LLM API 调用。为此项目实现了 LLM Mock 框架，在浏览器侧拦截 /webapi/chat/ 请求并回放符合 LobeHub 流式协议的 SSE 响应。本文以 e2e/docs/llm-mock.md 为骨架，结合 e2e/src/mocks/llm/index.ts 与 e2e/src/mocks/llm/index.test.ts 源码，讲解其拦截原理、SSE 格式、配置项与实战用法，读完你可以在自己的 LobeHub E2E 场景中熟练注册、定制与重置 LLM Mock。

一、为什么 E2E 需要 LLM Mock

真实 LLM API 存在成本、网络不稳定、响应时序不可控等问题，无法作为测试断言的基础。LLM Mock 的价值在于：

• 确定性：对同一条用户消息永远返回同样的内容，便于精确断言；
• 低成本：测试不产生任何 API 计费；
• 可模拟极端时序：通过配置 responseDelay/streamDelay，可以模拟慢速流式、中途打断、停止生成等真实交互中难以稳定复现的行为；
• 可支撑回归验证：长文本、滚动吸附、冷启动等场景需要固定长度、固定节奏的响应内容。

在 LobeHub 中，E2E mock 框架分两层：e2e/src/mocks/index.ts 提供面向 tRPC/REST 的通用 MockManager，而 e2e/src/mocks/llm/index.ts 单独提供了面向流式对话的 LLMMockManager 与全局单例 llmMockManager。

二、核心原理：拦截 /webapi/chat/ 请求

1. 从文档描述到源码实现

原文档将核心原理概括为一句话：通过 Playwright 的 page.route() 拦截对 /webapi/chat/openai 的请求并返回预设的 SSE 流式响应。源码把这一描述落实得更精确、更通用——拦截前缀是 /webapi/chat/（对所有 provider 生效）：

// e2e/src/mocks/llm/index.ts 中注入页面脚本的关键判定
if (!url.pathname.startsWith('/webapi/chat/')) {
  return originalFetch(input, init); // 非聊天接口直接放行原始 fetch
}

对应的注释也写明了设计目标：

Intercepts /webapi/chat/[provider] requests and returns mock SSE responses. This allows E2E tests to run without real LLM API calls.

2. 关键细节：为什么不用 page.route().fulfill() 而用浏览器内 fetch 拦截

这是理解本实现最重要的一处源码事实：LLMMockManager.setup() 中没有依赖 page.route() 的 fulfill，而是绕开了它。源码注释给出了明确理由：

Playwright's route.fulfill buffers the entire body, so it cannot model token-by-token SSE delivery. Install a browser-side fetch interceptor that returns a real ReadableStream and honors the configured delays.

即 route.fulfill 会把整个响应体一次性缓冲，无法真实地逐 token（逐 chunk）推送 SSE，也就无法测试流式渲染过程。因此实现的链路分两步：

第一步：暴露 Node 侧绑定（page.exposeFunction）

await page.exposeFunction(LLM_MOCK_BINDING, (requestBody: string) =>
  this.createStreamPlan(requestBody),
);

其中 LLM_MOCK_BINDING = '__lobehubE2ELLMMock'。这一步把 Node 侧的单例逻辑（读请求体、按用户消息选响应、切分 SSE chunks）以函数形式暴露给浏览器页面。

第二步：注入浏览器端 fetch 拦截器（page.addInitScript）

addInitScript 在页面任何脚本运行前执行，通过替换 window.fetch 实现拦截，返回真实 ReadableStream 作为响应体：

// 关键结构（源码为字符串脚本）
window.fetch = async (input, init) => {
  // ...构造 request 并检查 url.pathname.startsWith('/webapi/chat/')
  const plan = await createStreamPlan(await request.clone().text()); // 调用浏览器绑定
  await wait(plan.responseDelay, request.signal);                     // 首包前延迟
  const body = new ReadableStream({
    start(controller) {
      for (const chunk of plan.chunks) {
        controller.enqueue(encoder.encode(chunk)); // 逐 chunk 写入
        await wait(plan.streamDelay, request.signal);
      }
      controller.close();
    },
  });
  return new Response(body, {
    status: 200,
    headers: {
      'Cache-Control': 'no-cache',
      'Content-Type': 'text/event-stream',
    },
  });
};

这里有两个值得留意的实现事实：

• 真实支持流式中断：ReadableStream 的 cancel() 与 request.signal 的 abort 监听，配合 wait(delay, signal) 辅助函数，可以在用户停止生成时立即中止后续 chunk 写入——这正是“停止生成”类测试能测通的原因；
• wait 实现考虑到了零延迟与中止：delay <= 0 时直接 resolve，abort 时以 AbortError reject，并用 AbortController 的 once 监听避免泄漏；
• 拦截器用原始 JavaScript 字符串注入而非 page.addInitScript(fn) 的函数形式，源码注释说明原因是 tsx 会给具名函数装饰模块级辅助函数，而该辅助函数在页面环境并不存在；
• 拦截器带有幂等保护：if (window.__lobehubE2ELLMFetchInstalled) return;，重复调用 setup() 不会重复注入。

第三步：匹配响应内容

createStreamPlan(requestBody) 解析请求体中的 messages，取最后一条 role === 'user' 消息作为匹配键，通过 getResponse() 命中自定义响应或回退到 defaultResponse，再由 buildSSEChunks() 生成 chunk 序列，最终返回 { chunks, responseDelay, streamDelay } 流式计划。每次拦截还会打印调试日志（🤖 LLM Request intercepted / ✅ LLM Response streaming），便于在 Cucumber 输出中定位问题。

三、SSE 响应格式：必须严格匹配 LobeHub 流式协议

1. 事件序列

LobeHub 使用特定的 SSE 事件格式，Mock 必须逐字段严格匹配，前端解析器才能正确消费。规范的事件顺序如下（与文档一致）：

// 1. 初始 data 事件（携带空 content 的消息元数据）
id: msg_xxx
event: data
data: {"id":"msg_xxx","model":"gpt-4o-mini","role":"assistant","type":"message","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}

// 2. 文本内容分块（text 事件，一条消息按 chunk 拆成多条）
id: msg_xxx
event: text
data: "Hello"

id: msg_xxx
event: text
data: "! I am"

// 3. 停止事件
id: msg_xxx
event: stop
data: "end_turn"

// 4. 使用量统计
id: msg_xxx
event: usage
data: {"totalTokens":100,...}

// 5. 最终停止
id: msg_xxx
event: stop
data: "message_stop"

2. 源码中的 SSE Builder

上述序列由 e2e/src/mocks/llm/index.ts 的 buildSSEChunks(content, chunkSize) 逐段构造（该函数被导出，可被单元测试直接引用）：

• data 事件：初始消息对象带 content: []、id、model: 'gpt-4o-mini'、role: 'assistant'、type: 'message'、空 stop_reason/stop_sequence 与初始 usage；
• text 事件：for (let i = 0; i < content.length; i += chunkSize) 按 chunkSize 切分内容，每个分片通过 JSON.stringify(chunk) 编码后作为 data（必须是 JSON 字符串字面量，因为前端 JSON.parse 后得到纯文本）；
• stop（end_turn）：结束本回合文本；
• usage 事件：回传完整使用量对象，字段包括 cost、inputCacheMissTokens、inputCachedTokens、totalInputTokens、totalOutputTokens、totalTokens，其中 totalOutputTokens 按 Math.ceil(content.length / 4) 估算；
• stop（message_stop）：消息级结束标志。

每个 chunk 都以 \n\n 空行结尾，这是 SSE 的事件分隔符；id 恒定为同一 msg_mock_${Date.now()}，保证整条消息在协议层自洽。

3. 流式分片参数

分片节奏由 LLMMockConfig 控制，该接口定义如下：

export interface LLMMockConfig {
  /** 未命中任何自定义响应时的兜底回复 */
  defaultResponse: string;
  /** 是否启用 LLM mock */
  enabled: boolean;
  /** 首包前的网络延迟模拟（毫秒） */
  responseDelay: number;
  /** 每条 text 事件承载的字符数（chunk 粒度） */
  streamChunkSize: number;
  /** 相邻 chunk 之间的推送间隔（毫秒），用于模拟逐 token 到达 */
  streamDelay: number;
}

默认配置如下：

const defaultConfig: LLMMockConfig = {
  defaultResponse: 'Hello! I am a mock AI assistant. How can I help you today?',
  enabled: true,
  responseDelay: 100,
  streamChunkSize: 10,
  streamDelay: 20,
};

• streamChunkSize: 10 意味着每条回复约每 10 个字符产生一条 text 事件；
• streamDelay: 20 使消息以接近真实流式的速度渲染，能验证“打字机”效果、滚动跟随等 UI 行为。

四、用法：在测试步骤中接入 Mock

1. 最小可运行示例

按原文档，接入只需两步——先注册响应，再在当前页面上启用：

import { llmMockManager, presetResponses } from '../../mocks/llm';

// 在测试步骤中设置 mock
llmMockManager.setResponse('hello', presetResponses.greeting);
await llmMockManager.setup(this.page);

这里的 import 路径对应于 e2e/src/mocks/llm/index.ts（从仓库根目录看即 e2e/src/mocks/llm）。在仓库真实代码中，该用法出现在对话步骤定义里，且 mock 注册发生在页面跳转之前：

// e2e/src/steps/agent/conversation.steps.ts
llmMockManager.setResponse('hello', presetResponses.greeting);
await llmMockManager.setup(this.page);
// setup 之后再导航，保证拦截器先于页面脚本生效
await this.page.goto('/agent/inbox', { waitUntil: 'domcontentloaded' });

2. setup() 的调用时机

从实现看，setup() 做两件事：exposeFunction 暴露绑定 + addInitScript 注入 fetch 拦截器。由于 addInitScript 在每次导航前都会重新执行，因此它天然覆盖了 page.goto() 之后所有页面资源请求。只要在业务页面脚本加载前调用过一次 setup()，后续 SPA 内部路由跳转产生的 /webapi/chat/ 请求也会被拦截。

3. 自定义响应

原文档给出了设置与清空自定义响应的示例，源码把语义定义得更细：

// 精确匹配：为某条用户消息设置响应（大小写不敏感，会 trim）
llmMockManager.setResponse('你好', '你好！我是 Lobe AI，有什么可以帮助你的？');

// 片段匹配：请求的最后一条 user 消息只要包含该片段即命中。
// 适用于系统提示词包裹了原始用户内容的场景
llmMockManager.setResponseContaining('测试对话内容', '测试对话');

// 清除所有自定义响应（精确 + 片段同时清空）
llmMockManager.clearResponses();

匹配算法在 getResponse(messages) 中：取最后一条 user 消息，先查 customResponses（精确匹配），未命中再遍历 customResponseFragments（includes 子串匹配），全部未命中则回退到 config.defaultResponse。精确匹配的 key 统一做 toLowerCase().trim() 归一化，因此 setResponse('Hello', ...) 对用户发送 ' hello ' 同样有效。

仓库真实用法示例见 e2e/src/steps/agent/conversation-mgmt.steps.ts：

llmMockManager.setResponseContaining('测试对话内容', '测试对话');

这正是片段匹配的典型场景——用户消息里除原始内容外还带有额外的格式文本，无法精确匹配。

4. 运行时配置调整与重置

除 setResponse 之外，管理器还提供两类配置方法：

// 合并覆盖部分配置：用于模拟慢流、快流等特殊时序
llmMockManager.setConfig({
  responseDelay: 0,       // 首包零延迟
  streamChunkSize: 1024,  // 每个 text 事件携带 1024 字符（近似一次性输出）
  streamDelay: 0,         // chunk 间零间隔
});

// 恢复出厂默认：应在 After 钩子中调用，防止本场景的时序覆盖泄漏到下一个场景
llmMockManager.resetConfig();

此外还有 enable() / disable() 两个开关，disable() 后 setup() 直接短路并打印 🔇 LLM mocks disabled。

五、内置预设响应

presetResponses 为常见断言场景提供了开箱即用的回复素材（源码）：

预设键	用途
greeting	问候回复，用于基础对话流程断言
codeHelp	代码帮助话术
error	错误兜底话术
nameIntro / nameRecall	多轮对话场景：前者表示“记住名字”，后者表示“回忆名字”，两者搭配可验证多轮记忆链路
regenerated	用于“重新生成回复”测试，与首次回复内容可区分
longArticle	长中文文章，用于“停止生成”测试——响应足够长，保证中途停止可被观测
longScrollArticle	由 60 段（30 段短文本 + 30 段长段落）组成的超长回复，确保内容高度必然超过视口高度，用于滚动吸附类场景

六、真实场景串讲：从 Step 定义到 Feature

下面结合仓库 feature 文件与 step 定义，看不同测试目标如何使用 LLM Mock。

1. 滚动吸附类测试（慢流 + 超长内容）

e2e/src/steps/agent/scroll.steps.ts 是 mock 使用最深入的场景之一。它的 sendPrompt 辅助函数在每次发消息前调用 llmMockManager.setResponse(prompt, response)，让“发送什么就回复什么”；随后用 expect.poll 等待用户消息从乐观 tmp_ id 重挂载为持久化 id，再校验助手的滚动行为。

在模拟“回复生成过程中用户继续操作”的场景时，它刻意调慢了流式：

llmMockManager.setConfig({ responseDelay: 4000, streamChunkSize: 40, streamDelay: 25 });
await sendPrompt(this, prompt, presetResponses.longScrollArticle);

responseDelay: 4000 让首包晚到 4 秒，从而在回复生成过程中穿插滚动、发送第二条消息等操作；测试结束处调用 llmMockManager.resetConfig() 恢复默认。驱动这些步骤的 feature 见 agent-scroll.feature（源码注释中标注其服务于 @AGENT-SCROLL-* 系列场景）。

2. Home 冷启动测试（零延迟快响应）

e2e/src/steps/home/chat-input.steps.ts 中，为了在冷启动场景快速拿到回复，它先 clearResponses() 清空残留，再用 setConfig 将响应压到零延迟、单 chunk（streamChunkSize: 1024 对短回复而言几乎一次性输出）：

llmMockManager.clearResponses();
llmMockManager.setConfig({
  responseDelay: 0,
  streamChunkSize: 1024,
  streamDelay: 0,
});
llmMockManager.setResponse('cold route home message', 'cold route response');
await llmMockManager.setup(this.page);

3. 多轮对话与消息操作

对话管理类场景（conversation-mgmt.steps.ts）依赖“记住用户名字”的预设：第一轮用 nameIntro 声明已记住，第二轮用 nameRecall 复述名字，从而在完全离线的 mock 环境下验证多轮上下文的保持。

七、单元测试：验证 SSE 生成的正确性

LLM Mock 不只有 E2E 使用，还配有 vitest 单元测试 e2e/src/mocks/llm/index.test.ts，从协议层兜底。它复用了 LobeHub 前端真实的流解析器——packages/utils/src/client/fetchEventSource/parse 中的 getLines/getMessages——把 buildSSEChunks 的输出解析回事件流：

• 跨网络边界测试：测试故意不把整个响应一次喂给解析器，而是按 7 字节切片逐段调用 onChunk，以验证 chunk 被任意 TCP 分片时仍可正确解析；
• JSON 敏感文本往返测试：对 '第一行\n第二行\r\n"quoted" \\ path 😀' 这类含换行、引号、反斜杠与 emoji 的内容，分别用 chunkSize = 1, 3, 8, 10, 64 验证，断言所有事件的 event 非空、data 可被 JSON.parse，且将所有 text 事件拼接后与原文完全一致；
• 多行滚动夹具测试：验证 presetResponses.longScrollArticle 以 chunkSize = 10 往返后内容无损——这是滚动类 E2E 场景可靠性的前提。

这些断言说明一个事实：text 事件的 data 必须是 JSON 编码的字符串字面量，因为测试正是通过 JSON.parse(data) 还原文本再拼接比对。写自定义响应时无需手动处理这些细节——buildSSEChunks 已统一处理，但理解这一点有助于排查“收到内容乱码/被截断”类问题。

八、常见问题排查

1. LLM Mock 未生效

原因：fetch 拦截器注册晚于页面业务脚本加载；或页面在 setup() 之前就已发起请求。

解决：确保在导航（page.goto）之前调用 await llmMockManager.setup(page)。由于注入走 addInitScript，只要在首次导航前 setup，之后 SPA 内的所有路由请求都会被覆盖。对应写法可参考 e2e/src/steps/agent/conversation.steps.ts。

2. 响应命中了兜底文案而非自定义内容

原因：用户消息与 setResponse 的 key 不一致，或请求体中没有 role === 'user' 的消息。匹配算法取最后一条 user 消息，且做 toLowerCase().trim() 归一化。

解决：改用片段匹配 setResponseContaining(fragment, response)，或先观察拦截日志（🤖 LLM Request intercepted (N messages)）确认实际请求体。若你的场景是系统提示词把用户内容包裹在中间，片段匹配是更稳的选择。

3. 测试之间的 mock 状态互相污染

原因：setResponse 与 setConfig 修改的都是全局单例 llmMockManager 的状态。

解决：在 Cucumber 的 After 钩子中调用 resetConfig() 与 clearResponses()。仓库在 scroll、chat-input 等步骤中都遵循了这一约定（见 scroll.steps.ts 的 llmMockManager.resetConfig()）。

4. 想测“停止生成/中断流式”却总是瞬间结束

原因：默认 streamChunkSize: 10 + streamDelay: 20 对短文本来说结束太快，来不及点“停止”。

解决：使用 setConfig({ responseDelay, streamChunkSize: 40, streamDelay: 25 }) 拉长输出窗口，并配合 presetResponses.longArticle/longScrollArticle 这类超长回复，为 UI 操作留出充足时间窗。

九、总结

LobeHub 的 LLM Mock 是一个“协议精确 + 时序可控 + 全程离线”的流式响应模拟方案：拦截维度上，用浏览器端 fetch 覆写替代 route.fulfill，从而获得真正的逐 chunk SSE 推送与中止能力；协议维度上，严格复刻 data → text × N → stop(end_turn) → usage → stop(message_stop) 五段事件序列；使用维度上，通过全局单例提供精确匹配、片段匹配、运行时配置与钩子重置等能力。配合 e2e/src/mocks/llm/index.test.ts 的协议级单测，保证了 mock 输出与前端真实流解析器始终兼容。掌握这套框架后，你既可以在现有 feature 基础上扩充断言，也可以为新对话场景快速搭建完全离线的、可复现的 LLM 交互测试。