Haystack Pipeline API 深度解析从组件图编排到同步/异步双引擎执行机制【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本篇技术指南基于 Haystack 仓库中 2.21 版本的 Pipeline API 参考文档docs-website/reference_versioned_docs/version-2.21/haystack-api/pipeline_api.md编写系统讲解Pipeline与AsyncPipeline两套编排引擎的完整 API构造参数、组件图的增删与连接、run/run_async/run_async_generator的执行语义、序列化/反序列化以及可视化调试接口。读完后你将能够独立完成 RAG 管线的搭建、理解并发执行图的调度细节并依据源码定位常见运行期错误PipelineRuntimeError、PipelineMaxComponentRuns等的根因。1. Pipeline 编排引擎定位组件 有向图 执行调度Haystack 的 Pipeline API 文档描述的核心对象是两个模块中的两个类模块async_pipeline中的AsyncPipelinePipeline 编排引擎的异步版本Manages components in a pipeline allowing for concurrent processing when the pipelines execution graph permits——当执行图允许时并发处理组件以最小化空闲时间、最大化资源利用率模块pipeline中的Pipeline同步版本的编排引擎Orchestrates component execution according to the execution graph, one after the other——严格按照执行图逐个顺序执行组件。从源码结构看这两个类共享同一套图管理能力当前仓库中Pipeline继承自PipelineBase见 haystack/core/pipeline/base.py后者内部使用 NetworkX 的MultiDiGraph存储组件节点与连接边self.graph networkx.MultiDiGraph()文档中列出的大部分方法add_component、connect、to_dict、dumps/loads、show/draw等都定义在基类中两个引擎的差异集中在run系列的执行调度上。1.1 构造参数__init__的三个关键配置文档给出的签名Pipeline与AsyncPipeline一致def __init__(metadata: Optional[dict[str, Any]] None, max_runs_per_component: int 100, connection_type_validation: bool True)参数默认值作用metadataNone任意元数据字典。注意文档的提醒如果希望把 Pipeline 保存到文件必须保证其中所有值都可序列化/反序列化max_runs_per_component100同一 Component 在本 Pipeline 中最多可运行几次达到上限即抛出PipelineMaxComponentRuns。这是防止循环管线死循环的保险丝connection_type_validationTrue连接组件时是否校验 socket 类型兼容性关闭后类型不匹配的连接也可以建立对应源码见 PipelineBase.init。此外文档还覆盖了两个常被忽略的等值与展示方法__eq__Pipeline 的相等性由类型 序列化形式共同定义。两个同类型 Pipeline 共享全部 metadata、节点与边即视为相等不要求使用相同的节点实例——这正是保存后再加载回来的 Pipeline 能与原对象相等的原因。源码实现非常直白return self.to_dict() other.to_dict()见 base.py__repr__返回包含 Metadata、Components、Connections 三段的人类可读文本表示。1.2 组件图管理add_component / remove_component / get_componentadd_component(name, instance)def add_component(name: str, instance: Component) - None文档说明组件添加后默认不与任何东西相连需要配合connect()使用组件名必须唯一。异常语义ValueError已存在同名组件PipelineValidationError给定对象不是 Component。当前仓库源码在文档版本的基础上又收紧了约束见 PipelineBase._validate_component可以作为排查add_component报错时的完整规则清单组件名不能包含.点号因为.是connect()中component_name.socket_name语法的一部分_debug是调试输出的保留名同一个组件实例只能被添加到一个Pipeline 中一次——源码通过给实例打上__haystack_added_to_pipeline__属性实现跨 Pipeline 的占用检测。文档 2.21 版写作 component instances can be reused if needed而现网实现已演进为实例不能跨 Pipeline 共享按当前仓库行为以新约束为准。remove_component(name)def remove_component(name: str) - Component按名称移除组件并返回被移除的实例所有连接到该组件的边一并删除名称不存在时抛ValueError。源码中base.py还能看到它做了两件更细致的清理清空相邻 socket 中对该组件的senders/receivers引用以及重置 variadic socket 的包装状态避免悬空引用。get_component / get_component_nameget_component(name)按名称取实例未找到抛ValueErrorget_component_name(instance)反向操作返回实例在本 Pipeline 中的名称若实例未加入本 Pipeline返回空字符串。2. connect建立类型安全的组件间连接文档签名def connect(sender: str, receiver: str) - PipelineBase规则与 1.2 节示例代码一致两端组件必须都已存在于 Pipeline若组件有多个输入/输出 socket使用component_name.connection_name显式指定例如 RAG 示例中的rag_pipeline.connect(retriever, prompt_builder.documents)——这里documents就是 prompt builder 的输入 socket 名失败时抛PipelineConnectError组件不存在、类型不匹配等。源码实现PipelineBase.connect补充了文档没有展开的两条规则值得在实操中记住组件不能连接到自身sender与receiver解析为同一组件时直接抛PipelineConnectError(Connecting a Component to itself is not supported.)多对一 list 输入的类型提升当多个 sender 连向同一个 list 类型的 receiver socket 时该 socket 会被提升为 lazy variadic socket 以接收全部来值。同步run下聚合后的列表按 sender 组件名字母序排列而不是connect()的调用顺序异步执行路径下则不保证顺序——因为不同分支的组件可能并行运行。RAG 管线的典型搭建流程继承自文档Pipeline.run用法示例retriever InMemoryBM25Retriever(document_storedocument_store) prompt_builder PromptBuilder(templateprompt_template) llm OpenAIGenerator(api_keySecret.from_token(api_key)) rag_pipeline Pipeline() rag_pipeline.add_component(retriever, retriever) rag_pipeline.add_component(prompt_builder, prompt_builder) rag_pipeline.add_component(llm, llm) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder, llm)3. 执行路径一Pipeline.run 同步执行def run(data: dict[str, Any], include_outputs_from: Optional[set[str]] None, *, break_point: Optional[Union[Breakpoint, AgentBreakpoint]] None, pipeline_snapshot: Optional[PipelineSnapshot] None ) - dict[str, Any]3.1 输入格式data以组件名 → 该组件输入参数字典的形式组织data { comp1: {input1: 1, input2: 2}, }文档同时指出当输入名在整条管线中唯一时支持简写形式data { input1: 1, input2: 2, }include_outputs_from是一组组件名集合指定哪些组件的中间输出需要并入最终结果对于在循环中被多次调用的组件只包含其最后一次产生的输出。完整 RAG 调用示例文档原文示例使用InMemoryBM25RetrieverPromptBuilderOpenAIGenerator# Write documents to InMemoryDocumentStore document_store InMemoryDocumentStore() document_store.write_documents([ Document(contentMy name is Jean and I live in Paris.), Document(contentMy name is Mark and I live in Berlin.), Document(contentMy name is Giorgio and I live in Rome.) ]) prompt_template Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{question}} Answer: retriever InMemoryBM25Retriever(document_storedocument_store) prompt_builder PromptBuilder(templateprompt_template) llm OpenAIGenerator(api_keySecret.from_token(api_key)) rag_pipeline Pipeline() rag_pipeline.add_component(retriever, retriever) rag_pipeline.add_component(prompt_builder, prompt_builder) rag_pipeline.add_component(llm, llm) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder, llm) question Who lives in Paris? results rag_pipeline.run( { retriever: {query: question}, prompt_builder: {question: question}, } ) print(results[llm][replies]) # Jean lives in Paris3.2 返回值与异常返回{组件名: 组件输出}字典include_outputs_from为None时字典只包含叶子组件没有出边的组件的输出。文档中llm恰好是叶子所以能看到results[llm]想同时拿检索结果传include_outputs_from{retriever, llm}。异常文档 Raises 列表逐条有源码对应ValueError输入非法——由validate_input校验见 4.2 节PipelineRuntimeError管线包含会导致卡死的不支持循环连接或某个 Component 执行失败/返回了不支持的输出类型。同步run中组件抛出的任意异常都会被包装见 Pipeline._run_component 中raise PipelineRuntimeError.from_exception(...)以及输出不是 Mapping 时的from_invalid_outputPipelineMaxComponentRuns组件达到max_runs_per_component上限PipelineBreakpointException命中break_point时抛出异常对象携带组件名、状态和局部结果。3.3 断点与快照break_point / pipeline_snapshot2.21 文档中run独有的两个参数是调试/恢复能力的入口break_point断点集合用于调试管线执行。源码层面断点命中时会先构造一个PipelineSnapshot记录组件访问计数、当前输入状态、原始输入数据等默认保存为 JSON 文件并附在异常上见 pipeline.pypipeline_snapshot先前保存的执行快照run可从断点处恢复。恢复时会先_validate_pipeline_snapshot_against_pipeline把快照与当前图比对然后重建component_visits与内部输入格式继续推进。这套机制对 Agent 类长管线尤其有用某步失败后可基于快照 新断点单步前进。4. 执行路径二AsyncPipeline 的三种异步接口4.1 AsyncPipeline.run同步外观、异步内核def run(data: dict[str, Any], include_outputs_from: Optional[set[str]] None, concurrency_limit: int 4) - dict[str, Any]文档描述Provides a synchronous interface... Internally, the pipeline components are executed asynchronously, but the method itself will block until the entire pipeline execution is complete.——内部组件异步执行可并发但方法本身阻塞到整条管线跑完需要非阻塞时改用run_async/run_async_generator。concurrency_limit默认 4允许并发运行的组件数上限。额外异常RuntimeError——如果在一个已有的 async 上下文中调用它应改用run_async。参数、返回、其余异常与Pipeline.run完全一致示例代码见文档结构上与 3.1 节的 RAG 示例相同仅把Pipeline()换成AsyncPipeline()并直接rag_pipeline.run(data)。4.2 run_async全异步执行async def run_async(data: dict[str, Any], include_outputs_from: Optional[set[str]] None, concurrency_limit: int 4) - dict[str, Any]文档给出的用法要点保留async def run_inner(data, include_outputs_from): return await rag_pipeline.run_async(datadata, include_outputs_frominclude_outputs_from) data { retriever: {query: question}, prompt_builder: {question: question}, } results asyncio.run(run_inner(data, include_outputs_from{retriever, llm})) print(results[llm][replies]) # [ChatMessage(_roleChatRole.ASSISTANT: assistant, _content[TextContent(textJean lives in Paris.)], ...)]返回与异常语义同Pipeline.rundata同样支持组件名嵌套与输入名平铺两种格式。4.3 run_async_generator增量输出与实时处理async def run_async_generator( data: dict[str, Any], include_outputs_from: Optional[set[str]] None, concurrency_limit: int 4) - AsyncIterator[dict[str, Any]]Executes the pipeline step by step asynchronously, yielding partial outputs when any component finishes.——每当任意组件完成就产出一个部分输出适合边执行边展示进度的场景。文档示例中的消费方式async def process_results(): async for partial_output in rag_pipeline.run_async_generator( datadata, include_outputs_from{retriever, llm} ): # Each partial_output contains the results from a completed component if retriever in partial_output: print(Retrieved documents:, len(partial_output[retriever][documents])) if llm in partial_output: print(Generated answer:, partial_output[llm][replies][0]) asyncio.run(process_results())底层调度机制源码级补充帮助理解concurrency_limit到底约束了什么当前仓库中run_async_generator的实现Pipeline.run_async_generator用asyncio.Semaphore(concurrency_limit)作为并发闸门每一轮都会重建优先级队列并挑选可运行组件普通组件按READY优先级调度然后继续塞满剩余并发额度while len(priority_queue) 0 and not ready_sem.locked()拥有HIGHEST优先级即带 GreedyVariadic 输入 socket的组件必须独占运行先等所有在途任务结束再单独执行防止下游组件在运行期间又喂入新输入见_run_component_in_isolation某个组件失败时_cancel_in_flight_tasks会取消所有在途任务并等待清理避免僵尸任务在后台继续跑生成器在finally中兜底取消在途任务因此消费方中途放弃迭代也不会泄漏任务。而run_async本身只是run_async_generator的薄封装迭代到最后一个partial并返回pipeline.py。concurrency_limit 1时抛ValueError。4.4 输入校验 validate_inputPipeline与AsyncPipeline都提供def validate_input(data: dict[str, Any]) - None文档列出它检查的四件事两条run路径在执行前都会调用它每个组件名确实存在于 Pipeline 中每个组件不缺少任何必填输入每个组件的每个输入 socket 只有一个输入来源可变长 variadic 除外组件不会收到另一个组件已经在提供的输入——即外部data与组件间连接冲突。不满足则抛ValueError。5. 输入/输出内省inputs 与 outputsdef inputs(include_components_with_connected_inputs: bool False) - dict[str, dict[str, Any]] def outputs(include_components_with_connected_outputs: bool False) - dict[str, dict[str, Any]]两者对称inputs返回{组件名: {输入 socket 描述}}socket 描述含类型与是否可选include_components_with_connected_inputsFalse时只列出存在未连接输入边的组件——也就是需要从外部 data 喂数据的组件。outputs同理描述输出 socketinclude_components_with_connected_outputsFalse时只列出存在未连接输出边的组件即结果会出现在 run 返回字典中的叶子组件。这两个方法是管线契约的自省接口服务端封装run之前可以先调用它们动态确定必填参数比硬编码参数名更稳健。6. 序列化与持久化to_dict / from_dict / dumps / dump / loads / load方法签名要点说明to_dict()- dict[str, Any]序列化为字典可作为中间表示或直接落盘from_dict(data, callbacksNone, **kwargs)classmethod反序列化kwargs支持components{name: instance}以便复用已有组件实例而不是新建dumps(marshallerDEFAULT_MARSHALLER)- str按 Marshaller 格式返回字符串表示默认YamlMarshallerdump(fp, marshallerDEFAULT_MARSHALLER)- None写入文件类对象fploads(data, marshallerDEFAULT_MARSHALLER, callbacksNone)classmethod从str/bytes/bytearray构造 Pipeline出错抛DeserializationErrorload(fp, marshallerDEFAULT_MARSHALLER, callbacksNone)classmethod从文件类对象读取并构造 Pipeline同样可能抛DeserializationErrorto_dict的产出结构可以从源码确认PipelineBase.to_dict包含metadata、max_runs_per_component、components逐组件序列化、connections每条边记录sender.sender_socket - receiver.receiver_socket以及connection_type_validation。也就是说三个构造参数都会被持久化加载后即得到等价管线——这正是 1.1 节__eq__基于序列化形式比较的基础。源码层面还有一个文档 2.21 版未列出的安全约束当前from_dict支持allowed_modules与unsafe参数默认只对haystack、haystack_integrations、builtins、typing、collections等模块白名单内的类放行base.py。如果你加载的是外部来源的序列化管线需要显式扩展信任模块列表。7. 可视化与遍历show / draw / walk / warm_up7.1 show 与 drawMermaid 渲染管线图def show(*, server_url: str https://mermaid.ink, params: Optional[dict] None, timeout: int 30, super_component_expansion: bool False) - Noneshow在 Jupyter notebook 中显示管线图draw(path...)则把图保存为指定路径的图片文件。两者共用同一套渲染参数server_urlMermaid 渲染服务地址默认https://mermaid.inkparams渲染自定义字典支持键formatimg/svg/pdf默认img、typejpeg/png/webp默认png、themedefault/neutral/dark/forest默认neutral、bgColor、width、height、scale1–3仅在指定 width/height 时生效、fitPDF 自适应、paper如a4、landscapetimeout请求 Mermaid 服务的超时秒数默认 30super_component_expansion为True且管线含 SuperComponent 时图中展开其内部结构而非黑盒显示。异常PipelineDrawingError——show在非 Jupyter 环境调用或渲染失败时draw在渲染/保存失败时同样抛出。7.2 walk 与 warm_updef walk() - Iterator[tuple[str, Component]] def warm_up() - Nonewalk以任意顺序逐个访问每个组件恰好一次产出(name, instance)元组文档明确不提供访问顺序保证。warm_up确保所有节点预热完成例如加载模型。文档强调节点自身负责保证该方法可在每次run前被重复调用而不重复初始化一切。当前源码中它已被run/run_async_generator在执行前自动调用见 pipeline.py 的self.warm_up()通常无需手动触发。7.3 validate_pipeline执行前的死锁预判staticmethod def validate_pipeline(priority_queue: FIFOPriorityQueue) - None文档说明其用途检查管线是否被阻塞、或没有有效入口点是则抛PipelineRuntimeError。从源码结构看pipeline.pyrun在把组件名填入优先级队列后会立刻执行该检查把配置错误导致的必然卡死提前到执行前暴露而不是在循环里空转。8. from_template 与版本演进AsyncPipeline 已并入 Pipelineclassmethod def from_template(cls, predefined_pipeline: PredefinedPipeline, template_params: Optional[dict[str, Any]] None) - PipelineBase从预定义模板PredefinedPipeline枚举创建 Pipelinetemplate_params用于渲染模板参数返回一个Pipeline实例。这是不写连接代码直接拿现成管线的快捷方式。重要版本演进提示本文主体依据的是 2.21 版参考文档其中AsyncPipeline还是独立类run系列带concurrency_limit参数。当前仓库代码已经合并了这两个类仓库中的迁移说明Merge-AsyncPipeline-into-Pipeline 发布说明明确写道AsyncPipeline已被移除其异步能力并入单一Pipeline类同步run 异步run_async、run_async_generator与stream迁移方式把AsyncPipeline()替换为Pipeline()即可已序列化的管线、SuperComponent、PipelineTool 会按Pipeline加载行为差异新的Pipeline.run是顺序执行且不接受concurrency_limit原先AsyncPipeline.run那种同步外观 并发内核的语义现在用asyncio.run(pipeline.run_async(...))保留Tracing 上同步与异步运行统一使用haystack.pipeline.run操作名以haystack.pipeline.execution_mode标签sync/async区分不再使用旧的haystack.async_pipeline.run操作名。因此在 2.21 及更早版本按本文第 3、4 节的 API 编写代码没有问题升级到合并版本后只需把类名统一为Pipeline其余方法名run_async、run_async_generator、concurrency_limit、include_outputs_from保持不变。9. API 速查表类别方法关键语义构造__init__metadata/max_runs_per_component100/connection_type_validationTrue图管理add_component/remove_component/get_component/get_component_name/connect名称唯一.语法指定 socket类型不匹配抛PipelineConnectError同步执行Pipeline.run顺序执行支持break_point/pipeline_snapshot叶子组件输出默认返回异步执行AsyncPipeline.run/run_async/run_async_generator并发执行concurrency_limit默认 4generator 版逐步产出部分结果内省inputs/outputs/walk查看需外部喂入的输入、可取回的输出、遍历全部组件校验validate_input/validate_pipeline输入合法性四重检查执行前死锁/入口点检查持久化to_dict/from_dict/dumps/dump/loads/load默认YamlMarshallerfrom_dict可复用组件实例可视化show/drawMermaid 服务渲染super_component_expansion展开超级组件预热warm_up确保所有节点就绪run前自动调用模板from_template由PredefinedPipeline模板快速建管线10. 小结Haystack 的 Pipeline API 把编排拆成了三层清晰的关注点用add_component/connect声明结构带 socket 级类型校验的有向图用run/run_async/run_async_generator选择执行策略同步顺序、同步外观的并发、原生异步或增量流式用dumps/loads、show/draw、inputs/outputs补齐工程化配套持久化、可视化、契约自省。理解include_outputs_from的叶子组件默认输出规则、concurrency_limit的并发闸门口径以及PipelineRuntimeError/PipelineMaxComponentRuns的触发条件再配合本文引用的源码路径haystack/core/pipeline/base.py、haystack/core/pipeline/pipeline.py就足以在任意 Haystack 项目里独立搭建、调试与迁移管线代码。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考