How to Build an Agentic RAG Pipeline with LangGraph and Map-Reduce

Build an agentic RAG pipeline by orchestrating parallel retrieval nodes (map) through LangGraph, aggregating results in a reduce node, and passing the consolidated context to an LLM for generation, allowing the agent to iteratively refine its search.

An agentic RAG pipeline combines autonomous decision-making with retrieval-augmented generation, enabling LLMs to dynamically fetch, evaluate, and synthesize information from multiple sources. According to the aishwaryanr/awesome-generative-ai-guide repository, implementing this pattern with LangGraph's map-reduce workflow provides a scalable, fault-tolerant architecture for production-grade AI agents. This approach treats each data source as an independent node that executes in parallel, then consolidates results before generation.

Understanding the Agentic RAG Architecture

The architecture described in resources/agentic_rag_101.md follows a clear separation of concerns between retrieval, aggregation, and generation.

The Map Phase (Parallel Retrieval)

In the map phase, the agent analyzes the user query and dispatches multiple retrievers simultaneously. Each retriever node operates independently, querying vector stores, APIs, or databases without blocking other operations. This parallelism reduces latency when querying heterogeneous data sources and ensures fault tolerance—if one retriever fails, others still contribute to the final answer.

The Reduce Phase (Aggregation)

The reduce node collects outputs from all map nodes, deduplicates overlapping documents, and respects the LLM's context window limits. As noted in the guide's step-by-step breakdown, this aggregation step ensures the LLM receives a clean, consolidated context rather than fragmented chunks from individual sources.

Core LangGraph Components

To implement this in LangGraph, you need four essential components working in concert:

• State: A dictionary tracking the question, retrieved documents, and final answer
• Map Nodes: Functions like retrieve_from_vector and retrieve_from_api that fetch data in parallel
• Reduce Node: A function such as aggregate_chunks that merges retrieval results into a single context string
• LLM Node: The generate_answer function that calls the language model with the aggregated context

Implementing the Map-Reduce Workflow

The following implementation mirrors the "Production-Grade AI Agents using LangGraph (Map-Reduce Implementation)" entry found in resources/60_ai_projects.md.

Step 1: Define Retrieval Functions (Map Nodes)

Create independent retrieval functions that accept the current state and return document lists. Each function acts as a map node in the LangGraph workflow.

from langgraph import Graph, GraphNode
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from openai import ChatCompletion

# Initialize your vector store (example setup)

vector_store = FAISS.load_local("faiss_index", OpenAIEmbeddings())

def retrieve_from_vector(state):
    """Query a FAISS vector store."""
    docs = vector_store.similarity_search(state["question"], k=4)
    return {"retrieved": docs}

def retrieve_from_api(state):
    """Call an external knowledge-base API (placeholder)."""
    # Example: response = requests.get(...).json()

    # For illustration we return an empty list.

    return {"retrieved": []}

Step 2: Create the Reducer Function

The reducer combines documents from all map nodes, removes duplicates, and prepares the final context string.

def aggregate_chunks(state):
    """Flatten and deduplicate retrieved docs."""
    all_docs = []
    for src in state["retrieved"]:
        all_docs.extend(src)  # src is a list of Document objects

    
    # Simple deduplication by text

    unique = {doc.page_content: doc for doc in all_docs}.values()
    context = "\n".join([d.page_content for d in unique][:8])  # respect token limit

    
    return {"context": context}

Step 3: Build the LangGraph Workflow

Wire the nodes together using LangGraph's graph API, directing parallel map nodes into a single reduce node.

def generate_answer(state):
    """Call the LLM with aggregated context."""
    prompt = f"""Answer the following question using ONLY the provided context.

Question: {state["question"]}

Context:
{state["context"]}

Answer:"""
    
    response = ChatCompletion.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return {"answer": response.choices[0].message.content.strip()}

# ------------------------------------------------------------

# Assemble the LangGraph workflow

# ------------------------------------------------------------

graph = Graph()

# Map phase – two parallel retrievers

graph.add_node(GraphNode(fn=retrieve_from_vector, name="vector"))
graph.add_node(GraphNode(fn=retrieve_from_api, name="api"))

# Reduce phase

graph.add_node(GraphNode(fn=aggregate_chunks, name="reduce"))

# LLM generation

graph.add_node(GraphNode(fn=generate_answer, name="llm"))

# Define edges (map → reduce → llm)

graph.add_edge("vector", "reduce")
graph.add_edge("api", "reduce")
graph.add_edge("reduce", "llm")

# Run the pipeline

def run_agentic_rag(question: str):
    init_state = {"question": question, "retrieved": []}
    final_state = graph.run(init_state)
    return final_state["answer"]

# Example usage

if __name__ == "__main__":
    q = "Why is my internet slow in the evenings?"
    print(run_agentic_rag(q))

Extending to Multi-Round Retrieval

For complex queries, the agent can evaluate its own answer and trigger additional retrieval cycles. By adding a maybe_refine node that checks answer quality, you create a loop where the graph conditionally returns to the map phase.

def maybe_refine(state):
    """Agent logic: if answer is unsatisfactory, trigger another map-reduce."""
    if "I don't know" in state["answer"] or len(state["answer"].split()) < 30:
        # Re-run map nodes with a higher k or additional sources

        state["retrieved"] = []  # reset

        state["question"] = state["question"] + " (more detail needed)"
        return {"continue": True}
    return {"continue": False}

# Add the refinement node and conditional loop

graph.add_node(GraphNode(fn=maybe_refine, name="refine"))
graph.add_edge("llm", "refine")
graph.add_edge("refine", "vector", condition=lambda s: s["continue"])
graph.add_edge("refine", "api", condition=lambda s: s["continue"])

This illustrates the agent's autonomy as described in the guide—the system can decide to perform additional retrievals until a confidence threshold is met.

Summary

• Use LangGraph to orchestrate parallel retrieval operations across multiple data sources using the Graph and GraphNode classes
• Implement the map phase with independent retriever nodes like retrieve_from_vector and retrieve_from_api that execute concurrently
• Aggregate results in a reduce node using aggregate_chunks to respect context limits and remove duplicates before LLM generation
• Connect components through the graph's add_edge method to create a deterministic workflow from parallel retrieval to final answer
• Reference the architecture in resources/agentic_rag_101.md and production examples in resources/60_ai_projects.md for advanced implementation patterns

Frequently Asked Questions

What is the advantage of using map-reduce for RAG instead of sequential retrieval?

Parallel retrieval reduces latency when querying multiple heterogeneous sources simultaneously. The pattern also provides fault tolerance—if one map node fails, others still contribute, and the reducer can handle missing pieces gracefully while the LLM node receives a complete context window.

How does LangGraph handle state between the map and reduce phases?

LangGraph maintains a shared state dictionary that persists across node executions, allowing the reduce node to access the retrieved list populated by all map nodes. Each node function receives the current state and returns partial updates that LangGraph merges automatically.

Can I add more than two retrievers to the map phase?

Yes, you can add unlimited retriever nodes by calling graph.add_node() for each new source, then connecting them all to the reduce node with graph.add_edge(). This modularity allows you to extend the pipeline without modifying existing reducer or LLM logic.

Where can I find the full production example referenced in the guide?

The complete implementation is listed under "Production-Grade AI Agents using LangGraph (Map-Reduce Implementation)" in resources/60_ai_projects.md within the aishwaryanr/awesome-generative-ai-guide repository. The guide also references an external starter repository at AIAnytime/Map-Reduce-implementation-using-LangGraph for additional reference code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}