5.1 基础问答系统

5.1 基础问答系统

从零搭建一个 Haystack 问答管道

在本节中，我们将从零开始搭建一个基于 Haystack 的基础问答系统。这是最经典的 RAG 应用场景，也是理解整个框架运作机制的最好起点。

1. 项目初始化

首先安装必要的依赖：

# requirements.txt
haystack-ai>=2.0
sentence-transformers>=3.0
langchain-community>=0.3
numpy>=1.24
​

创建项目目录结构：

qa_project/
├── config.yaml
├── main.py
├── document_store.py
├── pipeline.py
├── api_server.py
└── data/
    └── knowledge_base/
​

2. 配置管理

使用 YAML 配置文件管理所有参数，这是企业级项目的标准做法：

# config.yaml
document_store:
  type: in_memory
  embedding_model: sentence-transformers/all-MiniLM-L6-v2

retriever:
  top_k: 5
  score_threshold: 0.5

generator:
  model: gpt-3.5-turbo
  max_tokens: 512
  temperature: 0.3

pipeline:
  batch_size: 32
​

加载配置的工具函数：

import yaml
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class RetrieverConfig:
    top_k: int = 5
    score_threshold: float = 0.5

@dataclass
class GeneratorConfig:
    model: str = "gpt-3.5-turbo"
    max_tokens: int = 512
    temperature: float = 0.3

@dataclass
class PipelineConfig:
    batch_size: int = 32

@dataclass
class AppConfig:
    retriever: RetrieverConfig = field(default_factory=RetrieverConfig)
    generator: GeneratorConfig = field(default_factory=GeneratorConfig)
    pipeline: PipelineConfig = field(default_factory=PipelineConfig)

def load_config(path: str = "config.yaml") -> AppConfig:
    with open(path) as f:
        data = yaml.safe_load(f)
    return AppConfig(
        retriever=RetrieverConfig(**data.get("retriever", {})),
        generator=GeneratorConfig(**data.get("generator", {})),
        pipeline=PipelineConfig(**data.get("pipeline", {}))
    )
​

3. 文档存储与索引

搭建文档存储层，支持从本地文件批量导入知识：

from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack.components.writers import DocumentWriter
from haystack.components.converters import MarkdownToDocument, PyPDFToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.routers import FileTypeRouter
from haystack.components.joiners import DocumentJoiner
import os

class KnowledgeBase:
    """知识库管理器：负责文档的导入、处理和索引"""
    
    def __init__(self, embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"):
        self.document_store = InMemoryDocumentStore()
        self.embedding_model = embedding_model
    
    def import_documents(self, directory: str, chunk_size: int = 500, chunk_overlap: int = 50):
        """从目录批量导入文档"""
        indexing_pipeline = Pipeline()
        
        # 文件类型路由
        indexing_pipeline.add_component(
            "file_type_router",
            FileTypeRouter(mime_types=["text/plain", "text/markdown", "application/pdf"])
        )
        
        # 转换器
        indexing_pipeline.add_component("markdown_converter", MarkdownToDocument())
        indexing_pipeline.add_component("pdf_converter", PyPDFToDocument())
        
        # 合并
        indexing_pipeline.add_component("joiner", DocumentJoiner())
        
        # 清洗与分块
        indexing_pipeline.add_component("cleaner", DocumentCleaner())
        indexing_pipeline.add_component(
            "splitter",
            DocumentSplitter(split_by="word", split_length=chunk_size, split_overlap=chunk_overlap)
        )
        
        # 向量化与写入
        indexing_pipeline.add_component(
            "embedder",
            SentenceTransformersDocumentEmbedder(model=self.embedding_model)
        )
        indexing_pipeline.add_component("writer", DocumentWriter(document_store=self.document_store))
        
        # 连接管道
        indexing_pipeline.connect("file_type_router.text/plain", "markdown_converter")
        indexing_pipeline.connect("file_type_router.text/markdown", "markdown_converter")
        indexing_pipeline.connect("file_type_router.application/pdf", "pdf_converter")
        indexing_pipeline.connect("markdown_converter", "joiner")
        indexing_pipeline.connect("pdf_converter", "joiner")
        indexing_pipeline.connect("joiner", "cleaner")
        indexing_pipeline.connect("cleaner", "splitter")
        indexing_pipeline.connect("splitter", "embedder")
        indexing_pipeline.connect("embedder", "writer")
        
        # 收集文件
        files = []
        for root, _, filenames in os.walk(directory):
            for filename in filenames:
                filepath = os.path.join(root, filename)
                files.append(filepath)
        
        if not files:
            print("没有找到文档文件")
            return
        
        # 批量索引
        for file_path in files:
            try:
                result = indexing_pipeline.run({"file_type_router": {"sources": [file_path]}})
                print(f"已索引: {file_path}")
            except Exception as e:
                print(f"索引失败 {file_path}: {e}")
        
        print(f"索引完成，共处理 {len(files)} 个文件")
​

4. 搭建问答管道

核心 RAG 管道只有三个组件：检索器、提示构建器、生成器：

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack import Pipeline

def build_qa_pipeline(config: AppConfig, document_store: InMemoryDocumentStore):
    """构建问答管道"""
    pipeline = Pipeline()
    
    # 文本向量化
    pipeline.add_component(
        "text_embedder",
        SentenceTransformersTextEmbedder(model=config.retriever.top_k)
    )
    
    # 向量检索
    pipeline.add_component(
        "retriever",
        InMemoryEmbeddingRetriever(
            document_store=document_store,
            top_k=config.retriever.top_k,
            score_threshold=config.retriever.score_threshold
        )
    )
    
    # 提示模板
    prompt_template = """
请根据以下参考信息回答用户的问题。如果参考信息中没有相关内容，请如实告知。

参考信息：
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

用户问题：{{ question }}

请用简洁、准确的中文回答。
"""
    pipeline.add_component("prompt_builder", PromptBuilder(template=prompt_template))
    
    # LLM 生成
    pipeline.add_component(
        "generator",
        OpenAIGenerator(
            model=config.generator.model,
            max_tokens=config.generator.max_tokens,
            temperature=config.generator.temperature
        )
    )
    
    # 连接
    pipeline.connect("text_embedder", "retriever")
    pipeline.connect("retriever.documents", "prompt_builder.documents")
    pipeline.connect("prompt_builder", "generator")
    
    return pipeline
​

5. 运行问答

整合所有组件，创建一个完整的问答服务：

class QASystem:
    """基础问答系统"""
    
    def __init__(self, config_path: str = "config.yaml"):
        self.config = load_config(config_path)
        self.kb = KnowledgeBase()
        self.pipeline = None
    
    def initialize(self, knowledge_dir: str):
        """初始化系统"""
        print("正在索引知识库...")
        self.kb.import_documents(knowledge_dir)
        print("正在构建问答管道...")
        self.pipeline = build_qa_pipeline(self.config, self.kb.document_store)
        print("系统就绪！")
    
    def ask(self, question: str) -> dict:
        """向系统提问"""
        if not self.pipeline:
            raise RuntimeError("请先调用 initialize() 初始化系统")
        
        result = self.pipeline.run({
            "text_embedder": {"text": question},
            "prompt_builder": {"question": question}
        })
        
        return {
            "question": question,
            "answer": result["generator"]["replies"][0],
            "meta": result["generator"]["meta"]
        }

# 使用示例
if __name__ == "__main__":
    qa = QASystem()
    qa.initialize("data/knowledge_base")
    
    questions = [
        "什么是 RAG？",
        "Haystack 支持哪些文档存储后端？",
        "如何优化检索效果？"
    ]
    
    for q in questions:
        print(f"\nQ: {q}")
        result = qa.ask(q)
        print(f"A: {result['answer']}")
​

6. 添加 API 服务

使用 FastAPI 将问答系统封装为 HTTP API：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Haystack QA System")
qa_system = QASystem()

class Query(BaseModel):
    question: str

@app.post("/ask")
async def ask(query: Query):
    result = qa_system.ask(query.question)
    return result

@app.get("/health")
async def health():
    return {"status": "ok"}

@app.on_event("startup")
async def startup():
    qa_system.initialize("data/knowledge_base")
​

7. 效果评估

搭建基础评估框架来衡量问答质量：

from haystack.evaluation import SASEvaluator

def evaluate_qa(pipeline, test_questions: list[dict]):
    """评估问答系统准确率"""
    evaluator = SASEvaluator()
    
    results = []
    for item in test_questions:
        prediction = pipeline.run({
            "text_embedder": {"text": item["question"]},
            "prompt_builder": {"question": item["question"]}
        })
        
        score = evaluator.compute_score(
            prediction=prediction["generator"]["replies"][0],
            reference=item["answer"]
        )
        results.append({
            "question": item["question"],
            "score": score
        })
    
    avg_score = sum(r["score"] for r in results) / len(results)
    print(f"平均 SAS 评分: {avg_score:.2f}")
    return results
​

架构总结

本节搭建了一个功能完整的基础问答系统，涵盖了文档索引、检索、生成和 API 服务四个关键环节。在接下来的小节中，我们将在这个基础上逐步添加多轮对话能力、企业级部署等高级特性。