Haystack（AI 应用框架）

from haystack import Pipeline, component

@component
class FakeRetriever:
    @component.output_types(documents=list)
    def run(self):
        return {"documents": ["Haystack 会把组件串成管道"]}

@component
class FakePromptBuilder:
    @component.output_types(prompt=str)
    def run(self, documents: list):
        return {"prompt": f"请总结：{documents[0]}"}

@component
class FakeGenerator:
    @component.output_types(answer=str)
    def run(self, prompt: str):
        return {"answer": f"生成结果：{prompt}"}

# 用 add_component 添加组件，用 connect 定义数据流向
pipeline = Pipeline()
pipeline.add_component("retriever", FakeRetriever())
pipeline.add_component("prompt_builder", FakePromptBuilder())
pipeline.add_component("llm", FakeGenerator())
pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder.prompt", "llm.prompt")

pip install haystack-ai

# 基于 haystack-ai==2.12.1 验证（截至 2026-03）
from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.readers import ExtractiveReader

# 1. 创建文档存储，写入几条文档
doc_store = InMemoryDocumentStore()
doc_store.write_documents([
    Document(content="Python 是由 Guido van Rossum 于 1991 年创建的高级编程语言。"),
    Document(content="JavaScript 由 Brendan Eich 在 1995 年创建，主要用于网页开发。"),
    Document(content="Go 语言由 Google 在 2009 年发布，专注于并发和高性能。"),
])

# 2. 创建组件
retriever = InMemoryBM25Retriever(document_store=doc_store)
reader = ExtractiveReader(model="deepset/roberta-base-squad2")
reader.warm_up()  # 预加载模型（首次运行需下载，约 500MB）

# 3. 构建管道并连接组件
pipeline = Pipeline()
pipeline.add_component("retriever", retriever)
pipeline.add_component("reader", reader)
pipeline.connect("retriever.documents", "reader.documents")

# 4. 提问
query = "Python 是什么时候创建的？"
result = pipeline.run(data={
    "retriever": {"query": query},
    "reader": {"query": query}
})

# 5. 输出答案
answer = result["reader"]["answers"][0]
print(f"问题：{query}")
print(f"答案：{answer.data}")
print(f"置信度：{float(answer.score):.2%}")

问题：Python 是什么时候创建的？
答案：1991 年
置信度：89.34%

# 需要额外安装：pip install openai
# 需要设置环境变量：export OPENAI_API_KEY="sk-..."
import os
from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

# 文档存储
doc_store = InMemoryDocumentStore()
doc_store.write_documents([
    Document(content="Haystack 是 deepset 开源的 AI 编排框架，支持构建 RAG 和 Agent 应用。"),
    Document(content="LangChain 是一个通用 LLM 应用框架，主打链式调用和组件生态。"),
])

# 提示词模板
template = [ChatMessage.from_user(
    "根据以下文档回答问题。\n"
    "文档：\n{% for doc in documents %}{{ doc.content }}\n{% endfor %}\n"
    "问题：{{ question }}\n回答："
)]

# 构建 RAG 管道
rag = Pipeline()
rag.add_component("retriever", InMemoryBM25Retriever(document_store=doc_store))
rag.add_component("prompt", ChatPromptBuilder(template=template, required_variables=["question", "documents"]))
rag.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
rag.connect("retriever.documents", "prompt.documents")
rag.connect("prompt", "llm.messages")

# 执行查询
question = "Haystack 是什么？"
result = rag.run({
    "retriever": {"query": question},
    "prompt": {"question": question},
})
print(result["llm"]["replies"][0].text)