数据截至 (上游 commit 0ab3414c015b)
主线:一条聊天消息的一生
30 秒导读: 你在 LibreChat 里对一个 agent 发一句话,浏览器打的是
POST /api/agents/chat。 这一章端到端追这条请求:先过一串中间件(脱敏、审核、鉴权、装配),再进控制器主函数。 最关键的一点:控制器不会把回答顺着这条 HTTP 连接吐回去——它先建一个后台"生成作业", 立刻回一个streamId就结束 POST;前端再用另一条GET长连接(SSE)去订阅那个作业的输出。 生成和 HTTP 连接解耦,所以刷新、断网、切标签页都能重新接上同一次回答。
本章只讲流程编排与生命周期:中间件做了哪些前置活、请求怎么被拆成两段、中断(abort)怎么管、
消息和文件怎么被拼起来、标题什么时候生成。至于 provider 差异见 02,
AgentClient 内部与流式细节见 03,工具/MCP 见 04。
1. 先看全景: 一次对话其实是"两段式"
老式聊天后端是一段式:POST 进来 → 一边生成一边往这条连接里流 → 连接断了这次回答就没了。
LibreChat 现在走的是可续(resumable)模型,把一次对话拆成两条独立的 HTTP 请求:
前端 后端
│
│ ① POST /api/agents/chat (带 text / agent_id / conversationId)
├─────────────────────────────────► 过中间件 → 控制器
│ 建后台生成作业(job)
│ ◄───────────────────────────────── 立刻回 { streamId, conversationId, status:"started" }
│ (POST 到此结束,连接正常关闭——这不算 abort)
│
│ ┌─ 后台:initializeClient → sendMessage → 落库 ─┐
│ │ (脱离任何 HTTP 连接,自己跑) │
│ ② GET /api/agents/chat/stream/:streamId (SSE 长连接) │
├─────────────────────────────────► 订阅这个 job │
│ ◄═══ event: message / final / title ══════ 作业把 chunk 推给所有订阅者 ◄──────────┘
│
│ (想停就再打) ③ POST /api/agents/chat/abort { streamId }
├─────────────────────────────────► GenerationJobManager.abortJob()
怎么读这张图: 左边是浏览器,右边是后端。① 和 ② 是两条不同的 HTTP 请求,③ 是可选的停止请求。
三者共享一个身份证:streamId,而它恒等于 conversationId(会话 ID)。
这个设计带来两个直接后果,后面每一步都是围着它转的:
- POST 秒回、不等生成。 控制器在生成开始前就
sendGenerationJson(res, 200, { status: 'started' })回话了(request.js:1178-1185),因为工具加载 (尤其 MCP OAuth)可能要几秒,得让前端赶紧连上 SSE 才不漏事件。 - "连接关闭"不再等于"用户想停"。 一段式后端靠
res.on('close')判断用户中断;可续模型里 POST 本来就会正常关闭,所以中断改由专门的POST /chat/abort触发(index.js:223的abortJob)。
这一章主要讲第 ① 段(POST 控制器)的生命周期;② 的 SSE 回放/重连细节属于流式,归 03。
2. 第一段:POST 进来,先过中间件流水线
路由挂载在 api/server/routes/agents/index.js。真正处理聊天的 chat.js 被挂在 /chat 下,
而且排在 SSE 订阅、abort、状态查询这些 GET 路由后面——因为那些不需要跑 buildEndpointOption
这种重中间件,要先被截住(index.js:333-345)。
一条 POST /api/agents/chat 命中 chat.js 后,会按顺序穿过下面这串 router.use
(chat.js:72-78)。顺序是有讲究的:先便宜的过滤/审核,再花钱的鉴权和装配。
| 顺序 | 中间件 | 干什么 | 源码 |
|---|---|---|---|
| 1 | createMessageFilterPii | 按配置扫描用户输入里的 PII(个人身份信息),命中就按策略脱敏/拦截 | packages/api/src/middleware/messageFilterPii.ts:createMessageFilterPii |
| 2 | moderateText | 若开了 OPENAI_MODERATION,把文本发去审核接口,被 flag 就 denyRequest 拒掉 | moderateText.js:12 |
| 3 | checkAgentAccess | 角色级权限:这个用户有没有 AGENTS.USE 权限 | access.ts:generateCheckAccess(chat.js:28) |
| 4 | checkAgentResourceAccess | 资源级权限:这个用户能不能 VIEW 这个具体的 agent_id | canAccessAgentFromBody.js:149 |
| 5 | validateConvoAccess | 这个 conversationId 是不是属于该用户(防越权读别人会话) | validate/convoAccess.js:32 |
| 6 | buildEndpointOption | 解析请求体、套用 modelSpec 预设,产出 endpointOption(含 agent 的懒加载 Promise) | buildEndpointOption.js:28 |
几个不显然但重要的点,分开说:
脱敏和审核为什么要扫"合并后的字符串"。 用户可能把一句话拆进"引用块"和"正文",单看每段都干净,
拼起来才是敏感内容。所以两个中间件都不是只扫 req.body.text,而是先用 getReferencedQuotes 归一化引用,
再把 mergeQuotedText(text, quotes) 合并串也一起扫——和 AgentClient 最终喂给模型的那份逐字一致
(moderateText.js:52-55、messageFilterPii.ts 同注释)。
两级鉴权是分工的,不是重复。 checkAgentAccess 管"你这个人能不能用 agent 功能"(角色权限);
checkAgentResourceAccess 管"你能不能看这个特定 agent"(资源 ACL)。临时(ephemeral)agent 没有资源
ACL,canAccessAgentFromBody 直接放行到下一层(canAccessAgentFromBody.js:176-178)。
buildEndpointOption 是装配的关键一步,但它不加载 agent。 对 agents 端点,它调 agents.buildOptions
(buildEndpointOption.js:23、build.js:buildOptions)。注意 agent 字段塞进去的是一个 Promise
(build.js:11 的 loadAgent(...)),没有 await——真正把 agent 读出来是后面控制器里 initializeClient
的活(见 §4)。这样中间件层不为一次可能被并发闸门挡掉的请求白白读库。
// build.js —— agent 是懒加载 Promise,装配期不 await(示意,非源码)
const agentPromise = loadAgent({ req, spec, agent_id, endpoint, model_parameters })
.catch(() => undefined); // 读不到不抛,留给控制器判空
return removeNullishValues({ endpoint, agent_id, model_parameters, agent: agentPromise });
过完这 6 层,req.body.endpointOption 就绪,请求交给控制器(chat.js:80-82 的 controller)。
3. 第二段:控制器主函数 ResumableAgentController
转发壳 AgentController 已被移除,request.js 直接定义并导出主控制器 ResumableAgentController
(request.js:332、module.exports 在 :2020)。这个函数就是本章的心脏,把它按时间切成 7 个阶段来看:
阶段① 并发闸门 + 生成 conversationId/streamId
│ checkAndIncrementPendingRequest / crypto.randomUUID
▼
阶段② 建后台作业 job + 立刻 res.json({streamId}) ← POST 在这里"回话"
│ GenerationJobManager.createJob
▼
阶段③ 挂"全员离场"监听(断连保存半成品)
▼
阶段④ initializeClient —— 装 agent、加载工具、造 AgentClient
│ (可能耗时;abort 信号来自 job.abortController)
▼
阶段⑤ client.sendMessage —— 真正跑模型(内部细节见 03)
│ 并行:若够格,addTitle immediate 已同时起跑
▼
阶段⑥ 落库:先存 user 消息,再存 response 消息,再 emitDone
│ 顺序是硬要求——防止前端追问时 parentMessageId 还没落库
▼
阶段⑦ 完成/中断/替换 三种收尾 + disposeClient
下面逐阶段拆。
3.1 阶段①:并发闸门与 ID 分配
先防刷:checkAndIncrementPendingRequest(userId) 原子地给该用户的"在途请求数"加一,超限就
返回 429 并记一次 CONCURRENT violation(request.js:976-989;定时任务自动触发的请求经
exemptFromConcurrencyLimiter 豁免此闸)。Redis 在场时走一段 Lua
脚本保证原子(concurrency.ts:119-145);没 Redis 就退化成内存 get/set。
然后定 ID。前端新会话可能传空或字面量 "new",控制器把它当占位符,现场生成真 UUID;
而且约定 streamId === conversationId,后续订阅、abort、落库全靠这一个 ID 串起来
(request.js:552-555)。
还有一道前置守卫:如果用户想追问的父消息(parentMessageId)还是个没落库的"preliminary"父消息
(上一条回答还在存),直接回 409 让前端稍后重试(request.js:502-544 的 isUnpersistedPreliminaryParent)。
这正是可续模型的副作用之一——生成和落库是异步的,得防止孤儿 parentMessageId。
3.2 阶段 ②:建作业,然后 POST 立刻收工
// request.js:231-238 —— 建作业,秒回,不等生成(示意,非源码)
const job = await GenerationJobManager.createJob(streamId, userId, conversationId);
const jobCreatedAt = job.createdAt; // 记住创建时刻,用来识别"作业被替换"
req._resumableStreamId = streamId;
// 关键:立刻回 JSON,让前端赶紧连 SSE——工具加载(MCP OAuth)会先发事件
res.json({ streamId, conversationId, status: 'started' });
job 自带一个 abortController,它是整个生成期唯一的中断开关(取代老式的 res.on('close'))。
jobCreatedAt 会在后面反复用到:同一个 streamId 可能被一次更新的请求"抢占",靠比对创建时刻
就能认出"我这个作业已经是旧的了,别再往新作业上 emit"(见 3.6)。
3.3 阶段③:全员离场时保存半成品
res.json 之后 POST 连接就关了,但生成还在后台跑。要是所有 SSE 订阅者都断开了(比如用户关了页面
又没别的标签在看),不能让这半截回答凭空蒸发。所以控制器给 job 挂了个 allSubscribersLeft 监听
(request.js:1207-1275):把已聚合的内容过一遍 filterPersistableAbortContent,若有可存的,就以
unfinished: true 存一条部分回答(saveMessage)。partialResponseSaved 标志位防重复存。
3.4 阶段④:initializeClient —— 把 agent 真正装起来
到这里才 await endpointOption.agent(那个中间件层留下的 Promise),把 agent 读出来
(initialize.js:283)。这一步是装配总成,顺序大致是:
| 子步骤 | 干什么 | 源码符号 |
|---|---|---|
| 取 agent | await 懒加载的 agent Promise,判空 | initialize.js:283 |
| 校验模型 | agent 指定的模型在不在允许清单 | validateAgentModel(initialize.js:290) |
| 加载工具 | 只加载工具定义(事件驱动模式),不建完整实例 | createToolLoader → loadAgentTools(initialize.js:306、ToolService) |
| 装子 agent | 发现 handoff 连接的 agent、递归解析 subagent 图(带深度/节点上限) | discoverConnectedAgents、resolveSubagentTrees(initialize.js:425、839) |
| 造客户端 | 把上面所有东西塞进 new AgentClient(...) | initialize.js:930 |
本章不深入其中任何一步:agent/工具/子 agent 的装配细节属于 03 和
04。对生命周期而言,只需知道 initializeClient 返回 { client, userMCPAuthMap },
且它接收 job.abortController.signal——初始化期间就能被中断(request.js:1278-1339)。若初始化中途 abort,
直接 completeJob('...during initialization') 并 finishResumableRequest 收尾,不再往下走。
3.5 阶段⑤:sendMessage 跑模型(标题可能同时起跑)
拿到 client 后,先用 resolveTitleTiming 定标题时机(见 §5),再组 messageOptions 调
client.sendMessage(text, options)(request.js:1576-1620)。注意 progressOptions.res 传的是一组
假的 no-op 写方法(write: () => true 等)——因为可续模型下,真正的流出走 GenerationJobManager
往 SSE 推,而不是往这条早就关掉的 POST 连接写(request.js:1593-1600)。
sendMessage 内部怎么调模型、怎么流式,是 03 的事。生命周期只关心两个回调:
onStart(userMsg, respMsgId):模型刚开跑就把 user 消息和 responseMessageId 写进 job 元数据 (给中断/续接用),并emitChunk({ created: true, message })让订阅者先看到自己的消息回显 (request.js:1510-1574)。- 若够格,
addTitle(..., { immediate: true })在sendMessage并行起跑,不等回答完成 (request.js:1605-1618)——这是 immediate 时机的意义(§5)。
3.6 阶段⑥:落库顺序是硬约束
sendMessage 返回 response 后,收尾顺序被注释反复强调,不能乱:
1. 先存 user 消息 saveMessage(userMessage)
2. 再存 response saveMessage({ ...response })
3. 最后才 emitDone(final 事件)给订阅者
为什么?因为前端拿到 final 事件后可能立刻发追问,那条追问的 parentMessageId 指向这次的
response。如果 final 先到、response 还没落库,追问就会挂在一个数据库里不存在的父上,变成孤儿
(request.js:1854-1904 的两处 saveMessage + 注释 "CRITICAL: Save response message BEFORE emitting")。
在 emit 之前还有一道**"作业被替换"守卫**:重新取一次 job,比对 createdAt 和当初记下的 jobCreatedAt。
不等 = 这个 streamId 已被更新的请求抢占,当前这次是旧的——于是丢弃自己的标题、resolveConvoReady()
放行、finishResumableRequest 减计数,然后跳过 final emit 直接收尾(request.js:1823-1841)。
正常路径则构造 finalEvent 并 publishTerminalClaim(terminalClaim, finalEvent) 发布终局,再
finishResumableRequest(request.js:1939-1958、:1753)。 被用户中断的路径也 emit 一个带
unfinished: true 的 final(terminalWasAborted 时,request.js:1946),收尾则按 aborted 原因认领终局(:1204-1207)。
3.7 中断(abort)的完整生命周期
把散在各处的中断线索收拢成一张表——这是可续模型里最容易看晕的部分:
| 触发点 | 谁被 abort | 效果 |
|---|---|---|
| 用户点 Stop | POST /chat/abort → GenerationJobManager.abortJob(streamId) | 触发 job.abortController,并抢先存好部分回答(index.js:223、276+) |
| 初始化期中断 | job.abortController.signal(传给 initializeClient) | completeJob('...during initialization'),不进 sendMessage(request.js:1301-1339) |
| 作业被更新请求替换 | titleAbortController + titleDiscardController | 丢弃旧标题、跳过 final emit(request.js:1823-1841) |
| 标题自己的取消 | 独立的 titleAbortController / titleDiscardController | 见 §5,和 job 的中断刻意分开 |
一个关键设计:标题有自己独立的中断控制器,不跟 job.abortController 绑死。因为 completeJob 在
成功完成时也会触发 job 的 abort 信号来做清理——如果标题共用这个信号,一个只是比短回答慢一点的标题
就会被误杀(request.js:1462-1476 注释)。所以:用户 Stop 用 titleAbortController 取消在途标题;
只有被替换/失败才用 titleDiscardController 把已生成的标题也丢掉,免得旧标题盖掉新作业的会话。
清理统一走 disposeClient(client),而且要等 immediate 标题结算完再 dispose——否则会在标题还在
用 client 生成时就把它拆了(request.js:2001-2013 的 immediateTitlePromise.finally(...))。
老的一段式
_LegacyAgentController已在后续更新中从文件里整体删除,不再留作对照:它当年用res.on('close')判断中断、直接sendEvent(res, ...)往 POST 连接流回答。理解可续模型改了什么, 可对照文件里残留的注释(request.js:1191-1193明说不再用res.on('close')判中断)。
4. 消息与文件怎么被拼起来
用户传的 req.body.files 只是一串文件引用(带 file_id),不是完整文件对象。真正要挂到消息上的
"干净文件",由 buildMessageFiles 在回答落库前拼出来(request.js:1839-1845):
// utils/message.ts:buildMessageFiles —— 拿请求里的 file_id 去 attachments 里配对(示意,非源码)
const requestFileIds = new Set(requestFiles.map((f) => f.file_id)); // 请求声明的文件
const files = attachments // 初始化时解析出的附件
.filter((a) => a.file_id != null && requestFileIds.has(a.file_id)) // 只保留请求真的引用了的
.map(sanitizeFileForTransmit); // 剥掉不该外传的字段
思路很直白:以请求声明的 file_id 为准做交集,再对每个附件 sanitizeFileForTransmit 剥字段
(FILE_STRIP_FIELDS),防止把内部字段泄给前端。配对成功就挂到 userMessage.files,并删掉临时的
image_urls(request.js:1840-1844)。attachments 本身是 initializeClient 阶段解析好的,归属上下文工程
——细节见 05。
5. 标题什么时候生成:resolveTitleTiming
标题不是随回答一起来的,它是另一次(便宜的)模型调用。有两种时机,由 resolveTitleTiming 决定
(providers.ts:65):
| 时机 | 含义 | 触发点 |
|---|---|---|
immediate(默认) | 从用户第一句话就并行起跑,和回答同时生成 | request.js:1605-1618 |
final(旧行为) | 等整段回答完成后才生成 | request.js:2014-2027 |
时机的解析有优先级(providers.ts:72-113):endpoints.all.titleTiming 全局覆盖 → 否则按候选端点
依次查(公开的 agents 端点可覆盖底层 provider) → 再退到 provider/custom 配置 → 都没有就默认 immediate。
控制器拿到 client 后用 [endpointOption.endpoint, client.options.agent.endpoint] 两个候选去解析
(request.js:1348-1351)。
immediate 好处是标题早早就能推给前端,但带来一个时序难题:标题可能比会话行先生成好。而 saveConvo
用 noUpsert: true——会话行不存在时它是静默 no-op,标题就会丢(缓存里还有,但没落库)。解决办法是一个门闩:
addTitle(immediate) ──生成好标题──► 先写缓存 + onTitleGenerated 推给前端(live UI 立刻显示)
│
▼ await convoReady ← 卡在这里
控制器落库完 response(会话行此时才存在) ── resolveConvoReady() ──► 放行
│
▼
saveConvo(noUpsert) 这下能命中行了
convoReady 由控制器在会话确实落库后 resolveConvoReady()(request.js:1916-1918),addTitle 里
await convoReady 等它(title.js:147-149)。此外 title.js 还有一层 discardSignal:若这条流已被更新的
运行取代(或本轮失败),即使标题已生成也要丢掉,免得旧标题盖掉新作业现在拥有的会话——而且只在缓存里
仍是自己这份标题时才删缓存,防止误删替换流已写好的新标题(title.js:151-164)。标题模型调用本身还套了
45 秒超时(title.js:74-75)。
6. 边界与容易踩的点
- POST 正常关闭 ≠ 中断。 可续模型下别再指望
res.on('close')判断用户停止——那是旧路径的做法。 中断只认POST /chat/abort(request.js:1191-1193注释明确说明)。 streamId恒等于conversationId。 全流程用它当唯一键。新会话的"new"是占位符,会被换成真 UUID (request.js:552-555);abort 端点也要特意跳过"new"、必要时按用户查活跃作业兜底(index.js:234-251)。- 落库顺序不能省。 user → response → emit final,是防孤儿
parentMessageId的硬约束,不是风格问题。 - 标题的中断和 job 的中断是两套。 混用会导致"慢一点的标题被成功完成的清理信号误杀"或"旧标题盖掉新会话"。
- 鉴权是两级 + 会话归属校验三道。 少任何一道都可能越权;临时 agent 走的是放行分支,别以为它没被检查。
7. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 路由 + 中间件链 | api/server/routes/agents/chat.js | checkAgentAccess、checkAgentResourceAccess、router.post('/') |
| 路由挂载 + SSE 订阅 + abort | api/server/routes/agents/index.js | GET /chat/stream/:streamId、POST /chat/abort、chatRouter |
| 控制器主函数(可续) | api/server/controllers/agents/request.js | ResumableAgentController、AgentController |
| 旧一段式控制器(对照) | api/server/controllers/agents/request.js | _LegacyAgentController |
| 断连保存 / 中断收尾 | api/server/controllers/agents/request.js | allSubscribersLeft 监听、finishResumableRequest、createCloseHandler |
| 客户端装配 | api/server/services/Endpoints/agents/initialize.js | initializeClient、validateAgentModel、createToolLoader |
| endpointOption 装配 | api/server/services/Endpoints/agents/build.js | buildOptions(agent 懒加载 Promise) |
| endpoint 装配中间件 | api/server/middleware/buildEndpointOption.js | buildEndpointOption |
| 内容审核 / PII | api/server/middleware/moderateText.js、packages/api/src/middleware/messageFilterPii.ts | moderateText、createMessageFilterPii |
| 角色 / 资源鉴权 | packages/api/src/middleware/access.ts、api/server/middleware/accessResources/canAccessAgentFromBody.js | generateCheckAccess、canAccessAgentFromBody |
| 会话归属校验 | api/server/middleware/validate/convoAccess.js | validateConvoAccess |
| 并发闸门 | packages/api/src/middleware/concurrency.ts | checkAndIncrementPendingRequest、decrementPendingRequest |
| 消息文件组装 | packages/api/src/utils/message.ts | buildMessageFiles |
| 标题时机解析 | packages/api/src/endpoints/config/providers.ts | resolveTitleTiming |
| 标题生成 + 门闩 | api/server/services/Endpoints/agents/title.js | addTitle(convoReady / discardSignal) |