AI Agent for Subscription Box Services: Preference Collection, Box Curation, and Feedback

The Subscription Box Model

Subscription boxes deliver curated products on a recurring basis — beauty products, snacks, books, pet supplies, or clothing. The key challenge is curation: each box must feel personalized, avoid repeats, incorporate feedback, and surprise the customer positively. An AI agent manages this entire lifecycle from preference collection through curation to feedback processing.

Preference Profiling

The first interaction with a subscriber should build a detailed preference profile. This goes beyond simple category selection — it captures intensity, allergies, experience level, and variety tolerance.

flowchart LR
    CALLER(["Shopper"])
    subgraph TEL["Telephony"]
        SIP["Twilio SIP and PSTN"]
    end
    subgraph BRAIN["E-commerce AI Agent"]
        STT["Streaming STT<br/>Deepgram or Whisper"]
        NLU{"Intent and<br/>Entity Extraction"}
        TOOLS["Tool Calls"]
        TTS["Streaming TTS<br/>ElevenLabs or Rime"]
    end
    subgraph DATA["Live Data Plane"]
        CRM[("CRM and Notes")]
        CAL[("Calendar and<br/>Schedule")]
        KB[("Knowledge Base<br/>and Policies")]
    end
    subgraph OUT["Outcomes"]
        O1(["Order status answered"])
        O2(["Return RMA created"])
        O3(["Specialist handoff"])
    end
    CALLER --> SIP --> STT --> NLU
    NLU -->|Lookup| TOOLS
    TOOLS <--> CRM
    TOOLS <--> CAL
    TOOLS <--> KB
    NLU --> TTS --> SIP --> CALLER
    NLU -->|Resolved| O1
    NLU -->|Schedule| O2
    NLU -->|Escalate| O3
    style CALLER fill:#f1f5f9,stroke:#64748b,color:#0f172a
    style NLU fill:#4f46e5,stroke:#4338ca,color:#fff
    style O1 fill:#059669,stroke:#047857,color:#fff
    style O2 fill:#0ea5e9,stroke:#0369a1,color:#fff
    style O3 fill:#f59e0b,stroke:#d97706,color:#1f2937

from agents import Agent, Runner, function_tool
from typing import Optional
from datetime import datetime
import random

SUBSCRIBER_PROFILES = {}

@function_tool
def create_preference_profile(subscriber_id: str,
                               box_type: str,
                               preferences: str,
                               allergies: str = "",
                               experience_level: str = "beginner",
                               variety_tolerance: str = "moderate") -> str:
    """Create a detailed preference profile for a new subscriber."""
    profile = {
        "box_type": box_type,
        "preferences": [p.strip() for p in preferences.split(",")],
        "allergies": [a.strip() for a in allergies.split(",") if a.strip()],
        "experience_level": experience_level,
        "variety_tolerance": variety_tolerance,  # low, moderate, high
        "past_boxes": [],
        "item_ratings": {},
        "satisfaction_scores": [],
        "subscription_start": "2026-03-17",
        "boxes_received": 0,
        "skip_next": False,
    }
    SUBSCRIBER_PROFILES[subscriber_id] = profile
    return (
        f"Profile created for {subscriber_id}:\n"
        f"  Box type: {box_type}\n"
        f"  Preferences: {', '.join(profile['preferences'])}\n"
        f"  Allergies/Exclusions: {', '.join(profile['allergies']) or 'None'}\n"
        f"  Experience: {experience_level}\n"
        f"  Variety tolerance: {variety_tolerance}"
    )

@function_tool
def update_preferences(subscriber_id: str,
                       field: str, value: str) -> str:
    """Update a specific preference field for a subscriber."""
    profile = SUBSCRIBER_PROFILES.get(subscriber_id)
    if not profile:
        return "Subscriber not found."

    if field == "preferences":
        profile["preferences"] = [p.strip() for p in value.split(",")]
    elif field == "allergies":
        profile["allergies"] = [a.strip() for a in value.split(",")]
    elif field == "experience_level":
        profile["experience_level"] = value
    elif field == "variety_tolerance":
        profile["variety_tolerance"] = value
    else:
        return f"Unknown field: {field}"

    return f"Updated {field} to: {value}"

Item Catalog and Curation Engine

The curation engine selects items that match preferences, avoid known dislikes and allergens, and introduce appropriate variety.

# Simulated item catalog for a gourmet snack box
ITEM_CATALOG = [
    {"id": "ITM-001", "name": "Dark Chocolate Truffle Bar",
     "category": "chocolate", "tags": ["sweet", "premium"],
     "allergens": ["dairy", "soy"], "experience": "any"},
    {"id": "ITM-002", "name": "Spicy Sriracha Cashews",
     "category": "nuts", "tags": ["spicy", "savory", "protein"],
     "allergens": ["tree_nuts"], "experience": "intermediate"},
    {"id": "ITM-003", "name": "Organic Dried Mango Slices",
     "category": "dried_fruit", "tags": ["sweet", "healthy", "tropical"],
     "allergens": [], "experience": "any"},
    {"id": "ITM-004", "name": "Artisan Sourdough Crackers",
     "category": "crackers", "tags": ["savory", "artisan"],
     "allergens": ["gluten"], "experience": "any"},
    {"id": "ITM-005", "name": "Ghost Pepper Beef Jerky",
     "category": "jerky", "tags": ["spicy", "protein", "bold"],
     "allergens": [], "experience": "advanced"},
    {"id": "ITM-006", "name": "Lavender Honey Caramels",
     "category": "candy", "tags": ["sweet", "floral", "unique"],
     "allergens": ["dairy"], "experience": "any"},
    {"id": "ITM-007", "name": "Wasabi Pea Crunch Mix",
     "category": "snack_mix", "tags": ["spicy", "crunchy"],
     "allergens": ["soy"], "experience": "intermediate"},
    {"id": "ITM-008", "name": "Cold Brew Coffee Granola",
     "category": "granola", "tags": ["coffee", "sweet", "crunchy"],
     "allergens": ["gluten", "tree_nuts"], "experience": "any"},
]

@function_tool
def curate_box(subscriber_id: str, items_count: int = 5) -> str:
    """Curate a personalized box for a subscriber."""
    profile = SUBSCRIBER_PROFILES.get(subscriber_id)
    if not profile:
        return "Subscriber not found."

    # Get previously sent item IDs to avoid repeats
    sent_items = set()
    for box in profile["past_boxes"]:
        for item_id in box["items"]:
            sent_items.add(item_id)

    # Filter eligible items
    eligible = []
    for item in ITEM_CATALOG:
        # Skip already sent
        if item["id"] in sent_items:
            continue

        # Allergen check
        if any(a in item["allergens"] for a in profile["allergies"]):
            continue

        # Experience level filter
        exp_order = {"beginner": 0, "intermediate": 1, "advanced": 2}
        item_exp = exp_order.get(item["experience"], 0)
        sub_exp = exp_order.get(profile["experience_level"], 0)
        if item["experience"] != "any" and item_exp > sub_exp:
            continue

        # Score based on preference match
        score = 0
        for pref in profile["preferences"]:
            if pref.lower() in [t.lower() for t in item["tags"]]:
                score += 2
            if pref.lower() in item["category"].lower():
                score += 3

        # Check past ratings for category
        for rated_id, rating in profile["item_ratings"].items():
            rated_item = next(
                (i for i in ITEM_CATALOG if i["id"] == rated_id), None
            )
            if rated_item and rated_item["category"] == item["category"]:
                if rating >= 4:
                    score += 2
                elif rating <= 2:
                    score -= 3

        # Variety bonus
        if profile["variety_tolerance"] == "high":
            score += 1  # Slight boost for diversity
        eligible.append({"item": item, "score": score})

    eligible.sort(key=lambda x: x["score"], reverse=True)
    selected = eligible[:items_count]

    if len(selected) < items_count:
        return (
            f"Only {len(selected)} eligible items found. "
            f"Consider expanding the catalog or relaxing preferences."
        )

    box_id = f"BOX-{len(profile['past_boxes']) + 1:03d}"
    box_record = {
        "box_id": box_id,
        "items": [s["item"]["id"] for s in selected],
        "curated_date": datetime.now().isoformat(),
        "shipped": False,
        "feedback_received": False,
    }
    profile["past_boxes"].append(box_record)
    profile["boxes_received"] += 1

    lines = [f"Curated {box_id} for {subscriber_id}:"]
    for s in selected:
        item = s["item"]
        lines.append(
            f"  - {item['name']} ({item['category']}) "
            f"[score: {s['score']}]"
        )
    return "\n".join(lines)

Feedback Processing

After each box, collect item-level ratings and free-text feedback. Use this to refine future curation.

@function_tool
def submit_box_feedback(subscriber_id: str, box_id: str,
                        ratings: str,
                        overall_satisfaction: int,
                        comments: str = "") -> str:
    """Submit feedback for a received box. Ratings format: ITM-001:5,ITM-002:3"""
    profile = SUBSCRIBER_PROFILES.get(subscriber_id)
    if not profile:
        return "Subscriber not found."

    box = next(
        (b for b in profile["past_boxes"] if b["box_id"] == box_id), None
    )
    if not box:
        return f"Box {box_id} not found."

    # Parse and store individual ratings
    for rating_pair in ratings.split(","):
        parts = rating_pair.strip().split(":")
        if len(parts) == 2:
            item_id = parts[0].strip()
            score = int(parts[1].strip())
            profile["item_ratings"][item_id] = score

    profile["satisfaction_scores"].append(overall_satisfaction)
    box["feedback_received"] = True

    avg_satisfaction = (
        sum(profile["satisfaction_scores"])
        / len(profile["satisfaction_scores"])
    )

    return (
        f"Feedback recorded for {box_id}. "
        f"Overall satisfaction: {overall_satisfaction}/5. "
        f"Running average: {avg_satisfaction:.1f}/5. "
        f"Individual item ratings saved and will influence future boxes."
        f"{f' Comments noted: {comments}' if comments else ''}"
    )

Churn Prevention

Monitor subscriber engagement signals and flag at-risk accounts before they cancel.

@function_tool
def assess_churn_risk(subscriber_id: str) -> str:
    """Assess the churn risk for a subscriber based on engagement signals."""
    profile = SUBSCRIBER_PROFILES.get(subscriber_id)
    if not profile:
        return "Subscriber not found."

    risk_score = 0
    reasons = []

    # Low satisfaction trend
    scores = profile["satisfaction_scores"]
    if len(scores) >= 2:
        recent_avg = sum(scores[-2:]) / 2
        if recent_avg < 3.0:
            risk_score += 3
            reasons.append(
                f"Recent satisfaction declining ({recent_avg:.1f}/5)"
            )

    # Many low-rated items
    low_ratings = sum(1 for r in profile["item_ratings"].values() if r <= 2)
    if low_ratings >= 3:
        risk_score += 2
        reasons.append(f"{low_ratings} items rated 2 or below")

    # Skipped boxes
    if profile.get("skip_next"):
        risk_score += 2
        reasons.append("Has requested to skip next box")

    # No feedback submitted for recent box
    recent_boxes = profile["past_boxes"][-2:]
    unfeedback = sum(1 for b in recent_boxes if not b["feedback_received"])
    if unfeedback > 0:
        risk_score += 1
        reasons.append(f"{unfeedback} recent boxes without feedback")

    if risk_score >= 4:
        risk_level = "HIGH"
        action = (
            "Recommend: Send personalized retention offer "
            "(free upgrade or discount on next box)"
        )
    elif risk_score >= 2:
        risk_level = "MEDIUM"
        action = (
            "Recommend: Reach out to collect preferences update "
            "and address concerns"
        )
    else:
        risk_level = "LOW"
        action = "No immediate action needed"

    result = f"Churn risk for {subscriber_id}: {risk_level} (score: {risk_score})"
    if reasons:
        result += "\nSignals:\n" + "\n".join(f"  - {r}" for r in reasons)
    result += f"\n{action}"
    return result

Assembling the Subscription Box Agent

subscription_agent = Agent(
    name="Subscription Box Curator",
    instructions="""You manage a gourmet snack subscription box service.

    New subscribers:
    - Collect detailed preferences (sweet/savory/spicy, dietary restrictions)
    - Ask about experience level and variety tolerance
    - Create a preference profile

    Ongoing management:
    - Curate boxes that match preferences and avoid allergens
    - Never repeat items from previous boxes
    - Process feedback and incorporate it into future curation
    - Monitor satisfaction trends and flag churn risks
    - Handle skip requests and subscription modifications

    When curating, explain why each item was selected. If a subscriber
    gives low ratings, acknowledge it and adjust future selections.
    Proactively check churn risk for subscribers with declining
    satisfaction.""",
    tools=[create_preference_profile, update_preferences,
           curate_box, submit_box_feedback, assess_churn_risk],
)

result = Runner.run_sync(
    subscription_agent,
    "I just signed up for the snack box. I love spicy and savory snacks "
    "but I am allergic to tree nuts. I would say I am an intermediate "
    "snacker who likes variety.",
)
print(result.final_output)

FAQ

How do I prevent item fatigue in long-running subscriptions?

Track the complete history of items sent to each subscriber. Maintain a "cooldown" period — if you sent an item from a specific category in the last two boxes, deprioritize that category. For catalogs with limited items, partner with new vendors regularly to refresh the available pool. Consider introducing "throwback" items after a 6-month gap with a note like "back by popular demand" to reuse highly rated items.

What is the best way to handle dietary restriction changes mid-subscription?

Build the preference update into the agent flow so it takes effect immediately on the next box. When a subscriber reports a new allergy or dietary restriction, retroactively check the next queued box (if already curated but not shipped) and swap out any conflicting items. Send a confirmation that the change has been applied. Maintain an audit log of preference changes for food safety compliance.

How do I measure the effectiveness of the curation algorithm?

Track three core metrics: average box satisfaction score (target above 4.0 out of 5), item-level rating distribution (percentage rated 4 or higher), and churn rate by cohort month. Compare these against a control group receiving randomly curated boxes. A good curation algorithm should achieve at least a 15 to 20 percent improvement in satisfaction and a measurable reduction in monthly churn rate over the random baseline.

---

#SubscriptionBox #PreferenceEngine #CurationAI #ChurnPrevention #ECommerce #AgenticAI #LearnAI #AIEngineering