Hybrid RAG with LangGraph: Vector, Keyword & Metadata Retrieval in Action

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:

Web search
Knowledge base escalation
Service desk ticket creation

Dynamic Metadata Extraction

Replace prompt-based extraction with:

Lightweight classifiers
Structured output mode
Fine-tuned extraction models

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:

Explicit state management
Parallel execution
Checkpointing
Streaming
Observability

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.