Agentic AI for
Shipment Exception Management

Executive Summary

Every logistics operation produces a daily stream of shipping exceptions. How those exceptions are handled determines SLA performance, expedite cost, and customer satisfaction. The current model is structurally broken — and agents are the right fix.

This whitepaper presents a complete reference implementation for an agentic exception management system built on LangGraph, FastAPI, and LiteLLM. It covers the business problem, the 6-agent architecture, the data layer, the trust and safety framework, the evaluation strategy, and a concrete implementation roadmap.

The system is designed for operators at Supply Chain Autonomy Maturity Level 2 (Assisted) seeking to reach Level 3–4 (Automated / Agentic). It is not a black-box product — it is a reference implementation you can audit, fork, and adapt. Every design decision is documented in Architecture Decision Records. Every agent behavior is covered by an eval scenario before a line of agent code is written.

The Problem: Shipping Exceptions at Scale

Shipping exception management is a decision problem hiding inside a communication problem. It fails not because people are incompetent or systems are absent — it fails because the operating model is structurally wrong for the volume and variability it faces.

The compounding cost of manual exception management

Four Structural Failure Modes

The current workflow breaks along four structural fault lines, each of which compounds the others:

Exception Taxonomy

Shipping exceptions are not a single category. Each type has different triggers, different resolution paths, and different cost profiles. A system that treats them identically over-engineers simple cases and under-serves complex ones.

Maturity Model: Where Are You Today?

The Solution: Agentic Control Tower

The agentic future workflow inverts the current model: agents gather context continuously, propose decisions within policy bounds, and execute approved actions automatically — while humans focus exclusively on cases that genuinely require judgment.

Six Specialized Agents

Exception Lifecycle State Machine

The exception lifecycle is an explicit state machine. State names are immutable across every layer — ADRs, canonical docs, stack code, and eval scenarios use identical names. No aliases.

  detected ──▶ triaged ──▶ investigating ──▶ action_proposed ──▶ awaiting_human
                                                    │                  │
                                                    │                  │ operator approves
                                                    ▼                  ▼
                                              auto_resolved ◀──────────┘
                                                    │         (or failed)
                                                    ▼
                                                 closed

Autonomy Tiers

Data Layer: Bronze → Silver → Gold

The medallion architecture ensures that agents reason on structured, validated, enriched context — never on raw carrier payloads. Each layer has a defined latency budget, schema contract, and quality gate.

Data Quality Circuit Breaker

Trust & Safety: Guardrails and HITL

A production agentic system requires defense-in-depth. Seven guardrail layers operate at different points in the agent trajectory. Four "never" rules are deterministic constraints that no confidence score or operator approval can override.

7-Layer Guardrail Stack

Four "Never" Rules

These are hard constraints that operate independently of agent confidence, operator approval, or business urgency:

Evaluation: Eval-First Engineering

At 95% per-step accuracy, a 10-step agent trajectory succeeds only 60% of the time. The eval harness is scaffolded before any agent code. This is not process discipline — it is the only way to detect compounding errors during development, not after deployment.

Four-Pillar Eval Framework

pass@k vs pass^k

Carrier rebook, SLA breach escalation, prompt injection resistance, and the customs always-escalate rule all require pass^5 (5-of-5) before production graduation. A system that passes 4-of-5 on a safety-critical scenario is not ready.

Implementation Path

Build in this order. Stop at each gate. Do not add the next component until the current one's signal criterion passes.

What "Local-First" Means in Practice

The entire stack runs locally without cloud dependencies. Carrier adapters use fixture JSON from stack/evals/fixtures/. The LangGraph checkpoint store uses SQLite locally (Postgres in production). LiteLLM routes to a local model (Ollama) or a cloud API depending on the LITELLM_MODEL environment variable. The eval harness runs without any live carrier API credentials.

Stack Dependencies

Python 3.11+
FastAPI + uvicorn
LangGraph 0.2+
LiteLLM (model routing layer)
Pydantic v2
Langfuse (LLM observability)
OpenTelemetry SDK
Redis (worker queue — local: fakeredis)
SQLite (LangGraph checkpoint store — local)
pytest + pytest-asyncio

Business Case

The ROI from exception automation compounds: each automated investigation cycle reduces cost per case and accumulates labeled training data that improves future case routing. Payback: 6–18 months at 50+ cases/day.

KPI Before vs. After

ICP: Are You Ready?

You are the right fit if: You are a VP of Supply Chain, Head of Logistics Ops, or Engineering Lead at a 3PL, large manufacturer, or high-volume retailer. You handle 50–500 exception cases per day across multiple carriers. You have a TMS or ERP with an API you can query. You are at Maturity Level 2 (Assisted) and want to reach Level 3–4.

Not quite ready if: You don't have structured exception data (only emails and phone calls). Your carrier data arrives exclusively via manual entry with no EDI, API, or webhook. You handle fewer than 20 exceptions per day — the ROI isn't there yet. You have no engineering capacity to wire an adapter or run a LangGraph graph locally.

Code Implementation

The reference stack is a Python monorepo organized around a clean architecture: the domain layer (entities and state machine) has zero external dependencies, and all external systems are accessed through a typed adapter interface.

Project Setup

Dependencies are managed via pyproject.toml. The key runtime dependencies and their roles:

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
python -m domain.db_init          # initialize SQLite checkpoint store (local)
pytest tests/unit/ tests/state_machine/ -v   # no LLM or cloud credentials needed
python -m evals.run_scenario --scenario-id GOLD-001

Module Structure

stack/
├── api/              ← FastAPI: POST /webhooks/carrier/{code}, GET/POST /cases/{id}
├── domain/           ← ExceptionCase entity, state machine, AgentContext allowlist
├── orchestration/    ← StateGraph wiring, 6 agent nodes, interrupt_before=["execute"]
├── adapters/         ← AbstractCarrierAdapter + mocked FedEx/UPS/Maersk/TMS/ERP/EDI
├── worker/           ← SQS polling loop, visibility timeout extension, DLQ handler
├── evals/            ← EvalHarness ABC, 8 YAML scenarios, LLM-as-judge graders
├── observability/    ← 17 typed event emitters, MELT metrics, OTel span decorators
└── tests/            ← Unit / integration / state machine / E2E (4-layer pyramid)

Core Interface Contracts

AbstractCarrierAdapter

class AbstractCarrierAdapter(ABC):
    @abstractmethod
    def fetch_tracking_events(
        self, tracking_number: str
    ) -> list[NormalizedEvent]:
        """Fetch normalized events. Must be idempotent."""
        ...

    @abstractmethod
    def execute_action(
        self, action: CarrierActionRequest
    ) -> CarrierActionResult:
        """Execute approved action. Idempotency key required."""
        ...

ExceptionCase entity

class ExceptionCase(BaseModel):
    case_id: str
    state: CaseState = "detected"
    tracking_number: str
    carrier_code: str
    exception_type: ExceptionType      # CUSTOMS_HOLD | WEATHER_DELAY | ...
    autonomy_tier: Literal["LOW", "MEDIUM", "HIGH"] = "HIGH"
    risk_score: float = Field(0.5, ge=0.0, le=1.0)
    data_completeness_score: float = Field(0.0, ge=0.0, le=1.0)
    sla_deadline_utc: datetime | None = None
    financial_materiality_usd: float = 0.0

LangGraph graph wiring

graph = StateGraph(ExceptionCaseState)
# Register 6 agent nodes
graph.add_node("normalize",   normalization.run)
graph.add_node("assess_risk", risk_assessment.run)
graph.add_node("investigate", investigation.run)
graph.add_node("strategize",  strategy.run)
graph.add_node("execute",     execution.run)
graph.add_node("learn",       learning.run)

# Conditional edge: HITL gate at action_proposed
graph.add_conditional_edges(
    "strategize",
    should_route_to_human,
    {"awaiting_human": END, "execute": "execute"}
)

return graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute"]   # ← HITL serialization point
)

The interrupt_before=["execute"] call is the HITL mechanism. When the graph reaches this node for a case requiring operator review, LangGraph serializes the full graph state to the Postgres checkpoint store. The case waits in awaiting_human until the operator approves or rejects. The graph resumes from the checkpoint — no state is lost across process restarts or worker scaling events.

Testing Pyramid

State machine tests cover all 13 valid transitions and all 6 invalid transitions (each must raise InvalidStateTransitionError). These run without any credentials in under 2 seconds — they are the fastest regression signal in the system.

Deployment on AWS

The production topology maps onto five AWS services. The API service is lightweight — validation and queue dispatch only. The Worker service is heavy — it runs LangGraph including multi-step LLM calls. They scale independently.

Internet
    │
    ▼
ALB (HTTPS :443)
    │
    ▼
ECS Fargate — API Service (FastAPI webhook receiver)
  cpu: 512  memory: 1 GB  min: 2 tasks  scale: requests/target
    │  SQS SendMessage
    ▼
SQS — shipping-exception-queue
  VisibilityTimeout: 900s (customs hold budget)
    │  SQS ReceiveMessage (1 msg per task)
    ▼
ECS Fargate — Worker Service (LangGraph runner)
  cpu: 1024  memory: 4 GB  min: 1 task  scale: queue depth
    ├──▶  RDS Postgres    (LangGraph checkpoint store — awaiting_human state)
    ├──▶  AWS Bedrock     (Sonnet: investigation + strategy / Haiku: others)
    ├──▶  S3 Bronze       (raw carrier payloads, EDI messages, OCR extractions)
    └──▶  Langfuse        (LLM traces, token cost, eval grading)

Service Configuration and Tuning Knobs

Model Routing: Sonnet vs. Haiku

Monthly Cost Model

Terraform Module Structure

infra/
├── modules/
│   ├── ecs-api/     ← Task def, ALB, target group, auto-scaling
│   ├── ecs-worker/  ← Task def, SQS-triggered step scaling
│   ├── sqs/         ← Queue, DLQ, CloudWatch alarm
│   ├── rds/         ← Postgres instance, subnet group, Multi-AZ
│   └── iam/         ← API task role, Worker task role (least privilege)
└── envs/
    ├── dev/          ← SQLite locally; only RDS and SQS differ
    └── prod/         ← Full VPC, Multi-AZ RDS, autoscaling enabled

Glossary & State Machine Reference

Glossary

State Transition Reference

All 13 valid state transitions. Any transition not in this list raises InvalidStateTransitionError.