Spaces:
Sleeping
Sleeping
File size: 8,057 Bytes
bab4aa8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | import os
import json
import statistics
from datetime import datetime
from typing import List, Dict, Any
from groq import Groq
from dotenv import load_dotenv
from app.services.supabase import SupabaseQuerier
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
dotenv_path = os.path.join(base_dir, ".env")
load_dotenv(dotenv_path=dotenv_path, override=True)
class MissingRegularsAgent:
def __init__(self):
api_key = os.getenv("GROQ_API_KEY")
if api_key:
api_key = api_key.replace("your_groq_api_key_here", "").strip()
print(f"[MissingRegularsAgent] GROQ API Key: {api_key}")
if not api_key:
print("[WARNING] GROQ_API_KEY not detected, using mock key")
api_key = "gsk_mock_key_placeholder"
self.client = Groq(api_key=api_key)
self.model = os.getenv("GROQ_MODEL", "llama-3.1-8b-instant")
self.fallback_model = os.getenv("GROQ_FALLBACK_MODEL", "llama-3.3-70b-versatile")
self.supabase = SupabaseQuerier()
def _analyze_regularity(self, orders: List[Dict[str, Any]], current_cart_skus: List[str]) -> List[Dict[str, Any]]:
"""
Analyzes order history to find regular items missing from the current cart.
Applies a 3-layer filter: Frequency, Consistency (CV), and Timing (Due Date).
"""
if not orders:
return []
# 1. Extract dates and group by SKU
sku_history = {}
sku_metadata = {}
for order in orders:
try:
# The items column might be a JSON string or already parsed list
items_raw = order.get("items", [])
items = json.loads(items_raw) if isinstance(items_raw, str) else items_raw
# Parse created_at
created_at_str = order.get("created_at")
if not created_at_str:
continue
# Simple parsing assuming ISO format
created_at = datetime.fromisoformat(created_at_str.replace("Z", "+00:00")).date()
for item in items:
# Extract SKU from details if id is not the SKU
details = item.get("details", "")
sku = item.get("id")
if "SKU: " in details:
sku = details.split("SKU: ")[1].split(" ")[0]
if not sku:
continue
if sku not in sku_history:
sku_history[sku] = set()
sku_metadata[sku] = {
"sku": sku,
"name": item.get("name", "Unknown Item"),
"price": item.get("price", 0),
"imageUrl": item.get("imageUrl", "")
}
sku_history[sku].add(created_at)
except Exception as e:
print(f"[MissingRegularsAgent] Error parsing order: {e}")
continue
today = datetime.utcnow().date()
missing_regulars = []
# 2. Apply regularity logic
for sku, dates_set in sku_history.items():
dates = sorted(list(dates_set))
# Layer 1: Frequency Check (Must be bought on at least 3 distinct days)
if len(dates) < 3:
continue
# Calculate gaps between purchases
gaps = [(dates[i] - dates[i-1]).days for i in range(1, len(dates))]
if not gaps:
continue
avg_gap = statistics.mean(gaps)
# Layer 2: Consistency Check (CV <= 0.6)
if len(gaps) > 1:
std_dev = statistics.stdev(gaps)
cv = std_dev / avg_gap if avg_gap > 0 else 0
if cv > 0.6:
continue # Too erratic, not a consistent regular item
# Layer 3: Timing / Due Date Check
days_since_last = (today - dates[-1]).days
# If they bought it very recently, they don't need it yet.
# If days_since_last is close to or greater than avg_gap, they are due.
# We add a small buffer (e.g., -2 days) so we remind them slightly before they completely run out.
if days_since_last < (avg_gap - 2):
continue
# Final Filter: Is it already in the cart?
if sku in current_cart_skus:
continue
# Passed all layers! It's a missing regular.
item_data = sku_metadata[sku]
item_data["frequency"] = len(dates)
item_data["avg_gap_days"] = round(avg_gap, 1)
item_data["last_bought_days_ago"] = days_since_last
missing_regulars.append(item_data)
# Sort by most frequently bought
missing_regulars.sort(key=lambda x: x["frequency"], reverse=True)
return missing_regulars
async def analyze_cart(self, user_id: str, current_cart: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Main entry point for the route.
Fetches history, finds missing regulars, and gets the LLM to write a friendly reminder.
"""
# Fetch history (last 90 days)
orders = await self.supabase.get_order_history(user_id=user_id, days=90)
# Extract current SKUs
current_skus = [item.get("sku") or item.get("id") for item in current_cart]
# Find missing regulars via Python logic
missing_items = self._analyze_regularity(orders, current_skus)
if not missing_items:
return {
"response_text": "",
"missing_regulars": []
}
# Format items for the LLM prompt
cart_str = ", ".join([item.get("name", "") for item in current_cart]) if current_cart else "Empty cart"
missing_str = ", ".join([item["name"] for item in missing_items])
# LLM writes the phrasing
prompt = f"""You are a helpful, friendly AI shopping assistant.
The user is currently reviewing their shopping cart.
Current cart contains: {cart_str}
Based on our deterministic analysis of their past 90 days of orders, they regularly buy these items every few weeks, but forgot to add them today:
Missing Regulars: {missing_str}
Write a very brief, friendly 1-2 sentence reminder suggesting they might want to add these to their cart before checking out.
Do not mention the "90 days" or the algorithm. Just be natural and helpful, like "I noticed you usually grab..." or "Don't forget your usual...".
Respond ONLY with the message text. No JSON, no extra formatting."""
try:
try:
completion = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
max_tokens=150,
temperature=0.4
)
except Exception as inner_e:
if self.fallback_model:
print(f"[MissingRegularsAgent] Main model {self.model} failed: {inner_e}. Falling back to {self.fallback_model}...")
completion = self.client.chat.completions.create(
model=self.fallback_model,
messages=[{"role": "user", "content": prompt}],
max_tokens=150,
temperature=0.4
)
else:
raise inner_e
response_text = completion.choices[0].message.content.strip()
except Exception as e:
print(f"[MissingRegularsAgent] LLM Generation failed completely: {e}")
response_text = "It looks like you might have forgotten a few of your regular items. Would you like to add them?"
return {
"response_text": response_text,
"missing_regulars": missing_items
}
|