🚚 Shipment Exception Desk

Concept

Shipment Exception Desk is an operations-focused AI workflow for logistics exception handling. Instead of directly drafting responses from raw complaint text, the system follows a structured triage pipeline: classify issue type, calculate policy compensation, evaluate escalation need, generate the right communication, and record everything in a session ledger.

The design deliberately combines LLM reasoning with deterministic business rules. Classification and narrative drafting are delegated to LangChain model chains, while monetary calculations and escalation thresholds remain exact and auditable in Python.

Theory & Concepts

1. Hybrid architecture: probabilistic AI + deterministic policy

The project uses the model for language-heavy tasks (classification and writing) and uses plain code for exact rule execution. This split keeps behavior stable for compliance-sensitive operations while still benefiting from LLM flexibility.

2. Tier-aware escalation gating

Premium and standard customers use different escalation thresholds. Unknown reports auto-escalate. This makes escalation decisions both explainable and easy to tune:

• Standard: escalate above $100 compensation.
• Premium: escalate above $50 compensation.
• Unknown: escalate immediately for manual review.

3. Session-level operational analytics

Each processed claim is appended to an in-memory ledger. A daily summary endpoint aggregates totals, escalation rate, and costliest category by cumulative payout, turning each run into an auditable operations snapshot.

Request flow

Code flow

Pipeline Orchestration

ESCALATION_THRESHOLDS = {
    "standard": 100.0,
    "premium": 50.0,
}


def evaluate_escalation(category: str, compensation_amount: float, customer_tier: str) -> tuple[bool, str]:
    """Tier-aware escalation decision."""
    tier_normalized = customer_tier.strip().lower()
    threshold = ESCALATION_THRESHOLDS.get(tier_normalized, 100.0)

    if category == "unknown":
        return True, "Unclassifiable or garbled report requires manual operations review."

    if compensation_amount > threshold:
        return (
            True,
            f"Compensation amount (${compensation_amount:.2f}) exceeds {tier_normalized.capitalize()} tier threshold (${threshold:.2f}).",
        )

    return (
        False,
        f"Compensation (${compensation_amount:.2f}) is within {tier_normalized.capitalize()} tier auto-approval limit (${threshold:.2f}).",
    )


def process_exception(report_text: str, shipment_value: float, customer_tier: str = "standard", log_to_session: bool = True):
    steps = []
    customer_tier = customer_tier.strip().lower()

    # Step 1: LLM category classification
    category = classify_chain.invoke({"report_text": report_text})
    steps.append(f"Step 1 [Classify]: Report classified as '{category.upper()}' via LLM.")

    # Step 2: Deterministic compensation routing
    if category == "delayed":
        comp_result = calculate_delay_compensation(shipment_value)
    elif category == "damaged":
        comp_result = calculate_damage_compensation(shipment_value)
    elif category == "lost":
        comp_result = calculate_lost_compensation(shipment_value)
    else:
        comp_result = calculate_unknown_compensation(shipment_value)

    comp_amount = float(comp_result.get("amount", 0.0))

    # Step 3: Escalation gate
    escalated, escalation_reason = evaluate_escalation(
        category=category,
        compensation_amount=comp_amount,
        customer_tier=customer_tier,
    )

    # Step 4: Conditional draft generation
    if escalated:
        draft = escalate_chain.invoke({
            "customer_tier": customer_tier.capitalize(),
            "shipment_value": f"{shipment_value:.2f}",
            "category": category,
            "compensation_amount": f"{comp_amount:.2f}",
            "escalation_reason": escalation_reason,
            "report_text": report_text,
        })
    else:
        draft = draft_email_chain.invoke({
            "customer_tier": customer_tier.capitalize(),
            "shipment_value": f"{shipment_value:.2f}",
            "category": category,
            "compensation_amount": f"{comp_amount:.2f}",
            "report_text": report_text,
        })

    result = {
        "category": category,
        "compensation_amount": comp_amount,
        "escalated": escalated,
        "escalation_reason": escalation_reason,
        "draft": draft,
        "steps": steps,
    }

    # Step 5: Session ledger append
    if log_to_session:
        log_exception(result)

    return result

Compensation Policy Tools

def calculate_delay_compensation(shipment_value: float, days_delayed: int = 1):
    """20% payout, with a minimum courtesy credit and cap."""
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    if shipment_value == 0:
        return {
            "category": "delayed",
            "amount": 0.0,
            "currency": "USD",
            "reason": "Shipment value is $0.00; no compensation issued.",
        }

    base_compensation = shipment_value * 0.20
    compensation = max(base_compensation, 15.0)
    compensation = min(compensation, shipment_value)

    return {
        "category": "delayed",
        "amount": round(compensation, 2),
        "currency": "USD",
        "reason": "Delay compensation with minimum courtesy credit and shipment-value cap",
    }


def calculate_damage_compensation(shipment_value: float, damage_severity: str = "partial"):
    """50% for partial damage, 100% for total/severe damage."""
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    severity = damage_severity.strip().lower()
    rate = 1.0 if severity in ("total", "severe", "complete") else 0.50
    return {
        "category": "damaged",
        "amount": round(shipment_value * rate, 2),
        "currency": "USD",
    }


def calculate_lost_compensation(shipment_value: float):
    """Lost shipments get full replacement value."""
    if shipment_value < 0:
        raise ValueError("Shipment value cannot be negative.")

    return {
        "category": "lost",
        "amount": round(shipment_value, 2),
        "currency": "USD",
    }


def calculate_unknown_compensation(shipment_value: float):
    """Unknown/garbled reports are routed to manual review."""
    return {
        "category": "unknown",
        "amount": 0.0,
        "currency": "USD",
        "reason": "Unclassified exception: no automatic compensation; manual review required",
    }

Session Aggregation

_SESSION_RECORDS = []


def log_exception(record):
    """Append one processed exception to the in-memory daily ledger."""
    entry = dict(record)
    if "timestamp" not in entry:
        entry["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    _SESSION_RECORDS.append(entry)
    return entry


def generate_daily_summary():
    """Aggregate session records into operational metrics."""
    total_exceptions = len(_SESSION_RECORDS)
    total_compensation = sum(float(r.get("compensation_amount", 0.0)) for r in _SESSION_RECORDS)
    escalated_count = sum(1 for r in _SESSION_RECORDS if r.get("escalated"))
    escalation_rate = (escalated_count / total_exceptions * 100.0) if total_exceptions else 0.0

    categories = ["delayed", "damaged", "lost", "unknown"]
    category_breakdown = {}

    for cat in categories:
        cat_records = [r for r in _SESSION_RECORDS if r.get("category") == cat]
        category_breakdown[cat] = {
            "count": len(cat_records),
            "total_compensation": round(sum(float(r.get("compensation_amount", 0.0)) for r in cat_records), 2),
            "escalated": sum(1 for r in cat_records if r.get("escalated")),
        }

    payout_per_category = {
        cat: category_breakdown[cat]["total_compensation"] for cat in categories
    }

    max_payout = max(payout_per_category.values()) if payout_per_category else 0.0
    costliest_category = max(payout_per_category, key=lambda k: payout_per_category[k]) if max_payout > 0 else "None"

    return {
        "total_exceptions": total_exceptions,
        "total_compensation": round(total_compensation, 2),
        "escalated_count": escalated_count,
        "escalation_rate": round(escalation_rate, 2),
        "costliest_category": costliest_category,
        "category_breakdown": category_breakdown,
    }

API route

@bp.route("/api/triage", methods=["POST"])
def triage_report():
    """Process an incoming shipment exception report."""
    data = request.get_json(force=True) or {}
    report_text = (data.get("report_text") or "").strip()
    shipment_value = data.get("shipment_value")
    customer_tier = (data.get("customer_tier") or "standard").strip().lower()

    # Step 1: Input validation
    if not report_text:
        return jsonify({"detail": "Report text cannot be empty."}), 400

    try:
        shipment_value = float(shipment_value)
    except (TypeError, ValueError):
        return jsonify({"detail": "Shipment value must be a valid number."}), 400

    if shipment_value < 0:
        return jsonify({"detail": "Shipment value cannot be negative."}), 400

    if customer_tier not in {"standard", "premium"}:
        return jsonify({"detail": "Customer tier must be standard or premium."}), 400

    # Step 2: Execute full pipeline
    try:
        result = process_exception(
            report_text=report_text,
            shipment_value=shipment_value,
            customer_tier=customer_tier,
            log_to_session=True,
        )
        return jsonify(result)
    except Exception as exc:
        return jsonify({"detail": str(exc)}), 500


@bp.route("/api/log", methods=["GET"])
def fetch_log():
    """Return all triage records logged in the current session."""
    return jsonify(get_triage_log())


@bp.route("/api/summary", methods=["GET"])
def fetch_summary():
    """Return aggregated summary metrics and category breakdown."""
    return jsonify(generate_daily_summary())


@bp.route("/api/reset", methods=["POST"])
def reset_session():
    """Clear session records."""
    clear_session()
    return jsonify({"status": "session_cleared"})