Haystack Pipeline.run_async 怎样限制并发组件数量

调用Pipeline.run_async执行异步流水线的时候，想要限制并发组件的数量，不用修改Haystack的底层源码，通过设置全局异步信号量或者自定义组件的执行调度规则就能实现，操作起来门槛很低。

通过给异步运行时绑定自定义的并发信号量，就能把同一时间并行执行的流水线组件数量严格控制在你指定的数值内，不会额外引入兼容性问题。

Haystack 的 Pipeline.run_async 可以在一次异步管线执行中限制“同时运行的组件数量”。做法是在调用处传入 concurrency_limit，例如设置为 2；官方 API 参考中的默认值是 4，小于 1 会抛出 ValueError。它控制的是 Pipeline 内部的组件调度，不是每个组件内部线程池，也不是所有请求共享的全局限流器。

官方地址：https://docs.haystack.deepset.ai/reference/pipeline-api

concurrency_limit 管的到底是什么

run_async 接收 data、可选的 include_outputs_from 和 concurrency_limit。其中最后一个参数描述的是一次 Pipeline 执行里允许同时运行的组件数量。它应该被理解为“这张组件图的调度边界”，而不是“把所有下游请求限制为 N 个”。

例如一张图里有检索、提示构建和生成等组件。某些组件必须等待上游输出，另一些分支则可能同时就绪。限制值为 2 时，调度器最多让两个符合条件的组件并行运行；当一个完成后，才可能补入下一个。若整个图本来只有一条依赖链，把数字调得更大也不会创造额外并行度。

适用场景：让并行分支有统一上限

当多个分支会共同访问模型服务、向量检索或外部 API 时，可以先从较小的上限开始，让单次管线的资源峰值更可控。下面的示例把限制放在 run_async 调用点，示例数据只表达组件输入结构：

import asyncio

async def run_pipeline(pipeline, question: str):
    # 只限制这一次 Pipeline 执行中同时运行的组件数量。
    data = {
        "retriever": {"query": question},
        "prompt_builder": {"question": question},
    }
    return await pipeline.run_async(
        data=data,
        include_outputs_from={"retriever", "generator"},
        concurrency_limit=2,
    )

async def main(pipeline):
    # 由调用方负责进入异步事件循环，避免阻塞当前线程。
    result = await run_pipeline(pipeline, "如何设置并发上限")
    return result

这里的 2 不是越小越好。分支多、单个组件占用外部连接较多时，较小值有助于控制峰值；如果图是串行的，限制值主要是安全边界。生产代码还应根据下游服务的配额、连接池和请求超时来选择，不要把它当成吞吐量保证。

快速试用：不要把组件上限写成全局限流

常见误解是：设置了 concurrency_limit=2，应用里就永远只有两个组件在运行。实际上，这个参数跟着一次 run_async 调用生效。如果上层同时创建三次管线执行，每次执行都可能拥有自己的两个组件名额，总体压力仍可能叠加。

import asyncio

async def run_one(pipeline, gate, question: str):
    # Semaphore 限制的是多个 Pipeline 调用的总并发，不替代内部上限。
    async with gate:
        return await pipeline.run_async(
            data={"retriever": {"query": question}},
            concurrency_limit=2,  # 每次调用内部最多同时运行两个组件。
        )

async def run_many(pipeline, questions: list[str]):
    gate = asyncio.Semaphore(3)  # 应用级最多同时占用三个 Pipeline 调用名额。
    tasks = [run_one(pipeline, gate, item) for item in questions]
    return await asyncio.gather(*tasks)

这两个数字处在不同层级：concurrency_limit 管一次管线里的组件，Semaphore 管多个管线调用。若多个调用最终共享同一个模型服务，还要把服务端配额和客户端连接池纳入总量估算。

和 run()、外层 Semaphore 对比

如果只需要异步地跑一张图，直接使用 run_async 并设置上限即可；如果要批量处理问题，先决定外层允许多少次调用，再决定每次管线内部的组件上限。两者相乘只是粗略的压力上界，实际仍取决于图的依赖关系。

采用风险与排查清单

1. 先确认报错是否来自 concurrency_limit ；这属于参数错误，不是模型服务超时。
2. 确认需要限制的是单次 Pipeline，还是所有请求；后者应补充调用方级别的信号量。
3. 看组件依赖图是否真的存在并行分支；纯串行图不应通过盲目增大数值来“提速”。
4. 把模型服务、检索后端、连接池和超时一起纳入容量评估；该参数不会替你限制组件内部线程池。

另外，include_outputs_from 只影响结果中保留哪些组件输出，不会把并发上限变成更大的值。调度问题与返回结果裁剪是两件事，排查时要分开。

常见问题

把 concurrency_limit 设置为 1 是否等于关闭异步？

不等于。它仍然是异步入口，只是一次只允许一个符合条件的组件运行；调用方仍可使用 await，而整个 Pipeline 的完成结果仍由异步调用返回。

为什么把 concurrency_limit 调大后耗时没有下降？

可能是组件图本身没有足够的并行分支，也可能瓶颈在下游服务、连接池或组件内部工作。这个参数只给 Pipeline 调度提供上限，不承诺线性提速。

run_async 和 run_async_generator 的限制参数一样吗？

官方 API 参考都提供 concurrency_limit，但两者的返回形态不同：前者返回完整结果字典，后者按异步生成器产生部分输出。本文的总量控制边界仍以单次调用和外层信号量为准。