Posted inAI

Xây dựng AI Pipeline dạng Modular với Haystack 2.0: Giải pháp thay thế sạch gọn cho LangChain trong Production RAG

AI tutorial - IT technology blog
AI tutorial - IT technology blog

So sánh cách tiếp cận: Explicit DAG vs. Lớp trừu tượng ma thuật của Framework

Đầu năm ngoái, nhóm của tôi tiếp nhận một dịch vụ tìm kiếm enterprise đa người dùng (multi-tenant) lập chỉ mục cho khoảng 2,5 triệu tài liệu nội bộ. Mỗi khi quá trình retrieval bị đình trệ hoặc độ trễ p99 tăng vọt quá 4.000ms, việc tìm ra nguyên nhân thực sự là một cơn ác mộng. Quá trình trace lỗi đòi hỏi phải lục tung các wrapper class lồng nhau, opaque callback và các dictionary ngữ cảnh đầy bí ẩn. Chúng tôi đã lãng phí hàng giờ vật lộn với cơ chế nội bộ của framework thay vì tinh chỉnh prompt và cải thiện chất lượng kết quả.

Sự ức chế đó đã thúc đẩy chúng tôi đánh giá Haystack 2.0, phiên bản được deepset viết lại hoàn toàn cho engine điều phối của họ. Kiến trúc này mang lại sự khác biệt rõ rệt trong cách dữ liệu di chuyển giữa các bước:

  • Cách tiếp cận Chain nguyên khối (ví dụ: LangChain đời đầu): Xây dựng dựa trên hệ thống kế thừa sâu và trạng thái chia sẻ ngầm định (implicit shared state). Các bước truyền qua lại những payload dictionary lỏng lẻo, nơi các key xuất hiện và biến mất mà không có ràng buộc rõ ràng. Cách này giúp làm prototype rất nhanh. Tuy nhiên, việc debug các nhánh logic động hoặc lỗi kiểu dữ liệu (type errors) trên production sẽ nhanh chóng trở nên hỗn loạn.
  • Cách tiếp cận Explicit DAG (Haystack 2.0): Mô hình hóa retriever, prompt template và LLM client thành các node độc lập trong một Đồ thị có hướng không chu trình (DAG). Mỗi node đều khai báo kiểu dữ liệu input và output chặt chẽ bằng type hints chuẩn của Python. Bạn kết nối trực tiếp các socket thông qua pipeline.connect("node_a.output_key", "node_b.input_key"). Nếu kiểu dữ liệu xung đột, đồ thị sẽ báo lỗi ngay khi build—trước rất lâu khi một payload không hợp lệ có cơ hội chạm tới production.

Ưu và nhược điểm: Đánh giá Haystack 2.0 cho các hệ thống thực tế

Chuyển từ các chain ngầm định sang graph tường minh mang lại nhiều lợi ích kỹ thuật rõ rệt, nhưng nó cũng đòi hỏi những đánh đổi có chủ đích.

Ưu điểm

  • Luồng dữ liệu có thể dự đoán: Do các kết nối socket được khai báo từng dòng rõ ràng, bạn luôn biết chính xác mỗi node nhận vào và trả ra những gì. Sẽ không có chuyện các key ẩn trong dictionary vô tình ghi đè lên input của prompt ở các bước sau.
  • Viết Custom Component đơn giản: Để viết một bộ lọc theo domain hay một bộ cắt gọn token, bạn chỉ cần một class thông thường đi kèm decorator @component và một method run() có khai báo type annotation. Bạn không bao giờ phải kế thừa từ các base class lồng nhau phức tạp.
  • Khả năng tuần tự hóa và kiểm tra trực quan: Các pipeline có thể xuất mượt mà sang định dạng YAML hoặc biểu đồ Mermaid ngay từ đầu. Bạn hoàn toàn có thể audit cấu trúc đồ thị trong các bước kiểm tra CI/CD trước khi đưa vào xử lý request thực tế.
  • Dung lượng gọn nhẹ: Package lõi haystack-ai rất tinh gọn. Các tích hợp bên thứ ba cho các vector database như Qdrant hay Milvus nằm trong các package tùy chọn riêng biệt.

Nhược điểm cần cân nhắc

  • Khởi tạo ban đầu tốn công hơn: Để kết nối một pipeline gồm 6 node, bạn cần viết 5 câu lệnh pipeline.connect() rõ ràng. Bạn phải đánh đổi sự tiện lợi của những dòng code ngắn dạng prototype để lấy một kiến trúc có tính thực thi cao.
  • Hệ sinh thái có trọng tâm: Các nhà cung cấp phổ biến (OpenAI, HuggingFace, Cohere, Pinecone) đều hoạt động rất mượt mà. Tuy nhiên, các công cụ mới từ cộng đồng thường chưa có sẵn connector, đòi hỏi bạn phải tự viết khoảng 15 dòng code wrapper.

Kiến trúc khuyến nghị: Tách biệt Ingestion và Retrieval

Hãy tránh việc gom cả indexing và truy vấn runtime vào chung một script nguyên khối. Việc tách chúng thành hai đồ thị chuyên biệt sẽ giúp cô lập các sự cố vận hành và giảm thiểu mức tiêu thụ tài nguyên:

  1. Ingestion Pipeline: Phân tích file thô, chia nhỏ văn bản thành các chunk 300 từ, tạo embedding theo batch và ghi vector vào kho lưu trữ. Pipeline này chạy dưới dạng background worker bất đồng bộ.
  2. Query Pipeline: Chuyển đổi truy vấn của người dùng trong runtime thành vector, truy xuất top-k kết quả phù hợp, điền vào prompt template và gọi LLM. Pipeline này chạy trực tiếp trong luồng API yêu cầu độ trễ thấp.

Coi document store như hạ tầng dùng chung trong khi tách rời chu kỳ ingestion và serving sẽ giúp độ trễ luôn ổn định. Việc re-index batch dưới nền sẽ không gây nghẽn các lượt tìm kiếm của người dùng.

Thiết lập môi trường ảo Python 3.10+ và cài đặt các dependency cần thiết:

# Tạo và kích hoạt môi trường ảo
python3 -m venv .venv
source .venv/bin/activate

# Cài đặt Haystack 2.x và thư viện local embedding gọn nhẹ
pip install --upgrade pip
pip install haystack-ai fastembed-haystack
export OPENAI_API_KEY="sk-your-openai-api-key"

Hướng dẫn triển khai: RAG dạng Modular và xử lý tùy biến

Dưới đây là một hệ thống RAG hoàn chỉnh từ đầu đến cuối sử dụng local embedding và cách kết nối graph tường minh.

Bước 1: Xây dựng Document Ingestion Pipeline

Chúng ta cấu hình một InMemoryDocumentStore, phân tách tài liệu thành các chunk 100 từ, tính toán embedding cục bộ bằng FastEmbed (sử dụng bge-small-en-v1.5) và ghi các vector vào bộ nhớ.

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack_integrations.components.embedders.fastembed import FastembedDocumentEmbedder

# 1. Khởi tạo Document Store dùng chung
document_store = InMemoryDocumentStore(embedding_similarity_function="cosine")

# 2. Khởi tạo từng component riêng lẻ
splitter = DocumentSplitter(split_by="word", split_length=100, split_overlap=20)
embedder = FastembedDocumentEmbedder(model="BAAI/bge-small-en-v1.5")
writer = DocumentWriter(document_store=document_store)

# 3. Ráp các node vào DAG
ingestion_pipeline = Pipeline()
ingestion_pipeline.add_component("splitter", splitter)
ingestion_pipeline.add_component("embedder", embedder)
ingestion_pipeline.add_component("writer", writer)

# 4. Kết nối các socket tường minh (node.output -> node.input)
ingestion_pipeline.connect("splitter.documents", "embedder.documents")
ingestion_pipeline.connect("embedder.documents", "writer.documents")

# Chạy indexing với dữ liệu mẫu
raw_docs = [
    Document(content="Haystack 2.0 sử dụng đồ thị có hướng để kết nối các component tường minh và tạo ra pipeline có tính tất định."),
    Document(content="Kiến trúc RAG modular tách biệt ingestion pipeline khỏi query pipeline trong runtime để đảm bảo độ tin cậy."),
    Document(content="Các custom component trong Haystack chỉ cần decorator @component và các tham số hàm run có định kiểu.")
]

ingestion_pipeline.run({"splitter": {"documents": raw_docs}})
print(f"Đã lập chỉ mục thành công {document_store.count_documents()} document chunk.")

Bước 2: Thiết kế Query và Generation Pipeline

Query pipeline sẽ chạy khi có request từ người dùng. Nó tạo embedding cho truy vấn, tìm kiếm các kết quả phù hợp nhất, định dạng prompt và gửi prompt tới OpenAI.

from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack_integrations.components.embedders.fastembed import FastembedTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

# Định nghĩa template chặt chẽ với quy tắc dự phòng (fallback) rõ ràng
prompt_template = """
Dựa vào các tài liệu ngữ cảnh sau đây, hãy trả lời câu hỏi một cách chính xác.
Nếu câu trả lời không có trong ngữ cảnh, hãy ghi rõ 'Thông tin không có trong ngữ cảnh.'

Ngữ cảnh:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

Câu hỏi: {{ query }}
Trả lời:
"""

# Khởi tạo các query component
text_embedder = FastembedTextEmbedder(model="BAAI/bge-small-en-v1.5")
retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=2)
prompt_builder = PromptBuilder(template=prompt_template)
generator = OpenAIGenerator(model="gpt-4o-mini")

# Xây dựng DAG truy vấn
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", text_embedder)
query_pipeline.add_component("retriever", retriever)
query_pipeline.add_component("prompt_builder", prompt_builder)
query_pipeline.add_component("generator", generator)

# Kết nối các component một cách tường minh
query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
query_pipeline.connect("retriever.documents", "prompt_builder.documents")
query_pipeline.connect("prompt_builder.prompt", "generator.prompt")

# Thực thi truy vấn
user_query = "Haystack 2.0 kết nối các component như thế nào?"
response = query_pipeline.run({
    "text_embedder": {"text": user_query},
    "prompt_builder": {"query": user_query}
})

print("Kết quả từ LLM:")
print(response["generator"]["replies"][0])

Bước 3: Tạo Custom Component để đảm bảo an toàn và định tuyến

Các custom node chính là điểm sáng giúp Haystack phát huy tối đa sức mạnh trong production. Nếu điểm retrieval tụt xuống dưới ngưỡng chấp nhận được, việc định tuyến truy vấn thẳng tới luồng fallback sẽ giúp tiết kiệm chi phí token LLM và ngăn chặn tình trạng hallucination.

from typing import List
from haystack import component
from haystack.dataclasses import Document

@component
class RelevanceFilter:
    def __init__(self, min_score: float = 0.65):
        self.min_score = min_score

    @component.output_types(passed_docs=List[Document], fallback=bool)
    def run(self, documents: List[Document]):
        # Loại bỏ các chunk có độ tin cậy thấp trước khi đưa vào prompt cho mô hình
        filtered = [doc for doc in documents if (doc.score or 0.0) >= self.min_score]
        if not filtered:
            return {"passed_docs": [], "fallback": True}
        return {"passed_docs": filtered, "fallback": False}

Đặt RelevanceFilter vào giữa retriever và prompt builder sẽ ngăn chặn ngữ cảnh không liên quan làm loãng prompt của bạn. Trong các tác vụ thực tế của chúng tôi, bước kiểm tra đơn giản này đã giúp cắt giảm 18% lượng token không cần thiết.

Lời khuyên thực chiến cho môi trường Production

  • Kiểm tra cấu trúc Graph trong CI: Xuất schema của pipeline bằng query_pipeline.dumps(), hoặc tạo sơ đồ kiến trúc với query_pipeline.draw(path="graph.png") để review các thay đổi kết nối ngay trong pull request.
  • Minh bạch hóa các Input dùng chung: Khi hai component ở phía hạ nguồn cùng cần dữ liệu từ một nguồn phía trên, hãy chỉ định các key ánh xạ socket rõ ràng để tránh nhầm lẫn khi định tuyến.
  • Duy trì tính chất Stateless: Hãy coi các instance pipeline là các executor phi trạng thái (stateless). Việc gọi run() đồng thời trên nhiều worker thread là hoàn toàn an toàn, miễn là các custom component của bạn không chỉnh sửa các thuộc tính instance nội bộ trong quá trình thực thi.
Share:
Tags: