Building Logistics Tools with LLMs: A Step-by-Step Guide

Last month our warehouse team spent six hours a day parsing carrier delay emails into spreadsheets. We fixed it with a small LLM agent that reads raw shipment notifications and outputs structured incident reports with customer-ready emails. In this guide we will build that same pipeline using Oxlo.ai so you can drop it into your own ops stack.

What you'll need

Python 3.10 or newer, an Oxlo.ai API key, and the OpenAI SDK. Install dependencies with:

pip install openai pydantic

Step 1: Set up the client

I always start by verifying the endpoint and my key. A one-line ping against a cheap model prevents thirty minutes of debugging later.

from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Say 'Oxlo.ai connection OK' and nothing else."},
    ],
)
print(response.choices[0].message.content)

Step 2: Define the schema

Raw carrier emails are unstructured noise. We need a strict schema so downstream tools can consume the output without guessing.

from pydantic import BaseModel, Field
from typing import List, Optional

class ShipmentException(BaseModel):
    tracking_id: str = Field(description="The tracking or PRO number")
    carrier: str = Field(description="Shipping carrier name")
    original_eta: Optional[str] = Field(description="Originally promised delivery date")
    updated_eta: Optional[str] = Field(description="New estimated delivery date")
    delay_reason: str = Field(description="One-sentence reason for the delay")
    severity: str = Field(description="One of: low, medium, high, critical")
    affected_skus: List[str] = Field(default_factory=list)

Step 3: Extract structured fields

This is where the LLM does the heavy lifting. We pass the raw email and a prompt that forbids hallucination, then parse the JSON output into our Pydantic model. I use qwen-3-32b here because it handles structured extraction reliably.

import json

EXTRACTION_PROMPT = """You are a logistics data extractor.
Read the raw shipment message below and extract the fields as JSON.
Use null for unknown dates. Severity must be one of: low, medium, high, critical.
Do not guess tracking IDs. If the carrier name is ambiguous, write exactly what appears."""

def extract_exception(raw_text: str) -> ShipmentException:
    response = client.chat.completions.create(
        model="qwen-3-32b",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": EXTRACTION_PROMPT},
            {"role": "user", "content": raw_text},
        ],
    )
    data = json.loads(response.choices[0].message.content)
    return ShipmentException(**data)

Step 4: Classify business impact

Not every delay is critical. A missed hospital supply run is different from a delayed promotional flyer. We run a second inference to classify severity and decide whether to trigger rerouting. I use deepseek-v3.2 for this reasoning step.

REROUTE_PROMPT = """You are a logistics operations analyst.
Given the extracted shipment exception, decide:
1. Should we reroute? (yes/no)
2. Priority score (1-10)
3. Recommended action: one sentence.
Respond as JSON with keys: reroute (bool), priority (int), action (str)."""

def classify_impact(exception: ShipmentException):
    payload = exception.model_dump_json()
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": REROUTE_PROMPT},
            {"role": "user", "content": payload},
        ],
    )
    return json.loads(response.choices[0].message.content)

Step 5: Draft communications

Once we know what happened and how bad it is, we draft the actual communication. The model writes the customer email and a short internal checklist for the warehouse team. I use llama-3.3-70b here for clean, general-purpose output.

EMAIL_PROMPT = """You are a logistics communications assistant.
Write a brief customer email about the delay and an internal ops checklist.
Tone: professional, no excuses, include new ETA if known.
Respond as JSON with keys: subject (str), customer_email (str), ops_checklist (list of str)."""

def draft_response(exception: ShipmentException, impact: dict):
    context = f"Exception: {exception.model_dump_json()}\nImpact: {json.dumps(impact)}"
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": EMAIL_PROMPT},
            {"role": "user", "content": context},
        ],
    )
    return json.loads(response.choices[0].message.content)

Step 6: Assemble the agent

Now we wire the three stages into one callable agent. Here is the system prompt that governs the whole pipeline, followed by the orchestration code.

SYSTEM_PROMPT = """You are a shipment exception handler.
Your job is to turn raw carrier notifications into structured incident reports.
Always output valid JSON. Never invent tracking numbers.
If a date is missing, use null. Keep customer emails under 150 words."""

def handle_exception(raw_email: str):
    # Stage 1: extract
    exception = extract_exception(raw_email)
    
    # Stage 2: classify
    impact = classify_impact(exception)
    
    # Stage 3: draft comms
    comms = draft_response(exception, impact)
    
    return {
        "incident": exception.model_dump(),
        "impact": impact,
        "communications": comms,
    }

Run it

Here is a realistic carrier delay message passed through the full pipeline.

RAW_EMAIL = """Subject: Delay Alert - Load #884921
Hi, this is Mike from QuickFreight. Your shipment on PRO 884921-A is running 
about 2 days behind due to a blown tire outside Denver. Original delivery was 
scheduled for 2024-09-15 but we are now looking at 2024-09-17. Items on board: 
SKU-4412, SKU-4413. Let us know if you need anything."""

result = handle_exception(RAW_EMAIL)
print(json.dumps(result, indent=2))

Example output:

{
  "incident": {
    "tracking_id": "884921-A",
    "carrier": "QuickFreight",
    "original_eta": "2024-09-15",
    "updated_eta": "2024-09-17",
    "delay_reason": "Blown tire outside Denver",
    "severity": "medium",
    "affected_skus": ["SKU-4412", "SKU-4413"]
  },
  "impact": {
    "reroute": false,
    "priority": 4,
    "action": "Monitor recovery and confirm updated ETA with driver."
  },
  "communications": {
    "subject": "Delivery Update for Shipment 884921-A",
    "customer_email": "Your shipment 884921-A is delayed by approximately two days...",
    "ops_checklist": [
      "Verify tire replacement receipt",
      "Confirm driver availability for 09-17 delivery"
    ]
  }
}

Wrap up

The pipeline runs three separate inferences, which on Oxlo.ai costs the same per request regardless of how long the raw email is. That makes it practical to feed in entire email threads or attached PDF text without watching token meters spin. Two concrete next steps: wire this function to an email IMAP inbox or carrier webhook so it runs hands-free, and add an Oxlo.ai function-calling tool that POSTs critical incidents directly to a Slack channel.