Approach Comparison: Explicit DAGs vs. Magical Framework Abstractions
Early last year, my team inherited a multi-tenant enterprise search service indexing roughly 2.5 million corporate docs. Whenever retrieval stalled or p99 latency spiked past 4,000ms, finding the culprit was painful. Tracing meant digging through nested wrapper classes, opaque callbacks, and magic context dictionaries. We wasted hours fighting framework internals instead of tuning prompts and improving relevance.
That friction led us to evaluate Haystack 2.0, deepset’s complete rewrite of their orchestration engine. The architecture differs sharply in how data moves between steps:
- The Monolithic Chain Approach (e.g., early LangChain): Builds on deep inheritance hierarchies and implicit shared state. Steps pass loose dictionary payloads where keys appear and disappear without clear contracts. Rapid prototyping is quick. Yet debugging dynamic branching or type errors in production quickly becomes messy.
- The Explicit DAG Approach (Haystack 2.0): Models retrievers, prompt templates, and LLM clients as isolated nodes in a Directed Acyclic Graph (DAG). Every node declares strict input and output types using standard Python type hints. You connect sockets directly via
pipeline.connect("node_a.output_key", "node_b.input_key"). If types clash, the graph fails at build time—long before an invalid payload hits production.
Pros and Cons: Evaluating Haystack 2.0 for Real Systems
Moving from implicit chains to explicit graphs brings tangible engineering benefits, but it also demands deliberate trade-offs.
The Advantages
- Predictable Data Flow: Because socket connections are declared line by line, you always know what each node consumes and yields. No hidden dictionary keys overwrite prompt inputs downstream.
- Straightforward Custom Components: Writing a domain filter or token trimmer requires only a standard class with a
@componentdecorator and a type-annotatedrun()method. You never have to inherit from deeply nested base classes. - Native Serialization and Inspection: Pipelines export cleanly to YAML or Mermaid diagrams out of the box. You can audit graph topology during CI/CD checks before running requests.
- Lightweight Footprint: The core
haystack-aipackage stays lean. Third-party integrations for vector databases like Qdrant and Milvus live in isolated, optional packages.
The Trade-offs
- More Initial Setup: Wiring a 6-node pipeline requires 5 explicit
pipeline.connect()statements. You trade quick one-liner prototypes for enforceable architecture. - Targeted Ecosystem: Mainstream providers (OpenAI, HuggingFace, Cohere, Pinecone) work smoothly out of the box. However, newer community tools often lack pre-built connectors, requiring you to write 15 lines of custom wrapper code.
Recommended Architecture: Splitting Ingestion and Retrieval
Avoid assembling monolithic pipelines that bundle indexing and runtime querying into a single script. Decoupling them into two dedicated graphs isolates operational failures and cuts resource usage:
- Ingestion Pipeline: Parses raw files, splits text into 300-word chunks, generates embeddings in batches, and writes vectors to storage. Runs as an asynchronous background worker.
- Query Pipeline: Converts runtime user queries to vectors, fetches top-k candidates, populates prompt templates, and calls the LLM. Runs inside your low-latency API path.
Treating document stores as shared infrastructure while decoupling ingestion and serving cycles keeps latency stable. Background batch re-indexing will not block incoming user searches.
Set up a Python 3.10+ virtual environment and install the required dependencies:
# Create and activate virtual environment
python3 -m venv .venv
source .venv/bin/activate
# Install Haystack 2.x and lightweight local embedding provider
pip install --upgrade pip
pip install haystack-ai fastembed-haystack
export OPENAI_API_KEY="sk-your-openai-api-key"Implementation Guide: Modular RAG and Custom Processing
Here is an end-to-end RAG setup using local embeddings and explicit graph wiring.
Step 1: Building the Document Ingestion Pipeline
We configure an InMemoryDocumentStore, partition documents into 100-word chunks, compute embeddings locally with FastEmbed (using bge-small-en-v1.5), and write the vectors to memory.
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. Initialize the shared Document Store
document_store = InMemoryDocumentStore(embedding_similarity_function="cosine")
# 2. Instantiate individual components
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. Assemble the DAG
ingestion_pipeline = Pipeline()
ingestion_pipeline.add_component("splitter", splitter)
ingestion_pipeline.add_component("embedder", embedder)
ingestion_pipeline.add_component("writer", writer)
# 4. Connect explicit sockets (node.output -> node.input)
ingestion_pipeline.connect("splitter.documents", "embedder.documents")
ingestion_pipeline.connect("embedder.documents", "writer.documents")
# Run indexing with sample data
raw_docs = [
Document(content="Haystack 2.0 uses directed graphs for explicit component connection and deterministic pipelines."),
Document(content="Modular RAG architectures separate ingestion pipelines from runtime query pipelines for reliability."),
Document(content="Custom components in Haystack require only a @component decorator and typed run parameters.")
]
ingestion_pipeline.run({"splitter": {"documents": raw_docs}})
print(f"Indexed {document_store.count_documents()} document chunks successfully.")Step 2: Designing the Query and Generation Pipeline
The query pipeline runs on incoming user requests. It embeds the query, searches for top matches, formats a prompt, and streams the prompt to 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
# Define a strict template with explicit fallback rules
prompt_template = """
Given the following context documents, answer the query accurately.
If the answer cannot be found in the context, state 'Information not available in context.'
Context:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}
Question: {{ query }}
Answer:
"""
# Instantiate query components
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")
# Build the query DAG
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)
# Connect components explicitly
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")
# Execute query
user_query = "How does Haystack 2.0 connect components?"
response = query_pipeline.run({
"text_embedder": {"text": user_query},
"prompt_builder": {"query": user_query}
})
print("LLM Output:")
print(response["generator"]["replies"][0])Step 3: Creating Custom Components for Safety and Routing
Custom nodes make Haystack shine in production. If retrieval scores drop below acceptable thresholds, routing queries straight to a fallback prevents wasted LLM token costs and hallucinations.
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]):
# Drop low-confidence chunks before prompting the model
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}
Placing RelevanceFilter between the retriever and prompt builder stops irrelevant context from polluting your prompt. In our workloads, this simple check cut unnecessary token usage by 18%.
Practical Production Tips
- Audit Graph Structure in CI: Export your pipeline schema with
query_pipeline.dumps(), or generate an architectural chart usingquery_pipeline.draw(path="graph.png")to review connection changes during pull requests. - Disambiguate Shared Inputs: When two downstream components need the same upstream data, specify explicit socket mapping keys to prevent routing confusion.
- Preserve Statelessness: Treat pipeline instances as stateless executors. Calling
run()across concurrent worker threads is safe as long as your custom components avoid modifying internal instance attributes during execution.
