一条消息的旅程：从平台事件到进入流水线

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:26`
`resolve_pipeline_uuid`	按路由规则把消息分流到某条流水线，或标记丢弃	`botmgr.py:80`
`MessageAggregator`	防抖：把用户连发的多条消息并成一条	`src/langbot/pkg/pipeline/aggregator.py:54`
`QueryPool`	请求池：造 `Query` 对象、缓存供插件回调、通知调度器	`src/langbot/pkg/pipeline/pool.py:13`
`Controller`	调度循环：双层信号量下选中 query 并异步跑流水线	`src/langbot/pkg/pipeline/controller.py:12`

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:467）。

生命周期。 RuntimeBot.run() 把 adapter.run_async() 包一层异常捕获，交给 task_mgr.create_task 作为一个 platform-adapter 后台任务常驻（botmgr.py:384-393）；shutdown() 则 adapter.kill() 再取消该任务（botmgr.py:395）。

3.2 入站回调里发生的事（以私聊为例）

on_friend_message 收到事件后，在把消息交给下游前做三件事，顺序很重要：

推 webhook 并决定是否跳过。 若配了 webhook_pusher，先推送；webhook 可以返回"跳过流水线"，此时消息直接被丢掉不入池（botmgr.py:216-219）。
算路由。 取消息纯文本和元素类型，调 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:261-320）。适配器若实现了 get_launcher_id，还能自定义会话标识（botmgr.py:225-228）。

3.3 路由规则：按序匹配，首命中即停

resolve_pipeline_uuid 遍历该机器人配置的 pipeline_routing_rules，从上到下逐条匹配，第一条命中就返回它的 pipeline_uuid；一条都不中，就回退到默认流水线 use_pipeline_uuid（botmgr.py:109-136）。

四种规则类型：

规则 `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:117-118），坏配置不会中断路由。
message_has_element 走独立分支，不经过 _match_operator——它只认 eq/neq 两种（botmgr.py:129-134）。
返回值是元组 (pipeline_uuid, routed_by_rule)，routed_by_rule 标记"是否被规则命中"（vs 回退默认），这个布尔会一路带进 Query（见 §6），供下游区分。

3.4 丢弃机制：`discard`

有一条特殊的目标流水线值 PIPELINE_DISCARD = '__discard__'（botmgr.py:77）。当某条路由规则把消息导向它，回调不会入池，而是静默丢弃——但会先调 _record_discarded_message 在监控系统里记一笔（状态 discarded），并保证该会话在会话监视器里存在（botmgr.py:138-197）。这样"被规则过滤掉的消息"仍然可见、可审计，而不是凭空消失。

4. 第二关：防抖聚合（`MessageAggregator`）

4.1 要解决的小问题

聊天里人常常连发："在吗" → "帮我看个东西" → "就是这个 bug"。若每条都独立触发一次 LLM，既浪费又答非所问。聚合层做的就是防抖（debounce）：短时间内同一会话的多条消息，等一等、并成一条再处理。

4.2 思路：延时定时器 + 上限强刷

add_message 的逻辑（aggregator.py:119）：

查配置。 从目标流水线的 trigger.message-aggregation 读 enabled 和 delay。默认关闭，默认延时 1.5s，且 delay 被夹到 [1.0, 10.0] 秒（aggregator.py:91-115）。
没开就直通。 聚合关闭时，消息直接 query_pool.add_query，一点不耽误（aggregator.py:139-152）。
开了就进缓冲。 按 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:207）：只有 1 条就原样入池；多条则 _merge_messages 把各条的 message_chain 用换行 Plain('\n') 拼成一条（aggregator.py:242-279）。

两个精巧处：

message_event 保留第一条的原件，只单独传合并后的 message_chain——这样 message_id 等元数据不丢，回复 / 引用仍能对上（aggregator.py:267-278 注释明确说明）。
routed_by_rule 取所有被并消息的 any(...)（aggregator.py:278）：只要其中一条是被规则命中的，合并结果就算被规则路由。
并发安全：缓冲增删都在 self.lock 内，但定时器只在锁内 cancel、不在锁内 await（aggregator.py:169-193），避免持锁等待。flush_all() 供关停时把所有缓冲刷干净、不丢消息（aggregator.py:281）。

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:34-66）：

# 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:68-73），所以调度器能用 async with self.ap.query_pool: 直接拿锁。

6. Query 对象承载什么

Query 是这条消息贯穿整个流水线的载体——从入池到最终回复，都揣着它。add_query 构造时填入的字段（pool.py:48-61）：

字段	含义
`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:17-19；全局信号量在 Controller.__init__ 建（controller.py:22），会话信号量在 SessionManager.get_session 首次创建会话时建（sessionmgr.py:38）。

为什么要两层？ 只有全局上限，同一个人连发的两条消息可能被两个协程同时处理，回复会乱序、上下文会打架。只有每会话串行，又挡不住"一万个不同会话同时涌入压垮系统"。两层叠加：跨会话可并行到 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:50）。这是个高效的事件驱动循环，不空转。

7.4 选中之后：全局限流下异步跑

挑中的 query 被包进 _process_query 协程，交给 task_mgr.create_task 异步跑（controller.py:80-88）——循环立刻回去挑下一个，不阻塞。协程内部：

# 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-73）。

pipeline.run(query) 之后的世界——责任链、生成器分叉、各阶段——是 02-pipeline-engine.md 的主题。

8. 边界与局限（这一层刻意不做什么）

不碰 LLM、不碰业务。 这四关纯粹是调度/并发/路由，一个 token 都不生成。
路由是"首命中即停"的线性扫描。 规则顺序即优先级；没有权重、没有"最佳匹配"。规则多且靠后命中时是 O(n) 逐条比。
聚合默认关闭。 不配 message-aggregation.enabled 就没有防抖，消息逐条直通请求池。
会话默认严格串行（concurrency.session=1）。 这是刻意的正确性取舍；调大能让同会话并发，但要自担乱序/上下文竞争风险。
丢弃分两种，语义不同： webhook 请求跳过（skip_pipeline，botmgr.py:222）是不记账的静默跳过；路由 __discard__ 是记账的可审计丢弃（botmgr.py:236）。
正则规则容错但静默。 非法正则不报错、直接判不匹配（botmgr.py:73-74），配错了不会有告警。

9. 代码地图（导航索引）

主题	文件路径	符号名
机器人运行时 / 事件监听注册	`src/langbot/pkg/platform/botmgr.py:26`	`RuntimeBot`, `RuntimeBot.initialize`
私聊 / 群聊入站回调	`src/langbot/pkg/platform/botmgr.py:200`	`on_friend_message`, `on_group_message`
路由规则解析（首命中即停）	`src/langbot/pkg/platform/botmgr.py:80`	`resolve_pipeline_uuid`
单条规则算子求值	`src/langbot/pkg/platform/botmgr.py:57`	`_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:119`	`MessageAggregator.add_message`
聚合配置 / 默认值与夹取	`src/langbot/pkg/pipeline/aggregator.py:85`	`_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:13,34`	`QueryPool`, `add_query`, `condition`
待处理与缓存队列	`src/langbot/pkg/pipeline/pool.py:20,22`	`queries`, `cached_queries`
调度循环 / 选中出池	`src/langbot/pkg/pipeline/controller.py:24`	`Controller.consumer`
全局信号量	`src/langbot/pkg/pipeline/controller.py:17,22`	`Controller.semaphore`
每会话信号量创建	`src/langbot/pkg/provider/session/sessionmgr.py:38`	`SessionManager.get_session`
并发上限配置	`src/langbot/templates/config.yaml:17`	`concurrency.pipeline`, `concurrency.session`

相关章节： 消息进入 pipeline.run 之后 → 02-pipeline-engine.md；流水线内的本地 Agent 循环 → 03-local-agent-loop.md；全景与阅读地图 → index.md。

1. 这段路径是干嘛的（零基础也能懂）​

2. 顶层全景（一条消息怎么流下来）​

3. 第一关：平台适配层与路由（RuntimeBot）​

3.1 适配器把"平台方言"翻成统一事件​

3.2 入站回调里发生的事（以私聊为例）​

3.3 路由规则：按序匹配，首命中即停​

3.4 丢弃机制：__discard__​

4. 第二关：防抖聚合（MessageAggregator）​

4.1 要解决的小问题​

4.2 思路：延时定时器 + 上限强刷​

4.3 并单怎么并​

5. 第三关：请求池（QueryPool）​

6. Query 对象承载什么​

7. 第四关：调度与双层信号量（Controller）​

7.1 一句话​

7.2 两层信号量分别管什么​

7.3 调度循环怎么挑（选中即出池）​

7.4 选中之后：全局限流下异步跑​

8. 边界与局限（这一层刻意不做什么）​

9. 代码地图（导航索引）​