วิธีสร้าง RAG Pipeline ด้วย Haystack 2.0 ใน Python

# Import the package
import pandas as pd
# Load the CSV
df = pd.read_csv("bbc-news-data.csv", sep="\t")
# Inspect the first 5 rows
df.head(5)

# Import the package
from haystack import Document
# Instantiate a collector
docs = []
# Loop through the records
for row in df.itertuples():
    doc = Document(
        content=row.content,
        meta={
            "title": row.title,
            "category": row.category,
            "filename": row.filename
        }
    )
    docs.append(doc)
 
# Inspect one Document
docs[0]

# Import the package
from haystack.components.preprocessors import RecursiveDocumentSplitter
# Create a splitter
splitter = RecursiveDocumentSplitter(
    split_length=180,
    split_overlap=30,
    split_unit="word",
    separators=[
        "\n\n",
        "sentence",
        "\n",
        " ",
    ]
)

# Import the package
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersDocumentEmbedder
)
# Create a document embedder
document_embedder = SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2",
    meta_fields_to_embed=["title"],
)

# Import the package
from haystack.document_stores.in_memory import InMemoryDocumentStore
# Create a vector store
vector_store = InMemoryDocumentStore(
    embedding_similarity_function="cosine"
)

# Import the package
from haystack.components.writers import DocumentWriter
# Create a document writer
document_writer = DocumentWriter(
    document_store=vector_store
)

# Import the package
from haystack import Pipeline
# Instantiate a Pipeline object
indexing_pipeline = Pipeline()

# Add the components
indexing_pipeline.add_component("splitter", splitter)
indexing_pipeline.add_component("embedder", document_embedder)
indexing_pipeline.add_component("writer", document_writer)

# Connect the components
indexing_pipeline.connect("splitter.documents", "embedder.documents")
indexing_pipeline.connect("embedder.documents", "writer.documents")

# Display the pipeline
indexing_pipeline.show()

# Run the indexing pipeline
indexing_result = indexing_pipeline.run(
    {
        "splitter": {
            "documents": docs
        }
    }
)
# Display the result
print(indexing_result)

# Import the package
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder
)
# Create a question embedder
question_embedder = SentenceTransformersTextEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
)

# Import the package
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
# Create a retriever
retriever = InMemoryEmbeddingRetriever(
    document_store=vector_store,
    top_k=5
)

# Set the system prompt
system_prompt = """
You are a helpful assistant for a historical BBC News Archive.
Answer only from the retrieved BBC article passages.
Rules:
- Be concise.
- Do not use outside knowledge.
- Do not invent facts that are not supported by the passages.
- The archive is historical, so do not present its content as current news.
- Cite factual claims using source labels such as [1] or [2].
- If the retrieved passages do not contain enough evidence, say that clearly.
- Response in short bullet points.
- At the end of the answer, include a "Sources:" section listing only the sources that were actually cited in the answer. Format each source as:
  - [n] <Title>
"""

# Set the user prompt
user_prompt = """
Question:
{{ question }}
Retrieved BBC archive passages:
{% for doc in docs %}
[{{ loop.index }}]
Title: {{ doc.meta["title"] }}
Category: {{ doc.meta["category"] }}
Filename: {{ doc.meta["filename"] }}
Passage:
{{ doc.content }}
{% endfor %}
Write a concise answer using only the retrieved passages.
After the answer, include:
Sources:
{% raw %}- [n] <Title>{% endraw %}
where each title corresponds to the cited passage, and list only the sources that were actually cited in the answer.
"""

# Import the packages
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage
# Create a prompt builder
prompt_builder = ChatPromptBuilder(
    template=[
        ChatMessage.from_system(system_prompt.strip()),
        ChatMessage.from_user(user_prompt.strip())
    ],
    required_variables=["question", "docs"],
)

# Import the packages
import os
from dotenv import load_dotenv
# Load the .env content
load_dotenv()
# Get the API key
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")

# Import the packages
from haystack.utils import Secret
from haystack_integrations.components.generators.google_genai import (
    GoogleGenAIChatGenerator
)
# Create a generator
generator = GoogleGenAIChatGenerator(
    model="gemini-2.5-flash",
    api_key=Secret.from_token(GEMINI_API_KEY),
    generation_kwargs={
        "temperature": 0.2
    }
)

# Import the package
from haystack import Pipeline
# Instantiate a Pipeline object
querying_pipeline = Pipeline()

# Add the components
querying_pipeline.add_component("question_embedder", question_embedder)
querying_pipeline.add_component("retriever", retriever)
querying_pipeline.add_component("prompt_builder", prompt_builder)
querying_pipeline.add_component("generator", generator)

# Connect the components
querying_pipeline.connect("question_embedder.embedding", "retriever.query_embedding")
querying_pipeline.connect("retriever.documents", "prompt_builder.docs")
querying_pipeline.connect("prompt_builder.prompt", "generator.messages")

# Display the pipeline
querying_pipeline.show()

# Create the function
def run_query(query: str):
    # Run the pipeline
    result = querying_pipeline.run(
        {
            "question_embedder": {
                "text": query
            },
            "prompt_builder": {
                "question": query
            }
        }
    )
    # Return the result
    return result["generator"]["replies"][0].text

# Set the question
my_question = "What concerns did users raise about Google’s AutoLink feature?"
# Run the query
query_result = run_query(query=my_question)
# Display the result
print(query_result)