Product Recommendation Agent

Amazon attributes 35% of revenue to its recommendation engine. Netflix saved $1 billion per year in customer retention through personalized recommendations. In e-commerce, showing the right product at the right time is worth its weight in gold.

Why LLM-Powered Recommendations?

Recommendation Agent Architecture

         Customer Input
        "I need a birthday gift for my tech-loving brother"
                │
                ▼
        ┌───────────────┐
        │  Intent       │
        │  Analyzer     │ ─── Extract: gift, male, tech, birthday
        └───────┬───────┘
                │
                ▼
        ┌───────────────┐
        │  Customer     │
        │  Profile      │ ─── Load: past purchases, preferences
        └───────┬───────┘
                │
                ▼
        ┌───────────────┐
        │  Product      │
        │  Search       │ ─── Query: tech products, gift-worthy
        └───────┬───────┘
                │
                ▼
        ┌───────────────┐
        │  Ranking &    │
        │  Filtering    │ ─── Apply: budget, inventory, preferences
        └───────┬───────┘
                │
                ▼
        ┌───────────────┐
        │  Response     │
        │  Generation   │ ─── Format: personalized recommendations
        └───────────────┘

Step 1: Define the State

from typing import TypedDict, Annotated, Optional, Literal
from langchain_core.messages import BaseMessage
import operator

class ProductRecommendationState(TypedDict):
    # Conversation
    messages: Annotated[list[BaseMessage], operator.add]

    # Customer profile
    customer_id: str
    customer_name: str
    purchase_history: list[dict]  # Past orders
    browsing_history: list[str]   # Recently viewed SKUs
    preferences: dict             # Size, color, brand preferences

    # Current request
    query: str
    intent: Optional[dict]  # Parsed intent from query

    # Search parameters (extracted from intent)
    category: Optional[str]
    price_min: Optional[float]
    price_max: Optional[float]
    occasion: Optional[str]
    recipient: Optional[str]  # self, gift

    # Results
    candidate_products: list[dict]
    filtered_products: list[dict]
    recommendations: list[dict]

    # Response
    explanation: str

Step 2: Build the Agent Components

Intent Analyzer

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import json

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

INTENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """Analyze the customer's product request and extract structured information.

Customer Profile:
- Recent purchases: {purchase_history}
- Browsing history: {browsing_history}
- Known preferences: {preferences}

Extract the following in JSON format:
{{
    "intent_type": "browse" | "search" | "gift" | "replenish" | "compare",
    "category": "electronics" | "clothing" | "home" | "beauty" | etc.,
    "price_range": {{"min": number or null, "max": number or null}},
    "occasion": "birthday" | "holiday" | "everyday" | null,
    "recipient": "self" | "gift_male" | "gift_female" | "gift_child" | null,
    "attributes": ["wireless", "compact", "premium", etc.],
    "urgency": "immediate" | "flexible" | null
}}"""),
    ("human", "{query}")
])

def analyze_intent(state: ProductRecommendationState) -> dict:
    """Parse customer query to understand intent"""
    # Summarize purchase history
    purchase_summary = [
        f"{p['name']} (${p['price']})"
        for p in state.get("purchase_history", [])[-5:]
    ]

    response = llm.invoke(
        INTENT_PROMPT.format(
            purchase_history=purchase_summary,
            browsing_history=state.get("browsing_history", [])[-10:],
            preferences=state.get("preferences", {}),
            query=state["query"]
        )
    )

    intent = json.loads(response.content)

    return {
        "intent": intent,
        "category": intent.get("category"),
        "price_min": intent.get("price_range", {}).get("min"),
        "price_max": intent.get("price_range", {}).get("max"),
        "occasion": intent.get("occasion"),
        "recipient": intent.get("recipient")
    }

Product Search

# Mock product catalog - in production, use Elasticsearch or similar
PRODUCT_CATALOG = [
    {
        "sku": "TECH-001",
        "name": "Wireless Noise-Canceling Headphones",
        "category": "electronics",
        "price": 299.99,
        "attributes": ["wireless", "premium", "noise-canceling"],
        "rating": 4.8,
        "in_stock": True,
        "gift_score": 0.9  # Good for gifting
    },
    {
        "sku": "TECH-002",
        "name": "Portable Bluetooth Speaker",
        "category": "electronics",
        "price": 79.99,
        "attributes": ["wireless", "portable", "waterproof"],
        "rating": 4.5,
        "in_stock": True,
        "gift_score": 0.85
    },
    {
        "sku": "TECH-003",
        "name": "Smart Watch Fitness Tracker",
        "category": "electronics",
        "price": 199.99,
        "attributes": ["wearable", "fitness", "smart"],
        "rating": 4.6,
        "in_stock": True,
        "gift_score": 0.8
    },
    {
        "sku": "HOME-001",
        "name": "Smart Home Hub",
        "category": "electronics",
        "price": 129.99,
        "attributes": ["smart", "home-automation"],
        "rating": 4.3,
        "in_stock": False,  # Out of stock
        "gift_score": 0.7
    },
    {
        "sku": "ACC-001",
        "name": "Premium Leather Wallet",
        "category": "accessories",
        "price": 89.99,
        "attributes": ["leather", "premium", "compact"],
        "rating": 4.7,
        "in_stock": True,
        "gift_score": 0.95
    }
]

def search_products(state: ProductRecommendationState) -> dict:
    """Search product catalog based on extracted intent"""
    intent = state.get("intent", {})
    candidates = []

    for product in PRODUCT_CATALOG:
        score = 0

        # Category match
        if state.get("category"):
            if product["category"] == state["category"]:
                score += 3
            elif state["category"] in ["tech", "electronics"] and product["category"] == "electronics":
                score += 3

        # Price range
        price_min = state.get("price_min", 0)
        price_max = state.get("price_max", float('inf'))
        if price_min <= product["price"] <= price_max:
            score += 2

        # Attribute match
        query_attrs = intent.get("attributes", [])
        matching_attrs = set(query_attrs) & set(product["attributes"])
        score += len(matching_attrs)

        # Gift score (if buying for someone else)
        if state.get("recipient") and state["recipient"] != "self":
            score += product["gift_score"] * 2

        # Rating boost
        score += product["rating"] / 2

        candidates.append({
            **product,
            "relevance_score": round(score, 2)
        })

    # Sort by relevance
    candidates.sort(key=lambda x: x["relevance_score"], reverse=True)

    return {"candidate_products": candidates[:10]}  # Top 10

Filter & Rank

def filter_and_rank(state: ProductRecommendationState) -> dict:
    """Apply business rules and personalization"""
    candidates = state.get("candidate_products", [])
    customer_purchases = state.get("purchase_history", [])

    filtered = []
    for product in candidates:
        # Exclude out-of-stock (but note it for transparency)
        if not product["in_stock"]:
            product["availability_note"] = "Currently out of stock"

        # Exclude recently purchased (don't recommend same item)
        purchased_skus = [p["sku"] for p in customer_purchases]
        if product["sku"] in purchased_skus:
            continue

        # Boost if customer has shown category interest
        if product["category"] in [p.get("category") for p in customer_purchases]:
            product["relevance_score"] += 1

        filtered.append(product)

    # Separate in-stock and out-of-stock
    in_stock = [p for p in filtered if p["in_stock"]]
    out_of_stock = [p for p in filtered if not p["in_stock"]]

    # Primary: in-stock, sorted by score
    # Secondary: out-of-stock alternatives
    final = in_stock[:5]  # Top 5 in-stock
    if out_of_stock:
        final.append({**out_of_stock[0], "is_alternative": True})

    return {"filtered_products": final}

Generate Recommendations

RECOMMENDATION_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a personal shopping assistant. Generate personalized product recommendations.

Customer: {customer_name}
Their Request: {query}
Intent Analysis: {intent}

Products to recommend:
{products}

Guidelines:
1. Lead with the BEST match and explain why
2. For gifts: mention gift-worthiness and who it's good for
3. Note any out-of-stock items and suggest alternatives
4. If price is a concern, highlight value
5. Be conversational and helpful, not salesy

Format your response as:
1. Brief acknowledgment of their need
2. Top recommendation with explanation
3. 2-3 alternatives with brief reasons
4. Any relevant notes (stock, shipping, etc.)"""),
    ("human", "Generate recommendations")
])

def generate_recommendations(state: ProductRecommendationState) -> dict:
    """Generate personalized recommendation response"""
    products = state.get("filtered_products", [])

    # Format products for the LLM
    product_text = "\n".join([
        f"- {p['name']} (${p['price']}) - Rating: {p['rating']}/5 - {', '.join(p['attributes'])}"
        + (f" [OUT OF STOCK]" if not p['in_stock'] else "")
        for p in products
    ])

    response = llm.invoke(
        RECOMMENDATION_PROMPT.format(
            customer_name=state.get("customer_name", "there"),
            query=state["query"],
            intent=state.get("intent", {}),
            products=product_text
        )
    )

    # Structure the recommendations
    recommendations = [
        {
            "sku": p["sku"],
            "name": p["name"],
            "price": p["price"],
            "in_stock": p["in_stock"],
            "relevance_score": p["relevance_score"]
        }
        for p in products[:3]
    ]

    return {
        "recommendations": recommendations,
        "explanation": response.content,
        "messages": [{"role": "assistant", "content": response.content}]
    }

Step 3: Build the Graph

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# Build workflow
workflow = StateGraph(ProductRecommendationState)

workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("search_products", search_products)
workflow.add_node("filter_and_rank", filter_and_rank)
workflow.add_node("generate_recommendations", generate_recommendations)

workflow.add_edge(START, "analyze_intent")
workflow.add_edge("analyze_intent", "search_products")
workflow.add_edge("search_products", "filter_and_rank")
workflow.add_edge("filter_and_rank", "generate_recommendations")
workflow.add_edge("generate_recommendations", END)

# Compile with memory for session persistence
memory = MemorySaver()
recommendation_agent = workflow.compile(checkpointer=memory)

Step 4: Run Recommendations

# Customer profile
customer_profile = {
    "customer_id": "CUST-789",
    "customer_name": "Alex",
    "purchase_history": [
        {"sku": "TECH-005", "name": "USB-C Hub", "price": 49.99, "category": "electronics"},
        {"sku": "TECH-006", "name": "Laptop Stand", "price": 79.99, "category": "electronics"}
    ],
    "browsing_history": ["TECH-001", "TECH-002", "TECH-003"],
    "preferences": {"brands": ["Sony", "Bose"], "budget": "mid-range"}
}

# Get recommendations
result = recommendation_agent.invoke({
    **customer_profile,
    "messages": [],
    "query": "I need a birthday gift for my tech-loving brother, budget around $100-200",
    "candidate_products": [],
    "filtered_products": [],
    "recommendations": [],
    "explanation": ""
}, {"configurable": {"thread_id": "session_alex_123"}})

print(result["explanation"])

I'd be happy to help you find the perfect tech gift for your brother!

**Top Pick: Smart Watch Fitness Tracker ($199.99)**
This is an excellent choice for a tech enthusiast - it combines fitness tracking
with smart notifications, and at 4.6 stars, it's highly rated. Perfect for
someone who appreciates gadgets and wants to stay active.

**Great Alternatives:**
1. **Wireless Noise-Canceling Headphones ($299.99)** - Slightly above budget but
   exceptional quality. If he's into music or works from home, these are a
   premium gift he'll use daily.

2. **Portable Bluetooth Speaker ($79.99)** - Under budget with room for a small
   accessory! Waterproof and portable, great for someone who enjoys music on
   the go.

All items are in stock and can ship in time for his birthday!

Follow-up Conversations

# Continue the conversation
result = recommendation_agent.invoke({
    "messages": [{"role": "user", "content": "Does the smart watch work with iPhone?"}],
    "query": "Does the smart watch work with iPhone?"
}, {"configurable": {"thread_id": "session_alex_123"}})

# Agent remembers context and can answer about the previously recommended watch

Key Takeaways

Next up: Inventory Agent - Building autonomous stock monitoring and reorder systems.