Introduction
Single-strategy retrieval systems rarely survive production workloads. Vector search excels at semantic understanding but struggles with exact matches, acronyms, or structured constraints. Keyword/BM25 search guarantees lexical precision but misses conceptual relevance. Metadata filtering adds business context (department, date, status, version) but isn't a retrieval strategy on its own.
Hybrid retrieval solves this by orchestrating multiple retrieval paradigms, fusing their results, and applying intelligent reranking. When combined with LangGraph, you get a stateful, observable, and fault-tolerant RAG pipeline that can branch, retry, stream, and adapt to real-time constraints.
In this article, we'll build an end-to-end hybrid RAG system using LangGraph, anchored in a real-world enterprise IT support use case.
Real-World Use Case: Enterprise IT Support Knowledge Assistant
Scenario
An IT operations engineer submits the following query:
"How do I fix Error 503 on Kubernetes clusters after upgrading to v1.28? Looking for resolved tickets from the infrastructure team."
Challenges
Semantic intent: "fix", "Kubernetes upgrade", "cluster instability"
Exact matches: Error 503, v1.28, Kubernetes
Business constraints: team=infrastructure, status=resolved, created_after=2025-01-01
Latency requirement: <800ms end-to-end for real-time agent assistance
Solution Architecture
Query Analyzer: Extracts intent, keywords, and metadata filters
Parallel Retrieval: Vector + BM25 + Strict Metadata Filter
Fusion & Rerank: Reciprocal Rank Fusion (RRF) + Cross-Encoder
Generation: LLM synthesizes answer with citations
Orchestration: LangGraph manages state, parallelism, streaming, and checkpoints
Architecture Overview
LangGraph excels in this architecture because:
Nodes run in parallel when edges converge.
State is explicitly typed and accumulates results safely.
Checkpointing enables retries, streaming, and human-in-the-loop fallbacks.
Observability is built-in through LangGraph traces.
Step-by-Step LangGraph Implementation
1. Dependencies & Setup
pip install langgraph langchain langchain-core langchain-openai rank_bm25 numpy
2. State Definition
LangGraph requires explicit state typing. We use Annotated with operator.add so parallel retrieval nodes accumulate documents instead of overwriting them.
from typing import TypedDict, List, Dict, Any, Annotated
import operator
from langchain_core.documents import Document
class HybridRAGState(TypedDict):
query: str
metadata_filters: Dict[str, Any]
vector_docs: Annotated[List[Document], operator.add]
keyword_docs: Annotated[List[Document], operator.add]
metadata_docs: Annotated[List[Document], operator.add]
reranked_docs: List[Document]
answer: str
citations: List[Dict[str, str]]
3. Node Implementations
Query Analysis & Metadata Extraction
from langchain_openai import ChatOpenAI
import json
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def analyze_query(state: HybridRAGState):
prompt = f"""
Extract metadata filters and search intent from this query:
"{state['query']}"
Return ONLY a JSON object with keys:
department, status, min_date, error_code, component.
Use null for missing fields.
Example:
{{
"department": "infrastructure",
"status": "resolved",
"min_date": "2025-01-01",
"error_code": "503",
"component": "kubernetes"
}}
"""
response = llm.invoke(prompt)
filters = json.loads(response.content)
return {"metadata_filters": filters}
Parallel Retrieval Nodes
In production, replace these with actual FAISS, Elasticsearch, OpenSearch, or pgvector implementations.
Vector Retrieval
def retrieve_vector(state: HybridRAGState):
docs = vector_store.similarity_search_with_score(
query=state["query"],
k=5,
filter=state["metadata_filters"]
)
return {
"vector_docs": [doc for doc, _ in docs]
}
Keyword Retrieval
def retrieve_keyword(state: HybridRAGState):
docs = bm25_retriever.invoke(
state["query"],
k=5
)
return {"keyword_docs": docs}
Metadata Retrieval
def retrieve_metadata(state: HybridRAGState):
docs = metadata_store.query(
filters=state["metadata_filters"],
limit=3
)
return {"metadata_docs": docs}
Fusion & Reranking (RRF + Cross-Encoder)
Reciprocal Rank Fusion
import numpy as np
from sentence_transformers import CrossEncoder
reranker = CrossEncoder(
"cross-encoder/ms-marco-MiniLM-L-6-v2"
)
def reciprocal_rank_fusion(
vector_docs,
keyword_docs,
metadata_docs,
k=60
):
scores = {}
def rrf_rank(docs, start_rank=1):
for i, doc in enumerate(docs):
doc_id = doc.metadata.get(
"id",
doc.page_content[:50]
)
rank = start_rank + i
scores[doc_id] = (
scores.get(doc_id, 0)
+ 1 / (k + rank)
)
rrf_rank(vector_docs)
rrf_rank(keyword_docs, 10)
rrf_rank(metadata_docs, 20)
sorted_ids = sorted(
scores,
key=scores.get,
reverse=True
)
doc_map = {
d.metadata.get(
"id",
d.page_content[:50]
): d
for d in (
vector_docs +
keyword_docs +
metadata_docs
)
}
return [
doc_map[doc_id]
for doc_id in sorted_ids
if doc_id in doc_map
]
Fuse and Rerank
def fuse_and_rerank(state: HybridRAGState):
fused = reciprocal_rank_fusion(
state["vector_docs"],
state["keyword_docs"],
state["metadata_docs"]
)
if len(fused) > 10:
pairs = [
(state["query"], doc.page_content)
for doc in fused[:10]
]
scores = reranker.predict(pairs)
reranked = sorted(
fused[:10],
key=lambda x: scores[fused.index(x)],
reverse=True
)
return {"reranked_docs": reranked}
return {"reranked_docs": fused}
Generation with Citations
def generate_response(state: HybridRAGState):
context = "\n---\n".join([
f"[{i+1}] {doc.metadata.get('source', 'KB')}\n{doc.page_content}"
for i, doc in enumerate(
state["reranked_docs"][:5]
)
])
prompt = f"""
You are an IT support assistant.
Answer the query using ONLY
the provided context.
If unsure, say so.
Cite sources using [1], [2], etc.
Query:
{state['query']}
Context:
{context}
"""
response = llm.invoke(prompt)
citations = [
{
"source": doc.metadata.get(
"source",
"Unknown"
),
"id": doc.metadata.get("id", "")
}
for doc in state["reranked_docs"][:5]
]
return {
"answer": response.content,
"citations": citations
}
4. Graph Construction & Execution
from langgraph.graph import StateGraph, END
workflow = StateGraph(HybridRAGState)
# Add nodes
workflow.add_node("analyze", analyze_query)
workflow.add_node("vector", retrieve_vector)
workflow.add_node("keyword", retrieve_keyword)
workflow.add_node("metadata", retrieve_metadata)
workflow.add_node("fuse", fuse_and_rerank)
workflow.add_node("generate", generate_response)
# Entry point
workflow.set_entry_point("analyze")
# Parallel retrieval
workflow.add_edge("analyze", "vector")
workflow.add_edge("analyze", "keyword")
workflow.add_edge("analyze", "metadata")
# Fusion
workflow.add_edge("vector", "fuse")
workflow.add_edge("keyword", "fuse")
workflow.add_edge("metadata", "fuse")
# Generation
workflow.add_edge("fuse", "generate")
workflow.add_edge("generate", END)
# Compile with checkpointing
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(
checkpointer=checkpointer
)
5. Run & Stream
query = """
How do I fix Error 503 on Kubernetes clusters
after upgrading to v1.28?
Looking for resolved tickets from
the infrastructure team.
"""
initial_state = {
"query": query,
"metadata_filters": {},
"vector_docs": [],
"keyword_docs": [],
"metadata_docs": [],
"reranked_docs": [],
"answer": "",
"citations": []
}
# Streaming execution
for event in app.stream(
initial_state,
stream_mode="updates"
):
print(
f"Step: {list(event.keys())[0]}"
)
# Final output
result = app.invoke(initial_state)
print("\nAnswer:", result["answer"])
print("Citations:", result["citations"])
Production Considerations
| Concern | LangGraph Solution |
|---|
| Latency | Parallel retrieval reduces search time significantly. RRF and lightweight reranking help maintain sub-second response times. |
| Resilience | MemorySaver or PostgresSaver enables retries, state rollback, and approval workflows. |
| Streaming | stream_mode="messages" and stream_mode="updates" provide incremental delivery. |
| Cost Control | Conditional routing can skip unnecessary retrievers based on query characteristics. |
| Evaluation | Track Hit@K, MRR, latency, and citation accuracy using LangSmith traces. |
Advanced Patterns
Conditional Routing
Use add_conditional_edges() to route requests to specialized retrievers based on query intent.
Fallback Chains
If reranked_docs is empty, branch to:
Dynamic Metadata Extraction
Replace prompt-based extraction with:
Example:
response_format={"type": "json_object"}
Caching
Wrap retrievers with:
LangChain Cache
Redis
Memory Cache
This avoids repeated vector and BM25 lookups for identical queries.
Conclusion
Hybrid retrieval is not simply about running multiple searches. It is about orchestrating retrieval strategies intelligently and combining their strengths.
LangGraph provides the ideal foundation through:
By combining:
Vector search for semantic recall
Keyword/BM25 retrieval for exact-match precision
Metadata filtering for business constraints
Reciprocal Rank Fusion and reranking for result optimization
LangGraph for production-grade orchestration
you can build a RAG system that handles real-world ambiguity, scales efficiently, and produces reliable, citation-backed answers.
Summary
Single retrieval strategies often fail in production because they cannot simultaneously satisfy semantic understanding, exact-match requirements, and business constraints. A Hybrid RAG architecture combines vector search, BM25 retrieval, metadata filtering, fusion, and reranking to improve relevance and accuracy. Using LangGraph as the orchestration layer adds parallelism, checkpointing, observability, streaming, and fault tolerance, creating a scalable and production-ready retrieval system capable of supporting demanding enterprise workloads.