Haystack Cheat Sheet — RAG & NLP Pipelines — Haystack Refere

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

# Create document store
doc_store = InMemoryDocumentStore()
docs = [
    Document(content="Haystack is a framework for NLP pipelines."),
    Document(content="Pipelines are DAGs of components.")
]
doc_store.write_documents(docs)

# Build pipeline
pipeline = Pipeline()
retriever = InMemoryBM25Retriever(document_store=doc_store)
prompt_builder = PromptBuilder(
    template="""Answer based on these docs: {% for doc in documents %}
{{ doc.content }}{% endfor %}
Question: {{ query }}"""
)
generator = OpenAIGenerator(
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4o"
)

# Connect components
pipeline.add_component("retriever", retriever)
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("generator", generator)

pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder.prompt", "generator.prompt")

# Run
result = pipeline.run({
    "retriever": {"query": "What is Haystack?"},
    "prompt_builder": {"query": "What is Haystack?"}
})
print(result["generator"]["replies"][0])

from haystack import component
import os
from openai import OpenAI

@component
class CustomSummarizer:
    def __init__(self, max_length: int = 100):
        self.max_length = max_length
        self.client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    
    @component.output_types(summary=str)
    def run(self, text: str):
        """Summarize text using OpenAI."""
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "user", "content": f"Summarize in {self.max_length} chars: {text}"}
            ]
        )
        return {"summary": response.choices[0].message.content}

# Use in pipeline
pipeline = Pipeline()
summarizer = CustomSummarizer(max_length=50)
pipeline.add_component("summarizer", summarizer)

result = pipeline.run({"summarizer": {"text": "Long document text here..."}})
print(result["summarizer"]["summary"])

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

# Setup
doc_store = InMemoryDocumentStore()
doc_store.write_documents([
    Document(content="Python is a programming language."),
    Document(content="RAG stands for Retrieval-Augmented Generation.")
])

pipeline = Pipeline()
retriever = InMemoryBM25Retriever(document_store=doc_store)
prompt = PromptBuilder(
    template="""Chat history: {{ chat_history }}
Docs: {% for doc in documents %}{{ doc.content }}{% endfor %}
User: {{ query }}
Assistant:"""
)
generator = OpenAIGenerator(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o")

pipeline.add_component("retriever", retriever)
pipeline.add_component("prompt", prompt)
pipeline.add_component("generator", generator)

pipeline.connect("retriever.documents", "prompt.documents")
pipeline.connect("prompt.prompt", "generator.prompt")

# Run conversation
chat_history = ""
queries = ["What is Python?", "Is RAG related to Python?"]

for query in queries:
    result = pipeline.run({
        "retriever": {"query": query},
        "prompt": {"query": query, "chat_history": chat_history}
    })
    response = result["generator"]["replies"][0]
    chat_history += f"User: {query}\nAssistant: {response}\n"
    print(f"Q: {query}\nA: {response}\n")

from haystack import Pipeline, Document
from haystack.components.document_splitters import DocumentSplitter, SentenceSplitter
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.embedders import OpenAIDocumentEmbedder
import os

# Create document store
doc_store = InMemoryDocumentStore()

# Build pipeline
pipeline = Pipeline()

# Load or create documents
docs = [
    Document(content="This is a long document. " * 100)  # Simulate long text
]

# Add components
splitter = SentenceSplitter(sentences_per_page=5)
embedder = OpenAIDocumentEmbedder(
    api_key=os.environ["OPENAI_API_KEY"],
    model="text-embedding-3-small"
)
writer = DocumentWriter(document_store=doc_store)

pipeline.add_component("splitter", splitter)
pipeline.add_component("embedder", embedder)
pipeline.add_component("writer", writer)

pipeline.connect("splitter.documents", "embedder.documents")
pipeline.connect("embedder.documents", "writer.documents")

# Run
pipeline.run({"splitter": {"documents": docs}})
print(f"Stored {doc_store.document_count()} chunks")