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.
![Real-World Use Case]()
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.