Haystack Builders 组件完全指南：AnswerBuilder、PromptBuilder 与 ChatPromptBuilder 的用法与源码剖析

Haystack Builders 组件完全指南：AnswerBuilder、PromptBuilder 与 ChatPromptBuilder 的用法与源码剖析

【免费下载链接】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 的builders模块提供了三个开箱即用的提示词与答案构建组件：AnswerBuilder负责把 Generator 的输出解析成结构化的GeneratedAnswer对象（支持正则抽取与文档引用追踪），PromptBuilder与ChatPromptBuilder负责用 Jinja2 模板把变量渲染成文本提示词或 ChatMessage 消息列表，供 Generator/ChatGenerator 消费。本文以 builders_api.md 为骨架，结合 builders 源码 与 对应测试 展开讲解，读完你就能在 RAG、问答与多模态管线中熟练配置和使用这三个组件。

本文基于仓库中的 Haystack 2.x 版本（参考文档为 version-2.19）撰写。文中代码路径均以仓库根目录为基准。

一、Builders 模块概览

haystack/components/builders目录下共有三个组件（见init.py）：

三者都通过@component装饰器注册为 Haystack 组件，可被Pipeline.add_component()直接加入管线；也都可以在run()时动态覆盖模板，因此非常适合"先搭管线、后做 prompt engineering"的工作流。

二、AnswerBuilder：把 Generator 回复解析成结构化答案

2.1 核心作用

AnswerBuilder将"查询 + Generator 回复"转换成GeneratedAnswer对象。它用自定义正则表达式从回复中抽取答案文本；可选地接收 Generator 的元数据与输入文档，把文档挂到答案对象上；同时兼容文本型 Generator 与对话型 ChatGenerator。GeneratedAnswer数据类定义在 haystack/dataclasses/answer.py，包含四个字段：

• data：解析后的答案文本；
• query：原始查询；
• documents：被引用的文档列表；
• meta：生成器元数据（含all_messages原始回复）。

2.2 构造参数

def __init__(pattern: Optional[str] = None, reference_pattern: Optional[str] = None, last_message_only: bool = False)

• pattern：抽取答案文本的正则。不传则把整个回复当作答案。正则最多只能有一个捕获组：有捕获组时取组内文本，无捕获组时取整个匹配。例如：
  • [^\n]+$可从"this is an argument.\nthis is an answer"中抽出this is an answer（取最后一行）；
  • Answer: (.*)可从"this is an argument. Answer: this is an answer"中抽出this is an answer。 源码在 answer_builder.py 中实现：re.search命中后，若match.lastindex为空则取group(0)，否则取group(1)；未命中返回空字符串。若正则含多个捕获组，构造或运行时都会抛出ValueError（见_check_num_groups_in_regex，answer_builder.py）。
• reference_pattern：解析文档引用的正则。不传则不解析，所有文档全部挂到答案上。引用按输入文档的从 1 开始的索引书写，例如\[(\d+)\]可从"this is an answer[1]"中抽出1。提供该参数后，返回文档副本的 meta 中会带有布尔键referenced。
• last_message_only：默认False使用全部消息作为答案；设为True则只取最后一条消息。

当前仓库源码还提供了两个文档中未出现的额外构造参数（answer_builder.py）：

• return_only_referenced_documents（默认True）：配合reference_pattern使用，只返回回复中真正被引用的文档；设为False返回全部文档，未引用的文档referenced=False。若未提供reference_pattern，该参数不生效。
• expand_reference_ranges（默认False）：设为True后支持[6-10]这样的引用区间展开为 6～10 号文档。开启后若使用默认reference_pattern，会自动切换到更宽的模式EXPANDED_REFERENCE_PATTERN = r"\[(\d+(?:[,-]\d+)*)\]"（answer_builder.py）。

2.3 run() 签名与参数

@component.output_types(answers=list[GeneratedAnswer]) def run(query: str, replies: Union[list[str], list[ChatMessage]], meta: Optional[list[dict[str, Any]]] = None, documents: Optional[list[Document]] = None, pattern: Optional[str] = None, reference_pattern: Optional[str] = None)

• query：喂给 Generator 的查询；
• replies：Generator 输出，字符串列表或ChatMessage列表均可；
• meta：Generator 返回的元数据列表，长度必须与replies一致，否则抛ValueError（answer_builder.py）；不传时答案不含元数据；
• documents：Generator 的输入文档。传入后每个文档副本的 meta 会带上source_index（输入列表中从 1 开始的序号），原输入文档不被修改（通过dataclasses.replace生成副本）；若同时提供reference_pattern，则从 Generator 输出中解析被引用的文档，再依据return_only_referenced_documents决定返回全部还是仅返回被引用文档；
• pattern/reference_pattern：可覆盖初始化时的默认值，实现"一次构建、多次不同解析规则"。

返回字典键为answers，值为GeneratedAnswer列表。

2.4 实战示例

最简用法（不带文档）：

from haystack.components.builders import AnswerBuilder builder = AnswerBuilder(pattern="Answer: (.*)") builder.run(query="What's the answer?", replies=["This is an argument. Answer: This is the answer."])

带文档与引用解析（参考 answer_builder.py 中的官方示例）：

from haystack import Document from haystack.components.builders import AnswerBuilder replies = ["The capital of France is Paris [2]."] docs = [ Document(content="Berlin is the capital of Germany."), Document(content="Paris is the capital of France."), Document(content="Rome is the capital of Italy."), ] builder = AnswerBuilder(reference_pattern="\\[(\\d+)\\]", return_only_referenced_documents=False) result = builder.run(query="What is the capital of France?", replies=replies, documents=docs)["answers"][0] print(f"Answer: {result.data}") # >> Answer: The capital of France is Paris print("References:") for doc in result.documents: if doc.meta["referenced"]: print(f"[{doc.meta['source_index']}] {doc.content}") # >> [2] Paris is the capital of France. print("Other sources:") for doc in result.documents: if not doc.meta["referenced"]: print(f"[{doc.meta['source_index']}] {doc.content}") # >> [1] / [3] ...

2.5 源码级细节与边界行为

• 1 基索引与越界保护：引用是 1 基的，因此[0]会被解析为idx = -1；源码显式做0 <= idx < len(documents)边界检查（answer_builder.py），越界时记录WARNING日志"Document index '{index}' referenced in Generator output is out of range."并跳过该文档，绝不会静默错配到最后一个文档。
• 回复来源区分：对ChatMessage回复取其.text与.meta，对字符串回复直接使用（answer_builder.py）。
• 引用区间展开的安全钳制：展开[1-999999999]这类超大区间时，会把末端钳制到文档总数，避免物化出巨大集合（answer_builder.py）。
• 测试佐证：test_answer_builder.py 覆盖了 meta 长度不匹配抛错、多捕获组抛错、运行时覆盖 pattern、return_only_referenced_documents两种取值、越界引用告警、按 source order 返回被引用文档（[3, 10, 50]→ 索引[3, 10, 50]）等场景，可作为你自行验证行为的参考。

三、PromptBuilder：Jinja2 文本提示词渲染器

3.1 核心作用

PromptBuilder使用 Jinja2 语法渲染提示词模板，把变量填充进模板后输出纯文本prompt，供文本补全类 Generator 使用。默认模板中的变量即组件的输入，均可选；缺失的可选变量在渲染时替换为空字符串。管线每次运行时都可以传入新模板，方便反复做 prompt engineering。

3.2 构造参数

def __init__(template: str, required_variables: Optional[Union[list[str], Literal["*"]]] = None, variables: Optional[list[str]] = None)

• template：Jinja2 模板字符串，例如"Summarize this document: {{ documents[0].content }}\nSummary:"。模板变量会通过_extract_template_variables_and_assignments自动推断为组件的输入 socket；{% set %}等已赋值的变量会被排除在输入之外（prompt_builder.py）。
• required_variables：必须提供的变量列表；设为"*"表示模板中所有变量都必填。当前仓库源码中该参数默认值为"*"（见 prompt_builder.py），显式传入变量列表后，未列出的变量变为可选并在缺失时渲染为空字符串；设为None则全部可选。若变量较多又显式设None，源码会打出一条WARNING，提示多分支管线中"全部可选"可能引发非预期行为（prompt_builder.py）。
• variables：显式声明输入变量列表，替代从template自动推断。典型用途是 prompt engineering 阶段让组件接收比默认模板更多的变量。

渲染环境是HaystackSandboxedEnvironment（沙箱化 Jinja2 环境），并尽可能加载Jinja2TimeExtension（需要arrow依赖，缺失时静默降级），见 prompt_builder.py。

3.3 run() 签名与行为

@component.output_types(prompt=str) def run(template: Optional[str] = None, template_variables: Optional[dict[str, Any]] = None, **kwargs)

• template：运行时覆盖默认模板；None则使用初始化模板；
• template_variables：运行时覆盖管线变量的字典，优先级高于 kwargs；
• kwargs：用于渲染模板的管线变量。

变量合并顺序为{**kwargs, **template_variables}，即template_variables优先（prompt_builder.py）。渲染前会调用_validate_variables校验必填变量，缺失时抛ValueError，错误信息会列出缺失变量、必填列表与已提供列表，方便排查（prompt_builder.py）。

3.4 实战示例

独立使用：

from haystack.components.builders import PromptBuilder template = "Translate the following context to {{ target_language }}. Context: {{ snippet }}; Translation:" builder = PromptBuilder(template=template) builder.run(target_language="spanish", snippet="I can't speak spanish.") # 输出 prompt："Translate the following context to Spanish. Context: I can't speak Spanish.; Translation:"

在 RAG 管线中使用（官方示例，prompt_builder.py）：

from haystack import Pipeline, Document from haystack.utils import Secret from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.builders.prompt_builder import PromptBuilder # 真实场景中 documents 可来自 retriever、web 等任意来源 documents = [Document(content="Joe lives in Berlin"), Document(content="Joe is a software engineer")] prompt_template = """ Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{query}} Answer: """ p = Pipeline() p.add_component(instance=PromptBuilder(template=prompt_template), name="prompt_builder") p.add_component(instance=OpenAIChatGenerator(api_key=Secret.from_env_var("OPENAI_API_KEY")), name="llm") p.connect("prompt_builder", "llm") question = "Where does Joe live?" result = p.run({"prompt_builder": {"documents": documents, "query": question}}) print(result)

注意：参考文档中的示例连接的是OpenAIGenerator，而当前仓库 prompt_builder.py 的官方示例已更新为OpenAIChatGenerator（PromptBuilder输出str，通过p.connect("prompt_builder", "llm")或p.connect("prompt_builder.prompt", "llm.prompt")均可接入，取决于 Generator 的输入 socket）。

运行时更换模板（prompt engineering）：

documents = [ Document(content="Joe lives in Berlin", meta={"name": "doc1"}), Document(content="Joe is a software engineer", meta={"name": "doc1"}), ] new_template = """ You are a helpful assistant. Given these documents, answer the question. Documents: {% for doc in documents %} Document {{ loop.index }}: Document name: {{ doc.meta['name'] }} {{ doc.content }} {% endfor %} Question: {{ query }} Answer: """ p.run({ "prompt_builder": { "documents": documents, "query": question, "template": new_template, }, })

运行时覆盖变量：

language_template = """ You are a helpful assistant. Given these documents, answer the question. Documents: {% for doc in documents %} Document {{ loop.index }}: Document name: {{ doc.meta['name'] }} {{ doc.content }} {% endfor %} Question: {{ query }} Please provide your answer in {{ answer_language | default('English') }} Answer: """ p.run({ "prompt_builder": { "documents": documents, "query": question, "template": language_template, "template_variables": {"answer_language": "German"}, }, })

language_template引入了未绑定任何管线变量的answer_language，未覆盖时使用 Jinja2default('English')兜底，示例将其覆盖为German。template_variables同样可以覆盖documents等常规管线变量。

3.5 序列化

PromptBuilder.to_dict()返回包含template、variables、required_variables的字典（prompt_builder.py），用于组件/管线 YAML 序列化。

四、ChatPromptBuilder：面向对话模型的消息渲染器

4.1 核心作用

ChatPromptBuilder用 Jinja2 语法把模板渲染成list[ChatMessage]，直接对接ChatGenerator的messages输入。模板可以是：

• ChatMessage对象列表（静态或动态）；
• 特殊字符串模板（借助ChatMessageExtension与{% message %}/{% endmessage %}标签、templatize_part过滤器构建结构化消息，甚至支持图片等多模态内容）。

模板变量默认必填（required_variables默认"*"）；未被列为必填的变量缺失时渲染为空字符串。variables与required_variables用于定义输入类型与必填约束。

4.2 构造参数

def __init__(template: Optional[Union[list[ChatMessage], str]] = None, required_variables: Optional[Union[list[str], Literal["*"]]] = None, variables: Optional[list[str]] = None)

• template：ChatMessage列表或字符串模板，可在init或run时提供；
• required_variables/variables：语义与 PromptBuilder 完全一致（当前仓库默认"*"）。

变量推断规则与 PromptBuilder 略有不同：列表模板只从USER与SYSTEM角色的消息文本中提取变量，ASSISTANT/TOOL消息不参与推断（chat_prompt_builder.py）。若USER/SYSTEM消息文本为None或列表模板中使用了templatize_part过滤器，会分别抛出NO_TEXT_ERROR_MESSAGE与FILTER_NOT_ALLOWED_ERROR_MESSAGE错误。渲染环境为HaystackSandboxedEnvironment+ChatMessageExtension，若安装了arrow还会追加Jinja2TimeExtension（chat_prompt_builder.py）。

4.3 run() 签名与行为

@component.output_types(prompt=list[ChatMessage]) def run(template: Optional[Union[list[ChatMessage], str]] = None, template_variables: Optional[dict[str, Any]] = None, **kwargs)

• template：运行时覆盖默认模板；None则用初始化模板；
• template_variables：覆盖管线变量（优先级高于 kwargs）；
• kwargs：管线变量。

run会先校验模板非空且列表元素均为ChatMessage，否则抛ValueError。对列表模板，只渲染USER/SYSTEM消息，并通过dataclasses.replace生成新消息副本，避免原地修改原始消息（chat_prompt_builder.py）；对字符串模板，则调用_render_chat_messages_from_str_template渲染后再按行json.loads还原成ChatMessage（chat_prompt_builder.py）。

4.4 实战示例

静态 ChatMessage 模板：

template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")] builder = ChatPromptBuilder(template=template) builder.run(target_language="spanish", snippet="I can't speak spanish.")

运行时覆盖静态模板：

template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")] builder = ChatPromptBuilder(template=template) builder.run(target_language="spanish", snippet="I can't speak spanish.") msg = "Translate to {{ target_language }} and summarize. Context: {{ snippet }}; Summary:" summary_template = [ChatMessage.from_user(msg)] builder.run(target_language="spanish", snippet="I can't speak spanish.", template=summary_template)

动态 ChatMessage 模板（多轮对话管线）：

from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack import Pipeline from haystack.utils import Secret # 不传 init 模板，全部变量在运行时提供 prompt_builder = ChatPromptBuilder() llm = OpenAIChatGenerator(api_key=Secret.from_token("<your-api-key>"), model="gpt-4o-mini") pipe = Pipeline() pipe.add_component("prompt_builder", prompt_builder) pipe.add_component("llm", llm) pipe.connect("prompt_builder.prompt", "llm.messages") location = "Berlin" language = "English" system_message = ChatMessage.from_system("You are an assistant giving information to tourists in {{language}}") messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")] res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "language": language}, "template": messages}}) print(res) # >> {'llm': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text= # "Berlin is the capital city of Germany and one of the most vibrant ...")], _name=None, _meta={'model': # 'gpt-4o-mini', 'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 27, 'completion_tokens': 681, # 'total_tokens': 708}})]}}

同一管线第二次运行时更换问题模板与变量：

messages = [system_message, ChatMessage.from_user("What's the weather forecast for {{location}} in the next {{day_count}} days?")] res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"}, "template": messages}}) print(res) # >> {'llm': {'replies': [ChatMessage(... text="Here is the weather forecast for Berlin in the next 5 days:\n\n... # 'prompt_tokens': 37, 'completion_tokens': 201, 'total_tokens': 238})]}}

字符串模板（含多模态图片内容）：

from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses.image_content import ImageContent template = """ {% message role="system" %} You are a helpful assistant. {% endmessage %} {% message role="user" %} Hello! I am {{user_name}}. What's the difference between the following images? {% for image in images %} {{ image | templatize_part }} {% endfor %} {% endmessage %} """ images = [ImageContent.from_file_path("apple.jpg"), ImageContent.from_file_path("orange.jpg")] builder = ChatPromptBuilder(template=template) builder.run(user_name="John", images=images)

字符串模板依赖ChatMessageExtension提供的{% message %}块标签与templatize_part过滤器（其实现见 jinja2_chat_extension.py）。templatize_part会把图片等结构化内容包裹为_TemplatizedPart，并带有每次环境随机生成的 nonce 哨兵标记，防止外部注入伪造内容（jinja2_chat_extension.py）。

4.5 序列化

ChatPromptBuilder同时实现了to_dict()与from_dict()（chat_prompt_builder.py）：to_dict会把列表模板逐个ChatMessage.to_dict()转成字典；from_dict反序列化时再把字典还原为ChatMessage列表，从而支持 YAML 管线配置的完整往返。

五、三个组件的选型与最佳实践

1. 选型要点：
   • 目标是文本补全型生成（如传统OpenAIGenerator）→ 用PromptBuilder，输出str；
   • 目标是对话式生成（如OpenAIChatGenerator）→ 用ChatPromptBuilder，输出list[ChatMessage]，天然支持 system/user/assistant 多轮角色；
   • 需要把生成结果后处理为带引用文档的答案→ 在 Generator 之后串接AnswerBuilder。
2. Prompt engineering 工作流：初始化时给一个默认模板，之后每次p.run()通过template参数传入新模板、用template_variables覆盖变量，无需重建管线。
3. 必填校验：生产环境建议显式列出required_variables，避免"*"全必填导致的多分支管线因缺失分支变量而中断，也避免None全可选掩盖拼写错误。当前仓库中两个 Prompt 组件的默认值均为"*"。
4. 引用追踪：RAG 场景中给AnswerBuilder同时配置reference_pattern与return_only_referenced_documents，既能保留引用来源，又能控制返回文档数量，减少下游 token 消耗。
5. 源码阅读入口：三个组件的完整实现见 answer_builder.py、prompt_builder.py、chat_prompt_builder.py；行为测试见 test_answer_builder.py、test_prompt_builder.py、test_chat_prompt_builder.py；pydoc 配置见 pydoc/builders_api.yml。

六、常见问题速查

以上三个组件共同构成了 Haystack 管线中"输入模板化 → 生成 → 结构化答案"的关键链路，是 RAG 问答、Agent 对话与多模态应用最常用的基础构件。

【免费下载链接】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