Ai Recommending

ai-recommending

Build an AI recommendation engine using DSPy. The core pattern is two-stage - embedding-based retrieval to get candidate items, then an LM re-ranker that personalizes the ranking using user profile signals and generates human-readable explanations.

Step 1 - Understand the recommendation task

Before writing code, clarify:

• What items are you recommending? Products, articles, support docs, videos, playlists?
• What signals do you have? Purchase history, click history, explicit ratings, topic tags, demographic signals?
• Cold-start scenario? New users with no history need a fallback strategy (popular items, content-based matching).
• How many results? Typically top-5 or top-10. More candidates are retrieved then re-ranked down.
• Latency budget? LM re-ranking adds ~500ms. If you need sub-100ms, do embedding-only retrieval.

Step 2 - Build candidate retrieval

Use embedding similarity to retrieve a broad candidate set before LM re-ranking.

import dspy
import numpy as np

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# Embed items and user profile using any embedding provider
# This example uses a local stub — swap in your embedding function
def embed(text: str) -> np.ndarray:
    """Replace with your embedding model call."""
    raise NotImplementedError("Plug in your embedding model here")

def retrieve_candidates(user_embedding: np.ndarray, item_embeddings: dict, top_n: int = 20) -> list[str]:
    """Return top-N item IDs by cosine similarity."""
    scores = {
        item_id: float(np.dot(user_embedding, item_emb) /
                       (np.linalg.norm(user_embedding) * np.linalg.norm(item_emb) + 1e-9))
        for item_id, item_emb in item_embeddings.items()
    }
    return sorted(scores, key=scores.get, reverse=True)[:top_n]

Step 3 - Build the LM re-ranker

Define a DSPy signature that takes a user profile and candidate items, then outputs ranked results with explanations.

class ReRankRecommendations(dspy.Signature):
    """Re-rank candidate items for a user based on their preferences and history.
    Return the top items as a ranked list with a short, friendly explanation for each.
    Do not reveal internal scoring. Explanations should reference why the item fits the user."""

    user_profile: str = dspy.InputField(
        desc="Anonymized user preference summary - interests, recent activity, preferred categories"
    )
    candidate_items: str = dspy.InputField(
        desc="Numbered list of candidate items with title and short description"
    )
    num_results: int = dspy.InputField(desc="Number of top items to return")
    ranked_items: list[str] = dspy.OutputField(
        desc="Item IDs in ranked order, most relevant first"
    )
    explanations: list[str] = dspy.OutputField(
        desc="One friendly sentence per item explaining why it was recommended"
    )

reranker = dspy.Predict(ReRankRecommendations)

Step 4 - Two-stage pipeline module

Combine retrieval and re-ranking into a single DSPy module.

class RecommendationPipeline(dspy.Module):
    def __init__(self, item_catalog: dict, item_embeddings: dict, num_candidates: int = 20):
        super().__init__()
        self.item_catalog = item_catalog      # {item_id: {"title": ..., "description": ...}}
        self.item_embeddings = item_embeddings  # {item_id: np.ndarray}
        self.num_candidates = num_candidates
        self.reranker = dspy.Predict(ReRankRecommendations)

    def forward(self, user_profile: str, user_embedding: np.ndarray, num_results: int = 5):
        # Stage 1 - retrieve candidates by embedding similarity
        candidate_ids = retrieve_candidates(user_embedding, self.item_embeddings, self.num_candidates)

        # Stage 2 - format candidates for LM re-ranking
        candidates_text = "\n".join(
            f"{i+1}. [{cid}] {self.item_catalog[cid]['title']} - {self.item_catalog[cid]['description']}"
            for i, cid in enumerate(candidate_ids)
            if cid in self.item_catalog
        )

        result = self.reranker(
            user_profile=user_profile,
            candidate_items=candidates_text,
            num_results=num_results,
        )

        # Pair ranked item IDs with their explanations
        recommendations = []
        for item_id, explanation in zip(result.ranked_items[:num_results], result.explanations[:num_results]):
            item_id = item_id.strip("[]").strip()
            if item_id in self.item_catalog:
                recommendations.append({
                    "item_id": item_id,
                    "title": self.item_catalog[item_id]["title"],
                    "explanation": explanation,
                })

        return dspy.Prediction(recommendations=recommendations)

Step 5 - Cold-start strategies

When a user has no history, fall back gracefully.

def build_user_profile(history: list[str], item_catalog: dict) -> str | None:
    """Build a text profile from user history. Returns None if history is empty."""
    if not history:
        return None
    titles = [item_catalog[iid]["title"] for iid in history if iid in item_catalog]
    return f"Previously engaged with - {', '.join(titles)}"

def get_popular_items(item_catalog: dict, popularity_scores: dict, top_n: int = 5) -> list[dict]:
    """Fallback - return most popular items when no user profile exists."""
    ranked = sorted(popularity_scores, key=popularity_scores.get, reverse=True)[:top_n]
    return [{"item_id": iid, "title": item_catalog[iid]["title"], "explanation": "Popular with other users"} for iid in ranked]

def recommend(pipeline, user_profile, user_embedding, item_catalog, popularity_scores, num_results=5):
    if user_profile is None or user_embedding is None:
        return get_popular_items(item_catalog, popularity_scores, num_results)
    result = pipeline(user_profile=user_profile, user_embedding=user_embedding, num_results=num_results)
    return result.recommendations

Step 6 - Evaluate recommendations

Use a DSPy judge to assess recommendation quality.

class RecommendationJudge(dspy.Signature):
    """Assess whether recommended items are relevant and well-explained for the given user profile."""
    user_profile: str = dspy.InputField()
    recommendations: str = dspy.InputField(desc="Recommended items with titles and explanations")
    relevance_score: float = dspy.OutputField(desc="0.0 to 1.0 - how well items match the profile")
    explanation_quality: float = dspy.OutputField(desc="0.0 to 1.0 - how friendly and helpful the explanations are")
    feedback: str = dspy.OutputField(desc="One sentence of actionable feedback")

judge = dspy.Predict(RecommendationJudge)

def evaluate_recommendations(user_profile: str, recommendations: list[dict]) -> dict:
    recs_text = "\n".join(f"- {r['title']}: {r['explanation']}" for r in recommendations)
    result = judge(user_profile=user_profile, recommendations=recs_text)
    return {
        "relevance": result.relevance_score,
        "explanation_quality": result.explanation_quality,
        "feedback": result.feedback,
    }

For offline evaluation, compute precision@k - the fraction of recommended items in the top-k that the user actually engaged with.

def precision_at_k(recommended_ids: list[str], relevant_ids: set[str], k: int) -> float:
    top_k = recommended_ids[:k]
    return len([iid for iid in top_k if iid in relevant_ids]) / k

Step 7 - Optimize with BootstrapFewShot

from dspy.teleprompt import BootstrapFewShot

def recommendation_metric(example, prediction, trace=None):
    """Reward when relevant items appear in top results."""
    relevant = set(example.relevant_item_ids)
    recommended = [r["item_id"] for r in prediction.recommendations]
    return precision_at_k(recommended, relevant, k=5)

trainset = [
    dspy.Example(
        user_profile="Interested in hiking and outdoor gear",
        user_embedding=np.zeros(768),  # placeholder - use real embeddings
        relevant_item_ids=["item_001", "item_004"],
    ).with_inputs("user_profile", "user_embedding"),
    # Add more labeled examples
]

optimizer = BootstrapFewShot(metric=recommendation_metric, max_bootstrapped_demos=4)
optimized_pipeline = optimizer.compile(
    RecommendationPipeline(item_catalog={}, item_embeddings={}),
    trainset=trainset,
)
optimized_pipeline.save("recommender_optimized.json")

Tradeoff table

When NOT to use LLM recommendations

• Large implicit signal datasets (millions of user-item interactions) - use collaborative filtering (ALS, BPR, matrix factorization). LMs cannot process interaction matrices.
• Simple popularity-based ranking - just sort by count. An LM adds latency with no benefit.
• Real-time with sub-10ms latency requirements - embedding similarity lookups only. LM calls cannot reliably meet this budget.
• Highly repetitive catalog updates (new items every second) - re-embedding is fast; re-prompting an LM per update is not.

Key patterns

• Always pass anonymized profile signals to the LM, not raw user data. Summarize history as interests and categories.
• Keep the candidate set to 20-50 items before LM re-ranking. Larger sets exceed context windows and reduce quality.
• Store item embeddings offline (precompute on catalog ingestion). Do not embed on every request.
• Use dspy.ChainOfThought instead of dspy.Predict for the re-ranker when explanation quality matters more than speed.
• For A/B testing, run both embedding-only and LM-reranked pipelines and measure click-through rate.

Gotchas

• Claude returns candidates unchanged - if the re-ranking prompt does not explicitly say "reorder this list", the model may echo items back in the original order. Add "Reorder the items below, placing the most relevant first" to the signature docstring.
• Claude builds the full pipeline in a single LM call - the model may try to both retrieve and rank in one prompt. Enforce the two-stage approach: retrieve candidates with embeddings first, then pass only the candidate subset to the LM.
• Claude uses `dspy.Assert` or `dspy.Suggest` for ranking constraints - assertion-based control flow does not work well for ordering tasks. Use dspy.Refine with a reward function that checks ranking quality instead.
• Claude includes user PII in the ranking prompt - pass anonymized profile signals (interest categories, behavioral patterns) rather than names, emails, or raw purchase records. Make this explicit in the signature field description.
• Claude generates explanations that leak ranking logic ("this item scored 0.87") - the signature docstring must say "Do not reveal internal scoring. Write friendly explanations referencing why the item fits the user."

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• `/ai-searching-docs` - embedding-based retrieval pattern; the retrieval stage in this skill follows the same approach
• `/dspy-retrieval` - DSPy retrieval module patterns for the candidate generation stage
• `/dspy-refine` - iterative refinement with feedback; use instead of assertions for re-ranking quality loops
• `/dspy-best-of-n` - sample N ranking outputs and pick the best; useful for high-stakes recommendation slots
• `/ai-scoring` - scoring and ranking individual items; composable with this pipeline
• `/ai-sorting` - LM-based sorting of a fixed list; simpler than full recommendation when you already have candidates
• `/ai-improving-accuracy` - optimize the re-ranker with BootstrapFewShot or MIPROv2 once you have labeled data
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

See examples.md for worked examples - product recommender, article recommender, and support article suggester.

ai-recommending - Examples

Example 1 - Product recommender for e-commerce

A user has a purchase history. Retrieve similar products from the catalog, then re-rank with the LM to surface the top 5 with friendly explanations.

import dspy
import numpy as np

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

# --- Item catalog ---
item_catalog = {
    "p001": {"title": "Trail Running Shoes", "description": "Lightweight shoes for off-road running"},
    "p002": {"title": "Waterproof Hiking Jacket", "description": "Breathable jacket for wet conditions"},
    "p003": {"title": "Trekking Poles", "description": "Collapsible carbon fibre poles"},
    "p004": {"title": "Running Socks 3-Pack", "description": "Moisture-wicking merino socks"},
    "p005": {"title": "Yoga Mat", "description": "Non-slip mat for studio or home use"},
    "p006": {"title": "Road Cycling Helmet", "description": "Aerodynamic helmet for road cyclists"},
}

# --- Stub embeddings (replace with your embedding model) ---
rng = np.random.default_rng(42)
item_embeddings = {pid: rng.random(64) for pid in item_catalog}

# User purchased trail shoes and hiking jacket — build profile
purchase_history = ["p001", "p002"]
history_titles = [item_catalog[pid]["title"] for pid in purchase_history]
user_profile = f"Recently purchased - {', '.join(history_titles)}. Interested in trail running and outdoor activities."

# Approximate user embedding as mean of purchased item embeddings
user_embedding = np.mean([item_embeddings[pid] for pid in purchase_history], axis=0)

# --- Re-ranker ---
class ReRankRecommendations(dspy.Signature):
    """Reorder the candidate items below, placing the most relevant first for this user.
    Do not reveal internal scoring. Write friendly explanations referencing why each item fits the user."""
    user_profile: str = dspy.InputField()
    candidate_items: str = dspy.InputField()
    num_results: int = dspy.InputField()
    ranked_items: list[str] = dspy.OutputField(desc="Item IDs in ranked order, most relevant first")
    explanations: list[str] = dspy.OutputField(desc="One friendly sentence per item")

reranker = dspy.Predict(ReRankRecommendations)

# --- Retrieve top-N candidates by cosine similarity ---
def cosine_sim(a, b):
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

candidate_ids = sorted(item_embeddings, key=lambda pid: cosine_sim(user_embedding, item_embeddings[pid]), reverse=True)[:6]
candidates_text = "\n".join(
    f"{i+1}. [{cid}] {item_catalog[cid]['title']} - {item_catalog[cid]['description']}"
    for i, cid in enumerate(candidate_ids)
)

result = reranker(user_profile=user_profile, candidate_items=candidates_text, num_results=5)

print("Top 5 recommendations:")
for item_id, explanation in zip(result.ranked_items[:5], result.explanations[:5]):
    item_id = item_id.strip("[]").strip()
    title = item_catalog.get(item_id, {}).get("title", item_id)
    print(f"  {title} - {explanation}")

Expected output (varies by LM):

Top 5 recommendations:
  Trekking Poles - Perfect for your trail running and hiking adventures.
  Running Socks 3-Pack - Great pairing with your trail running shoes.
  Waterproof Hiking Jacket - Complements your existing outdoor kit for wet days.
  Trail Running Shoes - A natural follow-up if you need a second pair or a size up.
  Road Cycling Helmet - Expands your outdoor activity range beyond trails.

---

Example 2 - Article recommender for a blog

A reader has viewed several articles. Recommend related articles from the content library based on their reading history.

import dspy
import numpy as np

lm = dspy.LM("anthropic/claude-sonnet-4-5-20250929")  # or "openai/gpt-4o-mini", etc.
dspy.configure(lm=lm)

article_catalog = {
    "a001": {"title": "Getting Started with DSPy", "description": "Intro to DSPy signatures and modules"},
    "a002": {"title": "Optimizing LM Pipelines", "description": "Using BootstrapFewShot and MIPROv2"},
    "a003": {"title": "Building RAG Systems", "description": "Retrieval-augmented generation patterns"},
    "a004": {"title": "DSPy Assertions Guide", "description": "Using Refine and BestOfN for quality control"},
    "a005": {"title": "Fine-tuning vs Prompting", "description": "When to fine-tune and when to prompt"},
    "a006": {"title": "Evaluating LLM Outputs", "description": "Metrics, judges, and eval frameworks"},
}

rng = np.random.default_rng(7)
article_embeddings = {aid: rng.random(64) for aid in article_catalog}

read_history = ["a001", "a003"]
history_titles = [article_catalog[aid]["title"] for aid in read_history]
user_profile = f"Read articles about - {', '.join(history_titles)}. Interested in practical DSPy usage and RAG systems."

user_embedding = np.mean([article_embeddings[aid] for aid in read_history], axis=0)

class ArticleRecommendation(dspy.Signature):
    """Reorder the candidate articles below, placing the most relevant first for this reader.
    Do not mention similarity scores. Explain why each article continues their learning journey."""
    user_profile: str = dspy.InputField()
    candidate_items: str = dspy.InputField()
    num_results: int = dspy.InputField()
    ranked_items: list[str] = dspy.OutputField(desc="Article IDs in ranked order")
    explanations: list[str] = dspy.OutputField(desc="One sentence per article - why it fits this reader")

reranker = dspy.Predict(ArticleRecommendation)

def cosine_sim(a, b):
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

# Exclude already-read articles from candidates
unread_ids = [aid for aid in article_catalog if aid not in read_history]
candidate_ids = sorted(unread_ids, key=lambda aid: cosine_sim(user_embedding, article_embeddings[aid]), reverse=True)[:5]
candidates_text = "\n".join(
    f"{i+1}. [{aid}] {article_catalog[aid]['title']} - {article_catalog[aid]['description']}"
    for i, aid in enumerate(candidate_ids)
)

result = reranker(user_profile=user_profile, candidate_items=candidates_text, num_results=3)

print("Recommended articles:")
for aid, explanation in zip(result.ranked_items[:3], result.explanations[:3]):
    aid = aid.strip("[]").strip()
    title = article_catalog.get(aid, {}).get("title", aid)
    print(f"  {title}")
    print(f"    {explanation}")

---

Example 3 - Support article suggester

A user submits a support ticket. Match the ticket text to relevant help docs before routing to a human agent, reducing ticket volume.

import dspy
import numpy as np

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

help_docs = {
    "h001": {"title": "How to reset your password", "description": "Step-by-step password reset instructions"},
    "h002": {"title": "Billing and invoice questions", "description": "How to download invoices and update payment methods"},
    "h003": {"title": "Cancelling your subscription", "description": "How to cancel, pause, or downgrade your plan"},
    "h004": {"title": "Two-factor authentication setup", "description": "Enable and manage 2FA on your account"},
    "h005": {"title": "Exporting your data", "description": "Download your account data in CSV or JSON format"},
    "h006": {"title": "Connecting third-party integrations", "description": "Set up Slack, Zapier, and other integrations"},
}

rng = np.random.default_rng(99)
doc_embeddings = {did: rng.random(64) for did in help_docs}

# Simulate ticket text as user profile signal
ticket_text = "I cannot log into my account. I forgot my password and the reset email is not arriving."
user_profile = f"Support ticket - {ticket_text}"

# In production, embed the ticket text with your embedding model
# Here we use a random vector as a placeholder
ticket_embedding = rng.random(64)

class SupportDocSuggestion(dspy.Signature):
    """Given a support ticket, reorder the candidate help articles below to surface the most useful ones first.
    Do not mention internal scores. Write a short sentence explaining how each article addresses the user's issue."""
    user_profile: str = dspy.InputField(desc="Support ticket text describing the user's problem")
    candidate_items: str = dspy.InputField(desc="Numbered list of help articles with descriptions")
    num_results: int = dspy.InputField()
    ranked_items: list[str] = dspy.OutputField(desc="Help doc IDs in ranked order, most relevant first")
    explanations: list[str] = dspy.OutputField(desc="One sentence per doc explaining how it helps")

reranker = dspy.Predict(SupportDocSuggestion)

def cosine_sim(a, b):
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

candidate_ids = sorted(doc_embeddings, key=lambda did: cosine_sim(ticket_embedding, doc_embeddings[did]), reverse=True)[:5]
candidates_text = "\n".join(
    f"{i+1}. [{did}] {help_docs[did]['title']} - {help_docs[did]['description']}"
    for i, did in enumerate(candidate_ids)
)

result = reranker(user_profile=user_profile, candidate_items=candidates_text, num_results=3)

print(f"Suggested articles for ticket: '{ticket_text[:60]}...'")
for did, explanation in zip(result.ranked_items[:3], result.explanations[:3]):
    did = did.strip("[]").strip()
    title = help_docs.get(did, {}).get("title", did)
    print(f"  [{did}] {title}")
    print(f"    {explanation}")

Expected output:

Suggested articles for ticket: 'I cannot log into my account. I forgot my password and...'
  [h001] How to reset your password
    Directly addresses your forgotten password and reset email issue.
  [h004] Two-factor authentication setup
    Relevant if 2FA is blocking login after the password reset.
  [h002] Billing and invoice questions
    Less likely to be relevant but included as a fallback.