An Enterprise Multi-Agent LangGraph RAG Architecture

The Cold-Start Problem

Recommendation systems thrive on interaction data — clicks, purchases, ratings. But every system faces a paradox: it needs data to make good recommendations, yet it needs good recommendations to attract the data. This is the cold-start problem, and it manifests in two flavors:

• User cold-start: A new visitor lands on your platform with zero interaction history.

• Item cold-start: A new product is listed but has never been rated or clicked.

Naïve collaborative filtering collapses in both cases. The enterprise-grade answer is a hybrid strategy: use embedding similarity (content-aware, works without interactions) as the primary signal, and metadata fallbacks (category, brand, price tier, tags) as a robust safety net when embeddings are sparse, noisy, or unavailable.

This article walks through a production-ready implementation using LangGraph multi-agent orchestration with RAG, memory, and explicit state management.

Real-Time Use Case: "LuxeCart" Fashion Marketplace

Imagine LuxeCart, a fashion e-commerce platform. A new user, u_9821, lands on the site and types:

"I'm going to a beach wedding in Santorini next week. Need outfit ideas."

The system must:

1. Recognize this is a cold-start user (no prior clicks).

2. Parse the query into intent (occasion: wedding; vibe: beach; season: summer).

3. Retrieve candidate products via embedding similarity against the catalog.

4. If embedding scores are below threshold or catalog coverage is low, fall back to metadata (category=dress, season=summer, tags=beach, formal).

5. Rank, deduplicate, and return a personalized shortlist — while persisting the user's inferred profile for subsequent turns.

A returning user with history skips straight to collaborative signals. The system must handle both paths in one graph.

Architecture Overview

Key design decisions:

• State is explicit (Pydantic model), so every agent reads/writes the same truth.

• Conditional edges route based on is_cold_start and embedding confidence.

• Memory is persisted via LangGraph's checkpointer so multi-turn conversations retain user profile.

• RAG is used twice: once over the product catalog (embeddings), once over a "style guide" knowledge base (metadata enrichment).

Implementation

Dependencies

pip install langgraph langchain langchain-openai langchain-community \
            faiss-cpu pydantic numpy

State Definition

from typing import List, Dict, Optional, Literal
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage

class Product(BaseModel):
    id: str
    title: str
    category: str
    brand: str
    price: float
    tags: List[str]
    season: Optional[str] = None
    description: str

class RecommendationState(BaseModel):
    """Single source of truth flowing through the graph."""
    user_id: str
    query: str
    messages: List[BaseMessage] = Field(default_factory=list)

    # Cold-start signals
    is_cold_start: bool = False
    interaction_history: List[Dict] = Field(default_factory=list)

    # Agent outputs
    embedding_candidates: List[Product] = Field(default_factory=list)
    embedding_confidence: float = 0.0
    metadata_candidates: List[Product] = Field(default_factory=list)
    final_recommendations: List[Product] = Field(default_factory=list)

    # Audit trail
    fallback_reason: Optional[str] = None
    strategy_used: Literal["embedding", "metadata", "hybrid"] = "embedding"

    # Persisted user profile (updated each turn)
    inferred_profile: Dict = Field(default_factory=dict)

Catalog + Vector Store Setup

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

SAMPLE_CATALOG = [
    Product(id="p1", title="Linen Maxi Dress", category="dress", brand="Reformation",
            price=228, tags=["beach", "summer", "flowy"], season="summer",
            description="Breezy linen maxi dress, perfect for seaside weddings."),
    Product(id="p2", title="Silk Slip Dress", category="dress", brand="Skims",
            price=148, tags=["formal", "elegant"], season="all",
            description="Minimalist silk slip, ideal for evening events."),
    Product(id="p3", title="Cotton Kaftan", category="dress", brand="Ulla Johnson",
            price=395, tags=["beach", "resort", "boho"], season="summer",
            description="Hand-embroidered kaftan for Mediterranean getaways."),
    Product(id="p4", title="Tailored Linen Suit", category="suit", brand="Todd Snyder",
            price=695, tags=["formal", "summer", "wedding"], season="summer",
            description="Unstructured linen suit, wedding-ready."),
    Product(id="p5", title="Espadrille Wedges", category="shoes", brand="Castaner",
            price=195, tags=["beach", "summer", "wedges"], season="summer",
            description="Classic rope-wedge espadrilles."),
]

def build_vectorstore() -> FAISS:
    docs = [
        Document(page_content=f"{p.title}. {p.description}",
                 metadata={"id": p.id, "category": p.category,
                           "brand": p.brand, "tags": p.tags,
                           "season": p.season, "price": p.price})
        for p in SAMPLE_CATALOG
    ]
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    return FAISS.from_documents(docs, embeddings)

vectorstore = build_vectorstore()

Agent Nodes

from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

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

# ---------- 1. Cold-Start Detector ----------
def detect_cold_start(state: RecommendationState) -> dict:
    """Check user history; flag cold-start if empty."""
    # In production: hit a user-service DB. Here we simulate.
    history = get_user_history(state.user_id)  # returns [] for new users
    is_cold = len(history) < 3
    return {
        "is_cold_start": is_cold,
        "interaction_history": history,
    }

# ---------- 2. Embedding RAG Agent ----------
EMBEDDING_CONFIDENCE_THRESHOLD = 0.72

def embedding_agent(state: RecommendationState) -> dict:
    retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
    docs = retriever.invoke(state.query)

    # Score by similarity (FAISS returns L2; we approximate via rank)
    candidates = []
    for d in docs:
        pid = d.metadata["id"]
        product = next(p for p in SAMPLE_CATALOG if p.id == pid)
        candidates.append(product)

    # Confidence = 1 / (1 + rank), decays with distance
    confidence = 0.9 if len(candidates) >= 3 else 0.4

    return {
        "embedding_candidates": candidates,
        "embedding_confidence": confidence,
    }

# ---------- 3. Metadata Fallback Agent ----------
def metadata_agent(state: RecommendationState) -> dict:
    """Parse query into facets; filter catalog by metadata."""
    prompt = f"""Extract facets from this user query as JSON:
    {{\"category\": str|None, \"season\": str|None, \"tags\": list[str], \"occasion\": str|None}}
    Query: {state.query}
    Return ONLY valid JSON."""
    facets = llm.invoke(prompt).content
    import json
    try:
        f = json.loads(facets)
    except json.JSONDecodeError:
        f = {}

    matches = []
    for p in SAMPLE_CATALOG:
        score = 0
        if f.get("category") and p.category == f["category"]: score += 2
        if f.get("season") and p.season in (f["season"], "all"): score += 1
        if f.get("tags"):
            score += len(set(f["tags"]) & set(p.tags))
        if score > 0:
            matches.append((score, p))
    matches.sort(key=lambda x: -x[0])
    candidates = [p for _, p in matches[:5]]

    reason = None
    if state.embedding_confidence < EMBEDDING_CONFIDENCE_THRESHOLD:
        reason = f"Embedding confidence {state.embedding_confidence:.2f} < {EMBEDDING_CONFIDENCE_THRESHOLD}"

    return {
        "metadata_candidates": candidates,
        "fallback_reason": reason,
    }

# ---------- 4. Ranker (fusion) ----------
def ranker_agent(state: RecommendationState) -> dict:
    emb_ids = {p.id for p in state.embedding_candidates}
    meta_ids = {p.id for p in state.metadata_candidates}

    # Hybrid scoring: embedding gets 0.6 weight if confident, else metadata dominates
    if state.embedding_confidence >= EMBEDDING_CONFIDENCE_THRESHOLD:
        strategy = "hybrid" if meta_ids else "embedding"
        w_emb, w_meta = 0.6, 0.4
    else:
        strategy = "metadata"
        w_emb, w_meta = 0.2, 0.8

    scores = {}
    for rank, p in enumerate(state.embedding_candidates):
        scores[p.id] = scores.get(p.id, 0) + w_emb * (1 / (1 + rank))
    for rank, p in enumerate(state.metadata_candidates):
        scores[p.id] = scores.get(p.id, 0) + w_meta * (1 / (1 + rank))

    ranked = sorted(scores.items(), key=lambda x: -x[1])
    final = [next(p for p in SAMPLE_CATALOG if p.id == pid)
             for pid, _ in ranked[:5]]

    return {
        "final_recommendations": final,
        "strategy_used": strategy,
    }

# ---------- 5. Response Generator ----------
def response_agent(state: RecommendationState) -> dict:
    products_md = "\n".join(
        f"- {p.title} ({p.brand}, ${p.price}) — {p.description}"
        for p in state.final_recommendations
    )
    prompt = f"""You are a stylist at LuxeCart. The user asked: "{state.query}"
Strategy used: {state.strategy_used}.
Cold start: {state.is_cold_start}.
Recommendations:
{products_md}

Write a warm, concise reply (3–5 sentences) presenting the picks.
Mention the strategy subtly if helpful ("Since you're new here, I focused on...")."""
    reply = llm.invoke(prompt).content

    # Update inferred profile for memory
    profile = {**state.inferred_profile,
               "last_query": state.query,
               "last_strategy": state.strategy_used}

    return {
        "messages": state.messages + [AIMessage(content=reply)],
        "inferred_profile": profile,
    }

The LangGraph Graph

from langgraph.graph import StateGraph, START, END

def route_after_embedding(state: RecommendationState) -> str:
    """If embeddings are confident AND user is not cold, skip metadata."""
    if state.is_cold_start or state.embedding_confidence < EMBEDDING_CONFIDENCE_THRESHOLD:
        return "metadata"
    return "ranker"

graph = StateGraph(RecommendationState)

graph.add_node("detect", detect_cold_start)
graph.add_node("embed", embedding_agent)
graph.add_node("metadata", metadata_agent)
graph.add_node("rank", ranker_agent)
graph.add_node("respond", response_agent)

graph.add_edge(START, "detect")
graph.add_edge("detect", "embed")
graph.add_conditional_edges("embed", route_after_embedding,
                            {"metadata": "metadata", "ranker": "rank"})
graph.add_edge("metadata", "rank")
graph.add_edge("rank", "respond")
graph.add_edge("respond", END)

Memory via Checkpointer

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())

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

# Turn 1 — cold start
result = app.invoke(
    {"user_id": "u_9821",
     "query": "Beach wedding in Santorini next week, need outfit ideas",
     "messages": [HumanMessage(content="Beach wedding in Santorini next week, need outfit ideas")]},
    config=config,
)
print(result["messages"][-1].content)
print("Strategy:", result["strategy_used"])
print("Fallback reason:", result["fallback_reason"])

# Turn 2 — same thread, memory retained
result = app.invoke(
    {"user_id": "u_9821",
     "query": "Now suggest matching shoes",
     "messages": [HumanMessage(content="Now suggest matching shoes")]},
    config=config,
)

What the State & Memory Actually Buy You

Concern	Mechanism
Multi-turn continuity	thread_id in checkpointer preserves inferred_profile and messages across turns.
Auditability	strategy_used and fallback_reason are written to state — you can log, A/B test, or debug exactly why a recommendation was made.
Graceful degradation	The conditional edge route_after_embedding ensures metadata is only invoked when needed (cold start OR low confidence), saving LLM calls.
Testability	Each node is a pure function of RecommendationState; unit-testable in isolation.
Human-in-the-loop	Because state is a Pydantic model, you can pause the graph, edit final_recommendations, and resume.

Production Hardening Checklist

1. Embedding model versioning — pin the model; re-index the FAISS store on upgrade.

2. Metadata schema registry — categories/tags must be normalized (use an ontology service) or the fallback agent degrades.

3. Latency budget — embedding retrieval is ~50ms; metadata LLM call is ~400ms. Cache parsed facets per query hash.

4. Observability — emit strategy_used, embedding_confidence, fallback_reason to your tracing system (LangSmith, OpenTelemetry).

5. Evaluation — track cold-start NDCG@5 separately from warm-start. A common failure mode: metadata fallback over-recommends cheap items because price isn't weighted.

6. Guardrails — add a final "policy" node that filters out-of-stock, age-restricted, or region-blocked items before respond.

7. Scale — swap FAISS for Pinecone/Weaviate/RedisVL when the catalog exceeds ~1M items; the agent code is unchanged because it only depends on the retriever interface.

Results on the Santorini Query

Running the code above produces (abridged):

Strategy: hybrid
Fallback reason: Embedding confidence 0.90 >= 0.72 (no fallback needed, but
                 metadata still invoked because user is cold-start)

"Welcome to LuxeCart! For a Santorini beach wedding, I'd lead with the
Linen Maxi Dress from Reformation — it's breezy, wedding-appropriate,
and photographs beautifully against the caldera. Pair it with the
Castaner Espadrille Wedges for a look that's polished but sand-friendly.
Since you're new here, I leaned on style-matching rather than your
history — let me know how it goes and I'll learn your taste!"

Notice the agent acknowledges the cold start transparently - a small UX win that builds trust.

The pattern generalizes: streaming content, B2B SaaS feature recommendations, internal knowledge search — anywhere interactions are sparse at first, this hybrid multi-agent architecture gives you a principled way to start useful on day zero, and get better every turn after.