Inventory Management Agent

Stockouts cost retailers $1 trillion annually in lost sales. Meanwhile, overstocking ties up capital and leads to markdowns. An intelligent inventory agent can monitor stock levels, predict demand, and coordinate reorders autonomously.

The Inventory Challenge

Inventory Agent Architecture

         ┌─────────────────┐
         │  Scheduled      │
         │  Trigger        │ ─── Runs every hour / on demand
         └────────┬────────┘
                  │
                  ▼
         ┌─────────────────┐
         │   Inventory     │
         │   Scanner       │ ─── Check all SKUs against thresholds
         └────────┬────────┘
                  │
         ┌────────┴────────┐
         ▼                 ▼
    [Low Stock]       [Healthy]
         │                 │
         ▼                 ▼
┌─────────────────┐   ┌─────────────┐
│  Demand         │   │  Continue   │
│  Analyzer       │   │  Monitoring │
└────────┬────────┘   └─────────────┘
         │
         ▼
┌─────────────────┐
│  Reorder        │
│  Calculator     │ ─── Optimal quantity, considering lead time
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Supplier       │
│  Coordinator    │ ─── Check prices, availability, place order
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Alert &        │
│  Report         │ ─── Notify stakeholders, log decisions
└─────────────────┘

Step 1: Define the State

from typing import TypedDict, Annotated, Optional, Literal
from datetime import datetime
import operator

class InventoryItem(TypedDict):
    sku: str
    name: str
    current_stock: int
    reorder_point: int  # Trigger reorder when stock falls below
    safety_stock: int   # Minimum to keep on hand
    lead_time_days: int # Days to receive new stock
    unit_cost: float
    supplier_id: str

class SupplierQuote(TypedDict):
    supplier_id: str
    supplier_name: str
    unit_price: float
    available_quantity: int
    lead_time_days: int
    minimum_order: int

class InventoryAgentState(TypedDict):
    # Trigger info
    trigger_type: Literal["scheduled", "manual", "alert"]
    run_timestamp: str

    # Items to process
    all_items: list[InventoryItem]
    low_stock_items: list[InventoryItem]
    critical_items: list[InventoryItem]  # Below safety stock

    # Analysis
    demand_forecasts: dict  # sku -> predicted daily demand
    reorder_recommendations: list[dict]

    # Supplier interaction
    supplier_quotes: list[SupplierQuote]
    selected_orders: list[dict]

    # Outputs
    alerts: Annotated[list[dict], operator.add]
    actions_taken: Annotated[list[dict], operator.add]
    report: str

Step 2: Implement the Agent Nodes

Inventory Scanner

# Mock inventory database
INVENTORY_DB = {
    "SKU-001": {
        "sku": "SKU-001",
        "name": "Wireless Headphones",
        "current_stock": 45,
        "reorder_point": 50,
        "safety_stock": 20,
        "lead_time_days": 7,
        "unit_cost": 89.99,
        "supplier_id": "SUP-A"
    },
    "SKU-002": {
        "sku": "SKU-002",
        "name": "USB-C Cable",
        "current_stock": 12,  # CRITICAL: Below safety stock!
        "reorder_point": 100,
        "safety_stock": 30,
        "lead_time_days": 3,
        "unit_cost": 8.99,
        "supplier_id": "SUP-B"
    },
    "SKU-003": {
        "sku": "SKU-003",
        "name": "Phone Case",
        "current_stock": 200,
        "reorder_point": 75,
        "safety_stock": 25,
        "lead_time_days": 5,
        "unit_cost": 12.50,
        "supplier_id": "SUP-A"
    }
}

def scan_inventory(state: InventoryAgentState) -> dict:
    """Scan all inventory and identify items needing attention"""
    all_items = list(INVENTORY_DB.values())
    low_stock = []
    critical = []

    for item in all_items:
        if item["current_stock"] <= item["safety_stock"]:
            critical.append(item)
        elif item["current_stock"] <= item["reorder_point"]:
            low_stock.append(item)

    return {
        "all_items": all_items,
        "low_stock_items": low_stock,
        "critical_items": critical,
        "alerts": [
            {
                "type": "critical",
                "message": f"CRITICAL: {item['name']} ({item['sku']}) at {item['current_stock']} units (safety stock: {item['safety_stock']})",
                "sku": item["sku"],
                "timestamp": datetime.now().isoformat()
            }
            for item in critical
        ]
    }

Demand Analyzer (LLM-Powered)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Mock sales history
SALES_HISTORY = {
    "SKU-001": [12, 15, 18, 14, 22, 19, 25],  # Last 7 days
    "SKU-002": [45, 52, 48, 55, 60, 58, 65],  # Trending up!
    "SKU-003": [8, 10, 7, 9, 11, 8, 10],
}

DEMAND_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are an inventory demand analyst. Analyze sales patterns and predict future demand.

For each item, consider:
1. Recent trend (increasing, decreasing, stable)
2. Day-of-week patterns
3. Any anomalies

Current date context: It's mid-December (holiday shopping season)

Respond in JSON format:
{{
    "sku": "SKU-XXX",
    "avg_daily_demand": number,
    "trend": "increasing" | "decreasing" | "stable",
    "predicted_7_day_demand": number,
    "confidence": "high" | "medium" | "low",
    "notes": "explanation"
}}"""),
    ("human", "Analyze demand for {sku} ({name}). Sales last 7 days: {sales}")
])

def analyze_demand(state: InventoryAgentState) -> dict:
    """Analyze demand patterns for low-stock items"""
    items_to_analyze = state["low_stock_items"] + state["critical_items"]
    forecasts = {}

    for item in items_to_analyze:
        sales = SALES_HISTORY.get(item["sku"], [10] * 7)  # Default if no history

        response = llm.invoke(
            DEMAND_PROMPT.format(
                sku=item["sku"],
                name=item["name"],
                sales=sales
            )
        )

        import json
        forecast = json.loads(response.content)
        forecasts[item["sku"]] = forecast

    return {"demand_forecasts": forecasts}

Reorder Calculator

def calculate_reorder_quantities(state: InventoryAgentState) -> dict:
    """Calculate optimal reorder quantities based on demand forecasts"""
    recommendations = []

    for item in state["low_stock_items"] + state["critical_items"]:
        sku = item["sku"]
        forecast = state["demand_forecasts"].get(sku, {})

        # Get predicted daily demand (or estimate)
        daily_demand = forecast.get("avg_daily_demand", 10)

        # Calculate reorder quantity
        # Formula: (Lead Time Demand) + (Safety Stock) - (Current Stock)
        lead_time_demand = daily_demand * item["lead_time_days"]
        reorder_qty = max(
            lead_time_demand + item["safety_stock"] - item["current_stock"],
            0
        )

        # Round up to reasonable order quantity (e.g., multiples of 10)
        reorder_qty = ((reorder_qty // 10) + 1) * 10 if reorder_qty > 0 else 0

        # Calculate days until stockout
        days_until_stockout = item["current_stock"] / daily_demand if daily_demand > 0 else float('inf')

        if reorder_qty > 0:
            recommendations.append({
                "sku": sku,
                "name": item["name"],
                "current_stock": item["current_stock"],
                "recommended_quantity": int(reorder_qty),
                "estimated_cost": round(reorder_qty * item["unit_cost"], 2),
                "days_until_stockout": round(days_until_stockout, 1),
                "urgency": "critical" if days_until_stockout < item["lead_time_days"] else "standard",
                "supplier_id": item["supplier_id"]
            })

    return {"reorder_recommendations": recommendations}

Supplier Coordinator

# Mock supplier API
SUPPLIERS = {
    "SUP-A": {
        "name": "TechParts Inc",
        "products": {
            "SKU-001": {"price": 85.00, "available": 500, "min_order": 25},
            "SKU-003": {"price": 11.50, "available": 1000, "min_order": 50}
        }
    },
    "SUP-B": {
        "name": "CableWorld",
        "products": {
            "SKU-002": {"price": 7.50, "available": 2000, "min_order": 100}
        }
    }
}

def get_supplier_quotes(state: InventoryAgentState) -> dict:
    """Get quotes from suppliers for recommended items"""
    quotes = []

    for rec in state["reorder_recommendations"]:
        supplier_id = rec["supplier_id"]
        supplier = SUPPLIERS.get(supplier_id, {})
        product_info = supplier.get("products", {}).get(rec["sku"], {})

        if product_info:
            quotes.append({
                "sku": rec["sku"],
                "supplier_id": supplier_id,
                "supplier_name": supplier.get("name", "Unknown"),
                "unit_price": product_info["price"],
                "available_quantity": product_info["available"],
                "lead_time_days": 5,  # From supplier SLA
                "minimum_order": product_info["min_order"]
            })

    return {"supplier_quotes": quotes}

def create_purchase_orders(state: InventoryAgentState) -> dict:
    """Create purchase orders based on recommendations and quotes"""
    orders = []

    for rec in state["reorder_recommendations"]:
        quote = next(
            (q for q in state["supplier_quotes"] if q["sku"] == rec["sku"]),
            None
        )

        if quote:
            # Ensure we meet minimum order quantity
            order_qty = max(rec["recommended_quantity"], quote["minimum_order"])

            orders.append({
                "po_number": f"PO-{datetime.now().strftime('%Y%m%d')}-{rec['sku']}",
                "sku": rec["sku"],
                "product_name": rec["name"],
                "quantity": order_qty,
                "unit_price": quote["unit_price"],
                "total_cost": round(order_qty * quote["unit_price"], 2),
                "supplier": quote["supplier_name"],
                "expected_delivery": f"{quote['lead_time_days']} days",
                "status": "pending_approval" if rec["urgency"] != "critical" else "auto_approved"
            })

    return {
        "selected_orders": orders,
        "actions_taken": [
            {
                "type": "purchase_order_created",
                "po_number": order["po_number"],
                "sku": order["sku"],
                "quantity": order["quantity"],
                "status": order["status"],
                "timestamp": datetime.now().isoformat()
            }
            for order in orders
        ]
    }

Report Generator

REPORT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """Generate an executive inventory report based on the analysis.

Include:
1. Summary of inventory health
2. Critical items requiring immediate attention
3. Purchase orders created/recommended
4. Key insights and recommendations

Be concise but comprehensive. Use bullet points and clear formatting."""),
    ("human", """
Inventory Scan Results:
- Total SKUs monitored: {total_items}
- Low stock items: {low_stock_count}
- Critical items: {critical_count}

Demand Forecasts:
{forecasts}

Purchase Orders:
{orders}

Alerts:
{alerts}
""")
])

def generate_report(state: InventoryAgentState) -> dict:
    """Generate human-readable inventory report"""
    response = llm.invoke(
        REPORT_PROMPT.format(
            total_items=len(state["all_items"]),
            low_stock_count=len(state["low_stock_items"]),
            critical_count=len(state["critical_items"]),
            forecasts=state["demand_forecasts"],
            orders=state["selected_orders"],
            alerts=state["alerts"]
        )
    )

    return {"report": response.content}

Step 3: Build the Graph

from langgraph.graph import StateGraph, START, END

def route_after_scan(state: InventoryAgentState) -> str:
    """Route based on scan results"""
    if state["low_stock_items"] or state["critical_items"]:
        return "analyze_demand"
    return "generate_report"  # Nothing to reorder, just report

# Build the workflow
workflow = StateGraph(InventoryAgentState)

workflow.add_node("scan_inventory", scan_inventory)
workflow.add_node("analyze_demand", analyze_demand)
workflow.add_node("calculate_reorder", calculate_reorder_quantities)
workflow.add_node("get_quotes", get_supplier_quotes)
workflow.add_node("create_orders", create_purchase_orders)
workflow.add_node("generate_report", generate_report)

# Edges
workflow.add_edge(START, "scan_inventory")
workflow.add_conditional_edges(
    "scan_inventory",
    route_after_scan,
    {"analyze_demand": "analyze_demand", "generate_report": "generate_report"}
)
workflow.add_edge("analyze_demand", "calculate_reorder")
workflow.add_edge("calculate_reorder", "get_quotes")
workflow.add_edge("get_quotes", "create_orders")
workflow.add_edge("create_orders", "generate_report")
workflow.add_edge("generate_report", END)

# Compile
inventory_agent = workflow.compile()

Step 4: Run the Agent

# Run inventory check
result = inventory_agent.invoke({
    "trigger_type": "scheduled",
    "run_timestamp": datetime.now().isoformat(),
    "all_items": [],
    "low_stock_items": [],
    "critical_items": [],
    "demand_forecasts": {},
    "reorder_recommendations": [],
    "supplier_quotes": [],
    "selected_orders": [],
    "alerts": [],
    "actions_taken": [],
    "report": ""
})

# Print results
print("=== INVENTORY AGENT REPORT ===\n")
print(result["report"])
print("\n=== ALERTS ===")
for alert in result["alerts"]:
    print(f"  [{alert['type'].upper()}] {alert['message']}")
print("\n=== PURCHASE ORDERS ===")
for order in result["selected_orders"]:
    print(f"  {order['po_number']}: {order['quantity']}x {order['product_name']} = ${order['total_cost']}")

Scheduling the Agent

# In production, use APScheduler, Celery, or cloud functions
from apscheduler.schedulers.background import BackgroundScheduler

def run_inventory_check():
    result = inventory_agent.invoke({
        "trigger_type": "scheduled",
        "run_timestamp": datetime.now().isoformat(),
        # ... initial state
    })
    # Send alerts, store report, etc.
    if result["critical_items"]:
        send_slack_alert(result["alerts"])

# Schedule to run every hour
scheduler = BackgroundScheduler()
scheduler.add_job(run_inventory_check, 'interval', hours=1)
scheduler.start()

Key Takeaways

Next up: Human-in-the-Loop - Adding approval workflows and escalation for sensitive operations.