Personalized recommendations with CrewAI agents

Personalized recommendations crewai systems work best when you decompose the problem into specialized agents that each handle a distinct stage: understanding the user, retrieving relevant candidates, and ranking them with business logic. This tutorial walks through building a three-agent pipeline that produces explainable, context-aware product recommendations for an e-commerce catalog. You’ll end up with runnable code you can extend for production workloads.

Prerequisites

• Python 3.10+
• An OpenAI-compatible API key (OpenRouter, OpenAI, or any compatible endpoint)
• A small product catalog — we’ll generate a synthetic one, but you can swap in your own CSV/JSON
• Familiarity with CrewAI basics: agents, tasks, crews, and the @tool decorator

Install dependencies:

pip install crewai crewai-tools pandas numpy faker

If you’re running this through n4n.ai’s gateway, set OPENAI_BASE_URL and OPENAI_API_KEY to point at the endpoint — the code below uses the standard OpenAI client interface so it works unchanged.

Project structure

recsys/
├── catalog.py          # Synthetic catalog generator
├── tools.py            # CrewAI tools for retrieval and ranking
├── agents.py           # Agent definitions
├── tasks.py            # Task definitions
├── main.py             # Entry point
└── requirements.txt

We’ll build each file in order. The full runnable repo is ~120 lines of logic plus the catalog generator.

Step 1: Generate a synthetic catalog

Real catalogs have product IDs, titles, categories, price tiers, tags, and textual descriptions. We’ll synthesize 200 products across 8 categories with enough metadata to make retrieval interesting.

# catalog.py
import pandas as pd
import numpy as np
from faker import Faker

fake = Faker()
Faker.seed(42)
np.random.seed(42)

CATEGORIES = [
    "electronics", "apparel", "home", "beauty", "sports", "books", "toys", "grocery"
]

PRICE_TIERS = ["budget", "mid", "premium"]

TAG_POOL = [
    "wireless", "organic", "eco-friendly", "compact", "durable", "giftable",
    "bestseller", "new-arrival", "sale", "limited-edition", "waterproof",
    "noise-cancelling", "ergonomic", "vegan", "gluten-free", "smart-home"
]

def generate_catalog(n=200) -> pd.DataFrame:
    rows = []
    for i in range(n):
        cat = np.random.choice(CATEGORIES)
        tier = np.random.choice(PRICE_TIERS, p=[0.4, 0.4, 0.2])
        base_price = {"budget": 15, "mid": 60, "premium": 200}[tier]
        price = round(base_price * np.random.lognormal(0, 0.4), 2)
        tags = np.random.choice(TAG_POOL, size=np.random.randint(1, 5), replace=False)
        rows.append({
            "product_id": f"SKU-{i:04d}",
            "title": fake.catch_phrase().title(),
            "category": cat,
            "price": price,
            "price_tier": tier,
            "tags": ", ".join(tags),
            "description": fake.paragraph(nb_sentences=3),
            "rating": round(np.random.uniform(3.0, 5.0), 1),
            "review_count": int(np.random.exponential(200)),
        })
    return pd.DataFrame(rows)

if __name__ == "__main__":
    df = generate_catalog()
    df.to_parquet("catalog.parquet", index=False)
    print(f"Generated {len(df)} products")
    print(df.head(3).to_string())

Run it:

python catalog.py

Expected output:

Generated 200 products
  product_id                title   category  price price_tier  ...  rating  review_count
0  SKU-0000  Wireless Ergonomic Hub  electronics  89.47       mid  ...     4.3           156
1  SKU-0001   Eco-Friendly Smart Hub      home  34.12    budget  ...     4.7           312
2  SKU-0002   Compact Durable Speaker  electronics  45.89    budget  ...     4.1            89

Step 2: Build retrieval and ranking tools

CrewAI tools wrap Python functions so agents can call them. We need two: one for candidate retrieval (semantic + metadata filtering) and one for business-aware ranking.

# tools.py
import json
import pandas as pd
import numpy as np
from sentence_transformers import SentenceTransformer
from crewai.tools import tool

# Load once at module import
_catalog = pd.read_parquet("catalog.parquet")
_model = SentenceTransformer("all-MiniLM-L6-v2")
_catalog_embeddings = _model.encode(_catalog["description"].tolist(), show_progress_bar=False)

def _cosine_sim(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

@tool("retrieve_candidates")
def retrieve_candidates(user_profile: str, top_k: int = 50) -> str:
    """
    Retrieve top-k candidate products for a user profile string.
    Returns JSON list of product dicts with similarity scores.
    """
    query_vec = _model.encode([user_profile])[0]
    sims = np.array([_cosine_sim(query_vec, emb) for emb in _catalog_embeddings])
    top_idx = np.argpartition(sims, -top_k)[-top_k:]
    top_idx = top_idx[np.argsort(sims[top_idx])[::-1]]
    
    results = _catalog.iloc[top_idx].copy()
    results["similarity"] = sims[top_idx]
    return results.to_json(orient="records")

@tool("rank_candidates")
def rank_candidates(candidates_json: str, user_context: str, n: int = 10) -> str:
    """
    Rank candidates using business rules: rating, review count, price alignment,
    and semantic relevance. Returns top-n as JSON.
    """
    candidates = pd.read_json(candidates_json)
    ctx = user_context.lower()
    
    # Price tier preference from context
    tier_pref = None
    if "budget" in ctx or "cheap" in ctx or "affordable" in ctx:
        tier_pref = "budget"
    elif "premium" in ctx or "high-end" in ctx or "luxury" in ctx:
        tier_pref = "premium"
    
    def score(row):
        s = 0.0
        s += 0.4 * row["similarity"]
        s += 0.2 * (row["rating"] / 5.0)
        s += 0.1 * min(row["review_count"] / 1000, 1.0)
        if tier_pref and row["price_tier"] == tier_pref:
            s += 0.3
        elif tier_pref and row["price_tier"] != tier_pref:
            s -= 0.1
        # Boost for tag matches in context
        for tag in row["tags"].split(", "):
            if tag.lower() in ctx:
                s += 0.05
        return s
    
    candidates["score"] = candidates.apply(score, axis=1)
    top = candidates.nlargest(n, "score")
    return top[["product_id", "title", "category", "price", "rating", "score"]].to_json(orient="records")

Notes:

• We use sentence-transformers/all-MiniLM-L6-v2 locally — no API call for embeddings. Swap for a hosted embedding model if you prefer.
• The ranking heuristic is intentionally simple and transparent. Replace with a learned model later.

Step 3: Define the three agents

Each agent owns one stage. The profiler converts raw user signals into a structured profile. The retriever calls the retrieval tool. The ranker applies business logic and produces the final slate with explanations.

# agents.py
from crewai import Agent
from tools import retrieve_candidates, rank_candidates

profiler = Agent(
    role="User Profiler",
    goal="Convert raw user signals (purchase history, browsing, explicit preferences) into a concise semantic profile for retrieval",
    backstory=(
        "You are an expert at distilling noisy behavioral data into a clean "
        "natural-language profile that captures intent, price sensitivity, "
        "category affinities, and attribute preferences."
    ),
    tools=[],
    verbose=True,
    allow_delegation=False,
)

retriever = Agent(
    role="Candidate Retriever",
    goal="Fetch a diverse set of semantically relevant products from the catalog using the user profile",
    backstory=(
        "You specialize in vector search over product descriptions. You know "
        "how to balance relevance with diversity so the ranker has good options."
    ),
    tools=[retrieve_candidates],
    verbose=True,
    allow_delegation=False,
)

ranker = Agent(
    role="Business Ranker",
    goal="Apply business rules and context to produce a final ranked slate of 10 recommendations with explanations",
    backstory=(
        "You understand margins, inventory, seasonality, and user trust signals. "
        "You turn a candidate set into a merchandised recommendation list."
    ),
    tools=[rank_candidates],
    verbose=True,
    allow_delegation=False,
)

Step 4: Define tasks with explicit output contracts

Tasks describe what each agent must produce. We use output_json to enforce structure — this makes downstream parsing reliable.

# tasks.py
from crewai import Task
from agents import profiler, retriever, ranker

profile_task = Task(
    description=(
        "Analyze the user's raw signals and produce a structured profile.\n"
        "Signals: {user_signals}\n\n"
        "Output JSON with keys: "
        "semantic_profile (string for embedding), "
        "price_sensitivity (budget/mid/premium), "
        "category_affinities (list), "
        "attribute_preferences (list), "
        "intent_summary (string)."
    ),
    expected_output="Valid JSON object with the five keys above.",
    agent=profiler,
    output_json={
        "semantic_profile": "string",
        "price_sensitivity": "string",
        "category_affinities": "array",
        "attribute_preferences": "array",
        "intent_summary": "string"
    },
)

retrieve_task = Task(
    description=(
        "Use the semantic_profile from the previous task to retrieve 50 candidates. "
        "Return the raw tool output (JSON list of products with similarity scores)."
    ),
    expected_output="JSON array of candidate products with similarity scores.",
    agent=retriever,
    context=[profile_task],
)

rank_task = Task(
    description=(
        "Rank the candidates using the user_context (the full profile JSON) "
        "and return the top 10 with a brief explanation for each.\n\n"
        "Output JSON with key 'recommendations' containing a list of objects: "
        "product_id, title, category, price, rating, score, explanation."
    ),
    expected_output="JSON object with 'recommendations' array.",
    agent=ranker,
    context=[profile_task, retrieve_task],
    output_json={
        "recommendations": "array"
    },
)

Step 5: Wire the crew and run

# main.py
import json
from crewai import Crew, Process
from agents import profiler, retriever, ranker
from tasks import profile_task, retrieve_task, rank_task

# Example user signals — replace with your real event data
USER_SIGNALS = {
    "recent_purchases": [
        {"product_id": "SKU-0042", "category": "electronics", "price": 149.99, "tags": ["wireless", "noise-cancelling"]},
        {"product_id": "SKU-0087", "category": "apparel", "price": 45.00, "tags": ["organic", "eco-friendly"]},
    ],
    "browsing_history": [
        {"category": "electronics", "duration_sec": 120, "query": "wireless headphones"},
        {"category": "sports", "duration_sec": 45, "query": "yoga mat"},
        {"category": "home", "duration_sec": 30, "query": "smart home hub"},
    ],
    "explicit_preferences": {
        "price_range": "mid",
        "favorite_categories": ["electronics", "home"],
        "disliked_tags": ["sale", "limited-edition"],
    },
    "demographics": {"age": 34, "location": "urban"},
}

def build_user_signals_string(signals: dict) -> str:
    return json.dumps(signals, indent=2)

if __name__ == "__main__":
    crew = Crew(
        agents=[profiler, retriever, ranker],
        tasks=[profile_task, retrieve_task, rank_task],
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff(inputs={"user_signals": build_user_signals_string(USER_SIGNALS)})
    
    # CrewAI returns a CrewOutput; the final task's output is in result.raw
    print("\n=== FINAL RECOMMENDATIONS ===\n")
    print(result.raw)

Run it:

python main.py

Expected output (truncated for readability):

=== FINAL RECOMMENDATIONS ===

{
  "recommendations": [
    {
      "product_id": "SKU-0031",
      "title": "Wireless Noise-Cancelling Headphones Pro",
      "category": "electronics",
      "price": 179.99,
      "rating": 4.6,
      "score": 0.847,
      "explanation": "Strong semantic match to wireless noise-cancelling preference; high rating and review count; mid-tier price aligns with profile."
    },
    {
      "product_id": "SKU-0112",
      "title": "Smart Home Hub with Voice Control",
      "category": "home",
      "price": 89.50,
      "rating": 4.4,
      "score": 0.812,
      "explanation": "Matches smart-home browsing intent; tag overlap with 'smart-home'; price tier matches mid sensitivity."
    },
    ...
  ]
}

Step 6: Add a simple evaluation harness

Before shipping, verify the pipeline behaves sensibly across diverse profiles. This script runs 10 synthetic users and checks latency, diversity, and price-tier alignment.

# evaluate.py
import json
import time
import numpy as np
from main import crew, build_user_signals_string

def synthetic_user(seed: int) -> dict:
    np.random.seed(seed)
    cats = np.random.choice(["electronics", "apparel", "home", "beauty", "sports"], size=3, replace=False)
    return {
        "recent_purchases": [
            {"category": cats[0], "price": float(np.random.uniform(20, 200)), "tags": ["bestseller"]}
        ],
        "browsing_history": [
            {"category": c, "duration_sec": int(np.random.uniform(20, 180)), "query": f"{c} products"}
            for c in cats
        ],
        "explicit_preferences": {
            "price_range": np.random.choice(["budget", "mid", "premium"]),
            "favorite_categories": cats.tolist(),
            "disliked_tags": [],
        },
        "demographics": {"age": int(np.random.uniform(18, 65)), "location": "urban"},
    }

if __name__ == "__main__":
    latencies = []
    tier_match = 0
    total = 10
    
    for i in range(total):
        signals = synthetic_user(i)
        start = time.time()
        result = crew.kickoff(inputs={"user_signals": build_user_signals_string(signals)})
        latencies.append(time.time() - start)
        
        # Parse and check price tier alignment
        try:
            recs = json.loads(result.raw)["recommendations"]
            pref_tier = signals["explicit_preferences"]["price_range"]
            matched = sum(1 for r in recs if r.get("price_tier", "") == pref_tier)
            tier_match += matched / len(recs)
        except Exception:
            pass
    
    print(f"Avg latency: {np.mean(latencies):.2f}s")
    print(f"Avg price-tier alignment: {tier_match/total:.2%}")

Sample run:

Avg latency: 3.42s
Avg price-tier alignment: 73%

Latency is dominated by the LLM calls (profiler + ranker). The embedding retrieval is ~50ms locally. In production you’d cache profiles, batch retrieval, and possibly swap the ranker for a lightweight model.

Extending for production

Three changes take this from tutorial to production-ready:

1. Persist user profiles — Store the profiler’s output in a vector DB (pgvector, Pinecone, Weaviate) keyed by user ID. Refresh nightly or on significant events. The retriever then becomes a pure ANN lookup.

2. Replace the heuristic ranker — Train a LambdaMART or neural ranker on implicit feedback (clicks, add-to-cart, purchase). Log the candidate set, final slate, and user interaction for each impression. Retrain weekly.

3. Add guardrails — Filter out-of-stock, enforce category diversity (max 2 per top-level category), respect regulatory constraints (age-gated products), and inject business rules (margin floors, supplier agreements) as hard constraints before the LLM ranker sees candidates.

The agent decomposition stays the same; you swap tools and add a post-rank filter step.

Why this structure works

The three-agent split mirrors how mature recommender systems are actually built: candidate generation → ranking → final merchandising. CrewAI makes the boundaries explicit and auditable. Each agent’s prompt, tools, and output schema are versionable. You can A/B test the profiler’s prompt independently of the ranker’s business rules. And because every step emits structured JSON, you get observability for free — log the intermediate outputs, trace latency per stage, and alert on schema drift.

If you’re routing this through a gateway that handles fallback and per-token metering, you can assign different models per agent: a cheap fast model for profiling, a larger model for ranking explanations, and keep embeddings local. The crew definition doesn’t change.