What Are Seasonal AI Agent Businesses?
Seasonal AI agent businesses are purpose-built artificial intelligence services that operate during predictable peak-demand windows throughout the year. Unlike always-on SaaS products, these agents spin up to handle intense, time-bound workloads — tax filing season from January to April, holiday retail surges from November through December, back-to-school shopping in August, or summer camp registration in spring. The agents combine large language models with domain-specific knowledge bases, structured workflows, and often human-in-the-loop oversight to deliver services that would otherwise require seasonal staffing explosions.
Think of a tax preparation AI agent that interviews clients, extracts relevant data from uploaded W-2s and 1099s, cross-references the latest IRS publications via retrieval-augmented generation (RAG), and produces a complete filing package — all while maintaining compliance with IRS circular 230 guidelines when supervised by a licensed preparer. Or a holiday retail agent that handles gift recommendations across thousands of SKUs, manages inventory-aware suggestions, and resolves pre-shipment customer service inquiries at 3 AM when human agents are asleep.
The core insight: seasonal businesses have historically faced a brutal hiring-firing cycle. AI agents invert this by letting you scale software — not people — to meet the curve. You invest in building the agent once, then rent compute during peak season at a fraction of the cost of seasonal staff.
Why Seasonal AI Agents Matter
Market Opportunity
The numbers are staggering. The US tax preparation industry alone generates over $12 billion annually, concentrated in a 14-week window. Holiday retail e-commerce exceeds $200 billion in Q4, with customer service ticket volumes tripling between Black Friday and Christmas. Each spike creates a temporary labor market crunch where businesses pay premium wages for mediocre service. An AI agent that handles even 30% of this overflow at 90% accuracy captures enormous value while dramatically reducing marginal cost.
Beyond the obvious economics, seasonal AI agents solve a structural problem: knowledge workers don't want 3-month contracts. Certified tax preparers, retail merchandisers, and admissions counselors invest years building expertise — they seek full-time roles. AI agents let you bottle that expertise permanently, updating it annually as regulations or catalogs change, without relying on transient human availability.
The Technical Leverage Point
What makes seasonal agents uniquely suited to current AI capabilities is their bounded scope. A tax agent needs to master a finite set of forms, deductions, and rules that change only once per year. A holiday retail agent operates on a fixed product catalog with predictable query patterns (gift suitability, shipping deadlines, return policies). This boundedness means you can build highly reliable agents using today's LLMs without solving artificial general intelligence. The seasonality also provides natural retraining windows — you update your knowledge base in October for holiday retail, or in December for the upcoming tax season.
Core Architecture for Seasonal AI Agents
Most seasonal AI agents follow a layered architecture that separates knowledge retrieval, reasoning, and action execution. Here is a production-ready scaffold that works across tax, retail, and other seasonal domains:
# seasonal_agent_core.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import json
import asyncio
@dataclass
class SeasonalAgentConfig:
"""Configuration for a seasonal AI agent deployment."""
domain: str # e.g., "tax_prep", "holiday_retail"
season_start: datetime # First day of operational window
season_end: datetime # Last day of operational window
knowledge_base_version: str # e.g., "irs_2025_v2", "holiday_catalog_2024"
llm_model: str = "gpt-4o"
embedding_model: str = "text-embedding-3-large"
max_concurrent_sessions: int = 50
human_escalation_threshold: float = 0.85 # Confidence below this → escalate
@dataclass
class RetrievalContext:
"""Documents retrieved from the seasonal knowledge base."""
chunks: List[str]
metadata: Dict[str, str]
retrieval_score: float
class SeasonalKnowledgeBase:
"""Manages the vector store and document ingestion pipeline."""
def __init__(self, config: SeasonalAgentConfig):
self.config = config
self.vector_store = None # Would initialize Pinecone, Weaviate, etc.
self.document_index = {}
async def ingest_documents(self, documents: List[Dict]) -> None:
"""Ingest and chunk seasonal reference documents."""
for doc in documents:
chunks = self._chunk_document(doc)
embeddings = await self._embed_chunks(chunks)
await self._store_in_vector_db(chunks, embeddings, doc.get('metadata', {}))
async def retrieve(self, query: str, top_k: int = 5) -> List[RetrievalContext]:
"""Retrieve relevant seasonal knowledge for a query."""
query_embedding = await self._embed_query(query)
results = await self._similarity_search(query_embedding, top_k)
return [
RetrievalContext(
chunks=[r['text']],
metadata=r['metadata'],
retrieval_score=r['score']
)
for r in results
]
def _chunk_document(self, doc: Dict) -> List[str]:
text = doc.get('content', '')
# Chunk with overlap for context preservation
chunk_size = 800
overlap = 100
chunks = []
for i in range(0, len(text), chunk_size - overlap):
chunks.append(text[i:i + chunk_size])
return chunks
async def _embed_chunks(self, chunks: List[str]) -> List[List[float]]:
# Calls embedding API — OpenAI, Cohere, etc.
# Placeholder: would use openai.Embedding.create()
return [[0.0] * 1536 for _ in chunks] # Replace with actual embeddings
async def _embed_query(self, query: str) -> List[float]:
return [0.0] * 1536 # Replace with actual embedding
async def _similarity_search(self, embedding, top_k):
return [{'text': 'Sample result', 'score': 0.95, 'metadata': {}}]
async def _store_in_vector_db(self, chunks, embeddings, metadata):
pass
This scaffold separates concerns: the knowledge base handles document lifecycle, while the agent logic (which we'll build next) focuses on conversation management and action execution. The configuration object centralizes seasonal parameters so you can swap between tax season and holiday season without architectural changes.
Building a Tax Preparation AI Agent
Knowledge Base Setup with RAG
A tax preparation agent must ground its responses in authoritative IRS publications, state tax codes, and your firm's prior-year client patterns. Here's how to build the ingestion pipeline for annual tax document updates:
# tax_knowledge_ingestion.py
import asyncio
from seasonal_agent_core import SeasonalKnowledgeBase, SeasonalAgentConfig
from datetime import datetime
async def build_tax_knowledge_base():
config = SeasonalAgentConfig(
domain="tax_prep",
season_start=datetime(2025, 1, 15),
season_end=datetime(2025, 4, 15),
knowledge_base_version="irs_2025_v3"
)
kb = SeasonalKnowledgeBase(config)
# IRS publications to ingest
tax_documents = [
{
"content": open("irs_pub_17_2025.txt").read(),
"metadata": {
"source": "irs_pub_17",
"section": "filing_requirements",
"last_updated": "2025-01-02"
}
},
{
"content": open("irs_pub_503_2025.txt").read(),
"metadata": {
"source": "irs_pub_503",
"section": "child_dependent_care",
"last_updated": "2025-01-05"
}
},
{
"content": open("form_1040_instructions_2025.txt").read(),
"metadata": {
"source": "form_1040_instructions",
"section": "full",
"last_updated": "2024-12-20"
}
},
# State-specific tax codes
{
"content": open("ca_ftb_2025.txt").read(),
"metadata": {
"source": "california_ftb",
"section": "state_residency_rules",
"last_updated": "2025-01-10"
}
}
]
# Ingest all documents
await kb.ingest_documents(tax_documents)
print(f"Ingested {len(tax_documents)} tax documents for season {config.knowledge_base_version}")
return kb
# Run ingestion
asyncio.run(build_tax_knowledge_base())
Client Interview Agent
The heart of tax prep is the structured client interview — gathering dependents, income sources, deductions, and life changes. This agent orchestrates a multi-turn conversation while maintaining state across sessions:
# tax_interview_agent.py
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
import json
import asyncio
class InterviewStage(Enum):
GREETING = "greeting"
PERSONAL_INFO = "personal_info"
DEPENDENTS = "dependents"
INCOME_W2 = "income_w2"
INCOME_1099 = "income_1099"
INCOME_OTHER = "income_other"
DEDUCTIONS = "deductions"
CREDITS = "credits"
REVIEW = "review"
COMPLETE = "complete"
@dataclass
class ClientTaxProfile:
"""Accumulates client data during the interview."""
session_id: str
first_name: str = ""
last_name: str = ""
ssn_last4: str = ""
filing_status: str = ""
dependents: List[Dict] = field(default_factory=list)
w2_income: List[Dict] = field(default_factory=list)
1099_income: List[Dict] = field(default_factory=list)
other_income: List[Dict] = field(default_factory=list)
deductions: Dict[str, float] = field(default_factory=dict)
credits: List[str] = field(default_factory=list)
current_stage: InterviewStage = InterviewStage.GREETING
completed_stages: List[InterviewStage] = field(default_factory=list)
class TaxInterviewAgent:
"""Orchestrates the client tax interview with LLM-powered guidance."""
def __init__(self, kb: SeasonalKnowledgeBase, llm_client):
self.kb = kb
self.llm = llm_client
self.stage_prompts = self._build_stage_prompts()
def _build_stage_prompts(self) -> Dict[InterviewStage, str]:
return {
InterviewStage.GREETING: """You are a friendly tax preparation assistant.
Greet the client warmly, explain the interview process (it takes about
15-20 minutes), and ask for their first and last name to begin.
Do NOT ask for full SSN — only last 4 digits when needed for verification.""",
InterviewStage.PERSONAL_INFO: """Collect filing status (Single, Married Filing Jointly,
Married Filing Separately, Head of Household, Qualifying Widow(er)),
and verify the client's address on file. Ask about any life changes
in the past year: marriage, divorce, new child, moved states.""",
InterviewStage.DEPENDENTS: """Ask about dependents: children, elderly parents,
or others supported. For each dependent, collect: name, relationship,
months lived with client, and whether they provided more than 50% of support.
Reference IRS Publication 501 dependency tests via RAG as needed.""",
InterviewStage.INCOME_W2: """Collect all W-2 income. For each employer, ask for:
employer name, wages from Box 1, federal tax withheld from Box 2,
and state tax withheld. Ask if client has the physical W-2 or digital copy.
Offer to parse uploaded W-2 images if available.""",
InterviewStage.INCOME_1099: """Collect all 1099 income: 1099-NEC, 1099-MISC,
1099-INT, 1099-DIV, 1099-G, 1099-R. For each form, collect payer name
and amount. Ask about business expenses related to 1099-NEC income.""",
InterviewStage.DEDUCTIONS: """Discuss deductions. Check if client itemizes or
takes standard deduction. Gather: mortgage interest (Form 1098),
charitable contributions, medical expenses, state/local taxes paid.
Use RAG to check IRS Pub 17 for current deduction limits.""",
InterviewStage.CREDITS: """Screen for tax credits: Child Tax Credit,
Earned Income Tax Credit, American Opportunity Credit, Lifetime Learning Credit,
Child and Dependent Care Credit. Use IRS guidelines to determine eligibility
thresholds based on income and dependents already collected.""",
InterviewStage.REVIEW: """Summarize all collected information in a structured
format. Present the summary to the client and ask them to verify accuracy.
Highlight any potential issues: missing forms, income discrepancies,
deduction opportunities they may have missed."""
}
async def process_response(self, profile: ClientTaxProfile,
user_message: str) -> Dict[str, Any]:
"""Process a user message and determine next stage transition."""
# Retrieve relevant tax knowledge for this stage
retrieval_query = f"{profile.current_stage.value} {user_message}"
knowledge = await self.kb.retrieve(retrieval_query, top_k=3)
# Build context-aware prompt
context = f"""
Current interview stage: {profile.current_stage.value}
Client profile so far: {json.dumps(profile.__dict__, default=str, indent=2)}
Retrieved IRS guidance: {json.dumps([k.chunks for k in knowledge])}
Client message: {user_message}
Task: Extract relevant tax information from the client's message.
Update the profile fields appropriately. Determine if the current stage
is complete enough to move to the next stage. Respond helpfully.
"""
response = await self.llm.chat_completion(
system_prompt=self.stage_prompts[profile.current_stage],
user_message=context,
temperature=0.3 # Lower temperature for factual extraction
)
# Parse structured extraction from LLM response
extracted = self._parse_extraction(response)
# Update profile with extracted data
profile = self._update_profile(profile, extracted)
# Determine stage transition
next_stage = self._determine_stage_transition(profile, extracted)
return {
"agent_response": extracted.get("natural_response", response),
"updated_profile": profile,
"next_stage": next_stage,
"extracted_data": extracted
}
def _parse_extraction(self, llm_response: str) -> Dict:
"""Parse the structured extraction from the LLM response."""
# In production, use function calling or structured output parsing
try:
# Look for JSON block in response
if "json" in llm_response:
json_start = llm_response.find("json") + 7
json_end = llm_response.find("", json_start)
return json.loads(llm_response[json_start:json_end])
return {"natural_response": llm_response}
except json.JSONDecodeError:
return {"natural_response": llm_response}
def _update_profile(self, profile: ClientTaxProfile,
extracted: Dict) -> ClientTaxProfile:
"""Apply extracted data to the profile."""
if "first_name" in extracted:
profile.first_name = extracted["first_name"]
if "filing_status" in extracted:
profile.filing_status = extracted["filing_status"]
if "dependents" in extracted:
profile.dependents.extend(extracted["dependents"])
if "w2_entries" in extracted:
profile.w2_income.extend(extracted["w2_entries"])
if "deductions" in extracted:
profile.deductions.update(extracted["deductions"])
return profile
def _determine_stage_transition(self, profile: ClientTaxProfile,
extracted: Dict) -> InterviewStage:
"""Logic for advancing through interview stages."""
stage_order = list(InterviewStage)
current_idx = stage_order.index(profile.current_stage)
# Check if current stage is complete
if extracted.get("stage_complete", False):
next_idx = current_idx + 1
if next_idx < len(stage_order):
return stage_order[next_idx]
return profile.current_stage
# Usage example
async def run_interview():
profile = ClientTaxProfile(session_id="client_12345")
agent = TaxInterviewAgent(kb=None, llm_client=None) # Inject real instances
# Simulate conversation loop
responses = [
"Hi, I'm Jane Smith.",
"I'm married filing jointly, no life changes this year.",
"We have two children, ages 8 and 12, both lived with us all year.",
"I have one W-2 from Acme Corp, wages $85,000, federal withheld $12,400.",
"My husband has a W-2 from Beta Inc, wages $72,000, federal withheld $10,200.",
"We have mortgage interest of $14,300 from our home.",
"We paid $8,500 in state and local taxes.",
"Yes, everything looks correct in the summary."
]
for msg in responses:
result = await agent.process_response(profile, msg)
print(f"Agent: {result['agent_response'][:200]}...")
profile = result['updated_profile']
if result['next_stage'] == InterviewStage.COMPLETE:
print("Interview complete! Ready for return preparation.")
break
asyncio.run(run_interview())
Tax Form Generation Agent
Once the interview is complete, the agent generates populated tax forms. This requires precise mapping of profile data to IRS form fields with validation rules:
# tax_form_generator.py
from typing import Dict, List, Tuple
import math
class Form1040Generator:
"""Generates populated Form 1040 from ClientTaxProfile."""
def __init__(self, tax_year: int = 2024):
self.tax_year = tax_year
self.standard_deductions = {
"single": 14600,
"married_filing_jointly": 29200,
"married_filing_separately": 14600,
"head_of_household": 21900
}
self.tax_brackets = self._load_tax_brackets()
def _load_tax_brackets(self) -> List[Dict]:
"""Load current year tax brackets. Updated annually."""
return [
{"rate": 0.10, "min": 0, "max": 23200},
{"rate": 0.12, "min": 23201, "max": 94300},
{"rate": 0.22, "min": 94301, "max": 201050},
{"rate": 0.24, "min": 201051, "max": 383900},
{"rate": 0.32, "min": 383901, "max": 487450},
{"rate": 0.35, "min": 487451, "max": 731200},
{"rate": 0.37, "min": 731201, "max": float('inf')}
]
def calculate_agi(self, profile) -> float:
"""Calculate Adjusted Gross Income."""
total_w2 = sum(w2['wages'] for w2 in profile.w2_income)
total_1099 = sum(form['amount'] for form in profile.1099_income
if form.get('form_type') == '1099-NEC')
other_income = sum(item['amount'] for item in profile.other_income)
# Subtract deductible part of self-employment tax for 1099-NEC
se_deduction = total_1099 * 0.9235 * 0.153 * 0.5 if total_1099 > 0 else 0
return total_w2 + total_1099 + other_income - se_deduction
def calculate_taxable_income(self, agi: float, profile) -> float:
"""Determine taxable income after deductions."""
standard_deduction = self.standard_deductions.get(
profile.filing_status.lower().replace(" ", "_"), 14600
)
itemized_total = sum([
profile.deductions.get('mortgage_interest', 0),
profile.deductions.get('charitable', 0),
profile.deductions.get('medical', 0),
profile.deductions.get('salt', 0) # State and local taxes capped at $10K
])
deduction = max(standard_deduction, itemized_total)
taxable = max(0, agi - deduction)
return taxable
def calculate_tax(self, taxable_income: float) -> float:
"""Compute tax using progressive brackets."""
total_tax = 0.0
remaining = taxable_income
for bracket in self.tax_brackets:
bracket_range = bracket['max'] - bracket['min']
if remaining <= 0:
break
taxable_in_bracket = min(remaining, bracket_range)
total_tax += taxable_in_bracket * bracket['rate']
remaining -= taxable_in_bracket
return round(total_tax, 2)
def generate_form_1040(self, profile) -> Dict[str, Any]:
"""Generate a complete Form 1040 data structure."""
agi = self.calculate_agi(profile)
taxable = self.calculate_taxable_income(agi, profile)
total_tax = self.calculate_tax(taxable)
# Calculate credits
credits = self._calculate_credits(profile, agi)
total_credits = sum(credits.values())
tax_after_credits = max(0, total_tax - total_credits)
# Calculate payments
federal_withheld = sum(w2.get('federal_withheld', 0)
for w2 in profile.w2_income)
return {
"tax_year": self.tax_year,
"filing_status": profile.filing_status,
"line_1_wages": sum(w2['wages'] for w2 in profile.w2_income),
"line_2_interest_dividends": sum(
form['amount'] for form in profile.1099_income
if form.get('form_type') in ('1099-INT', '1099-DIV')
),
"line_8_agi": round(agi, 2),
"line_12_standard_or_itemized": max(
self.standard_deductions.get(profile.filing_status.lower(), 14600),
sum(profile.deductions.values())
),
"line_15_taxable_income": round(taxable, 2),
"line_16_tax": total_tax,
"line_18_sch2_total": 0, # Additional taxes from Schedule 2
"line_21_total_credits": total_credits,
"line_22_tax_after_credits": tax_after_credits,
"line_25_federal_withheld": federal_withheld,
"line_33_total_payments": federal_withheld,
"line_34_overpaid": max(0, federal_withheld - tax_after_credits),
"line_37_amount_owed": max(0, tax_after_credits - federal_withheld),
"generated_at": str(datetime.now()),
"requires_human_review": agi > 200000 or len(profile.1099_income) > 3
}
def _calculate_credits(self, profile, agi: float) -> Dict[str, float]:
credits = {}
# Child Tax Credit
qualifying_children = sum(
1 for dep in profile.dependents
if dep.get('age', 99) < 17 and dep.get('months_lived', 12) >= 6
)
if qualifying_children > 0 and agi < 400000:
credits['child_tax_credit'] = qualifying_children * 2000
# Earned Income Tax Credit (simplified)
if agi < 60000 and qualifying_children > 0:
credits['eitc'] = min(qualifying_children * 3500, 7430)
return credits
# Usage
from datetime import datetime
generator = Form1040Generator(tax_year=2024)
form_data = generator.generate_form_1040(profile)
print(json.dumps(form_data, indent=2, default=str))
Building a Holiday Retail AI Agent
Gift Recommendation Engine
Holiday retail agents face different challenges: they must understand product catalogs, interpret vague customer preferences ("something for my dad who likes golf but already has everything"), and respect inventory constraints. Here's a recommendation agent that combines semantic search with constraint-based filtering:
# holiday_retail_agent.py
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
import numpy as np
@dataclass
class Product:
sku: str
name: str
category: str
price: float
inventory_count: int
description: str
tags: List[str]
avg_rating: float
shipping_deadline_days: int # Days needed to ship before Christmas
age_range: Tuple[int, int] # Suitable age range
gender_target: str # "unisex", "male_leaning", "female_leaning"
class HolidayGiftRecommender:
"""Multi-strategy gift recommendation engine for holiday retail."""
def __init__(self, product_catalog: List[Product], llm_client, embedding_client):
self.catalog = product_catalog
self.llm = llm_client
self.embeddings = embedding_client
self.catalog_embeddings = self._precompute_embeddings()
def _precompute_embeddings(self) -> Dict[str, List[float]]:
"""Embed all product descriptions for semantic search."""
embeddings = {}
for product in self.catalog:
text = f"{product.name} {product.description} {' '.join(product.tags)}"
embeddings[product.sku] = self.embeddings.embed(text)
return embeddings
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
a_array = np.array(a)
b_array = np.array(b)
return np.dot(a_array, b_array) / (np.linalg.norm(a_array) * np.linalg.norm(b_array))
async def recommend(self, customer_query: str,
constraints: Dict = None) -> List[Dict]:
"""Generate personalized gift recommendations."""
# Step 1: Parse customer intent
intent = await self._parse_gift_intent(customer_query)
# Step 2: Semantic search over catalog
query_embedding = self.embeddings.embed(customer_query)
similarities = [
(sku, self._cosine_similarity(query_embedding, emb))
for sku, emb in self.catalog_embeddings.items()
]
ranked_skus = sorted(similarities, key=lambda x: x[1], reverse=True)
# Step 3: Apply hard constraints
candidates = []
for sku, sim_score in ranked_skus[:50]:
product = next(p for p in self.catalog if p.sku == sku)
# Inventory check
if product.inventory_count < 1:
continue
# Budget constraint
if constraints and 'max_price' in constraints:
if product.price > constraints['max_price']:
continue
# Shipping deadline
if constraints and 'days_until_christmas' in constraints:
if product.shipping_deadline_days > constraints['days_until_christmas']:
continue
# Age appropriateness
if intent.get('recipient_age'):
age = intent['recipient_age']
if age < product.age_range[0] or age > product.age_range[1]:
continue
candidates.append({
'product': product,
'similarity_score': sim_score,
'reason': ''
})
# Step 4: Generate natural language explanations
top_candidates = candidates[:5]
for candidate in top_candidates:
explanation = await self._generate_explanation(
customer_query=customer_query,
product=candidate['product'],
intent=intent
)
candidate['reason'] = explanation
return top_candidates
async def _parse_gift_intent(self, query: str) -> Dict:
"""Extract structured intent from natural language query."""
prompt = f"""
Parse this gift shopping query into structured intent:
Query: "{query}"
Extract: recipient_relationship, recipient_age (estimate if unclear),
recipient_interests (list), budget_range, gift_category_preference,
occasion (christmas, birthday, anniversary), desired_sentiment.
Return valid JSON only.
"""
response = await self.llm.chat_completion(prompt, temperature=0.2)
try:
return json.loads(response)
except:
return {"recipient_interests": [query], "occasion": "christmas"}
async def _generate_explanation(self, customer_query: str,
product: Product, intent: Dict) -> str:
prompt = f"""
Explain why this product makes a great gift for the customer's needs.
Customer query: "{customer_query}"
Product: {product.name} - {product.description}
Price: ${product.price}
Rating: {product.avg_rating}/5
Write a warm, personal 2-3 sentence explanation that connects the
product features to the recipient's interests and the holiday spirit.
"""
return await self.llm.chat_completion(prompt, temperature=0.7)
async def handle_inventory_question(self, product_sku: str) -> Dict:
"""Answer inventory and shipping questions."""
product = next((p for p in self.catalog if p.sku == product_sku), None)
if not product:
return {"status": "not_found", "message": "Product not found in catalog"}
return {
"sku": product.sku,
"name": product.name,
"in_stock": product.inventory_count > 0,
"available_count": product.inventory_count,
"ships_by": f"{product.shipping_deadline_days} days",
"guaranteed_christmas": product.shipping_deadline_days <= 5,
"price": product.price
}
# Example catalog and usage
sample_catalog = [
Product(
sku="GFT-001", name="Premium Golf Swing Analyzer",
category="Electronics", price=129.99, inventory_count=47,
description="Attaches to any club, provides real-time swing metrics via Bluetooth.",
tags=["golf", "tech", "sports", "training"],
avg_rating=4.3, shipping_deadline_days=3,
age_range=(12, 99), gender_target="unisex"
),
Product(
sku="GFT-002", name="Artisan Hot Sauce Collection",
category="Food", price=39.99, inventory_count=0,
description="Set of 6 globally-sourced small-batch hot sauces in gift box.",
tags=["food", "gourmet", "spicy", "gift_set"],
avg_rating=4.7, shipping_deadline_days=2,
age_range=(18, 99), gender_target="unisex"
),
Product(
sku="GFT-003", name="Cozy Weighted Blanket - 15lb",
category="Home", price=79.99, inventory_count=153,
description="Premium cotton weighted blanket with removable duvet cover.",
tags=["home", "wellness", "cozy", "sleep"],
avg_rating=4.5, shipping_deadline_days=4,
age_range=(8, 99), gender_target="unisex"
)
]
async def demo_recommendation():
recommender = HolidayGiftRecommender(
product_catalog=sample_catalog,
llm_client=None, # Inject real client
embedding_client=None # Inject real client
)
results = await recommender.recommend(
"I need a gift for my father-in-law who loves golf but is impossible to shop for. Budget around $150.",
constraints={"max_price": 150, "days_until_christmas": 10}
)
for r in results:
print(f"Recommended: {r['product'].name} - {r['reason']}")
asyncio.run(demo_recommendation())
Multi-Channel Deployment with Scaling
Seasonal retail agents must handle spikes across web chat, messaging platforms, and voice. Here's a deployment pattern using asynchronous queues and auto-scaling workers:
# retail_agent_deployment.py
import asyncio