Parent Document Retrieval in a Multi-Agent LangGraph Enterprise System

Introduction: The Chunking Paradox in Enterprise RAG

Retrieval-Augmented Generation (RAG) has matured from a novelty to a critical enterprise infrastructure. However, production teams consistently hit the Chunking Paradox: small chunks are necessary for precise vector retrieval, but large chunks are necessary for coherent LLM generation. When you retrieve a 100-token snippet about "Q3 revenue variance," the model lacks the surrounding context of accounting methodologies, regional adjustments, and forward-looking guidance needed to answer accurately.

Parent Document Retriever (PDR) solves this by decoupling the retrieval unit from the generation unit. You embed and search over granular child chunks, but return the full parent document (or larger section) to the LLM. This article demonstrates an end-to-end implementation of PDR within a stateful, multi-agent LangGraph architecture designed for enterprise financial analysis. We move beyond simple chains to build a system with memory, conditional routing, and human-in-the-loop validation.

Real-Time Use Case: Institutional Earnings Call Analysis

The Scenario

A Tier-1 Investment Bank’s research desk needs to analyze quarterly earnings calls and 10-K filings. Analysts ask complex questions like: "How does management's commentary on supply chain headwinds in the Q3 call compare to the risk factors outlined in the latest 10-K, and what is the net impact on FY26 guidance?"

Why Simple RAG Fails Here

Semantic Fragmentation: Supply chain risks are mentioned in passing during Q&A (small chunk) but detailed in Item 1A of the 10-K (large chunk). Vector search retrieves the Q&A mention but misses the structured risk disclosure.
Temporal State: The analyst’s question references "FY26 guidance." The system must remember that the previous turn established the baseline as "consensus estimates," not "company guidance."
Multi-Source Synthesis: Answering requires retrieving from two distinct corpora (transcripts vs. filings), validating consistency, and synthesizing a unified view.

The Solution Architecture

We implement a LangGraph Multi-Agent System with three specialized agents:

Retrieval Agent: Uses Parent Document Retriever to fetch context from both transcripts and filings.
Analyst Agent: Synthesizes retrieved parent documents into a draft response.
Compliance Agent: Validates claims against source text and checks for hallucination.

The graph maintains a shared state containing conversation memory, retrieved documents, and intermediate drafts.

Technical Implementation

Prerequisites

pip install langgraph langchain-community langchain-openai \
            chromadb tiktoken pydantic

Step 1: Define the Shared State

In LangGraph, state is the single source of truth passed between nodes. For enterprise RAG, we need typed, validated state.

from typing import Annotated, List, Optional, Sequence
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.documents import Document
import operator

class AgentState(TypedDict):
    """Shared state for the multi-agent RAG system."""
    # Conversation history with reducer pattern
    messages: Annotated[list, add_messages]
    
    # Retrieved parent documents (accumulated across agents)
    retrieved_docs: Annotated[List[Document], operator.add]
    
    # Current query being processed
    current_query: str
    
    # Intermediate analysis draft
    draft_response: Optional[str]
    
    # Compliance validation result
    is_compliant: bool
    
    # Metadata for audit trail
    session_id: str
    sources_cited: List[str]

Step 2: Build the Parent Document Retriever

This is the core differentiator. We use RecursiveCharacterTextSplitter for parents and a finer splitter for children, linked via metadata.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

def build_parent_document_retriever(pdf_paths: list[str]):
    """
    Builds a PDR optimized for financial documents.
    - Parent chunks: ~4000 tokens (full sections/paragraphs)
    - Child chunks: ~400 tokens (precise retrieval targets)
    """
    # Load documents
    docs = []
    for path in pdf_paths:
        loader = PyPDFLoader(path)
        docs.extend(loader.load())
    
    # Parent splitter: preserves section coherence
    parent_splitter = RecursiveCharacterTextSplitter(
        chunk_size=4000,
        chunk_overlap=200,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    
    # Child splitter: optimized for embedding precision
    child_splitter = RecursiveCharacterTextSplitter(
        chunk_size=400,
        chunk_overlap=50,
        separators=["\n", ". ", " ", ""]
    )
    
    vectorstore = Chroma(
        collection_name="financial_pdr",
        embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
    )
    
    store = InMemoryStore()
    
    retriever = ParentDocumentRetriever(
        vectorstore=vectorstore,
        docstore=store,
        child_splitter=child_splitter,
        parent_splitter=parent_splitter,
    )
    
    # Index documents (children go to vectorstore, parents to docstore)
    retriever.add_documents(docs, ids=None)
    
    return retriever

⚠️ Production Note: Replace InMemoryStore with Redis, PostgreSQL, or S3-backed storage for persistence. Chroma should be replaced with Pinecone, Weaviate, or pgvector in production.

Step 3: Define the Agents as Graph Nodes

Each agent is a function that reads/writes to AgentState.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# --- RETRIEVAL NODE ---
async def retrieval_node(state: AgentState) -> dict:
    """Uses PDR to fetch relevant parent documents."""
    query = state["current_query"]
    
    # PDR returns full parent documents, not tiny chunks
    docs = await retriever.ainvoke(query)
    
    # Deduplicate by parent doc ID
    seen_ids = set()
    unique_docs = []
    for doc in docs:
        doc_id = doc.metadata.get("doc_id", id(doc))
        if doc_id not in seen_ids:
            seen_ids.add(doc_id)
            unique_docs.append(doc)
    
    return {
        "retrieved_docs": unique_docs,
        "sources_cited": [d.metadata.get("source", "unknown") for d in unique_docs]
    }

# --- ANALYST NODE ---
analyst_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a senior equity research analyst. 
    Synthesize the provided source documents into a precise answer.
    Always cite sources using [Source: filename, page X] format.
    If documents conflict, note the discrepancy explicitly."""),
    ("human", "Query: {query}\n\nSources:\n{context}")
])

async def analyst_node(state: AgentState) -> dict:
    context = "\n\n---\n\n".join(
        f"[{d.metadata.get('source','')} p.{d.metadata.get('page','')}]\n{d.page_content}"
        for d in state["retrieved_docs"]
    )
    
    response = await analyst_prompt | llm | (lambda x: x.content)
    draft = await response.ainvoke({
        "query": state["current_query"],
        "context": context
    })
    
    return {"draft_response": draft}

# --- COMPLIANCE NODE ---
compliance_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a compliance reviewer. Verify that EVERY claim 
    in the draft is directly supported by the source documents.
    Return JSON: {"compliant": bool, "issues": list[str]}"""),
    ("human", "Draft:\n{draft}\n\nSources:\n{context}")
])

async def compliance_node(state: AgentState) -> dict:
    context = "\n\n".join(d.page_content for d in state["retrieved_docs"])
    
    result = await compliance_prompt | llm.with_structured_output(
        method="json_schema"
    ) | (lambda x: x)
    
    validation = await result.ainvoke({
        "draft": state["draft_response"],
        "context": context
    })
    
    return {"is_compliant": validation.get("compliant", False)}

Step 4: Assemble the LangGraph with Conditional Routing

from langgraph.graph import StateGraph, START, END

def route_after_compliance(state: AgentState) -> str:
    """Conditional edge: retry analysis if non-compliant."""
    if state["is_compliant"]:
        return "finalize"
    # Allow max 2 retries to prevent infinite loops
    retry_count = sum(1 for m in state["messages"] if m.type == "tool" and m.name == "compliance_retry")
    if retry_count < 2:
        return "retry_analyst"
    return "finalize_with_warning"

workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("retrieve", retrieval_node)
workflow.add_node("analyze", analyst_node)
workflow.add_node("comply", compliance_node)

# Define edges
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "analyze")
workflow.add_edge("analyze", "comply")

# Conditional routing based on compliance check
workflow.add_conditional_edges(
    "comply",
    route_after_compliance,
    {
        "finalize": END,
        "retry_analyst": "analyze",
        "finalize_with_warning": END
    }
)

app = workflow.compile()

Step 5: Execute with Memory Persistence

For enterprise use, thread-based memory ensures continuity across turns.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app_with_memory = workflow.compile(checkpointer=checkpointer)

# First turn
config = {"configurable": {"thread_id": "earnings-call-2024-q3"}}

result = await app_with_memory.ainvoke(
    {
        "current_query": "What did CFO say about margin compression drivers?",
        "session_id": "analyst-desk-001",
        "messages": [],
        "retrieved_docs": [],
        "sources_cited": [],
        "is_compliant": False,
        "draft_response": None
    },
    config=config
)

print(result["draft_response"])
print(f"Sources: {result['sources_cited']}")
print(f"Compliant: {result['is_compliant']}")

# Second turn - memory automatically provides prior context
follow_up = await app_with_memory.ainvoke(
    {
        "current_query": "Compare that to last quarter's explanation",
        # Other fields reset per-turn; messages persist via checkpointer
        "retrieved_docs": [],
        "sources_cited": [],
        "is_compliant": False,
        "draft_response": None
    },
    config=config
)

Visualizing the Agent Flow

Why This Matters for Enterprise Adoption

Capability	Naive RAG	PDR + Multi-Agent LangGraph
Context Coherence	Fragmented snippets	Full sections preserved
Multi-Source Reasoning	Single retriever	Specialized retrieval per corpus
Factual Grounding	No verification	Compliance agent validates every claim
Conversational Memory	Stateless or windowed	Persistent thread state with reducers
Auditability	Black box	Full state traceability per session
Failure Recovery	Silent failures	Conditional retry with bounded loops

Key Design Decisions Explained

Why PDR over reranking? Reranking still operates on small chunks. PDR guarantees the LLM sees complete paragraphs/sections. Reranking can be added on top of child retrieval for even better precision.
Why separate Compliance Agent? Embedding validation into the analyst prompt creates sycophancy—the model grades its own work. A separate agent with a distinct system prompt and structured output provides genuine adversarial checking.
Why operator.add reducer on retrieved_docs? In multi-step graphs, each node may contribute additional documents. The additive reducer accumulates them rather than overwriting, enabling progressive context enrichment.
Why bounded retries? Unbounded compliance loops are a production anti-pattern. Two retries balances quality with latency SLAs. After exhaustion, the system returns results with a warning flag for human review.

Production Hardening Checklist

Before deploying this architecture:

Replace in-memory stores with persistent vector DB and KV store
Add rate limiting at the retrieval node to protect downstream APIs
Implement token budget tracking in state to prevent context overflow
Add observability via LangSmith/Langfuse tracing on every node invocation
Cache PDR results for repeated queries (semantic cache layer)
Redact PII before indexing financial documents
Set up evaluation harness using RAGAS or custom compliance benchmarks
Configure checkpoint TTL to manage storage costs for long-lived threads

Conclusion

Parent Document Retriever is not just a retrieval optimization—it is an architectural enabler for enterprise RAG. By combining PDR's context preservation with LangGraph's stateful multi-agent orchestration, you build systems that don't just retrieve information, but reason over it with the rigor that regulated industries demand. The pattern demonstrated here—specialized retrieval, adversarial validation, persistent memory, and conditional flow control—is transferable beyond finance to legal contract analysis, medical literature review, and technical documentation systems. The key insight is that advanced RAG is no longer about better embeddings; it's about better orchestration. The complete code above is runnable as-is with your OpenAI API key and sample PDFs. Start with the earnings call use case, adapt the splitters to your document structure, and extend the graph with domain-specific agents as your requirements evolve.