AI Digital Logistics Agent Development

AI Logistics Agent — Digital Employee

AI Logistics Agent automates supply chain operational tasks: route planning, shipment tracking, carrier interaction, exception handling (delays, damage, shortages), KPI monitoring and reporting. Human operator connects only to non-standard situations requiring negotiations.

Shipment Monitoring and Tracking

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Optional
import operator

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

class ShipmentState(TypedDict):
    shipment_id: str
    shipment_data: dict
    tracking_history: list[dict]
    anomalies: Annotated[list, operator.add]
    actions_taken: Annotated[list, operator.add]
    escalation_required: bool
    escalation_reason: Optional[str]

class ShipmentMonitor:

    async def check_shipment(self, shipment_id: str) -> ShipmentState:
        """Complete shipment status check"""

        # Get data
        shipment = await logistics_db.get_shipment(shipment_id)
        tracking = await carrier_api.get_tracking(shipment["tracking_number"])
        expected_eta = shipment["expected_delivery"]
        current_eta = tracking.get("estimated_delivery")

        anomalies = []

        # Check delay
        if current_eta and current_eta > expected_eta:
            delay_hours = (current_eta - expected_eta).total_seconds() / 3600
            anomalies.append({
                "type": "delivery_delay",
                "severity": "high" if delay_hours > 24 else "medium",
                "details": f"Delay {delay_hours:.0f} hours, new date: {current_eta}",
            })

        # No updates
        last_update = tracking.get("last_event_time")
        hours_since_update = (datetime.now() - last_update).total_seconds() / 3600 if last_update else 99
        if hours_since_update > 48:
            anomalies.append({
                "type": "no_tracking_update",
                "severity": "medium",
                "details": f"No updates {hours_since_update:.0f} hours",
            })

        # LLM analysis on anomalies
        escalation_required = False
        escalation_reason = None

        if anomalies:
            assessment = await self.assess_anomalies(shipment, anomalies)
            escalation_required = assessment["requires_escalation"]
            escalation_reason = assessment.get("reason")

        return ShipmentState(
            shipment_id=shipment_id,
            shipment_data=shipment,
            tracking_history=tracking.get("events", []),
            anomalies=anomalies,
            actions_taken=[],
            escalation_required=escalation_required,
            escalation_reason=escalation_reason,
        )

    async def assess_anomalies(self, shipment: dict, anomalies: list) -> dict:
        """LLM assesses escalation necessity"""

        response = await llm.ainvoke(f"""Assess shipment situation.
Shipment: {json.dumps(shipment, ensure_ascii=False)}
Anomalies: {json.dumps(anomalies, ensure_ascii=False)}

Determine:
1. Does situation require immediate escalation to manager?
2. What automatic actions can be taken?
3. Should recipient be notified?

Return JSON: {{"requires_escalation": bool, "reason": "...", "auto_actions": [...], "notify_recipient": bool}}""")

        return json.loads(response.content)

Route Optimization

class RouteOptimizer:

    async def optimize_delivery_routes(
        self,
        deliveries: list[dict],    # [{id, address, time_window, weight}]
        vehicles: list[dict],      # [{id, capacity, location}]
        date: str,
    ) -> dict:
        """Optimizes delivery routes (Vehicle Routing Problem)"""

        # For small tasks — via LLM with reasoning
        if len(deliveries) <= 20:
            return await self.llm_route_optimizer(deliveries, vehicles)

        # For large — algorithmic approach + LLM for exceptions
        return await self.algorithmic_route_optimizer(deliveries, vehicles)

Exception Handling

class ExceptionHandler:

    EXCEPTION_PLAYBOOKS = {
        "delivery_delay": {
            "auto_actions": ["notify_recipient", "update_crm", "rebook_if_urgent"],
            "escalate_if": lambda hours: hours > 72,
        },
        "damaged_goods": {
            "auto_actions": ["create_claim", "notify_sender", "photo_request"],
            "escalate_always": True,
        },
        "customs_hold": {
            "auto_actions": ["get_customs_details", "notify_broker"],
            "escalate_if": lambda days: days > 3,
        },
        "address_not_found": {
            "auto_actions": ["contact_recipient", "check_database"],
            "escalate_if": lambda attempts: attempts > 2,
        },
    }

    async def handle_exception(self, exception: dict) -> dict:
        exception_type = exception["type"]
        playbook = self.EXCEPTION_PLAYBOOKS.get(exception_type)

        if not playbook:
            return await self.generic_exception_handler(exception)

        actions_taken = []

        # Execute automatic actions
        for action in playbook.get("auto_actions", []):
            result = await self.execute_action(action, exception)
            actions_taken.append({"action": action, "result": result})

        # Check if escalation needed
        escalate = playbook.get("escalate_always", False)
        if not escalate and "escalate_if" in playbook:
            escalate_fn = playbook["escalate_if"]
            escalate = escalate_fn(exception.get("delay_hours") or exception.get("hold_days") or exception.get("attempts", 0))

        if escalate:
            await self.escalate_to_manager(exception, actions_taken)

        return {"actions_taken": actions_taken, "escalated": escalate}

Analytics and KPI Monitoring

class LogisticsAnalytics:

    async def daily_kpi_report(self) -> str:
        """Daily logistics KPI report"""

        # Data from database
        metrics = await asyncio.gather(
            self.get_on_time_delivery_rate(),
            self.get_damage_rate(),
            self.get_carrier_performance(),
            self.get_cost_per_shipment(),
            self.get_exception_rate(),
        )

        report = await llm.ainvoke(f"""Create KPI report for logistics.

Metrics:
- On-time delivery: {metrics[0]['rate']:.1%} (target: {metrics[0]['target']:.1%})
- Damage rate: {metrics[1]['rate']:.3%}
- Top carriers by performance: {metrics[2]}
- Cost per shipment: {metrics[3]['avg']:,.0f} rubles
- Exception rate: {metrics[4]['rate']:.2%}

Format: brief summary (3 sentences), deviations from norm, recommendations.""")

        return report.content

Practical Case Study: FMCG Distributor, 500 Shipments/Day

Situation: 4 logistics operators, 500 shipments per day, operators spent 70% of time on tracking and exception handling.

AI Agent Took Over:

• Automatic tracking every 2 hours via carrier APIs
• Recipient notifications on status and delays
• Automatic claim opening on damage
• Rebooking for delays > 4 hours (for critical shipments)
• Daily KPI report and weekly carrier scorecard

Results:

• Manual exception processing: 180 cases/day → 45 (agent auto-closed)
• On-time delivery rate: 82% → 89% (better routing + proactive exception management)
• Confirmed claims (properly documented): 62% → 91%
• Logistics operators' operational time: -55%

Timeline

• Shipment tracking and monitoring: 2–3 weeks
• Exception handler with playbooks: 2–3 weeks
• Route optimization: 2–3 weeks
• Analytics and KPI reports: 1 week
• Carrier integration: 1–2 weeks
• Total: 8–12 weeks