Building an Intelligent Supply Chain Exception Agent: Architecture and Results

import json
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import ChatOpenAI
from dataclasses import dataclass
from typing import Literal


@dataclass
class SupplyChainException:
    exception_id: str
    exception_type: Literal["delay", "quantity_mismatch", "po_non_ack", "cost_variance"]
    supplier_id: str
    affected_po_numbers: list[str]
    financial_exposure_usd: float
    raw_details: dict


# --- Tool definitions ---

@tool("get_inventory_buffer")
def get_inventory_buffer(sku_id: str, dc_id: str) -> str:
    """
    Returns current inventory buffer in days for a given SKU at a distribution centre.
    Input: sku_id (str), dc_id (str)
    """
    # Production: query your WMS or inventory API
    mock_buffers = {
        ("SKU-4421", "DC-WEST"): {"buffer_days": 6, "units_on_hand": 4200},
        ("SKU-4421", "DC-EAST"): {"buffer_days": 2, "units_on_hand": 1100},
    }
    data = mock_buffers.get((sku_id, dc_id), {"buffer_days": 14, "units_on_hand": 9999})
    return json.dumps(data)


@tool("get_supplier_tier")
def get_supplier_tier(supplier_id: str) -> str:
    """
    Returns supplier criticality tier (1=critical single-source, 2=dual-source, 3=commodity).
    Input: supplier_id (str)
    """
    mock_tiers = {
        "SUP-TW-0041": {"tier": 1, "category": "display_panels", "lead_time_days": 45,
                        "has_secondary": False},
        "SUP-CN-0182": {"tier": 2, "category": "pcb_assembly", "lead_time_days": 21,
                        "has_secondary": True},
    }
    data = mock_tiers.get(supplier_id, {"tier": 3, "category": "commodity",
                                        "lead_time_days": 7, "has_secondary": True})
    return json.dumps(data)


@tool("get_production_schedule_impact")
def get_production_schedule_impact(po_numbers: str, delay_days: int) -> str:
    """
    Assesses production line impact given delayed POs and delay length.
    Input: po_numbers (comma-separated str), delay_days (int)
    """
    po_list = [p.strip() for p in po_numbers.split(",")]
    # Production: query your MRP/ERP for production schedule dependencies
    return json.dumps({
        "affected_production_runs": ["PR-2025-1142", "PR-2025-1143"],
        "earliest_line_stop_days": max(0, 4 - delay_days),
        "estimated_downtime_cost_usd": delay_days * 28500 if delay_days > 4 else 0,
    })


# --- Triage Agent ---

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

triage_agent = Agent(
    role="Supply Chain Triage Specialist",
    goal="Classify exception severity and assess risk based on inventory buffers, "
         "supplier criticality, and production schedule impact.",
    backstory="""You are a senior supply chain analyst with expertise in exception
    management. You have deep knowledge of procurement risk, inventory management,
    and supplier relationship dynamics. Your assessments drive automated resolution
    vs. human escalation decisions.""",
    tools=[get_inventory_buffer, get_supplier_tier, get_production_schedule_impact],
    llm=llm,
    verbose=True,
    max_iter=8,
)


def build_triage_task(exception: SupplyChainException) -> Task:
    return Task(
        description=f"""Triage the following supply chain exception and produce a structured assessment.

Exception ID: {exception.exception_id}
Type: {exception.exception_type}
Supplier: {exception.supplier_id}
Affected POs: {', '.join(exception.affected_po_numbers)}
Financial Exposure: ${exception.financial_exposure_usd:,.0f}
Details: {json.dumps(exception.raw_details, indent=2)}

Steps:
1. Check supplier tier for {exception.supplier_id}
2. Check inventory buffer for affected SKUs (use SKU from details if available)
3. Assess production schedule impact
4. Classify severity: critical (Tier-1 supplier + <3 days buffer + >$50K exposure),
   high (Tier-1 OR <5 days buffer OR >$10K), medium, low
5. Recommend: auto_resolve | human_review | urgent_escalation

Return JSON with: severity, reasoning, buffer_days, production_impact_usd,
recommended_action, escalation_reason (if applicable)""",
        expected_output="""JSON object with: severity (critical|high|medium|low),
        reasoning (str), buffer_days (int), production_impact_usd (float),
        recommended_action (auto_resolve|human_review|urgent_escalation),
        escalation_reason (str or null)""",
        agent=triage_agent,
    )


def triage_exception(exception: SupplyChainException) -> dict:
    task = build_triage_task(exception)
    crew = Crew(
        agents=[triage_agent],
        tasks=[task],
        process=Process.sequential,
        verbose=True,
    )
    result = crew.kickoff()
    try:
        # CrewAI result.raw contains the task output
        return json.loads(result.raw)
    except json.JSONDecodeError:
        return {"severity": "high", "recommended_action": "human_review",
                "escalation_reason": "Triage parsing failed — manual review required"}


# --- Usage ---

if __name__ == "__main__":
    exc = SupplyChainException(
        exception_id="EXC-2025111701",
        exception_type="delay",
        supplier_id="SUP-TW-0041",
        affected_po_numbers=["PO-442219", "PO-442220"],
        financial_exposure_usd=87500.0,
        raw_details={
            "sku_id": "SKU-4421",
            "dc_id": "DC-EAST",
            "original_delivery": "2025-11-22",
            "new_delivery": "2025-11-28",
            "delay_days": 6,
            "reason": "Port congestion at Kaohsiung"
        }
    )
    result = triage_exception(exc)
    print(json.dumps(result, indent=2))