Engineering LLMs for Real-World Applications

#engineering #oxlo #ai

We are building a support ticket triage agent that reads raw customer emails, classifies urgency and category, and either drafts a first reply or escalates to a human. It is a concrete piece of infrastructure you can drop into a cron job or webhook today. The whole thing runs on Oxlo.ai using standard Python and the OpenAI SDK.

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: Set up the Oxlo.ai client

Import the SDK and point it at the Oxlo.ai endpoint. I keep my key in an environment variable in production, but for this tutorial you can paste it directly.

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="llama-3.3-70b",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "ping"},
    ],
)
print(response.choices[0].message.content)

Step 2: Write the system prompt

The system prompt is the contract between you and the model. It forces JSON output so we do not have to guess at formatting. I treat this as a config file I iterate on independently of the code.

SYSTEM_PROMPT = """You are a support ticket triage agent.
Read the customer email and produce a JSON object with exactly these keys:
- urgency: one of "low", "medium", "high", "critical"
- category: one of "billing", "technical", "account", "general"
- summary: a one-sentence summary of the issue
- reply_draft: a polite, concise first reply if urgency is low or medium, otherwise an empty string
- escalate: true if urgency is high or critical, else false

Rules:
- If the user mentions downtime, data loss, or security breach, set urgency to critical and escalate to true.
- Keep reply_draft under 120 words.
- Output ONLY valid JSON. No markdown code fences, no explanations.
"""

Step 3: Call the model and parse the result

Now we wrap the API call in a small function. I use llama-3.3-70b because it follows structured instructions reliably at low temperature. We strip whitespace and parse the JSON.

import json

def triage_ticket(raw_email: str) -> dict:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": raw_email},
        ],
        temperature=0.2,
        max_tokens=512,
    )
    content = response.choices[0].message.content.strip()
    return json.loads(content)

# Test with a single ticket
raw = """Subject: Can't log in
Hey, I forgot my password and the reset link is not arriving. I need to get into my account today for a demo. Please help ASAP."""

result = triage_ticket(raw)
print(json.dumps(result, indent=2))

Step 4: Add routing logic

Parsing is useless without action. This router checks the escalate flag and either prints an escalation notice or returns the draft reply. In production, you would post to Slack or your CRM instead of printing.

def handle_ticket(ticket_id: str, raw_email: str):
    analysis = triage_ticket(raw_email)

    if analysis.get("escalate"):
        print(f"[{ticket_id}] ESCALATE -> {analysis['summary']}")
        return {"action": "escalate", "reason": analysis["summary"]}

    print(f"[{ticket_id}] REPLY -> category: {analysis['category']}")
    return {
        "action": "reply",
        "draft": analysis["reply_draft"],
        "category": analysis["category"],
    }

outcome = handle_ticket("TKT-001", raw)
print(json.dumps(outcome, indent=2))

Step 5: Process a batch

In practice, you will pull unread emails from a mailbox or webhook queue. Here is a small batch that exercises both the reply and escalate paths.

tickets = [
    ("TKT-002", "Subject: Invoice wrong\nI was charged twice this month. Can you fix it before my card is charged again?"),
    ("TKT-003", "Subject: API down\nAll our requests are returning 503 errors. This is blocking our checkout flow and losing us revenue."),
    ("TKT-004", "Subject: Feature request\nIt would be nice to have dark mode in the dashboard. No rush."),
]

for tid, body in tickets:
    try:
        handle_ticket(tid, body)
    except Exception as e:
        print(f"[{tid}] ERROR: {e}")

Run it

Save everything into triage_agent.py, replace YOUR_OXLO_API_KEY, and run the script. Because Oxlo.ai uses flat per-request pricing, ingesting a long email thread with a verbose system prompt costs the same as a short ping. That makes this triage pattern cheap to run at scale compared to token-based billing.

Here is the complete file for copy and paste.

from openai import OpenAI
import json

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

SYSTEM_PROMPT = """You are a support ticket triage agent.
Read the customer email and produce a JSON object with exactly these keys:
- urgency: one of "low", "medium", "high", "critical"
- category: one of "billing", "technical", "account", "general"
- summary: a one-sentence summary of the issue
- reply_draft: a polite, concise first reply if urgency is low or medium, otherwise an empty string
- escalate: true if urgency is high or critical, else false

Rules:
- If the user mentions downtime, data loss, or security breach, set urgency to critical and escalate to true.
- Keep reply_draft under 120 words.
- Output ONLY valid JSON. No markdown code fences, no explanations.
"""

def triage_ticket(raw_email: str) -> dict:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": raw_email},
        ],
        temperature=0.2,
        max_tokens=512,
    )
    content = response.choices[0].message.content.strip()
    return json.loads(content)

def handle_ticket(ticket_id: str, raw_email: str):
    analysis = triage_ticket(raw_email)

    if analysis.get("escalate"):
        print(f"[{ticket_id}] ESCALATE -> {analysis['summary']}")
        return {"action": "escalate", "reason": analysis["summary"]}

    print(f"[{ticket_id}] REPLY -> category: {analysis['category']}")
    return {
        "action": "reply",
        "draft": analysis["reply_draft"],
        "category": analysis["category"],
    }

if __name__ == "__main__":
    tickets = [
        ("TKT-002", "Subject: Invoice wrong\nI was charged twice this month. Can you fix it before my card is charged again?"),
        ("TKT-003", "Subject: API down\nAll our requests are returning 503 errors. This is blocking our checkout flow and losing us revenue."),
        ("TKT-004", "Subject: Feature request\nIt would be nice to have dark mode in the dashboard. No rush."),
    ]

    for tid, body in tickets:
        try:
            handle_ticket(tid, body)
        except Exception as e:
            print(f"[{tid}] ERROR: {e}")

Expected output looks like this.

[TKT-002] REPLY -> category: billing
[TKT-003] ESCALATE -> API returning 503 errors blocking checkout flow
[TKT-004] REPLY -> category: general

Wrap-up

Swap in kimi-k2.6 or deepseek-v3.2 on Oxlo.ai if you need stronger reasoning for complex multi-issue tickets. Next, wire this into a real email webhook or Slack slash command, and add a tiny SQLite cache so you do not reprocess the same thread ID twice.

DEV Community