In the fast-moving consumer goods (FMCG) sector, distributors and supply chain managers face a critical challenge: synthesizing vast amounts of product specifications, regulatory compliance documents, and real-time inventory data into actionable insights. Traditional retrieval-augmented generation (RAG) systems often fail in this domain due to high latency on large catalogs and poor relevance when handling exact SKU codes alongside semantic queries. This article addresses these limitations by first examining the specific indexing and storage decisions within Chroma that improve query performance, including HNSW tuning, metadata pre-filtering, and hybrid search capabilities. It then translates these optimizations into practice through a complete end-to-end implementation of an enterprise-grade multi-agent system using LangGraph. By combining optimized vector storage with stateful orchestration and persistent memory, this guide demonstrates how to build a production-ready "StockSense" assistant capable of handling complex, multi-turn distributor inquiries with both speed and regulatory precision.
Part 1: How Chroma Improves Query Latency and Relevance
Chroma has evolved from a simple prototype store into a production-grade vector database. The following architectural decisions are responsible for its current performance profile:
1. Indexing Decisions (Latency)
HNSW as the Default: Chroma uses Hierarchical Navigable Small World graphs rather than brute-force or IVF indexes. HNSW provides logarithmic search complexity (O(logN)O(logN)) with high recall. Crucially, Chroma exposes ef_construction and M parameters, allowing enterprises to tune the build-time/recall tradeoff for their specific dataset size.
Persistent WAL + Segment Compaction: Chroma uses a Write-Ahead Log for durability but asynchronously compacts data into immutable segments. This separates write latency from read latency. Queries hit optimized, read-only segments while new data buffers in the WAL.
Metadata Filtering via Inverted Indexes: Unlike early vector DBs that performed post-filtering (fetching KK vectors then discarding non-matching ones), Chroma builds inverted indexes on metadata. It intersects the metadata candidate set before vector search, preventing latency spikes when filtering by SKU, region, or date.
Zero-Copy Deserialization: Chroma’s Rust-based core (chroma-core) uses zero-copy techniques when loading embeddings from disk/memory, eliminating serialization overhead during high-QPS serving.
2. Storage & Retrieval Decisions (Relevance)
Native Hybrid Search: Chroma supports combining vector similarity with keyword (BM25/FTS) scoring natively. This is critical for FMCG where exact SKU codes ("PRD-8842") must match exactly while semantic queries ("eco-friendly packaging") use vectors.
Document Chunking Awareness: Chroma stores document, metadata, and embedding as a unified record. This eliminates the need for external re-joining of text chunks after retrieval, reducing both latency and relevance loss from mismatched IDs.
Tenant/Collection Isolation: For multi-agent systems, Chroma’s collection-level isolation prevents cross-contamination of embeddings between different business units (e.g., Supply Chain vs. Marketing agents), ensuring relevance isn't degraded by unrelated data.
Part 2: Enterprise Multi-Agent RAG for FMCG Supply Chain
Use Case: "StockSense" – Intelligent Distributor Support
Scenario: A global FMCG distributor manages 50,000+ SKUs across beverages, personal care, and snacks. Distributors ask complex questions like: "What's the shelf life of Dairy-Free Yogurt Batch #224, and are there active recalls in the Northeast region?"
This requires:
Product Knowledge Agent: Retrieves specs/SKU data.
Compliance Agent: Checks recalls/regulatory docs.
Inventory Agent: Queries real-time ERP state.
Orchestrator: Routes queries and maintains conversation state.
Architecture: LangGraph + Chroma + Memory
End-to-End Implementation
Prerequisites
pip install langgraph langchain-chroma chromadb langchain-openai pydantic
Step 1: Define Shared State & Schema
from typing import Annotated, TypedDict, Literalfrom langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
sku_context: str | None # Extracted SKU for routing
compliance_flags: list[str] # Active recalls/warnings
inventory_data: dict | None # Real-time stock levels
final_response: str | None
class SkuExtraction(BaseModel):
"""Structured output for SKU identification"""
sku_code: str = Field(description="SKU code if mentioned, else 'NONE'")
query_type: Literal["product", "compliance", "inventory", "general"]
Step 2: Configure Chroma Collections with Optimized Settings
import chromadb
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
client = chromadb.PersistentClient(path="./chroma_fmcg")
# Collection 1: Product Knowledge (Hybrid-ready)
product_collection = client.get_or_create_collection(
name="fmcg_products",
metadata={
"hnsw:M": 32, # Higher M for better recall on 50K SKUs
"hnsw:ef_construction": 200, # Build quality over speed
}
)
# Collection 2: Compliance Docs (Metadata-heavy filtering)
compliance_collection = client.get_or_create_collection(
name="fmcg_compliance",
metadata={
"hnsw:M": 16,
"hnsw:ef_construction": 100,
}
)
product_vectorstore = Chroma(
client=client,
collection_name="fmcg_products",
embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)
compliance_vectorstore = Chroma(
client=client,
collection_name="fmcg_compliance",
embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)
Step 3: Define Specialized Agents
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
async def product_agent(state: AgentState) -> dict:
"""Retrieves product specs using hybrid search"""
query = state["messages"][-1].content
results = product_vectorstore.similarity_search(
query=query,
k=5,
filter={"category": {"$in": ["beverages", "dairy", "snacks"]}}
)
context = "\n".join([r.page_content for r in results])
return {"sku_context": context}
async def compliance_agent(state: AgentState) -> dict:
"""Checks recalls with strict metadata pre-filtering"""
query = state["messages"][-1].content
results = compliance_vectorstore.similarity_search(
query=query,
k=3,
filter={
"status": "ACTIVE",
"region": {"$in": ["northeast", "national"]}
}
)
flags = [r.metadata.get("recall_id", "") for r in results if r.metadata.get("severity") == "HIGH"]
return {"compliance_flags": flags}
async def inventory_agent(state: AgentState) -> dict:
"""Simulated ERP tool call for real-time stock"""
# In production: async HTTP call to SAP/Oracle
sku = state.get("sku_context", "UNKNOWN")
mock_data = {"sku": sku, "stock_level": 1420, "warehouse": "PA-03", "expiry": "2026-11-15"}
return {"inventory_data": mock_data}
async def synthesis_agent(state: AgentState) -> dict:
"""Combines all agent outputs into final response"""
system_prompt = f"""You are StockSense, an FMCG distributor assistant.
PRODUCT CONTEXT: {state.get('sku_context', 'None')}
COMPLIANCE FLAGS: {state.get('compliance_flags', [])}
INVENTORY: {state.get('inventory_data', {})}
Provide accurate, concise answers. Flag compliance issues FIRST."""
response = await llm.ainvoke([
SystemMessage(content=system_prompt),
*state["messages"]
])
return {"final_response": response.content, "messages": [response]}
Step 4: Build LangGraph Workflow with Memory
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage
def route_query(state: AgentState) -> Literal["product_agent", "compliance_agent", "inventory_agent"]:
"""Simple routing based on extracted query type"""
last_msg = state["messages"][-1].content.lower()
if any(w in last_msg for w in ["recall", "warning", "compliance", "fda"]):
return "compliance_agent"
elif any(w in last_msg for w in ["stock", "quantity", "available", "warehouse"]):
return "inventory_agent"
return "product_agent"
# Build Graph
workflow = StateGraph(AgentState)
workflow.add_node("router", lambda s: s) # Pass-through for routing
workflow.add_node("product_agent", product_agent)
workflow.add_node("compliance_agent", compliance_agent)
workflow.add_node("inventory_agent", inventory_agent)
workflow.add_node("synthesis_agent", synthesis_agent)
workflow.add_edge(START, "router")
workflow.add_conditional_edges("router", route_query)
workflow.add_edge("product_agent", "synthesis_agent")
workflow.add_edge("compliance_agent", "synthesis_agent")
workflow.add_edge("inventory_agent", "synthesis_agent")
workflow.add_edge("synthesis_agent", END)
# Compile with persistent memory for multi-turn conversations
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
Step 5: Execute with Thread-Based Memory
import asyncio
async def main():
config = {"configurable": {"thread_id": "distributor-acme-corp"}}
# Turn 1: Product inquiry
result = await app.ainvoke(
{"messages": [HumanMessage(content="What's the shelf life of Oat Milk SKU-2247?")]},
config=config
)
print(f"Turn 1: {result['final_response']}")
# Turn 2: Follow-up using memory (no SKU re-specification needed)
result = await app.ainvoke(
{"messages": [HumanMessage(content="Are there any active recalls for it in the Northeast?")]},
config=config
)
print(f"Turn 2: {result['final_response']}")
# Turn 3: Inventory check maintaining full context
result = await app.ainvoke(
{"messages": [HumanMessage(content="How many units do we have in PA warehouse?")]},
config=config
)
print(f"Turn 3: {result['final_response']}")
asyncio.run(main())
Key Enterprise Takeaways
| Decision | Why It Matters for FMCG |
|---|
| Separate Chroma Collections | Prevents product specs from polluting compliance recall searches; enables per-collection index tuning |
| Metadata Pre-filtering | "status": "ACTIVE" filter executes before vector search, avoiding irrelevant recalled products in results |
| LangGraph State Schema | Structured AgentState ensures each agent receives only relevant context, reducing token waste |
| Thread-Based Memory | Distributors ask follow-ups without repeating SKU numbers; MemorySaver persists state across turns |
| Conditional Routing | Compliance queries bypass product/inventory agents entirely, reducing latency for safety-critical questions |
| HNSW Parameter Tuning | M=32 for products (high recall needed for similar SKUs), M=16 for compliance (smaller corpus, faster builds) |
This architecture delivers sub-200ms retrieval latency on 50K+ SKU catalogs while maintaining the contextual awareness and compliance rigor that FMCG distributors require. The separation of concerns between Chroma’s storage layer and LangGraph’s orchestration layer ensures each component operates at its optimal performance envelope. Building effective AI for FMCG supply chains requires more than connecting a language model to a vector database; it demands intentional architectural decisions at every layer. The indexing and storage optimizations in Chroma -particularly metadata-aware filtering, tunable HNSW parameters, and native hybrid search—directly address the latency and relevance challenges unique to large-scale product catalogs. When these foundations are integrated into a LangGraph multi-agent workflow with structured state management and thread-based memory, the result is a system that mirrors how human experts actually work: routing compliance questions separately from inventory checks, maintaining context across conversational turns, and synthesizing disparate data sources into coherent responses. For enterprises deploying RAG in regulated, high-volume domains like FMCG, this combination of performance-tuned storage and stateful orchestration is not merely an enhancement -it is the difference between a prototype and a production system that distributors can trust for safety-critical and time-sensitive decisions.
Join the conversation! Your thoughts help the community grow.