上一篇结束时，一条消息已经被归一成统一的 Query，躺在 QueryPool 里等待调度。这一篇我们进入全系列技术上最核心的部分：pkg/pipeline。它要回答三个问题——谁来消费这些 Query？并发怎么控制才不会一个用户刷爆机器人？以及，一条 Query 如何流过一条由 12 个 Stage 组成的处理链、并且天然支持流式输出？最后一个问题的答案，是这个项目里最精妙的一段控制流。

一、这一层解决什么问题

平台层只负责「把消息塞进池子」，真正的处理交给流水线。流水线引擎需要解决三件事：

  1. 调度与并发控制——QueryPool 里可能同时堆着很多 Query，必须按规则取出来处理，同时防止「单个会话刷屏」和「全局过载」两种情况。
  2. 可配置的处理流程——不同 pipeline 有不同的触发规则、安全策略、AI 行为、输出格式。处理流程要能按配置拼装，而不是写死。
  3. 流式输出——LLM 是逐 token 产出的，机器人要能「边生成边回复」，而不是等整段生成完再发。这要求处理链中的某个环节能多次产出结果。

对应的三个主角分别是：Controller（调度）、PipelineManager / RuntimePipeline（拼装与执行）、以及 RuntimePipeline._execute_from_stage（责任链 + 异步生成器混合执行模型）。

二、并发模型：双层信号量

回顾第二篇，运行态有一个名为 query-controller 的系统任务，它跑的就是 Controller.run() → consumer()。这是一个永不退出的消费循环：

# src/langbot/pkg/pipeline/controller.py:24
async def consumer(self):
    while True:
        selected_query: pipeline_query.Query = None

        async with self.ap.query_pool:                 # 持有 QueryPool 的锁
            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:                                       # 没有可处理的 query
                await self.ap.query_pool.condition.wait()   # 挂起，等被唤醒
                continue
        # ...拿到 selected_query 后，启动处理任务（见下）

这里藏着第一层信号量——session._semaphore。每个会话（session）自带一个信号量，控制「同一个会话同时能有几条消息在处理」。consumer 遍历池子时，会跳过那些会话已经占满的 Query，只挑会话尚有余量的。这就保证了单个用户/群不会霸占所有处理能力。

注意 QueryPool 本身是个异步上下文管理器，内部用 asyncio.Condition 协调（src/langbot/pkg/pipeline/pool.py:25）。当池子里没有「可处理」的 query 时，consumer 调 condition.wait() 把自己挂起；一旦有新消息入池（add_query 里 notify_all）或某个处理完成释放了会话信号量，它会被唤醒重新扫描。这是一个标准的「生产者-消费者 + 条件变量」模型。

选中一个 Query 后，真正的处理被丢进一个独立任务，外面再套第二层信号量：

# src/langbot/pkg/pipeline/controller.py:55
async def _process_query(selected_query):
    async with self.semaphore:                          # ② 全局流水线并发上限
        pipeline_uuid = selected_query.pipeline_uuid
        if pipeline_uuid:
            pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
            if pipeline:
                await pipeline.run(selected_query)
            # ... 找不到 pipeline 则丢弃并告警

    async with self.ap.query_pool:
        (await self.ap.sess_mgr.get_session(selected_query))._semaphore.release()  # 释放会话信号量
        self.ap.query_pool.condition.notify_all()       # 唤醒其他消费者

self.ap.task_mgr.create_task(_process_query(selected_query), kind='query', ...)

self.semaphore 在 Controller 构造时按配置创建：

# src/langbot/pkg/pipeline/controller.py:22
self.semaphore = asyncio.Semaphore(self.ap.instance_config.data['concurrency']['pipeline'])

两层信号量各司其职：

flowchart TD
    POOL["QueryPool.queries<br>(条件变量协调)"]
    LOOP{"consumer 循环<br>扫描池子"}
    S1{"该 query 的<br>会话信号量<br>有余量?"}
    PICK["取出 query<br>acquire 会话信号量"]
    WAIT["condition.wait()<br>挂起等待唤醒"]
    S2["_process_query 任务<br>acquire 全局信号量"]
    RUN["pipeline.run(query)"]
    REL["释放会话信号量<br>notify_all 唤醒消费者"]

    POOL --> LOOP --> S1
    S1 -- 否，跳过 --> LOOP
    S1 -- 全部占满 --> WAIT
    S1 -- 是 --> PICK --> S2 --> RUN --> REL --> POOL
    WAIT -. 被唤醒 .-> LOOP