Haystack - Framework NLP untuk Produksi

Daftar Isi

Pendahuluan

Haystack adalah framework NLP open-source yang dirancang untuk membangun sistem pencarian dan tanya-jawab siap produksi. Dikembangkan oleh deepset, Haystack menyediakan arsitektur modular berbasis pipeline yang memudahkan penggabungan retriever, reader, generator, dan komponen NLP lainnya menjadi aplikasi yang powerful.

Dengan Haystack 2.x, framework ini telah dirancang ulang dengan sistem pipeline fleksibel yang mendukung alur kerja AI ekstraktif maupun generatif. Baik Anda membangun sistem pencarian dokumen sederhana maupun pipeline Retrieval-Augmented Generation (RAG) yang kompleks, Haystack menyediakan blok pembangun dan penghubung untuk menggabungkannya.

Tutorial ini mencakup konsep inti Haystack, memandu Anda membangun pipeline praktis, menunjukkan cara membuat komponen kustom, dan mendemonstrasikan cara men-deploy sistem Anda sebagai REST API yang siap untuk lalu lintas produksi.

Prasyarat

• Python 3.9 atau lebih tinggi
• Pemahaman dasar konsep NLP (embedding, retrieval, generation)
• API key OpenAI (untuk komponen generator)

pip install haystack-ai "sentence-transformers>=2.2.0" "opensearch-haystack" fastapi uvicorn

export OPENAIAPIKEY="sk-kunci-anda-di-sini"

Memahami Arsitektur Haystack

Haystack 2.x dibangun di atas tiga konsep inti:

Komponen

Komponen adalah blok pembangun Haystack. Setiap komponen melakukan tugas spesifik seperti mengambil dokumen, menghasilkan teks, atau mengonversi format file. Komponen mendeklarasikan input dan output secara eksplisit, memungkinkan framework memvalidasi koneksi saat konstruksi pipeline.

Pipeline

Pipeline menghubungkan komponen menjadi graf berarah asiklik (DAG). Data mengalir dari satu komponen ke komponen berikutnya berdasarkan koneksi input/output yang dideklarasikan. Pipeline bisa bersifat linear atau bercabang.

Document Store

Document store adalah lapisan persistensi. Mereka menyimpan dokumen beserta metadata dan embedding terkait, menyediakan antarmuka pengambilan untuk komponen pipeline.

from haystack import Pipeline, Document
from haystack.components.writers import DocumentWriter
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter

Pipeline sederhana yang membersihkan dan memecah dokumen
preprocessing = Pipeline()
preprocessing.addcomponent("pembersih", DocumentCleaner())

preprocessing.addcomponent("pemecah", DocumentSplitter(
    splitby="sentence",

    splitlength=3,
    splitoverlap=1,

))
preprocessing.connect("pembersih", "pemecah")

Jalankan pipeline
dokumenlist = [
    Document(content="""Haystack adalah framework NLP open-source. Framework ini menyediakan
    komponen modular untuk membangun sistem pencarian. Framework mendukung pendekatan
    ekstraktif dan generatif. Anda dapat menggabungkan beberapa komponen menjadi pipeline.
    Pipeline dapat di-deploy sebagai REST API.""")
]

hasil = preprocessing.run({"pembersih": {"documents": dokumenlist}})

for dok in hasil["pemecah"]["documents"]:
    print(f"  Potongan: {dok.content[:80]}...")

from haystack.documentstores.inmemory import InMemoryDocumentStore

Buat document store in-memory
store = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")


Tulis dokumen
dokumen = [
    Document(
        content="Python adalah bahasa pemrograman serbaguna yang digunakan dalam data science.",
        meta={"kategori": "pemrograman", "level": "pemula"}
    ),
    Document(
        content="Container Docker menyederhanakan deployment dan scaling aplikasi.",
        meta={"kategori": "devops", "level": "menengah"}
    ),
    Document(
        content="PostgreSQL adalah sistem database relasional open-source yang powerful.",
        meta={"kategori": "database", "level": "menengah"}
    ),
    Document(
        content="Kubernetes mengorkestrasikan aplikasi terkontainerisasi dalam skala besar.",
        meta={"kategori": "devops", "level": "lanjut"}
    ),
    Document(
        content="FastAPI adalah framework web Python modern untuk membangun REST API.",
        meta={"kategori": "pemrograman", "level": "menengah"}
    ),
    Document(
        content="Redis menyediakan caching in-memory untuk aplikasi berkinerja tinggi.",
        meta={"kategori": "database", "level": "menengah"}
    ),
    Document(
        content="Model machine learning membutuhkan dataset besar untuk pelatihan yang efektif.",
        meta={"kategori": "ai", "level": "pemula"}
    ),
    Document(
        content="Arsitektur Transformers merevolusi pemrosesan bahasa alami.",
        meta={"kategori": "ai", "level": "lanjut"}
    ),
]

store.writedocuments(dokumen)
print(f"Dokumen di store: {store.countdocuments()}")

from opensearchhaystack.documentstore import OpenSearchDocumentStore

Hubungkan ke kluster OpenSearch
opensearchstore = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="dokumenhaystack",

    embeddingdim=384,
    returnembedding=True,

)

Tulis dokumen dengan pipeline
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder

pengindeksan = Pipeline()
pengindeksan.addcomponent("embedder", SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
))
pengindeksan.addcomponent("penulis", DocumentWriter(documentstore=opensearchstore))

pengindeksan.connect("embedder.documents", "penulis.documents")

Indeks dokumen
pengindeksan.run({"embedder": {"documents": dokumen}})

from haystack.components.embedders import (
    SentenceTransformersTextEmbedder,
    SentenceTransformersDocumentEmbedder,
)
from haystack.components.retrievers.inmemory import InMemoryEmbeddingRetriever

Pertama, embed dan simpan dokumen
docembedder = SentenceTransformersDocumentEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
)
docembedder.warmup()

dokdenganembedding = docembedder.run(documents=dokumen)

store = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")
store.writedocuments(dokdenganembedding["documents"])


Bangun pipeline pengambilan
pipelinepengambilan = Pipeline()
pipelinepengambilan.addcomponent("textembedder", SentenceTransformersTextEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
))
pipelinepengambilan.addcomponent("retriever", InMemoryEmbeddingRetriever(

    documentstore=store,
    topk=3,

))
pipelinepengambilan.connect("textembedder.embedding", "retriever.queryembedding")

Pencarian
hasil = pipelinepengambilan.run({"textembedder": {"text": "alat orkestrasikan container"}})

print("Dokumen yang diambil:")
for dok in hasil["retriever"]["documents"]:
    print(f"  [{dok.score:.4f}] {dok.content}")
    print(f"    Meta: {dok.meta}")

Pengambilan Kata Kunci BM25

from haystack.components.retrievers.inmemory import InMemoryBM25Retriever

BM25 tidak memerlukan embedding
bm25store = InMemoryDocumentStore()
bm25store.writedocuments(dokumen)

bm25retriever = InMemoryBM25Retriever(documentstore=bm25store, topk=3)

Pencarian
hasil = bm25retriever.run(query="database PostgreSQL")


print("Hasil BM25:")
for dok in hasil["documents"]:
    print(f"  [{dok.score:.4f}] {dok.content}")

from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import TransformersSimilarityRanker

pipelinehybrid = Pipeline()

Cabang pengambilan embedding
pipelinehybrid.addcomponent("textembedder", SentenceTransformersTextEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
))
pipelinehybrid.addcomponent("embeddingretriever", InMemoryEmbeddingRetriever(
    documentstore=store, topk=5
))

Cabang pengambilan BM25
pipelinehybrid.addcomponent("bm25retriever", InMemoryBM25Retriever(

    documentstore=bm25store, topk=5
))

Gabungkan dan rerank
pipelinehybrid.addcomponent("penggabung", DocumentJoiner())
pipelinehybrid.addcomponent("peranking", TransformersSimilarityRanker(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2", topk=3

))

Hubungkan komponen
pipelinehybrid.connect("textembedder.embedding", "embeddingretriever.queryembedding")

pipelinehybrid.connect("embeddingretriever", "penggabung")

pipelinehybrid.connect("bm25retriever", "penggabung")

pipelinehybrid.connect("penggabung", "peranking")

Jalankan pencarian hybrid
hasil = pipelinehybrid.run({

    "textembedder": {"text": "alat deployment untuk aplikasi"},
    "bm25retriever": {"query": "alat deployment untuk aplikasi"},

    "peranking": {"query": "alat deployment untuk aplikasi"},
})

print("Hasil pencarian hybrid:")
for dok in hasil["peranking"]["documents"]:
    print(f"  [{dok.score:.4f}] {dok.content}")

from haystack.components.readers import ExtractiveReader

reader = ExtractiveReader(model="deepset/roberta-base-squad2")
reader.warmup()

Ekstrak jawaban dari dokumen
hasil = reader.run(
    query="Apa itu PostgreSQL?",
    documents=[
        Document(content="PostgreSQL adalah sistem database relasional open-source yang powerful."),
        Document(content="Redis menyediakan caching in-memory untuk aplikasi berkinerja tinggi."),
    ],
    topk=2,

)

for jawaban in hasil["answers"]:
    print(f"  Jawaban: '{jawaban.data}' (skor: {jawaban.score:.4f})")
    print(f"  Konteks: {jawaban.document.content}")

from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

Buat prompt builder
templateprompt = """Berdasarkan dokumen konteks berikut, jawab pertanyaan tersebut.
Jika jawaban tidak ditemukan dalam konteks, katakan "Saya tidak memiliki informasi yang cukup."

Konteks:
{% for dok in documents %}
{{ dok.content }}
{% endfor %}


Pertanyaan: {{ question }}

Jawaban:"""

promptbuilder = PromptBuilder(template=templateprompt)
generator = OpenAIGenerator(model="gpt-4o")

Bangun pipeline QA generatif sederhana
pipelinegen = Pipeline()

pipelinegen.addcomponent("promptbuilder", promptbuilder)

pipelinegen.addcomponent("generator", generator)

pipelinegen.connect("promptbuilder", "generator")


Jalankan
dokkonteks = [
    Document(content="PostgreSQL adalah sistem database relasional open-source yang powerful."),
    Document(content="PostgreSQL mendukung tipe data lanjutan, full-text search, dan JSON."),
]

hasil = pipelinegen.run({

    "promptbuilder": {
        "documents": dokkonteks,

        "question": "Fitur apa saja yang didukung PostgreSQL?"
    }
})

print(f"Jawaban: {hasil['generator']['replies'][0]}")

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

def bangunpipelinerag(documentstore: InMemoryDocumentStore) -> Pipeline:

    """Bangun pipeline RAG lengkap."""

    promptrag = """Anda adalah asisten teknis yang membantu. Jawab pertanyaan berdasarkan
konteks yang diberikan. Berikan jawaban spesifik dan kutip informasi yang relevan.

Konteks:
{% for dok in documents %}
[Sumber {{ loop.index }}] {{ dok.content }}
{% endfor %}

Pertanyaan: {{ question }}

Instruksi:
Jawab hanya berdasarkan konteks yang diberikan
Referensikan sumber menggunakan notasi [Sumber N]
Jika konteks tidak memuat jawaban, nyatakan dengan jelas
Ringkas namun menyeluruh

Jawaban:"""

    pipeline = Pipeline()

    # Embedding query
    pipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(
        model="sentence-transformers/all-MiniLM-L6-v2"
    ))

    # Pengambilan dokumen
    pipeline.addcomponent("retriever", InMemoryEmbeddingRetriever(

        documentstore=documentstore,

        topk=5,
    ))

    # Konstruksi prompt
    pipeline.addcomponent("promptbuilder", PromptBuilder(template=promptrag))


    # Generasi LLM
    pipeline.addcomponent("generator", OpenAIGenerator(
        model="gpt-4o",
        generationkwargs={"temperature": 0.3, "maxtokens": 500},
    ))

    # Hubungkan pipeline
    pipeline.connect("textembedder.embedding", "retriever.queryembedding")
    pipeline.connect("retriever.documents", "promptbuilder.documents")

    pipeline.connect("promptbuilder", "generator")

    return pipeline


Bangun dan jalankan
rag = bangunpipelinerag(store)

pertanyaanlist = [

    "Apa alat terbaik untuk orkestrasikan container?",
    "Bahasa pemrograman mana yang terbaik untuk data science?",
    "Bagaimana cara meningkatkan performa database?",
]

for pertanyaan in pertanyaanlist:
    hasil = rag.run({
        "textembedder": {"text": pertanyaan},

        "promptbuilder": {"question": pertanyaan},
    })
    print(f"\nT: {pertanyaan}")
    print(f"J: {hasil['generator']['replies'][0]}")

RAG dengan Filter Metadata

def bangunragterfilter(documentstore: InMemoryDocumentStore) -> Pipeline:
    """Pipeline RAG dengan dukungan filter metadata."""

    promptrag = """Jawab pertanyaan berdasarkan konteks di bawah.

Konteks:
{% for dok in documents %}
{{ dok.content }} [{{ dok.meta.kategori }}/{{ dok.meta.level }}]
{% endfor %}


Pertanyaan: {{ question }}
Jawaban:"""

    pipeline = Pipeline()
    pipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(
        model="sentence-transformers/all-MiniLM-L6-v2"
    ))
    pipeline.addcomponent("retriever", InMemoryEmbeddingRetriever(

        documentstore=documentstore, topk=3
    ))
    pipeline.addcomponent("promptbuilder", PromptBuilder(template=promptrag))

    pipeline.addcomponent("generator", OpenAIGenerator(model="gpt-4o"))

    pipeline.connect("textembedder.embedding", "retriever.queryembedding")
    pipeline.connect("retriever.documents", "promptbuilder.documents")

    pipeline.connect("promptbuilder", "generator")

    return pipeline


ragterfilter = bangunragterfilter(store)


hasil = ragterfilter.run({
    "textembedder": {"text": "Alat DevOps apa yang harus saya pelajari?"},

    "retriever": {"filters": {"field": "meta.kategori", "operator": "==", "value": "devops"}},
    "promptbuilder": {"question": "Alat DevOps apa yang harus saya pelajari?"},
})

print(f"Jawaban: {hasil['generator']['replies'][0]}")

Komponen Kustom

Kekuatan Haystack terletak pada kemampuan perluasannya. Anda dapat membuat komponen kustom untuk tugas apa pun.

Membuat Komponen Kustom

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

@component
class FilterKonten:
    """Komponen kustom yang memfilter dokumen berdasarkan panjang konten dan keberadaan kata kunci."""

    def init(self, panjangminimal: int = 10, katakunciwajib: Optional[List[str]] = None):

        self.panjangminimal = panjangminimal

        self.katakunciwajib = katakunciwajib or []


    @component.outputtypes(documents=List[Document], jumlahterfilter=int)

    def run(self, documents: List[Document]) -> dict:
        terfilter = []
        for dok in documents:
            if len(dok.content) < self.panjangminimal:
                continue

            if self.katakunciwajib:
                kontenlower = dok.content.lower()

                if not any(kk.lower() in kontenlower for kk in self.katakunciwajib):
                    continue

            terfilter.append(dok)

        return {
            "documents": terfilter,
            "jumlahterfilter": len(documents) - len(terfilter),

        }


Gunakan komponen kustom dalam pipeline
pipeline = Pipeline()
pipeline.addcomponent("filter", FilterKonten(
    panjangminimal=20,

    katakunciwajib=["Python", "database"]

))

hasil = pipeline.run({"filter": {"documents": dokumen}})
print(f"Terfilter: {hasil['filter']['jumlahterfilter']} dokumen dihapus")

Pengaya Dokumen Kustom

@component
class PengayaMetadata:
    """Memperkaya dokumen dengan field metadata yang dihitung."""

    def init(self, hitungwaktubaca: bool = True, hitungjumlahkata: bool = True):
        self.hitungwaktubaca = hitungwaktubaca
        self.hitungjumlahkata = hitungjumlahkata

    @component.outputtypes(documents=List[Document])

    def run(self, documents: List[Document]) -> dict:
        diperkaya = []
        for dok in documents:
            metabaru = dict(dok.meta)

            if self.hitungjumlahkata:
                metabaru["jumlahkata"] = len(dok.content.split())

            if self.hitungwaktubaca:
                kata = len(dok.content.split())
                metabaru["waktubacamenit"] = round(kata / 200, 1)


            metabaru["adakode"] = "

content=dok.content, meta=metabaru, embedding=dok.embedding,

))

return {"documents": diperkaya}

Gunakan dalam pipeline

pipelinepengayaan = Pipeline()

pipelinepengayaan.addcomponent("pengaya", PengayaMetadata())

hasil = pipelinepengayaan.run({"pengaya": {"documents": dokumen}})

for dok in hasil["pengaya"]["documents"]:

print(f" {dok.content[:50]}... | Kata: {dok.meta.get('jumlahkata')} | "

f"Baca: {dok.meta.get('waktubacamenit')} menit")

Peranking Kustom dengan Logika Bisnis

def init(self, kategoriboost: Optional[List[str]] = None, faktorboost: float = 1.5):

self.kategoriboost = kategoriboost or []

self.faktorboost = faktorboost

@component.outputtypes(documents=List[Document])

def run(self, documents: List[Document], topk: int = 5) -> dict:

dokberskor = []

for dok in documents:

skor = dok.score or 0.0

if dok.meta.get("kategori") in self.kategoriboost:

skor = 1.1

dokberskor.append((dok, skor))

dokberskor.sort(key=lambda x: x[1], reverse=True)

hasildok = []

for dok, skor in dokberskor[:topk]:

hasildok.append(dok)

return {"documents": hasildok}

Evaluasi Pipeline

Mengevaluasi pipeline Anda sangat penting untuk memahami kekuatan dan kelemahannya.

def evaluasipengambilan(self, kasusuji: list[dict], k: int = 5) -> dict:

kasusuji: list dari {"query": str, "iddokrelevan": set}

"""

metrik = {"presisi": [], "recall": [], "mrr": []}

for kasus in kasusuji:

"textembedder": {"text": kasus["query"]},

})

dokdiambil = hasil.get("retriever", {}).get("documents", [])

iddiambil = {dok.id for dok in dokdiambil[:k]}

idrelevan = kasus["iddokrelevan"]

# Presisi@k

presisi = len(iddiambil & idrelevan) / len(iddiambil) if iddiambil else 0

metrik["presisi"].append(presisi)

# Recall@k

recall = len(iddiambil & idrelevan) / len(idrelevan) if idrelevan else 0

metrik["recall"].append(recall)

# MRR

mrr = 0

for i, dok in enumerate(dokdiambil[:k], 1):

if dok.id in idrelevan:

mrr = 1.0 / i

break

metrik["mrr"].append(mrr)

return {

nama: {"ratarata": float(np.mean(nilai)), "std": float(np.std(nilai))}

def evaluasigenerasi(self, kasusuji: list[dict]) -> dict:

skorlist = []

for kasus in kasusuji:

"textembedder": {"text": kasus["query"]},

"promptbuilder": {"question": kasus["query"]},

promptjuri = f"""Beri nilai jawaban berikut pada skala 1-5.

Jawaban yang diharapkan: {kasus['jawabandiharapkan']}

hasiljuri = juri.run(prompt=promptjuri)

skor = int(hasiljuri["replies"][0].strip())

skorlist.append(min(max(skor, 1), 5))

skorlist.append(1)

return {

"skorratarata": float(np.mean(skorlist)),

"skorminimum": int(min(skorlist)),

"skormaksimum": int(max(skorlist)),

"totaldievaluasi": len(skorlist),

Deployment REST API

Deploy pipeline Haystack Anda sebagai REST API siap produksi.

pipelinerag = None

documentstore = None

@app.onevent("startup")

async def startup():

global pipelinerag, documentstore

documentstore = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")

docembedder = SentenceTransformersDocumentEmbedder(

model="sentence-transformers/all-MiniLM-L6-v2"

)

docembedder.warmup()

hasil = docembedder.run(documents=dokumen)

documentstore.writedocuments(hasil["documents"])

pipelinerag = bangunpipelinerag(documentstore)

topk: int = Field(default=5, ge=1, le=20, description="Jumlah dokumen yang diambil")

filter: Optional[dict] = Field(default=None, description="Filter metadata")

class DokumenSumber(BaseModel):

konten: str

skor: float

metadata: dict

class ResponQuery(BaseModel):

jawaban: str

sumber: list[DokumenSumber]

query: str

@app.post("/query", responsemodel=ResponQuery)

if pipelinerag is None:

raise HTTPException(statuscode=503, detail="Pipeline belum diinisialisasi")

paramrun = {

"textembedder": {"text": permintaan.pertanyaan},

"promptbuilder": {"question": permintaan.pertanyaan},

"retriever": {"topk": permintaan.topk},

}

if permintaan.filter:

paramrun["retriever"]["filters"] = permintaan.filter

hasil = pipelinerag.run(paramrun)

dokretriever = hasil["retriever"]["documents"]

sumber = [

DokumenSumber(

konten=dok.content,

skor=dok.score or 0.0,

metadata=dok.meta,

)

for dok in dokretriever

raise HTTPException(statuscode=500, detail=str(e))

class PermintaanIndeks(BaseModel):

dokumen: list[dict] = Field(description="Dokumen dengan kunci 'content' dan 'meta' opsional")

@app.post("/indeks", statuscode=201)

async def indeksdokumen(permintaan: PermintaanIndeks):

"""Indeks dokumen baru ke document store."""

dokbaru = [

docembedder = SentenceTransformersDocumentEmbedder(

model="sentence-transformers/all-MiniLM-L6-v2"

)

docembedder.warmup()

hasil = docembedder.run(documents=dokbaru)

documentstore.writedocuments(hasil["documents"])

return {"terindeks": len(dokbaru), "total": documentstore.countdocuments()}

"pipelinesiap": pipelinerag is not None,

"dokumen": documentstore.countdocuments() if documentstore else 0,

}

if name == "main":

uvicorn.run(app, host="0.0.0.0", port=8000)

```

Praktik Terbaik

Kesimpulan

Haystack menyediakan framework yang powerful dan modular untuk membangun sistem NLP produksi. Arsitektur pipeline-nya memudahkan penggabungan retriever, reader, generator, dan komponen kustom menjadi aplikasi yang canggih.

Poin-poin kunci:

• Arsitektur komponen-pipeline Haystack 2.x memungkinkan sistem NLP yang bersih dan mudah dipelihara.
• Document store menyediakan persistensi fleksibel dengan dukungan embedding dan filter metadata.
• Pengambilan hybrid (embedding + BM25) dengan reranking memberikan kualitas pencarian terbaik.
• Komponen kustom memungkinkan Anda memperluas Haystack dengan logika spesifik domain.
• Evaluasi sistematis sangat penting -- ukur presisi pengambilan, recall, dan kualitas generasi.
• Integrasi FastAPI memudahkan deployment pipeline sebagai REST API produksi.

Haystack terus berkembang dengan cepat. Tetap mengikuti rilis terbaru untuk komponen baru, peningkatan performa, dan dukungan model yang lebih luas.