Haystack 是 deepset 开源的 Python 编排框架，定位是给生产级 LLM 应用搭 pipeline 和 agent 工作流。它最有辨识度的设计就一个：Pipeline 是一张有向多重图，节点是组件，边由 connect 显式声明。相比让你把逻辑塞进一个又一个回调，Haystack 把「数据从哪到哪」这件事摆在明面上，检索、路由、记忆、生成每一环都看得见、改得动。这篇拆它的连接机制，并给出一个能直接跑的 RAG pipeline。

Pipeline 与有向多重图

先纠正一个直觉：Haystack 的 pipeline 不是一条单向流水线，而是一张有向多重图（directed multigraph）。意思是同一对组件之间可以有多条边，图里也允许分支、汇合、循环、甚至孤立的组件。一个 pipeline 能干什么，完全由你放进去什么组件、怎么连决定——可以做索引、查询、预处理，也可以把多个用途塞进同一张图，用决策组件（比如 ConditionalRouter）分流。

这套图的执行是异步友好的：同步 run 按依赖顺序串行执行组件；异步方法里，没有依赖关系的组件会并行跑。这给复杂 pipeline 留了性能空间，也解释了为什么它敢把 agent 循环也塞进 pipeline 的图里。

组件与连接

组件的唯一契约是实现 run 方法，声明自己的输入输出。Haystack 把组件按能力拆成很多内置件：转换器（TextFileToDocument、HTMLToDocument）、预处理器（DocumentSplitter）、检索器（InMemoryBM25Retriever、InMemoryEmbeddingRetriever）、生成器（OpenAIChatGenerator、OpenAIGenerator）、构建器（ChatPromptBuilder）等等。

搭 pipeline 只有四步：建图、加组件、连边、run。关键在连边这一环——connect 在真正执行前就会校验两端的类型和输入输出名，配错了当场报详细错误，而不是跑到一半才炸：

from haystack import Pipeline

pipe = Pipeline()
pipe.add_component("fetcher", LinkContentFetcher())
pipe.add_component("converter", HTMLToDocument())
pipe.add_component("writer", DocumentWriter(document_store=document_store))

pipe.connect("fetcher.streams", "converter.sources")
pipe.connect("converter.documents", "writer.documents")

pipe.run({"fetcher": {"urls": ["https://example.com/faq"]}})

connect 的写法是 "产出组件.输出名" 到 "消费组件.输入名"。组件只有一个输出或输入时，可以只写组件名，框架自动推断；两边都只有一个口时，直接 connect("a", "b") 就行。这是它比手写编排省心的地方：连接是显式的，但又不啰嗦。

一个最小 RAG pipeline

检索增强问答是 Haystack 的主场。下面这段把一个 BM25 检索器、提示词构建器和生成器连起来，是能直接跑的完整例子：

from haystack import Pipeline, Document
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.dataclasses import ChatMessage

store = InMemoryDocumentStore()
store.write_documents([
    Document(content="知识库检索用 BM25 处理精确词匹配。"),
    Document(content="向量检索更适合语义相近但用词不同的查询。"),
])

retriever = InMemoryBM25Retriever(document_store=store)
builder = ChatPromptBuilder(template=[
    ChatMessage.from_system("只依据下面的文档回答：{% for d in documents %}{{ d.content }}{% endfor %}"),
    ChatMessage.from_user("{{ question }}"),
])
llm = OpenAIChatGenerator(model="gpt-4o-mini")

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("builder", builder)
rag.add_component("llm", llm)
rag.connect("retriever", "builder.documents")
rag.connect("builder", "llm")

result = rag.run({"retriever": {"query": "精确词匹配用哪个？"}, "builder": {"question": "精确词匹配用哪个？"}})
print(result["llm"]["replies"])

注意两个细节：run 可以同时给不同组件喂输入（retriever 拿 query、builder 拿 question）；生成器输出的是 replies 列表。模板里用 Jinja2 语法遍历 documents，把检索结果塞进提示词——这就是「显式控制 context 怎么组装」的直接体现。

条件路由

分支是图编排的价值点之一。用 ConditionalRouter 可以按上游结果决定走哪条路，比如检索不到就走联网搜索兜底：

from haystack.components.routers import ConditionalRouter

routes = [
    {"condition": "{{ replies | length == 0 }}", "output": "{{ query }}", "output_type": str, "output_name": "empty"},
    {"condition": "{{ replies | length > 0 }}", "output": "{{ replies }}", "output_type": str, "output_name": "hit"},
]
router = ConditionalRouter(routes)

条件写成 Jinja2 表达式，命中哪条就把结果送到对应命名的输出口，下游据此接不同的分支。这种把路由逻辑显式写成规则、而不是埋在 if-else 里的做法，让复杂 agent 的控制流也能被读出来、被测试。

异步与流式

同一个 pipeline 既能同步跑也能异步跑，生成器还能逐 token 流式输出。run_async 一次非阻塞调用拿完整结果，stream 返回一个句柄，用 async for 逐块消费。这对把 pipeline 包成 API 尤其有用——首 token 延迟和内存占用都能控。

序列化与可视化

每个组件实现 from_dict 和 to_dict，pipeline 也能整体序列化成 YAML 存盘、改完再加载，方便版本管理和分享。连接关系还可以直接画成 Mermaid 图核对，搭错边一眼就能看出来。这两个能力对「生产级」不是装饰，而是可维护性的一部分。

值不值得用

Haystack 的取舍很明确：它把「显式」放在第一位，代价是图里组件一多，连接关系要自己维护，上手比 LangChain 那种链式抽象略重。如果你在做的是检索、问答、多分支 agent，且希望控制流可读、可序列化、可并行，它的组件连接模型很对路；如果你只是想把几个 LLM 调用串起来图省事，图编排反而是多余的负担。它的定位是「生产级 RAG 与 agent 的底座」，而不是最短路径的 demo 脚手架。