数据截至 (上游 commit 7803d562546a)
一条消息的旅程：从平台事件到进入流水线
30 秒导读： 一条 QQ / 微信 / Discord 消息到达 LangBot 后，并不会立刻交给 LLM。它要先被路由到某条流水线（或被丢弃），可能被防抖聚合成一条，然后进请求池排队，最后由总控制器在两层信号量（全局并发上限 + 每会话串行）的约束下选中、异步跑起来。本章端到端追这条路径，讲清"消息真正进入流水线处理之前"发生的全部调度。
本章只讲入口到流水线门口这一段。流水线内部各阶段、以及本地 Agent 循环怎么调 LLM，分别在 02-pipeline-engine.md 和 03-local-agent-loop.md。先看 index.md 建立全局。
1. 这段路径是干嘛的（零基础也能懂）
问题场景。 你的机器人同时挂在好几个平台上 ，好几个群、几十个人都在给它发消息。这些消息同时涌进来。系统必须回答三个问题，才谈得上"处理"：
- 这条消息该走哪条流水线？（同一个机器人可以配多条流水线，按发信人 / 群 / 内容分流）
- 用户"啪啪啪"连发了 5 条短消息，要当1 次处理还是 5 次？
- 一堆消息一起来，能让它们无限并发吗？同一个人的两条消息能不能同时处理？
这一段就是回答这三个问题的调度层。 它夹在"平台适配器收到消息"和"流水线开始跑"之间，是纯粹的调度与并发逻辑，不碰任何 LLM。
一句话直觉。 把它想成餐厅的前厅：适配器是门口迎宾（把各平台的方言翻译成统一事件），路由是领位（决定你坐哪桌 = 哪条流水线），聚合是"同桌客人的连续点单先并单"，请求池是取餐排队区，控制器是限流的传菜口——总共只有 N 个传菜员（全局并发），且同一桌一次只上一道菜（每会话串行）。
2. 顶层全景（一条消息怎么流下来）
怎么读这张图： 从上到下是一条消息的时间顺序；每个方框是一个部件，右侧注明它在哪个文件。菱形是决策点。
平台(QQ/微信/Discord…) 原生事件
│
▼
┌───────────────────────────────┐
│ RuntimeBot 监听回调 │ src/langbot/pkg/platform/botmgr.py
│ on_friend/on_group_message │ (适配器把方言翻成统一 MessageEvent)
└───────────────────────────────┘
│
▼
◇ webhook 要跳过吗? ─── 是 ──▶ 丢弃, 不入池
│否
▼
┌───────────────────────────────┐
│ resolve_pipeline_uuid 路由 │ 按序匹配规则, 首命中即用
└───────────────────────────────┘
│
◇ 命中 __discard__ ? ── 是 ──▶ 记账后静默丢弃
│否
▼
┌───────────────────────────────┐
│ MessageAggregator 防抖聚合 │ src/langbot/pkg/pipeline/aggregator.py
│ (可关; 开则并单) │
└───────────────────────────────┘
│ (延时到 / 满10条 / 未开启)
▼
┌───────────────────────────────┐
│ QueryPool.add_query 造 Query │ src/langbot/pkg/pipeline/pool.py
│ 入 queries[], notify_all │
└───────────────────────────────┘
│ condition 唤醒
▼
┌───────────────────────────────┐
│ Controller.consumer 调度循环 │ src/langbot/pkg/pipeline/controller.py
│ 选一个会话空闲的 query │
│ 双层信号量: 全局 + 每会话 │
└───────────────────────────────┘
│ task_mgr.create_task
▼
pipeline.run(query) ── 进入 02 章
各部件一句话职责：
| 部件 | 干什么 | 在哪 |
|---|---|---|
RuntimeBot | 一个机器人实例；注册平台事件监听、跑适配器生命周期 | src/langbot/pkg/platform/botmgr.py:35 |
resolve_pipeline_uuid | 按路由规则把消息分流到某条流水线，或标记丢弃 | botmgr.py:122 |
MessageAggregator | 防抖：把用户连发的多条消息并成一条 | src/langbot/pkg/pipeline/aggregator.py:64 |
QueryPool | 请求池：造 Query 对象、缓存供插件回调、通知调度器 | src/langbot/pkg/pipeline/pool.py:114 |
Controller | 调度循环：双层信号量下选中 query 并异步跑流水线 | src/langbot/pkg/pipeline/controller.py:14 |
3. 第一关：平台适配层与路由（RuntimeBot）
3.1 适配器把"平台方言"翻成统一事件
每个平台（QQ、Discord、企业微信……）都有自己的消息格式。LangBot 用适配器把它们统一成 FriendMessage / GroupMessage 事件。RuntimeBot 是一个机器人实例的运行时包装，initialize() 里给 适配器注册两个监听回调：
# src/langbot/pkg/platform/botmgr.py:322-323 # 真实源码
self.adapter.register_listener(platform_events.FriendMessage, on_friend_message)
self.adapter.register_listener(platform_events.GroupMessage, on_group_message)
从这里起，下游代码只跟统一的 MessageEvent / MessageChain 打交道，不再关心是哪个平台。PlatformManager.load_bots_from_db() 负责从数据库把所有机器人加载成 RuntimeBot 并各自 initialize()（botmgr.py:828）。
生命周期。 RuntimeBot.run() 把 adapter.run_async() 包一层异常捕获，交给 task_mgr.create_task 作为一个 platform-adapter 后台任务常驻（botmgr.py:490-502）；shutdown() 则 adapter.kill() 再取消该任务（botmgr.py:504）。
3.2 入站回调里发生的事（以私聊为例）
on_friend_message 收到事件后，在把消息交给下游前做三件事，顺序很重要：
- 推 webhook 并决定是否跳过。 若配了
webhook_pusher，先推送；webhook 可以返回"跳过流水线"，此时消息直接被丢掉不入池（botmgr.py:302-308）。 - 算路由。 取消息纯文本和元素类型，调
resolve_pipeline_uuid得到目标流水线（下一节）。 - 丢弃判定 / 交棒。 若路由结果是丢弃标记就记账后
return；否则调msg_aggregator.add_message把消息交给聚合层。
# src/langbot/pkg/platform/botmgr.py:230-245 # 真实源码(节选)
message_text = str(event.message_chain)
element_types = [comp.type for comp in event.message_chain]
pipeline_uuid, routed_by_rule = self.resolve_pipeline_uuid(
'person', launcher_id, message_text, element_types
)
if pipeline_uuid == self.PIPELINE_DISCARD:
await self.logger.info('Person message discarded by routing rule')
await self._record_discarded_message(...)
return
on_group_message 是同构的一份，差别只在 launcher_type 是 GROUP、launcher_id 取群 id（botmgr.py:355-423）。适配器若实现了 get_launcher_id，还能自定义会话标识（botmgr.py:314-317）。
3.3 路由规则：按序匹配，首命中即停
resolve_pipeline_uuid 遍历该机器人配置的 pipeline_routing_rules，从上到下逐条匹配，第一条命中就返回它的 pipeline_uuid；一条都不中，就回退到默认流水线 use_pipeline_uuid（botmgr.py:151-178）。
四种规则类型：
规则 type | 拿 什么去比 | 说明 |
|---|---|---|
launcher_type | 会话类型字符串 "person" / "group" | 按私聊 / 群聊分流 |
launcher_id | 会话 / 群 id | 指定某个人或某个群走特定流水线 |
message_content | 消息纯文本 | 按内容（关键词 / 正则）分流 |
message_has_element | 消息是否含某类型元素 | Image/Voice/File/At… 只支持 eq(有)/neq(没有) |
前三种规则的比较交给 _match_operator，支持六种算子：
| 算子 | 含义 |
|---|---|
eq / neq | 相等 / 不等 |
contains / not_contains | 包含 / 不包含子串 |
starts_with | 前缀匹配 |
regex | 正则搜索（正则非法时返回 False，不抛错） |
# src/langbot/pkg/platform/botmgr.py:70-75 # 真实源码
elif operator == 'regex':
try:
return bool(re.search(expected, actual))
except re.error:
return False
return False
关键细节：
- 规则里缺
type或缺pipeline_uuid的直接跳过（botmgr.py:159-160），坏配置不会中断路由。 message_has_element走独立分支，不经过_match_operator——它只认eq/neq两种（botmgr.py:171-176）。- 返回值是元组
(pipeline_uuid, routed_by_rule)，routed_by_rule标记"是否被规则命中"（vs 回退默认），这个布尔会一路带进Query（见 §6），供下游区分。
3.4 丢弃机制：__discard__
有一条特殊的目标流水线值 PIPELINE_DISCARD = '__discard__'（botmgr.py:119）。当某条路由规则把消息导向它，回调不会入池，而是静默丢弃——但会先调 _record_discarded_message 在监控系统里记一笔（状态 discarded），并保证该会话在会话监视器里存在（botmgr.py:202-264）。这样"被规则过滤掉的消息"仍然可见、可审计，而不是凭空消失。
4. 第二关：防抖聚合（MessageAggregator）
4.1 要解决的小问题
聊天里人常常连发："在吗" → "帮我看个东西" → "就是这个 bug"。若每条都独立触发一次 LLM，既浪费又答非所问。聚合层做的就是防抖（debounce）：短时间内同一会话的多条消息，等一等、并成一条再处理。
4.2 思路：延时定时器 + 上限强刷
add_message 的逻辑（aggregator.py:139）：
- 查配置。 从目标流水线的
trigger.message-aggregation读enabled和delay。默认关闭，默认延时 1.5s，且 delay 被夹到[1.0, 10.0]秒（aggregator.py:106-137）。 - 没开就直通。 聚合关闭时，消息直接
query_pool.add_query，一点不耽误（aggregator.py:161-174）。 - 开了就进缓冲。 按
bot_uuid:launcher_type:launcher_id算会话 id，把消息塞进该会话的SessionBuffer；每来一条就取消旧定时器、重开一个新的——这正是防抖的精髓：只要还在连发，计时就不断重置。 - 两种触发刷新： 定时器 到点（
_delayed_flush），或缓冲攒够MAX_BUFFER_MESSAGES = 10条立即强刷（aggregator.py:25, 186-193）。
消息1 ─▶ 开0.0s定时器
消息2 ─▶ 取消旧定时器, 重开 ┐ 用户还在连发,
消息3 ─▶ 取消旧定时器, 重开 ┘ 计时一直被推后
…（delay 秒内无新消息）
└─▶ _delayed_flush 触发 ─▶ _merge_messages ─▶ add_query
(或中途攒满10条 ─▶ force_flush 立即并单)
4.3 并单怎么并
_flush_buffer 把该会话缓冲弹出（pop，aggregator.py:291）：只有 1 条就原样入池；多条则 _merge_messages 把各条的 message_chain 用换行 Plain('\n') 拼成一条（aggregator.py:322-367）。
两个精巧处：
message_event保留第一条的原件，只单独传合并后的message_chain——这样message_id等元数据不丢，回复 / 引用仍能对上（aggregator.py:355-366,message_event=base_msg.message_event原样透传）。routed_by_rule取所有被并消息的any(...)（aggregator.py:366）：只要其中一条是被规则命中的，合并结果就算被规则路由。- 并发安全：缓冲增删都在
self.lock内，但定时器只在锁内 cancel、不在锁内 await（aggregator.py:198、:285），避免持锁等待。flush_all()供关停时把所有缓冲刷干净、不丢消息（aggregator.py:369）。
5. 第三关：请求池（QueryPool）
聚合层的出口统一是 query_pool.add_query。请求池是"消息已就绪、等待被调度进流水线"的暂存区。它维护两份数据和一个通知机制：
| 成员 | 类型 | 作用 |
|---|---|---|
queries | list[Query] | 待处理队列，调度器从这里挑 |
cached_queries | dict[int, Query] | 按 query_id 缓存，供插件"回头调用" API，处理完才清 |
condition | asyncio.Condition | 生产/消费同步：入池后 notify_all 唤醒调度循环 |
query_id_counter | int | 自增，给每个 query 发唯一 id |
add_query 在 condition（复用 pool_lock）保护下造出 Query、入 queries 和 cached_queries、自增计数、notify_all（pool.py:248-330）：
# src/langbot/pkg/pipeline/pool.py:62-65 # 真实源码
self.queries.append(query)
self.cached_queries[query_id] = query
self.query_id_counter += 1
self.condition.notify_all()
QueryPool 本身实现了 __aenter__/__aexit__（pool.py:423-428），所以调度器能用 async with self.ap.query_pool: 直接拿锁。
6. Query 对象承载什么
Query 是这条消息贯穿整个流水线的载体——从入池到最终回复，都揣着它。add_query 构造时填入的字段（pool.py:281-308）：
| 字段 | 含义 |
|---|---|
query_id | 池内自增唯一 id |
bot_uuid | 哪个机器人收到的 |
launcher_type / launcher_id | 会话类型（person/group）与会话 id → 决定后续按哪个会话串行 |
sender_id | 发信人 id |
message_event | 原始平台事件（保留 message_id 等元数据，供回复/引用） |
message_chain | 消息内容（可能是聚合后的合并链） |
adapter | 收到消息的适配器，回复时用它发回去 |
pipeline_uuid | 路由定下的目标流水线 |
variables | 运行期变量字典，此刻先塞入 {'_routed_by_rule': ...} |
resp_messages / resp_message_chain | 预留的空回复容器，流水线跑的过程中往里填 |
也就是说，Query 入池时已经知道"谁、从哪、说了什么、该走哪条流水线"，而"答什么"是空的——流水线的任务就是把 resp_* 填满。
7. 第四关：调度与双层信号量（Controller）
7.1 一句话
控制器的 consumer() 是一个永不退出的循环，反复从池里挑一个"可以现在处理"的 query，异步跑它的流水线。挑选和放行受两层信号量约束——这是本章最核心的并发设计。
7.2 两层信号量分别管什么
| 信号量 | 建在哪 | 上限来自 | 防止什么 |
|---|---|---|---|
全局 self.semaphore | 控制器一个 | concurrency.pipeline（默认 20） | 全系统同时在跑的流水线总数 |
每会话 session._semaphore | 每个会话各一个 | concurrency.session（默认 1） | 同一会话的消息并发（默认严格串行） |
配置见 src/langbot/templates/config.yaml:47-78；全局信号量在 Controller.__init__ 建（controller.py:24），会话信号量在 SessionManager.get_session 首次创建会话时建（sessionmgr.py:323）。
为什么要两层？ 只有全局上限，同一个人连发的两条消息可能被两个协程同时处理，回复会乱序、上下文会打架。只有每会话串行，又挡不住"一万个不同会话同时涌入压垮系统"。两层叠加：跨会话可并行到 20 条，同一会话内默认一次只处理一条。
7.3 调度循环怎么挑（选中即出池）
# src/langbot/pkg/pipeline/controller.py:31-52 # 真实源码(节选)
async with self.ap.query_pool:
queries = self.ap.query_pool.queries
for query in queries:
session = await self.ap.sess_mgr.get_session(query)
if not session._semaphore.locked(): # 该会话当前空闲?
selected_query = query
await session._semaphore.acquire() # 占住会话
break
if selected_query:
queries.remove(selected_query) # 选中即从池移除
else: # 没有可跑的
await self.ap.query_pool.condition.wait() # 睡, 等 notify
continue
要点：
- 跳过忙会话。 遍历
queries，找到第一个"其会话信号量没被锁"的 query。会话默认上限 1，所以只要该会话有消息在处理，它的后续消息就被跳过、留在池里等。 - 选中即 acquire 并出池（
controller.py:41,48）——占住会话名额、从待处理列表移除，避免被重复挑走。 - 挑不到就睡。 若一圈下来没有可跑的（池空，或所有 query 的会话都满了），就
condition.wait()释放锁并挂起，直到add_query或某个流水线跑完时notify_all唤醒（controller.py:136）。这是个高效的事件驱动循环，不空转。
7.4 选中之后：全局限流下异步跑
挑中的 query 被包进 _process_query 协程，交给 task_mgr.create_task 异步跑（controller.py:156）——循环立刻回去挑下一个，不阻塞。协程内部：
# src/langbot/pkg/pipeline/controller.py:56-78 # 真实源码(节选)
async with self.semaphore: # 全局并发上限(默认20)
pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
if pipeline:
await pipeline.run(selected_query) # ← 进入流水线, 见 02 章
...
async with self.ap.query_pool:
(await self.ap.sess_mgr.get_session(selected_query))._semaphore.release()
self.ap.query_pool.condition.notify_all() # 会话腾出, 唤醒别人来挑
- 全局闸在这里。
async with self.semaphore才是全局 20 并发的真正卡点；会话信号量在挑选时已 acquire。 - 跑完必放、必唤醒。 无论流水线成功与否，
finally语义的结构保证会话信号量release、并notify_all唤醒调度循环去看看"这个会话的下一条能跑了吗、或有没有新消息"。 - 找不到流水线就丢。
pipeline_uuid为空或对应流水线不存在，只记警告、query 被丢弃（controller.py:66）。
pipeline.run(query) 之后的世界——责任链、生成器分叉、各阶段——是 02-pipeline-engine.md 的主题。
8. 边界与局限（这一层刻意不做什么）
- 不碰 LLM、不碰业务。 这四关纯粹是调度/并发/路由，一个 token 都不生成。
- 路由是"首命中即停"的线性扫描。 规则顺序即优先级；没有权重、没有"最佳匹配"。规则多且靠后命中时是 O(n) 逐条比。
- 聚合默认关闭。 不配
message-aggregation.enabled就没有防抖，消息逐条直通请求池。 - 会话默认严格串行（
concurrency.session=1）。 这是刻意的正确性取舍；调大能让同会话并发，但要自担乱序/上下文竞争风险。 - 丢弃分两种，语义不同： webhook 请求跳过（
skip_pipeline，botmgr.py:311）是不记账的静默跳过；路由__discard__是记账的可审计丢弃（botmgr.py:329）。 - 正则规则容错但静默。 非法正则不报错、直接判不匹配（
botmgr.py:115-116），配错了不会有告警。
9. 代码地图（导航索引）
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 机器人运行时 / 事件监听注册 | src/langbot/pkg/platform/botmgr.py:35 | RuntimeBot, RuntimeBot.initialize |
| 私聊 / 群聊入站回调 | src/langbot/pkg/platform/botmgr.py:285 | on_friend_message, on_group_message |
| 路由规则解析（首命中即停） | src/langbot/pkg/platform/botmgr.py:122 | resolve_pipeline_uuid |
| 单条规则算子求值 | src/langbot/pkg/platform/botmgr.py:99 | _match_operator |
| 丢弃标记 / 丢弃记账 | src/langbot/pkg/platform/botmgr.py:77,138 | PIPELINE_DISCARD, _record_discarded_message |
| 适配器加载 / 生命周期 | src/langbot/pkg/platform/botmgr.py:402,467 | PlatformManager, load_bots_from_db |
| 防抖聚合主入口 | src/langbot/pkg/pipeline/aggregator.py:139 | MessageAggregator.add_message |
| 聚合配置 / 默认值与夹取 | src/langbot/pkg/pipeline/aggregator.py:106 | _get_aggregation_config |
| 延时刷新 / 强制刷新 / 并单 | src/langbot/pkg/pipeline/aggregator.py:195,204,242 | _delayed_flush, _flush_buffer, _merge_messages |
| 请求池 / 造 Query / 通知 | src/langbot/pkg/pipeline/pool.py:114,248 | QueryPool, add_query, condition |
| 待处理与缓存队列 | src/langbot/pkg/pipeline/pool.py:20,22 | queries, cached_queries |
| 调度循环 / 选中出池 | src/langbot/pkg/pipeline/controller.py:109 | Controller.consumer |
| 全局信号量 | src/langbot/pkg/pipeline/controller.py:17,22 | Controller.semaphore |
| 每会话信号量创建 | src/langbot/pkg/provider/session/sessionmgr.py:323 | SessionManager.get_session |
| 并发上限配置 | src/langbot/templates/config.yaml:47 | concurrency.pipeline, concurrency.session |
相关章节： 消息进入 pipeline.run 之后 → 02-pipeline-engine.md；流水线内的本地 Agent 循环 → 03-local-agent-loop.md；全景与阅读地图 → index.md。