AI Agent for Retail: Automate Inventory, Pricing & Customer Experience

Retail runs on razor-thin margins. A 2% shrinkage rate, a 5% overstock problem, or a 15-minute delay in repricing can erase an entire quarter's profit. AI agents are changing this by making thousands of micro-decisions per hour that no human team could handle — from adjusting prices in real-time to predicting which products will sell out next Tuesday.

This guide covers 6 production-ready AI agent workflows for retail, with architecture decisions, code examples, and real ROI numbers from retailers who've deployed them.

1. Intelligent Inventory Management

Traditional inventory management relies on static reorder points and safety stock formulas. An AI agent replaces this with continuous demand sensing that factors in weather, events, social media trends, and competitor stock levels.

The Architecture

An inventory agent operates as a closed-loop system:

1. Demand sensing — Ingests POS data, foot traffic, weather forecasts, local events calendar, and social signals every 15 minutes
2. Demand forecasting — Ensemble model (LightGBM + DeepAR + external features) predicts demand at SKU-store-day granularity
3. Replenishment optimization — Calculates optimal order quantities considering lead times, MOQs, shelf life, and warehouse capacity
4. Allocation — Distributes incoming inventory across stores based on predicted sell-through rates

Demand Sensing Implementation

import lightgbm as lgb
import pandas as pd
from datetime import datetime, timedelta

class RetailDemandAgent:
    """AI agent for SKU-level demand forecasting."""

    def __init__(self, model_path: str):
        self.model = lgb.Booster(model_file=model_path)
        self.feature_cols = [
            'day_of_week', 'month', 'is_weekend', 'is_holiday',
            'price', 'promo_active', 'promo_depth',
            'temp_avg', 'precip_mm', 'local_events_count',
            'lag_7d', 'lag_14d', 'lag_28d',
            'rolling_7d_mean', 'rolling_28d_mean',
            'competitor_price_ratio', 'social_trend_score'
        ]

    def forecast(self, sku_id: str, store_id: str,
                 horizon_days: int = 14) -> pd.DataFrame:
        """Generate daily demand forecast for a SKU at a store."""
        features = self._build_features(sku_id, store_id, horizon_days)
        predictions = self.model.predict(features[self.feature_cols])

        # Prediction intervals via quantile regression
        lower = self.model_q10.predict(features[self.feature_cols])
        upper = self.model_q90.predict(features[self.feature_cols])

        return pd.DataFrame({
            'date': features['date'],
            'forecast': predictions.round(0).astype(int),
            'lower_bound': lower.round(0).astype(int),
            'upper_bound': upper.round(0).astype(int)
        })

    def calculate_reorder(self, sku_id: str, store_id: str) -> dict:
        """Determine if and how much to reorder."""
        forecast = self.forecast(sku_id, store_id, horizon_days=14)
        current_stock = self._get_current_stock(sku_id, store_id)
        lead_time = self._get_lead_time(sku_id)

        # Demand during lead time + safety window
        demand_during_lead = forecast.loc[
            :lead_time + 2, 'upper_bound'
        ].sum()

        # Dynamic safety stock based on forecast uncertainty
        forecast_std = (
            forecast['upper_bound'] - forecast['lower_bound']
        ).mean() / 2
        safety_stock = int(forecast_std * 1.65)  # 95% service level

        reorder_point = demand_during_lead + safety_stock

        if current_stock <= reorder_point:
            # Order up to cover forecast + safety
            order_qty = max(
                forecast['upper_bound'].sum() - current_stock + safety_stock,
                self._get_moq(sku_id)
            )
            return {
                'action': 'reorder',
                'quantity': int(order_qty),
                'urgency': 'high' if current_stock < demand_during_lead * 0.5 else 'normal',
                'expected_stockout_date': self._estimate_stockout(
                    current_stock, forecast
                )
            }

        return {'action': 'hold', 'days_of_supply': int(
            current_stock / max(forecast['forecast'].mean(), 0.1)
        )}

Smart Allocation

When a shipment arrives at the distribution center, the agent allocates units across stores using a sell-through optimization approach:

def allocate_shipment(self, sku_id: str, available_qty: int,
                      store_ids: list) -> dict:
    """Allocate incoming inventory across stores optimally."""
    allocations = {}
    store_scores = []

    for store_id in store_ids:
        forecast = self.forecast(sku_id, store_id, horizon_days=7)
        current = self._get_current_stock(sku_id, store_id)
        expected_demand = forecast['forecast'].sum()
        days_of_supply = current / max(expected_demand / 7, 0.1)

        # Priority score: lower stock coverage = higher priority
        # Weighted by store revenue contribution
        revenue_weight = self._get_store_revenue_share(store_id)
        score = (1 / max(days_of_supply, 0.1)) * revenue_weight

        store_scores.append({
            'store_id': store_id,
            'score': score,
            'min_need': max(0, int(expected_demand - current)),
            'current_dos': days_of_supply
        })

    # Two-pass allocation
    # Pass 1: Fulfill minimum needs (prevent stockouts)
    remaining = available_qty
    store_scores.sort(key=lambda x: x['current_dos'])

    for store in store_scores:
        alloc = min(store['min_need'], remaining)
        allocations[store['store_id']] = alloc
        remaining -= alloc

    # Pass 2: Distribute surplus by score
    if remaining > 0:
        total_score = sum(s['score'] for s in store_scores)
        for store in store_scores:
            bonus = int(remaining * store['score'] / total_score)
            allocations[store['store_id']] += bonus

    return allocations

2. Dynamic Pricing Engine

Dynamic pricing in retail isn't just about matching competitors. It's about finding the price elasticity sweet spot for each product, in each store, at each moment. An AI pricing agent processes competitor prices, inventory levels, demand signals, and margin targets to make thousands of pricing decisions per hour.

Price Elasticity Model

import numpy as np
from scipy.optimize import minimize_scalar

class PricingAgent:
    """AI agent for dynamic retail pricing."""

    def __init__(self, elasticity_model, competitor_monitor):
        self.elasticity = elasticity_model
        self.competitors = competitor_monitor
        self.min_margin = 0.15  # 15% floor
        self.max_change_pct = 0.10  # Max 10% change per day

    def optimize_price(self, sku_id: str, store_id: str) -> dict:
        """Find revenue-maximizing price within constraints."""
        current_price = self._get_current_price(sku_id, store_id)
        cost = self._get_unit_cost(sku_id)
        comp_prices = self.competitors.get_prices(sku_id)

        # Estimate demand at different price points
        elasticity = self.elasticity.predict(sku_id, store_id)
        base_demand = self._get_base_demand(sku_id, store_id)

        def revenue(price):
            # Log-linear demand model
            demand = base_demand * (price / current_price) ** elasticity
            margin = price - cost
            # Negative because we're minimizing
            return -(demand * margin)

        # Constraints
        min_price = max(
            cost / (1 - self.min_margin),  # Margin floor
            current_price * (1 - self.max_change_pct),  # Rate limit
            comp_prices['min'] * 0.95 if comp_prices else 0  # Comp floor
        )
        max_price = min(
            current_price * (1 + self.max_change_pct),
            comp_prices['avg'] * 1.15 if comp_prices else float('inf')
        )

        result = minimize_scalar(revenue, bounds=(min_price, max_price),
                                 method='bounded')

        optimal_price = round(result.x, 2)
        # Apply psychological pricing
        optimal_price = self._apply_price_ending(optimal_price)

        return {
            'sku_id': sku_id,
            'current_price': current_price,
            'recommended_price': optimal_price,
            'expected_demand_change': f"{((optimal_price/current_price)**elasticity - 1)*100:.1f}%",
            'expected_margin_change': f"{((optimal_price - cost)/(current_price - cost) - 1)*100:.1f}%",
            'competitor_position': self._calc_position(optimal_price, comp_prices),
            'confidence': self._calc_confidence(sku_id, store_id)
        }

    def _apply_price_ending(self, price: float) -> float:
        """Apply psychological price endings (.99, .95, .49)."""
        if price < 10:
            return round(price - 0.01, 2)  # $X.99
        elif price < 50:
            return round(price) - 0.01  # $XX.99
        else:
            base = round(price)
            return base - 0.05 if base % 10 < 5 else base - 0.01

Markdown Optimization

One of the highest-ROI applications: optimizing markdowns for seasonal and perishable goods. Instead of blanket 30% → 50% → 70% off schedules, the agent finds the minimum discount needed to clear inventory by the target date:

def optimize_markdown(self, sku_id: str, target_clear_date: str,
                      current_inventory: int) -> list:
    """Generate optimal markdown schedule to clear inventory."""
    days_remaining = (pd.Timestamp(target_clear_date) - pd.Timestamp.now()).days
    current_price = self._get_current_price(sku_id)
    cost = self._get_unit_cost(sku_id)

    # Simulate different markdown paths
    best_path = None
    best_revenue = 0

    for initial_discount in np.arange(0.05, 0.50, 0.05):
        for acceleration in np.arange(1.1, 2.0, 0.1):
            path = self._simulate_markdown_path(
                current_price, initial_discount, acceleration,
                days_remaining, current_inventory, sku_id
            )
            total_revenue = sum(p['revenue'] for p in path)
            units_sold = sum(p['units'] for p in path)

            if units_sold >= current_inventory * 0.95 and total_revenue > best_revenue:
                best_revenue = total_revenue
                best_path = path

    return {
        'schedule': best_path,
        'total_revenue': best_revenue,
        'vs_standard': f"{(best_revenue / self._standard_markdown_revenue(sku_id, current_inventory) - 1)*100:.1f}%",
        'estimated_clearance_rate': '95%+'
    }

3. Personalized Shopping Agent

The personalization agent creates a unified customer profile that follows shoppers across channels — website, mobile app, email, and in-store — to deliver contextually relevant recommendations, offers, and experiences.

Real-Time Recommendation Engine

from typing import Optional
import numpy as np

class PersonalizationAgent:
    """AI agent for personalized retail experiences."""

    def __init__(self, embedding_model, product_catalog, behavior_store):
        self.embeddings = embedding_model
        self.catalog = product_catalog
        self.behaviors = behavior_store

    def get_recommendations(self, customer_id: str,
                           context: dict,
                           n: int = 12) -> list:
        """Generate personalized product recommendations."""
        profile = self._build_profile(customer_id)

        # Multi-signal scoring
        candidates = self._get_candidates(profile, context, n * 5)

        scored = []
        for product in candidates:
            score = (
                0.35 * self._collaborative_score(profile, product) +
                0.25 * self._content_score(profile, product) +
                0.20 * self._contextual_score(context, product) +
                0.10 * self._trending_score(product) +
                0.10 * self._margin_score(product)
            )

            # Diversity penalty (avoid showing too many similar items)
            if scored:
                max_similarity = max(
                    self._product_similarity(product, s['product'])
                    for s in scored[:5]
                )
                score *= (1 - 0.3 * max_similarity)

            scored.append({'product': product, 'score': score})

        scored.sort(key=lambda x: x['score'], reverse=True)
        return scored[:n]

    def _build_profile(self, customer_id: str) -> dict:
        """Build unified customer profile from all touchpoints."""
        behaviors = self.behaviors.get_recent(customer_id, days=90)

        # Purchase patterns
        purchases = [b for b in behaviors if b['type'] == 'purchase']
        categories = {}
        for p in purchases:
            cat = p['product']['category']
            categories[cat] = categories.get(cat, 0) + 1

        # Price sensitivity
        avg_price_paid = np.mean([p['price'] for p in purchases]) if purchases else 0
        avg_discount_used = np.mean([
            p.get('discount_pct', 0) for p in purchases
        ]) if purchases else 0

        # Browse-to-buy ratio (intent signal)
        views = len([b for b in behaviors if b['type'] == 'view'])
        buys = len(purchases)

        return {
            'customer_id': customer_id,
            'embedding': self._compute_preference_embedding(behaviors),
            'top_categories': sorted(categories.items(), key=lambda x: -x[1])[:5],
            'price_sensitivity': 'high' if avg_discount_used > 20 else 'medium' if avg_discount_used > 10 else 'low',
            'avg_order_value': avg_price_paid,
            'engagement_level': 'high' if views > 50 else 'medium' if views > 15 else 'low',
            'conversion_rate': buys / max(views, 1),
            'last_purchase_days': self._days_since_last_purchase(purchases),
            'lifecycle_stage': self._classify_lifecycle(purchases, behaviors)
        }

    def _contextual_score(self, context: dict, product: dict) -> float:
        """Score based on current context (time, weather, location)."""
        score = 0.5  # Base

        # Time-of-day patterns
        hour = context.get('hour', 12)
        if product['category'] == 'breakfast' and 6 <= hour <= 10:
            score += 0.3
        elif product['category'] == 'dinner' and 16 <= hour <= 20:
            score += 0.3

        # Weather-driven recommendations
        if context.get('weather') == 'rain' and product['category'] in ['umbrellas', 'rain_gear', 'soup']:
            score += 0.2

        # In-store location (if available)
        if context.get('aisle') and product.get('aisle') == context['aisle']:
            score += 0.15  # Cross-sell in same aisle

        return min(score, 1.0)

Triggered Communications

The agent doesn't just recommend — it decides when and how to reach out:

• Browse abandonment — If a customer views a product 3+ times without purchasing, send a personalized nudge (not a generic email blast)
• Replenishment reminders — For consumables, predict when the customer will run out based on purchase frequency and send a timely reminder
• Price drop alerts — Notify price-sensitive customers when items in their wishlist or browse history go on sale
• Cross-sell timing — After a furniture purchase, wait 2-3 weeks (delivery + settling), then suggest matching accessories

4. AI Loss Prevention

Retail shrinkage costs the industry $112 billion annually in the US alone (NRF 2025). AI agents can detect theft, fraud, and operational losses that traditional methods miss — from self-checkout fraud patterns to organized retail crime rings.

Multi-Signal Anomaly Detection

from dataclasses import dataclass
from enum import Enum

class AlertSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ShrinkageAlert:
    alert_type: str
    severity: AlertSeverity
    store_id: str
    details: dict
    recommended_action: str

class LossPreventionAgent:
    """AI agent for retail loss prevention."""

    def __init__(self, transaction_model, video_model, inventory_model):
        self.txn_model = transaction_model
        self.video_model = video_model
        self.inv_model = inventory_model

    def analyze_transaction(self, transaction: dict) -> Optional[ShrinkageAlert]:
        """Real-time transaction analysis for fraud indicators."""
        risk_score = 0
        flags = []

        # Sweet-hearting detection (employee gives discounts to friends)
        if transaction.get('discount_pct', 0) > 20:
            employee = transaction['cashier_id']
            recent_discounts = self._get_employee_discounts(employee, hours=8)
            if len(recent_discounts) > 5:
                risk_score += 30
                flags.append('excessive_discounts')

        # Self-checkout fraud patterns
        if transaction.get('checkout_type') == 'self':
            items_scanned = transaction['items_scanned']
            items_bagged = transaction.get('bagging_area_items', 0)
            time_per_item = transaction['duration'] / max(len(items_scanned), 1)

            # Pass-around detection (item not scanned)
            if items_bagged > items_scanned * 1.2:
                risk_score += 40
                flags.append('potential_pass_around')

            # Ticket switching (scanning cheaper barcode)
            for item in items_scanned:
                weight_expected = self._get_expected_weight(item['sku'])
                weight_actual = item.get('scale_weight', weight_expected)
                if weight_actual > weight_expected * 1.5:
                    risk_score += 35
                    flags.append(f'weight_mismatch_{item["sku"]}')

            # Unusually fast scanning
            if time_per_item < 2.0 and len(items_scanned) > 10:
                risk_score += 20
                flags.append('speed_anomaly')

        # High-value void patterns
        voids = [i for i in transaction.get('voids', []) if i['price'] > 50]
        if len(voids) > 2:
            risk_score += 25
            flags.append('multiple_high_value_voids')

        if risk_score >= 50:
            severity = (AlertSeverity.CRITICAL if risk_score >= 80
                       else AlertSeverity.HIGH if risk_score >= 60
                       else AlertSeverity.MEDIUM)
            return ShrinkageAlert(
                alert_type='transaction_fraud',
                severity=severity,
                store_id=transaction['store_id'],
                details={'flags': flags, 'risk_score': risk_score,
                         'transaction_id': transaction['id']},
                recommended_action=self._get_recommended_action(flags, severity)
            )
        return None

    def detect_organized_retail_crime(self, store_id: str,
                                      window_days: int = 30) -> list:
        """Identify patterns suggesting organized retail crime."""
        anomalies = []

        # Spike detection by category
        category_sales = self._get_category_shrinkage(store_id, window_days)
        for category, data in category_sales.items():
            if data['shrinkage_rate'] > data['historical_avg'] * 2:
                anomalies.append({
                    'pattern': 'category_spike',
                    'category': category,
                    'current_rate': data['shrinkage_rate'],
                    'normal_rate': data['historical_avg'],
                    'estimated_loss': data['loss_amount']
                })

        # Geographic clustering (multiple nearby stores hit)
        nearby_stores = self._get_nearby_stores(store_id, radius_km=30)
        hit_stores = [s for s in nearby_stores
                     if self._recent_shrinkage_spike(s['id'])]
        if len(hit_stores) >= 3:
            anomalies.append({
                'pattern': 'geographic_cluster',
                'stores_affected': [s['id'] for s in hit_stores],
                'likely_orc': True,
                'recommended': 'Coordinate with law enforcement'
            })

        return anomalies

5. Workforce Scheduling Optimization

Labor is the largest controllable expense in retail (typically 10-15% of revenue). The scheduling agent optimizes coverage vs. cost by predicting customer traffic patterns and matching them with employee availability, skills, and labor regulations.

Traffic-Based Scheduling

from datetime import datetime, timedelta
from ortools.sat.python import cp_model

class WorkforceAgent:
    """AI agent for retail workforce scheduling."""

    def __init__(self, traffic_model, employee_db):
        self.traffic = traffic_model
        self.employees = employee_db

    def generate_schedule(self, store_id: str, week_start: str) -> dict:
        """Generate optimal weekly schedule for a store."""
        # Step 1: Predict hourly traffic for each day
        traffic_forecast = {}
        for day_offset in range(7):
            date = pd.Timestamp(week_start) + timedelta(days=day_offset)
            traffic_forecast[date.strftime('%Y-%m-%d')] = (
                self.traffic.predict_hourly(store_id, date)
            )

        # Step 2: Convert traffic to staffing requirements
        staff_needs = self._traffic_to_staffing(traffic_forecast, store_id)

        # Step 3: Solve constraint satisfaction problem
        employees = self.employees.get_available(store_id, week_start)
        model = cp_model.CpModel()

        # Decision variables: employee x day x shift
        shifts = {}
        for emp in employees:
            for day in range(7):
                for shift_type in ['morning', 'afternoon', 'evening', 'closing']:
                    var_name = f'{emp["id"]}_{day}_{shift_type}'
                    shifts[var_name] = model.NewBoolVar(var_name)

        # Constraints
        for emp in employees:
            # Max hours per week
            total_hours = sum(
                shifts[f'{emp["id"]}_{d}_{s}'] * self._shift_hours(s)
                for d in range(7)
                for s in ['morning', 'afternoon', 'evening', 'closing']
            )
            model.Add(total_hours <= emp['max_weekly_hours'])
            model.Add(total_hours >= emp.get('min_weekly_hours', 0))

            # No double shifts
            for day in range(7):
                model.Add(sum(
                    shifts[f'{emp["id"]}_{day}_{s}']
                    for s in ['morning', 'afternoon', 'evening', 'closing']
                ) <= 1)

            # Availability preferences
            for day, unavailable in emp.get('unavailable', {}).items():
                for shift in unavailable:
                    model.Add(shifts[f'{emp["id"]}_{day}_{shift}'] == 0)

            # Rest between shifts (11 hours minimum)
            for day in range(6):
                model.Add(
                    shifts[f'{emp["id"]}_{day}_closing'] +
                    shifts[f'{emp["id"]}_{day+1}_morning'] <= 1
                )

        # Coverage requirements
        for day in range(7):
            for shift_type in ['morning', 'afternoon', 'evening', 'closing']:
                needed = staff_needs[day][shift_type]
                assigned = sum(
                    shifts[f'{emp["id"]}_{day}_{shift_type}']
                    for emp in employees
                )
                model.Add(assigned >= needed)

        # Objective: minimize total labor cost while meeting coverage
        cost = sum(
            shifts[f'{emp["id"]}_{d}_{s}'] * emp['hourly_rate'] * self._shift_hours(s)
            for emp in employees
            for d in range(7)
            for s in ['morning', 'afternoon', 'evening', 'closing']
        )
        model.Minimize(cost)

        solver = cp_model.CpSolver()
        status = solver.Solve(model)

        if status == cp_model.OPTIMAL:
            return self._extract_schedule(solver, shifts, employees)
        return {'error': 'No feasible schedule found', 'status': str(status)}

The key differentiator: the traffic model doesn't just use historical averages. It incorporates leading indicators — weather forecasts, local events, marketing campaigns, nearby competitor promotions — that traditional scheduling tools miss.

6. Omnichannel Orchestration

Modern retail customers expect seamless experiences across online, mobile, and physical stores. The omnichannel agent manages order routing, inventory visibility, and fulfillment optimization across all channels.

Order Routing Optimization

When an online order comes in, the agent decides where to fulfill it from — warehouse, nearest store, or a store with excess inventory:

class OmnichannelAgent:
    """AI agent for cross-channel retail orchestration."""

    def route_order(self, order: dict) -> dict:
        """Find optimal fulfillment location for an order."""
        customer_location = order['shipping_address']
        items = order['items']

        # Get all fulfillment options
        options = []

        # Check warehouses
        for wh in self._get_warehouses():
            availability = self._check_availability(wh['id'], items)
            if availability['can_fulfill']:
                options.append({
                    'location': wh,
                    'type': 'warehouse',
                    'ship_cost': self._calc_shipping(wh, customer_location),
                    'delivery_days': self._estimate_delivery(wh, customer_location),
                    'pick_cost': wh['pick_cost_per_item'] * len(items),
                    'availability': availability
                })

        # Check stores (ship-from-store)
        nearby_stores = self._get_stores_near(customer_location, radius_km=50)
        for store in nearby_stores:
            availability = self._check_availability(store['id'], items)
            if availability['can_fulfill']:
                # Factor in opportunity cost of selling in-store
                opportunity_cost = sum(
                    self._store_demand_forecast(store['id'], item['sku']) * item['margin']
                    for item in items
                ) * 0.1  # 10% probability of lost in-store sale

                options.append({
                    'location': store,
                    'type': 'store',
                    'ship_cost': self._calc_shipping(store, customer_location),
                    'delivery_days': self._estimate_delivery(store, customer_location),
                    'pick_cost': store['pick_cost_per_item'] * len(items) * 1.5,  # Store picking is less efficient
                    'opportunity_cost': opportunity_cost,
                    'availability': availability
                })

        # BOPIS option (Buy Online, Pick Up In Store)
        for store in nearby_stores[:3]:
            availability = self._check_availability(store['id'], items)
            if availability['can_fulfill']:
                options.append({
                    'location': store,
                    'type': 'bopis',
                    'ship_cost': 0,
                    'delivery_days': 0,
                    'pick_cost': store['pick_cost_per_item'] * len(items),
                    'upsell_opportunity': self._estimate_bopis_upsell(store['id']),
                    'availability': availability
                })

        # Score and rank options
        for opt in options:
            total_cost = opt['ship_cost'] + opt['pick_cost'] + opt.get('opportunity_cost', 0)
            speed_score = max(0, 5 - opt['delivery_days']) / 5
            cost_score = 1 - (total_cost / max(o['ship_cost'] + o['pick_cost'] for o in options))
            upsell_score = opt.get('upsell_opportunity', 0) / 100

            opt['total_score'] = (
                0.40 * cost_score +
                0.35 * speed_score +
                0.15 * (1 if opt['type'] == 'bopis' else 0) +
                0.10 * upsell_score
            )

        best = max(options, key=lambda x: x['total_score'])
        return {
            'fulfillment_location': best['location']['id'],
            'fulfillment_type': best['type'],
            'estimated_delivery': best['delivery_days'],
            'total_fulfillment_cost': best['ship_cost'] + best['pick_cost'],
            'alternatives': sorted(options, key=lambda x: -x['total_score'])[:3]
        }

ROI Calculator

Here's what a mid-size retailer ($200M annual revenue, 50 stores) can expect:

Platform Comparison

Common Mistakes

• Over-optimizing for one metric — A pricing agent that maximizes margin can destroy customer loyalty. Always include customer lifetime value in the objective function.
• Ignoring cannibalization — Promoting Product A might steal sales from Product B with higher margins. The agent needs to model cross-product effects.
• Training on biased data — If historical data reflects stockouts (zero sales when product was unavailable), the model will underestimate demand. Use censored demand modeling.
• Forgetting the human in the loop — Store managers have local context that no model captures. Give them override capability and feed their overrides back into model training.
• Launching everything at once — Start with inventory optimization (easiest to measure, hardest to mess up), then add pricing, then personalization. Each builds on the data foundation of the previous.

Getting Started: Weekend MVP

You can build a useful retail AI agent in a weekend. Start with the highest-impact, lowest-risk workflow:

1. Day 1 morning — Export 12 months of POS data. Clean it (handle returns, zero-quantity days, holidays). Feature engineer: day-of-week, month, promo flags, lag features.
2. Day 1 afternoon — Train a LightGBM demand forecast model on your top 100 SKUs. Evaluate on the last 30 days. You'll likely get 15-25% MAPE, which is already better than most planners' gut feel.
3. Day 2 morning — Build the reorder logic. For each SKU, compare current stock to forecasted demand during lead time + safety stock. Generate a daily replenishment report.
4. Day 2 afternoon — Set up automated alerts: stockout warnings (3 days out), overstock alerts (60+ days of supply), and demand anomalies (actual > 2x forecast). Email to the buying team.

This MVP alone can reduce stockouts by 15-20% and overstock by 10-15% within the first month. Then iterate: add weather features, competitor monitoring, and markdown optimization.