Haystack Joiners 组件完全指南在多分支管道中高效合并答案、文档与列表【免费下载链接】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 中负责汇聚的组件家族——Joiners它们把来自多个上游分支、多个 Retriever 或多个 Generator 的输出合并成统一的列表是构建 RAG检索增强生成、多路召回、结果校验循环和对话系统等复杂管道时不可或缺的收口组件。读完本文你将掌握AnswerJoiner、DocumentJoiner、BranchJoiner、ListJoiner、StringJoiner五个组件的完整 API、全部 join 模式含 RRF 与分布融合的底层原理以及它们在真实管道中的连接与运行方式。一、Joiners 组件家族概览在 Haystack 中管道Pipeline通过组件 连接组织数据流一个组件可以有多个上游输入也可以把结果分发给多个下游。当多个分支需要汇聚到同一个下游组件时就需要 Joiner 类组件完成合并。从 haystack/components/joiners/init.py 的导入结构可以看出当前版本共提供 5 个 Joiner 组件按数据形态分为两类组件合并的数据类型输出形态典型场景AnswerJoiner多个list[Answer]单个list[Answer]合并多个 Generator 生成的答案DocumentJoiner多个list[Document]单个list[Document]多路检索BM25 向量结果融合BranchJoiner任意单一类型的单值该类型的单个值环路闭合、分支汇聚二选一/多选一ListJoiner多个同类型list单个扁平list合并多份消息列表、结果列表StringJoiner多个strlist[str]合并多个 PromptBuilder 生成的字符串它们的共同点是通过Variadic可变数量输入或GreedyVariadic类型注解接收任意数量的上游输入再按各自策略合并。下文的测试用例均可参考 test/components/joiners/ 目录下的对应测试文件。二、AnswerJoiner合并多个生成器的答案2.1 定位与使用场景AnswerJoiner用于把多份Answer对象列表合并成一份。典型场景是多模型投票或多路生成两个或多个LLM 各自针对同一问题生成答案由AnswerJoiner汇总成单一列表供下游评估器或展示层统一消费。以下示例摘自 answer_joiner.py 文档字符串它把两个OpenAIChatGenerator的回复分别交给两个AnswerBuilder构建答案再由AnswerJoiner合并from haystack.components.builders import AnswerBuilder from haystack.components.joiners import AnswerJoiner from haystack.core.pipeline import Pipeline from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage query Whats Natural Language Processing? messages [ChatMessage.from_system(You are a helpful, respectful and honest assistant. Be super concise.), ChatMessage.from_user(query)] pipe Pipeline() pipe.add_component(llm_1, OpenAIChatGenerator()) pipe.add_component(llm_2, OpenAIChatGenerator()) pipe.add_component(aba, AnswerBuilder()) pipe.add_component(abb, AnswerBuilder()) pipe.add_component(joiner, AnswerJoiner()) pipe.connect(llm_1.replies, aba) pipe.connect(llm_2.replies, abb) pipe.connect(aba.answers, joiner) pipe.connect(abb.answers, joiner) results pipe.run(data{llm_1: {messages: messages}, llm_2: {messages: messages}, aba: {query: query}, abb: {query: query}})2.2 构造参数与运行参数__init__签名见 answer_joiner.pydef __init__(join_mode: str | JoinMode JoinMode.CONCATENATE, top_k: int | None None, sort_by_score: bool False)参数默认值说明join_modeJoinMode.CONCATENATE合并模式。当前AnswerJoiner仅支持concatenate把多份答案列表直接拼接成一份。传入未知字符串会抛出ValueError错误信息会列出全部受支持模式top_kNone返回答案的最大数量。必须是None或大于 0否则构造时抛出ValueError(top_k must be greater than 0.)sort_by_scoreFalse若为True按 score 降序排序没有 score 的答案按-infinity处理run签名component.output_types(answerslist[AnswerType]) def run(answers: Variadic[list[AnswerType]], top_k: int | None None)answers可变数量的答案列表嵌套列表会被压平合并top_k本次运行的截断上限会覆盖实例化时的top_k传0表示不返回任何答案传负数会抛出ValueError返回值字典仅含一个键answers即合并后的答案列表。AnswerType在源码中被定义为GeneratedAnswer | ExtractedAnswer见 answer_joiner.py因此既能合并生成式答案也能合并抽取式答案。2.3 源码级实现细节从源码看AnswerJoiner的JoinMode枚举目前只定义了CONCATENATE concatenate一个成员answer_joiner.py构造函数会把 join 模式映射到具体的合并函数并缓存到self.join_mode_function。_concatenate实现极为轻量本质是itertools.chain.from_iterable的一层封装def _concatenate(self, answer_lists: list[list[AnswerType]]) - list[AnswerType]: return list(itertools.chain.from_iterable(answer_lists))值得注意的执行顺序answer_joiner.py是先拼接 → 再按需排序 → 最后截断。因此即使开启sort_by_scoreTrue排序也只作用于合并后的整体列表不会影响各分支内部的相对顺序。三、BranchJoiner环路闭合与分支汇聚3.1 定位二选一或多选一的闸门BranchJoiner与前几个组件的合并语义不同——它接收多个同类型输入但只把第一个到达的值转发给下游branch.py。它的两个典型用途环路处理Loop Handling当管道中存在校验失败 → 重新生成这类循环时BranchJoiner负责把首次数据与回流的修正数据合并成一条通路送回同一个下游组件。它让 Haystack 管道能够以显式连接的方式表达循环结构基于决策的合并Decision-Based Merging当ConditionalRouter、TextLanguageRouter等路由组件把不同查询分发给不同 Retriever 时各 Retriever 的结果由BranchJoiner收拢为单条输出流再送入PromptBuilder等下游。3.2 使用示例JSON 生成 模式校验的循环管道以下完整示例来自 version-2.22 的 joiners_api.md它把BranchJoiner用在了一个生成 JSON → 校验模式 → 失败则回流重新生成的闭环中import json from haystack import Pipeline from haystack.components.converters import OutputAdapter from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.joiners import BranchJoiner from haystack.components.validators import JsonSchemaValidator from haystack.dataclasses import ChatMessage # Define a schema for validation person_schema { type: object, properties: { first_name: {type: string, pattern: ^[A-Z][a-z]$}, last_name: {type: string, pattern: ^[A-Z][a-z]$}, nationality: {type: string, enum: [Italian, Portuguese, American]}, }, required: [first_name, last_name, nationality] } # Initialize a pipeline pipe Pipeline() # Add components to the pipeline pipe.add_component(joiner, BranchJoiner(list[ChatMessage])) pipe.add_component(generator, OpenAIChatGenerator()) pipe.add_component(validator, JsonSchemaValidator(json_schemaperson_schema)) pipe.add_component(adapter, OutputAdapter({{chat_message}}, list[ChatMessage], unsafeTrue)) # And connect them pipe.connect(adapter, joiner) pipe.connect(joiner, generator) pipe.connect(generator.replies, validator.messages) pipe.connect(validator.validation_error, joiner) result pipe.run( data{ generator: {generation_kwargs: {response_format: {type: json_object}}}, adapter: {chat_message: [ChatMessage.from_user(Create json from Peter Parker)]}} ) print(json.loads(result[validator][validated][0].text))运行结果示例 {first_name: Peter, last_name: Parker, nationality: American, name: Spider-Man, occupation: Superhero, age: 23, location: New York City}3.3 API 说明与类型约束__init__签名def __init__(type_: type)type_输入与输出共用的数据类型。BranchJoiner同一时刻只能管理一种数据类型——例如传入list[ChatMessage]那么它接收的所有上游输入和它的输出都必须是该类型见 branch.py。run签名def run(**kwargs) - dict[str, Any]输入关键字参数的类型必须与初始化时的type_一致返回值字典只含一个键value即第一个到达的那个输入值。3.4 源码级实现细节BranchJoiner的底层实现有两个值得关注的细节branch.py动态类型注册构造函数通过component.set_input_types(self, valueGreedyVariadic[type_])和component.set_output_types(self, valuetype_)动态注册输入/输出类型。注意它使用的是GreedyVariadic贪婪可变参数这是它与其余 Joiner 组件使用普通Variadic的关键区别——GreedyVariadic不会阻塞等待所有上游就绪而是只要有一个输入到达就立即触发执行这正是取第一个值语义能成立的原因严格校验run方法会检查kwargs[value]的长度若接收到的输入数量不等于 1立即抛出ValueErrorbranch.py。也就是说多个输入只是候选集合最终只能有一个值被放行。此外由于BranchJoiner承载的是单条数据流而非列表合并它的to_dict/from_dict需要使用serialize_type/deserialize_type对type_做类型序列化才能把list[ChatMessage]这类类型对象安全地存入 YAML/JSON 配置branch.py。四、DocumentJoiner多路检索结果融合含四种 join 模式4.1 定位与使用场景DocumentJoiner是 Joiner 家族中功能最丰富的一个专门合并多路检索返回的Document列表。最经典的用法是混合检索Hybrid Search一路 BM25 稀疏检索 一路向量稠密检索各自返回 top-k 文档由DocumentJoiner去重融合成单一列表兼顾关键词精确匹配与语义相似度。完整示例来自 document_joiner.py 的文档字符串这里保留文档原始写法使用SentenceTransformers嵌入器from haystack import Pipeline, Document from haystack.components.embedders import SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder from haystack.components.joiners import DocumentJoiner from haystack.components.retrievers import InMemoryBM25Retriever from haystack.components.retrievers import InMemoryEmbeddingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore document_store InMemoryDocumentStore() docs [Document(contentParis), Document(contentBerlin), Document(contentLondon)] embedder SentenceTransformersDocumentEmbedder(modelsentence-transformers/all-MiniLM-L6-v2) embedder.warm_up() docs_embeddings embedder.run(docs) document_store.write_documents(docs_embeddings[documents]) p Pipeline() p.add_component(instanceInMemoryBM25Retriever(document_storedocument_store), namebm25_retriever) p.add_component( instanceSentenceTransformersTextEmbedder(modelsentence-transformers/all-MiniLM-L6-v2), nametext_embedder, ) p.add_component(instanceInMemoryEmbeddingRetriever(document_storedocument_store), nameembedding_retriever) p.add_component(instanceDocumentJoiner(), namejoiner) p.connect(bm25_retriever, joiner) p.connect(embedding_retriever, joiner) p.connect(text_embedder, embedding_retriever) query What is the capital of France? p.run(data{query: query, text: query, top_k: 1})说明当前仓库源码的示例已改用OpenAITextEmbedder/OpenAIDocumentEmbedder见 document_joiner.py两套写法结构一致均可直接替换运行。连接方式上现代写法更推荐显式连接p.connect(text_embedder.embedding, embedding_retriever.query_embedding)。4.2 四种 join 模式详解JoinMode枚举定义了 4 种模式document_joiner.py模式枚举值合并策略是否使用weightsconcatenateconcatenate直接拼接按文档id去重重复文档保留 score 最高的那个忽略mergemerge对重复文档的 score 做加权求和并合并为一个文档使用reciprocal_rank_fusionreciprocal_rank_fusion基于倒排排名融合RRF重新赋分并去重使用distribution_based_rank_fusiondistribution_based_rank_fusion基于每个 Retriever 内部分数分布重标定后再去重忽略concatenate拼接去重内部实现为defaultdict(list)按文档id分组再用max(..., keyscore)保留最高分副本document_joiner.py。无 score 的文档按-inf参与比较因此有分数哪怕是 0的文档会胜出——测试用例test_run_with_concatenate_join_mode_keeps_zero_score_over_negative_duplicate等覆盖了这些边界情况见 test_document_joiner.py。merge加权求和为每个输入列表分配权重默认等权1/n重复文档的新 score Σ(原 score × 权重)最终用dataclasses.replace生成携带新分数的文档副本document_joiner.py。权重数组会在构造时被归一化除以总和且权重之和为 0 会直接抛出ValueErrordocument_joiner.py。reciprocal_rank_fusion倒排排名融合委托给 haystack/utils/misc.py 中的_reciprocal_rank_fusion。其核心公式为score(id) Σ wᵢ·n / (k rank)其中常数k 61——源码注释解释原始论文建议 60考虑到 Python 列表从 0 开始计数而论文按 1 计数故加 1misc.py。RRF 不依赖各检索器分数的绝对量纲天然对分数尺度差异鲁棒。distribution_based_rank_fusion基于分布的融合DBSF先在每个输入列表内部做分数重标定——计算该列表的均值与标准差以mean ± 3σ作为上下界做 min-max 归一化把分数映射到接近[0,1]的范围若某列表内所有分数相同无法提供区分信息则保留原分数最后再走 concatenate 逻辑按id去重、取最高分document_joiner.py。这样可以把不同 Retriever 的分数分布拉齐到可比尺度。4.3 构造参数与运行参数__init__签名def __init__(join_mode: str | JoinMode JoinMode.CONCATENATE, weights: list[float] | None None, top_k: int | None None, sort_by_score: bool True)参数默认值说明join_modeCONCATENATE上述 4 种模式之一也接受对应字符串weightsNone每个输入列表的重要性权重长度必须与输入列表数量一致仅对merge和reciprocal_rank_fusion生效concatenate与distribution_based_rank_fusion会忽略它top_kNone返回文档最大数量须为None或大于 0sort_by_scoreTrue是否按 score 降序排列无 score 文档按-inf处理此时会输出一条 info 级别日志见 document_joiner.pyrun签名component.output_types(documentslist[Document]) def run(documents: Variadic[list[Document]], top_k: int | None None)documents待合并的文档列表的列表top_k本次运行上限覆盖实例级top_k0表示不返回文档负数抛出ValueError返回值字典含一个键documents。4.4 与 weights 的配合weights是按列表而非按文档赋权的。例如两路检索中你更信任向量检索可以设weights[0.3, 0.7]在merge模式下重复文档的分数会据此偏斜在 RRF 模式下权重会进入融合公式wᵢ·n / (k rank)。权重会被自动归一化除以总和所以[0.3, 0.7]与[3, 7]效果相同但总和为 0如[-1, 1]会报错。五、ListJoiner把多个列表拼成扁平列表ListJoiner接收多个同类型列表按管道执行顺序先到的先拼拼接成一个扁平列表是处理list[ChatMessage]这类结构化列表的通用工具list_joiner.py。以下示例展示了一个问答 反馈管道主链路生成回答的同时把用户提示词、主回答和反馈 LLM 的回复三份消息列表全部汇入ListJoiner一次性拿到完整的消息轨迹from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack import Pipeline from haystack.components.joiners import ListJoiner user_message [ChatMessage.from_user(Give a brief answer the following question: {{query}})] feedback_prompt You are given a question and an answer. Your task is to provide a score and a brief feedback on the answer. Question: {{query}} Answer: {{response}} feedback_message [ChatMessage.from_system(feedback_prompt)] prompt_builder ChatPromptBuilder(templateuser_message) feedback_prompt_builder ChatPromptBuilder(templatefeedback_message) llm OpenAIChatGenerator() feedback_llm OpenAIChatGenerator() pipe Pipeline() pipe.add_component(prompt_builder, prompt_builder) pipe.add_component(llm, llm) pipe.add_component(feedback_prompt_builder, feedback_prompt_builder) pipe.add_component(feedback_llm, feedback_llm) pipe.add_component(list_joiner, ListJoiner(list[ChatMessage])) pipe.connect(prompt_builder.prompt, llm.messages) pipe.connect(prompt_builder.prompt, list_joiner) pipe.connect(llm.replies, list_joiner) pipe.connect(llm.replies, feedback_prompt_builder.response) pipe.connect(feedback_prompt_builder.prompt, feedback_llm.messages) pipe.connect(feedback_llm.replies, list_joiner) query What is nuclear physics? ans pipe.run(data{prompt_builder: {template_variables:{query: query}}, feedback_prompt_builder: {template_variables:{query: query}}}) print(ans[list_joiner][values])__init__签名def __init__(list_type_: type | None None)list_type_期望的列表元素类型如list[ChatMessage]。指定后所有输入列表必须符合该类型传None时组件退化为处理任意类型包括混合类型的列表。run签名def run(values: Variadic[list[Any]]) - dict[str, list[Any]]values多个列表返回{values: 扁平合并后的列表}。从实现看list_joiner.py合并逻辑只有一行list(chain(*values))list_type_的作用体现在__init__中通过component.set_output_types声明输出类型从而让管道在编译期就能校验连接的类型兼容性list_joiner.py。六、StringJoiner合并字符串为字符串列表StringJoiner是最轻量的 Joiner把来自不同组件的字符串收集成一个list[str]。典型场景是合并多个PromptBuilder生成的提示词交给下游统一处理。from haystack.components.joiners import StringJoiner from haystack.components.builders import PromptBuilder from haystack.core.pipeline import Pipeline from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage string_1 Whats Natural Language Processing? string_2 What is life? pipeline Pipeline() pipeline.add_component(prompt_builder_1, PromptBuilder(Builder 1: {{query}})) pipeline.add_component(prompt_builder_2, PromptBuilder(Builder 2: {{query}})) pipeline.add_component(string_joiner, StringJoiner()) pipeline.connect(prompt_builder_1.prompt, string_joiner.strings) pipeline.connect(prompt_builder_2.prompt, string_joiner.strings) print(pipeline.run(data{prompt_builder_1: {query: string_1}, prompt_builder_2: {query: string_2}}))输出 {string_joiner: {strings: [Builder 1: Whats Natural Language Processing?, Builder 2: What is life?]}}run签名component.output_types(stringslist[str]) def run(strings: Variadic[str])strings来自不同组件的字符串返回{strings: 合并后的字符串列表}。实现上StringJoiner甚至不需要__init__方法没有构造参数run只是把可变参数直接转成列表string_joiner.py因此它也没有自定义的to_dict/from_dict完全依赖组件默认序列化。七、序列化to_dict / from_dict除StringJoiner外其余四个 Joiner 都实现了标准的序列化接口用于把组件配置持久化为字典进而可写入 YAML/JSON 管道描述文件再在反序列化时还原to_dict()返回包含type、init_parameters的字典。AnswerJoiner会记录join_mode、top_k、sort_by_scoreanswer_joiner.pyDocumentJoiner额外记录归一化后的weightsdocument_joiner.pyListJoiner记录序列化后的list_type_list_joiner.py。from_dict(data)反序列化。BranchJoiner与ListJoiner需要先用deserialize_type把字符串形式的类型还原成真正的 Python 类型对象再调用default_from_dict完成构造branch.py、list_joiner.py。这一机制保证了包含 Joiner 的管道可以被安全地导出为可版本化、可复现的管道描述文件。八、选型建议与实战要点需求推荐组件合并多个 Generator 的答案AnswerJoiner融合 BM25 向量等多路检索结果DocumentJoiner默认concatenate需分数融合时用merge/reciprocal_rank_fusion/distribution_based_rank_fusion环路闭合、路由分支收拢只取一路BranchJoiner拼接多份同类型列表如消息轨迹ListJoiner收集多份字符串StringJoiner实战中还需注意三点BranchJoiner与列表型 Joiner 语义不同前者是多选一靠GreedyVariadic立即触发后者是全收下再合并。选错会导致管道行为完全不符预期top_k的覆盖规则run参数中的top_k始终覆盖实例参数且两处都有非负校验AnswerJoiner/DocumentJoiner的构造参数要求top_k 0而run参数允许0表示清空结果类型一致性BranchJoiner(type_)和ListJoiner(list_type_)在指定类型后连接类型会在管道编译期被校验对应测试见 test_branch_joiner.py 与 test_list_joiner.py。若不确定上游输出类型ListJoiner()不传list_type_是最宽松的兜底方案。通过合理组合这五个 Joiner你可以在 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），仅供参考