Mastering Messy Metadata: Flattening Nested Dicts in Enterprise LangGraph RAG

Introduction

In enterprise Retrieval-Augmented Generation (RAG) systems, the hardest part isn't usually the LLM—it's the data. Enterprise documents pulled from SharePoint, Confluence, or legacy ERPs come with deeply nested, highly variable, and often contradictory metadata schemas.

When indexing this data into a Vector Database, you are typically forced to flatten these nested dictionaries into key-value pairs. But what happens when schema evolution causes key collisions? For example, a nested object such as {"meta": {"author": "Alice"}} and a top-level field such as {"meta.author": "Bob"} both flatten to meta.author. Naive flattening overwrites data, leading to lost context and hallucinating agents.

This article explores how to build a robust dictionary flattening utility that resolves collisions using index suffixes and integrates it into an end-to-end Enterprise Multi-Agent LangGraph RAG pipeline with shared state and persistent memory.

Part 1: The Core Problem & The Solution

Why Naive Flattening Fails

Standard flattening functions simply overwrite duplicate keys. In an enterprise RAG context, if a financial report has both document.properties.creator and a legacy document.properties.creator string field, one gets silently dropped. The RAG agent then retrieves incomplete metadata, leading to poor filtering and retrieval.

The Solution: Collision-Aware Flattening

We need an algorithm that:

Recursively traverses deeply nested dictionaries with variable keys.
Joins keys using a separator (e.g., .).
Detects collisions in the flattened output.
Resolves collisions by appending an index suffix (_1, _2, etc.).

Part 2: Real-World Use Case

Scenario: An Enterprise Document Ingestion Pipeline

Goal: Ingest messy, nested JSON metadata from an enterprise CMS, normalize it for a Vector Database (such as Pinecone or Milvus), and store the key mapping in memory for future traceability.

Architecture

Ingestion Agent: Fetches raw, nested metadata from the CMS.
Normalization Agent: Uses the collision-aware flattening function to prepare metadata.
Indexing Agent: Upserts flattened metadata and document chunks into the Vector Database.
State: Shared LangGraph state holding raw metadata, flattened metadata, and processing logs.
Memory: Persistent memory storing the key mapping (flattened_key → original_nested_path) so agents can trace data lineage later.

Part 3: End-to-End Implementation

Prerequisites

pip install langgraph langchain-openai langchain-core pydantic

Step 1: The Core Utility (Collision-Aware Flattening)

This is the heart of the solution. It handles arbitrary nesting and guarantees no data loss on collisions.

# core/utils.py
from typing import Any, Dict

def flatten_dict_with_collisions(
    d: Dict[str, Any],
    parent_key: str = '',
    sep: str = '.'
) -> Dict[str, Any]:
    """
    Flattens a deeply nested dictionary.
    Handles key collisions by appending index suffixes (_1, _2, etc.).
    """
    items = {}

    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else str(k)

        if isinstance(v, dict):
            # Recursively flatten nested dictionaries
            sub_items = flatten_dict_with_collisions(v, new_key, sep=sep)

            for sub_k, sub_v in sub_items.items():
                final_key = sub_k
                counter = 1

                while final_key in items:
                    final_key = f"{sub_k}_{counter}"
                    counter += 1

                items[final_key] = sub_v

        else:
            final_key = new_key
            counter = 1

            while final_key in items:
                final_key = f"{new_key}_{counter}"
                counter += 1

            items[final_key] = v

    return items

Step 2: Core Interfaces & State

Following a strict layered architecture helps prevent circular dependencies and keeps business logic isolated from infrastructure concerns.

# core/interfaces.py
from typing import Protocol, Dict, Any

class IVectorDB(Protocol):
    async def upsert(
        self,
        doc_id: str,
        chunk: str,
        metadata: Dict[str, Any]
    ) -> None:
        ...

class ICMSClient(Protocol):
    async def fetch_metadata(self, doc_id: str) -> Dict[str, Any]:
        ...

class IMemoryStore(Protocol):
    async def save_key_mapping(
        self,
        doc_id: str,
        mapping: Dict[str, str]
    ) -> None:
        ...

# core/state.py
from typing import TypedDict, Annotated, List, Dict, Any
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class IngestionState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    doc_id: str
    raw_metadata: Dict[str, Any]
    flattened_metadata: Dict[str, Any]
    processing_logs: List[str]

Step 3: Infrastructure (Concrete Implementations)

Mock implementations simulate external enterprise systems.

# infrastructure/clients.py
from typing import Dict, Any

class MockCMSClient:
    """Simulates a CMS returning nested, colliding metadata."""

    async def fetch_metadata(self, doc_id: str) -> Dict[str, Any]:
        return {
            "document": {
                "properties": {
                    "author": "Alice Smith",
                    "created_date": "2024-01-01"
                },
                "properties.author": "Legacy System Override",
                "tags": ["finance", "Q3"]
            },
            "document.properties": {
                "author": "Bob Jones"
            }
        }

class MockVectorDB:
    async def upsert(
        self,
        doc_id: str,
        chunk: str,
        metadata: Dict[str, Any]
    ) -> None:
        print(
            f"[VectorDB] Upserted {doc_id} "
            f"with {len(metadata)} flat metadata keys."
        )

class MockMemoryStore:
    def __init__(self):
        self._store = {}

    async def save_key_mapping(
        self,
        doc_id: str,
        mapping: Dict[str, str]
    ) -> None:
        self._store[doc_id] = mapping
        print(f"[Memory] Saved key mapping for {doc_id}")

Step 4: Application (The Multi-Agent System)

The agents use shared state and injected dependencies.

Ingestion Agent

class IngestionAgent:
    def __init__(self, cms_client):
        self.cms_client = cms_client

    async def __call__(self, state):
        doc_id = state["doc_id"]

        raw_meta = await self.cms_client.fetch_metadata(doc_id)

        return {
            "raw_metadata": raw_meta,
            "processing_logs": (
                state.get("processing_logs", [])
                + [f"Fetched raw metadata for {doc_id}"]
            )
        }

Normalization Agent

class NormalizationAgent:
    def __init__(self, memory_store):
        self.memory_store = memory_store

    async def __call__(self, state):
        raw_meta = state["raw_metadata"]

        flat_meta = flatten_dict_with_collisions(raw_meta)

        key_mapping = {
            k: f"original_path_for_{k}"
            for k in flat_meta.keys()
        }

        await self.memory_store.save_key_mapping(
            state["doc_id"],
            key_mapping
        )

        return {
            "flattened_metadata": flat_meta,
            "processing_logs": (
                state["processing_logs"]
                + [
                    (
                        f"Flattened metadata. "
                        f"Original keys: {len(raw_meta)}, "
                        f"Flat keys: {len(flat_meta)}"
                    )
                ]
            )
        }

Indexing Agent

class IndexingAgent:
    def __init__(self, vectordb):
        self.vectordb = vectordb

    async def __call__(self, state):
        doc_id = state["doc_id"]
        flat_meta = state["flattened_metadata"]

        chunk = (
            "This is the content of the "
            "financial report..."
        )

        await self.vectordb.upsert(
            doc_id,
            chunk,
            flat_meta
        )

        return {
            "processing_logs": (
                state["processing_logs"]
                + [f"Successfully indexed {doc_id}"]
            )
        }

Step 5: Orchestration (LangGraph Workflow)

We wire the agents together using a sequential StateGraph.

# orchestration/graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

def build_ingestion_graph(
    ingestion_agent,
    normalization_agent,
    indexing_agent
):
    workflow = StateGraph(IngestionState)

    workflow.add_node("fetcher", ingestion_agent)
    workflow.add_node("normalizer", normalization_agent)
    workflow.add_node("indexer", indexing_agent)

    workflow.add_edge(START, "fetcher")
    workflow.add_edge("fetcher", "normalizer")
    workflow.add_edge("normalizer", "indexer")
    workflow.add_edge("indexer", END)

    memory = MemorySaver()

    return workflow.compile(checkpointer=memory)

Step 6: Composition Root (Execution)

The application entry point wires infrastructure, agents, and orchestration together.

# main.py
import asyncio

async def main():
    cms = MockCMSClient()
    vdb = MockVectorDB()
    mem = MockMemoryStore()

    fetcher = IngestionAgent(cms)
    normalizer = NormalizationAgent(mem)
    indexer = IndexingAgent(vdb)

    app = build_ingestion_graph(
        fetcher,
        normalizer,
        indexer
    )

    config = {
        "configurable": {
            "thread_id": "ingestion_run_001"
        }
    }

    initial_state = {
        "doc_id": "doc_99",
        "raw_metadata": {},
        "flattened_metadata": {},
        "processing_logs": []
    }

    async for event in app.astream(
        initial_state,
        config=config
    ):
        print(event)

if __name__ == "__main__":
    asyncio.run(main())

Part 4: Analyzing the Output

When the pipeline processes the intentionally messy metadata structure, collisions are preserved rather than overwritten.

Example Output

--- Starting Enterprise Ingestion Pipeline ---

[FETCHER] Fetched raw metadata for doc_99
[MEMORY] Saved key mapping for doc_99
[NORMALIZER] Flattened metadata. Original keys: 3, Flat keys: 6

--- Flattened Metadata (Notice Collision Suffixes) ---
document.properties.author: Alice Smith
document.properties.created_date: 2024-01-01
document.properties.author_1: Legacy System Override
document.tags.0: finance
document.tags.1: Q3
document.properties.author_2: Bob Jones
--------------------------------------------------

[INDEXER] Successfully indexed doc_99 to VectorDB

Why This Matters for RAG

No Data Loss

Notice that document.properties.author appears three times:

document.properties.author
document.properties.author_1
document.properties.author_2

Instead of silently discarding metadata, the system preserves every value. This allows downstream agents to reason over conflicting information.

Array Handling

Lists are flattened into indexed keys:

document.tags.0
document.tags.1

This structure aligns well with vector database metadata storage requirements.

Traceability

The key mapping stored in memory allows the system to trace any flattened field back to its original source path.

For example:

document.properties.author_2
→ document.properties.author
→ original source object

This lineage becomes critical when users ask where specific metadata originated.

Production Enhancements

While the example demonstrates the core concept, production systems should also:

Preserve original paths automatically during flattening.
Support nested arrays and mixed object/list structures.
Store mappings in Redis, PostgreSQL, or a dedicated metadata service.
Add schema versioning for evolving enterprise content models.
Include metadata validation before indexing.
Track source system provenance and ingestion timestamps.

Conclusion

In enterprise AI, the quality of your RAG system is ultimately limited by the quality of your data pipeline. By implementing collision-aware dictionary flattening and integrating it into a layered LangGraph-based ingestion architecture, organizations can safely normalize complex metadata without losing information.

This approach preserves every metadata variant, maintains complete traceability, supports vector database indexing requirements, and ensures that downstream agents always operate with the richest possible context. The result is a more accurate, auditable, and reliable enterprise RAG system that remains resilient even as metadata schemas evolve over time.