I built this exception triage agent for a freight forwarder that was manually copying carrier delay emails into spreadsheets. It reads raw unstructured notifications, extracts structured shipment data, classifies severity, and drafts both internal alerts and customer emails. If you manage logistics operations, this cuts response time from minutes to seconds.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Define the schema and system prompt
I always start by locking down the data model. Logistics data is messy, so I use Pydantic to enforce types and embed the schema directly into the system prompt. This guarantees the LLM knows exactly what fields to return.
import json
from pydantic import BaseModel, Field
from typing import Literal
class ShipmentException(BaseModel):
tracking_number: str = Field(description="The tracking or BOL number")
carrier: str = Field(description="Carrier name, e.g., FedEx, Maersk, DHL")
exception_type: str = Field(description="Category of exception")
estimated_delay_days: int = Field(description="Best estimate of additional transit days")
root_cause: str = Field(description="Underlying reason if stated")
affected_location: str = Field(description="Port, hub, or city affected")
severity: Literal["low", "medium", "high", "critical"] = Field(description="Business impact")
customer_facing_summary: str = Field(description="One sentence explanation for the customer")
schema_json = json.dumps(ShipmentException.model_json_schema(), indent=2)
SYSTEM_PROMPT = f"""You are a logistics exception parser. Your job is to read raw carrier or freight notifications and extract structured data.
Rules:
- Always respond with a single JSON object matching the provided schema.
- If a field is missing from the text, use your best judgment or set estimated_delay_days to 0.
- severity should reflect business impact: customs holds are high, weather delays at sea are medium, missed last-mile scans are low.
- customer_facing_summary must be polite, factual, and under 25 words.
Schema:
{schema_json}
"""Step 2: Extract structured data with Oxlo.ai
Next, I wire the prompt into an Oxlo.ai client call. I use llama-3.3-70b because it handles structured extraction reliably, and I set response_format to json_object to eliminate parsing drift. Oxlo.ai's request-based pricing means I do not need to count input tokens for these long carrier emails, which keeps costs predictable. See https://oxlo.ai/pricing for plan details.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def parse_exception(raw_text: str) -> dict:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": raw_text},
],
response_format={"type": "json_object"},
temperature=0.1,
)
content = response.choices[0].message.content
return json.loads(content)Step 3: Draft internal and customer alerts
Extracting fields is only half the battle. I want the agent to produce actionable outputs, so I add a severity router that drafts an internal ops alert and a customer-facing email based on the parsed exception.
def draft_notifications(parsed: dict) -> dict:
severity = parsed.get("severity", "medium")
summary = parsed.get("customer_facing_summary", "")
delay = parsed.get("estimated_delay_days", 0)
if severity == "critical":
tone = "Urgent executive escalation"
elif severity == "high":
tone = "Priority ops alert"
else:
tone = "Standard update"
internal = (
f"[{tone}] Shipment {parsed['tracking_number']} delayed {delay} days "
f"due to {parsed['root_cause']} at {parsed['affected_location']}."
)
external = (
f"Hi, we are writing to inform you that your shipment "
f"({parsed['tracking_number']}) is experiencing a delay. "
f"{summary} We now expect delivery in approximately {delay} additional business days. "
f"We apologize for the inconvenience."
)
return {
"internal_alert": internal,
"customer_email_draft": external,
"severity": severity,
"parsed": parsed,
}Step 4: Wire the pipeline together
Finally, I package everything into a single script that accepts raw notification text and prints a complete triage package. This is the version I actually run from a cron job that polls a carrier inbox.
if __name__ == "__main__":
raw_notification = (
"Subject: Exception Alert - Container MSCU1234567\n\n"
"Dear Partner,\n\n"
"Vessel MV Atlantic is delayed at Port of Rotterdam due to severe weather conditions.\n"
"New ETA at Felixstowe is 15 October. Original ETA was 12 October.\n"
"BOL: MSCU1234567\n"
"Carrier: Maersk Line\n\n"
"Regards,\n"
"Operations Desk"
)
parsed = parse_exception(raw_notification)
result = draft_notifications(parsed)
print(json.dumps(result, indent=2))Run it
Save the script as logistics_agent.py, export your OXLO_API_KEY, and run it. The sample carrier delay notice above produces output like this:
{
"internal_alert": "[Priority ops alert] Shipment MSCU1234567 delayed 3 days due to severe weather conditions at Port of Rotterdam.",
"customer_email_draft": "Hi, we are writing to inform you that your shipment (MSCU1234567) is experiencing a delay. Your shipment is delayed because of severe weather at the port. We now expect delivery in approximately 3 additional business days. We apologize for the inconvenience.",
"severity": "high",
"parsed": {
"tracking_number": "MSCU1234567",
"carrier": "Maersk Line",
"exception_type": "Vessel Delay",
"estimated_delay_days": 3,
"root_cause": "severe weather conditions",
"affected_location": "Port of Rotterdam",
"severity": "high",
"customer_facing_summary": "Your shipment is delayed because of severe weather at the port."
}
}Next steps
Replace the hardcoded sample with an IMAP or webhook listener that feeds real carrier emails directly into the agent. For international lanes, swap the model to qwen-3-32b on Oxlo.ai and add a second pass to translate customer emails into the recipient's local language.
Top comments (0)