Haystack 2.0: Xây dựng pipeline xử lý tài liệu với Document Indexing, Hybrid Retrieval và Q&A thông minh – ITFROMZERO

Table of Contents

Vấn đề với RAG pipeline thông thường

Mình từng xây một hệ thống Q&A nội bộ bằng LangChain — chạy được, nhưng khi cần thêm một bước xử lý mới vào giữa pipeline, mọi thứ bắt đầu rối. Code lồng callback vào callback, debug mất cả buổi chiều chỉ để hiểu data đang chạy qua đâu.

Haystack (của deepset) giải quyết đúng điểm đau này: pipeline là một đồ thị có hướng (DAG), từng component kết nối tường minh. Thêm một bước preprocessing? Chèn component vào. Muốn swap embedding model? Thay một dòng. Không cần đụng logic còn lại.

Bài viết này đi thẳng vào 3 phần thực tế: indexing tài liệu đúng cách, hybrid retrieval để tăng độ chính xác, và Q&A pipeline sẵn sàng production.

Quick Start — Pipeline Q&A hoạt động trong 5 phút

Cài đặt

pip install haystack-ai openai

Pipeline Q&A đơn giản nhất

from haystack import Document, Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
from haystack.document_stores.in_memory import InMemoryDocumentStore

# 1. Document store + indexing
store = InMemoryDocumentStore()
store.write_documents([
    Document(content="Haystack là framework Python để xây dựng AI pipeline xử lý tài liệu."),
    Document(content="BM25 là thuật toán tìm kiếm dựa trên từ khóa, phù hợp với văn bản kỹ thuật."),
    Document(content="Embedding retrieval dùng vector similarity để tìm tài liệu liên quan về mặt ngữ nghĩa."),
])

# 2. Xây pipeline
template = """
Dựa trên các tài liệu sau, trả lời câu hỏi bằng tiếng Việt:

{% for doc in documents %}
{{ doc.content }}
{% endfor %}

Câu hỏi: {{ question }}
"""

pipe = Pipeline()
pipe.add_component("retriever", InMemoryBM25Retriever(document_store=store))
pipe.add_component("prompt", PromptBuilder(template=template))
pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

pipe.connect("retriever", "prompt.documents")
pipe.connect("prompt", "llm")

# 3. Chạy
result = pipe.run({
    "retriever": {"query": "Haystack là gì?"},
    "prompt": {"question": "Haystack là gì?"}
})
print(result["llm"]["replies"][0])

Cấu trúc này rõ ràng hơn nhiều so với chain lồng nhau: mỗi component nhận input, xử lý, trả output — pipeline connect chúng lại. Dễ debug, dễ test từng bước.

Document Indexing — Làm đúng từ đầu

Phần lớn hệ thống RAG cho kết quả kém không phải vì LLM tệ, mà vì indexing sai. Tài liệu chunk quá lớn, không có metadata, hoặc text sau khi convert từ PDF chưa được clean.

Indexing pipeline đầy đủ cho PDF

from haystack import Pipeline
from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.embedders import OpenAIDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from pathlib import Path

store = InMemoryDocumentStore()

indexing_pipe = Pipeline()
indexing_pipe.add_component("converter", PyPDFToDocument())
indexing_pipe.add_component("cleaner", DocumentCleaner(
    remove_empty_lines=True,
    remove_extra_whitespaces=True,
))
indexing_pipe.add_component("splitter", DocumentSplitter(
    split_by="sentence",
    split_length=5,   # 5 câu mỗi chunk
    split_overlap=1,  # overlap 1 câu để không mất context
))
indexing_pipe.add_component("embedder", OpenAIDocumentEmbedder(
    model="text-embedding-3-small"
))
indexing_pipe.add_component("writer", DocumentWriter(document_store=store))

indexing_pipe.connect("converter", "cleaner")
indexing_pipe.connect("cleaner", "splitter")
indexing_pipe.connect("splitter", "embedder")
indexing_pipe.connect("embedder", "writer")

# Index tất cả PDF trong thư mục
pdf_files = list(Path("./docs").glob("*.pdf"))
indexing_pipe.run({"converter": {"sources": pdf_files}})
print(f"Đã index {store.count_documents()} chunks")

Với tài liệu kỹ thuật, split_by="sentence" và split_length=5 là combo mình hay dùng — chunk nhỏ giúp retrieval chính xác hơn, overlap 1 câu tránh mất context ở ranh giới chunk.

Thêm metadata để filter sau này

from haystack import Document

docs = [
    Document(
        content="Nội dung tài liệu kỹ thuật về Docker...",
        meta={
            "source": "docker-guide.pdf",
            "category": "devops",
            "language": "vi",
        }
    )
]
store.write_documents(docs)

# Retrieval với filter theo category
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
retriever = InMemoryBM25Retriever(document_store=store)
results = retriever.run(
    query="cấu hình Docker",
    filters={"field": "meta.category", "operator": "==", "value": "devops"}
)

Hybrid Retrieval — Kết hợp BM25 và Embedding

BM25 giỏi với từ khóa chính xác: tên riêng, mã lỗi, command cụ thể. Embedding retrieval giỏi với câu hỏi diễn đạt khác nhau nhưng cùng ý nghĩa. Hybrid kết hợp cả hai — đây là lý do retrieval quality tăng rõ rệt trong thực tế.

from haystack import Pipeline
from haystack.components.retrievers.in_memory import (
    InMemoryBM25Retriever,
    InMemoryEmbeddingRetriever
)
from haystack.components.embedders import OpenAITextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

template = """
Dựa trên tài liệu sau, trả lời câu hỏi:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}
Câu hỏi: {{ question }}
"""

query_pipe = Pipeline()
query_pipe.add_component("text_embedder", OpenAITextEmbedder(
    model="text-embedding-3-small"
))
query_pipe.add_component("embedding_retriever", InMemoryEmbeddingRetriever(
    document_store=store, top_k=10
))
query_pipe.add_component("bm25_retriever", InMemoryBM25Retriever(
    document_store=store, top_k=10
))
query_pipe.add_component("joiner", DocumentJoiner(
    join_mode="reciprocal_rank_fusion"  # RRF: kết hợp ranking tốt nhất
))
query_pipe.add_component("prompt", PromptBuilder(template=template))
query_pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

# Connect
query_pipe.connect("text_embedder.embedding", "embedding_retriever.query_embedding")
query_pipe.connect("embedding_retriever", "joiner")
query_pipe.connect("bm25_retriever", "joiner")
query_pipe.connect("joiner", "prompt.documents")
query_pipe.connect("prompt", "llm")

# Chạy
result = query_pipe.run({
    "text_embedder": {"text": "Cách debug lỗi container không start?"},
    "bm25_retriever": {"query": "Cách debug lỗi container không start?"},
    "prompt": {"question": "Cách debug lỗi container không start?"}
})
print(result["llm"]["replies"][0])

Số đo từ một dự án knowledge base nội bộ: hybrid retrieval cải thiện recall khoảng 15–20% so với embedding đơn thuần. Chênh lệch rõ nhất khi query chứa tên sản phẩm, mã lỗi, hoặc command cụ thể — đúng thứ BM25 xử lý tốt hơn embedding.

Nâng cao — Persistent Store và Production Tips

Dùng ChromaDB thay InMemory

pip install chroma-haystack

from chroma_haystack import ChromaDocumentStore

store = ChromaDocumentStore(
    collection_name="my_docs",
    persist_path="./chroma_db"  # Dữ liệu được lưu lại sau khi restart
)

Serialize pipeline để version và reuse

import yaml

# Lưu pipeline config
with open("pipeline.yaml", "w") as f:
    yaml.dump(query_pipe.to_dict(), f)

# Load lại sau — không cần rebuild từ đầu
from haystack import Pipeline
with open("pipeline.yaml") as f:
    pipe_dict = yaml.safe_load(f)
query_pipe = Pipeline.from_dict(pipe_dict)

Debug từng component độc lập

# Test retriever mà không cần chạy cả pipeline
retriever = InMemoryBM25Retriever(document_store=store)
docs = retriever.run(query="Docker", top_k=3)

print(f"Tìm được {len(docs['documents'])} tài liệu")
for doc in docs["documents"]:
    print(f"Score: {doc.score:.3f} | {doc.content[:100]}...")

Đây là thứ mình thấy Haystack làm tốt hơn LangChain: test từng component riêng lẻ. Khi hệ thống trả lời sai, mình khoanh vùng ngay được vấn đề ở indexing, retrieval, hay prompt — không cần log dump toàn bộ chain.

Tips thực tế khi chọn Haystack

Dùng Haystack khi cần pipeline rõ ràng, nhiều document source, cần serialize/version pipeline, hoặc team nhiều người maintain cùng lúc.
Dùng LangChain khi cần prototype nhanh với nhiều integration sẵn có và đã quen với LangChain ecosystem.
Dùng custom code khi pipeline chỉ 1–2 bước, không muốn phụ thuộc framework.

Dễ nhầm một điểm: Haystack 2.0 thay đổi kiến trúc hoàn toàn so với v1. Nếu tutorial bạn đọc có API trông khác — đó là v1 cũ. Package hiện tại là haystack-ai, không phải farm-haystack. V2 type-safe hơn, pipeline tường minh hơn, và debug dễ hơn hẳn.