We are going to build a low-latency customer support triage agent that classifies incoming tickets and drafts first responses in under 300 ms end to end. If you run a SaaS product, this is the difference between a user waiting and a user leaving. We will optimize it step by step on Oxlo.ai, using request-based pricing so that aggressive caching and short prompts do not require a cost rethink.
What you'll need
- Python 3.10+
pip install openai cachetools- An Oxlo.ai API key from https://portal.oxlo.ai
Step 1: Baseline blocking call
I started by writing the dumbest possible version so I had a number to beat. It sends a full ticket to Llama 3.3 70B and waits for the entire response.
import time
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
ticket = "I was charged twice for my subscription this month. Can you refund the duplicate?"
start = time.perf_counter()
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "You are a helpful support agent."},
{"role": "user", "content": ticket},
],
)
latency = time.perf_counter() - start
print(f"Response: {response.choices[0].message.content}")
print(f"Latency: {latency:.3f}s")Step 2: Stream tokens
Waiting for the full JSON blob feels slow even if the total time is identical. I switched to streaming so the user sees the first words immediately, and I measure time to first token separately.
import time
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
ticket = "I was charged twice for my subscription this month. Can you refund the duplicate?"
start = time.perf_counter()
first_token_time = None
chunks = []
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "You are a helpful support agent."},
{"role": "user", "content": ticket},
],
stream=True,
)
for chunk in response:
if first_token_time is None:
first_token_time = time.perf_counter() - start
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
print(f"Time to first token: {first_token_time:.3f}s")
print(f"Total: {time.perf_counter() - start:.3f}s")
print("".join(chunks))Step 3: Compress the system prompt
Long prompts add network serialization time and delay the first token. I rewrote the system prompt to be dense, then locked it in a constant.
import time
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a support triage agent. Rules:
1. Classify the ticket into Refund, Bug, or Account.
2. Draft a 1-sentence first response.
3. Output only valid JSON with keys: category, reply."""
ticket = "I was charged twice for my subscription this month. Can you refund the duplicate?"
start = time.perf_counter()
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": ticket},
],
)
latency = time.perf_counter() - start
print(f"Latency: {latency:.3f}s")
print(response.choices[0].message.content)Step 4: Cache identical requests
Because Oxlo.ai uses request-based pricing, a cache hit saves both money and latency. I added a 128-slot LRU cache keyed by the raw user message.
import time
from functools import lru_cache
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a support triage agent. Rules:
1. Classify the ticket into Refund, Bug, or Account.
2. Draft a 1-sentence first response.
3. Output only valid JSON with keys: category, reply."""
@lru_cache(maxsize=128)
def cached_triage(ticket_text: str):
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": ticket_text},
],
)
return response.choices[0].message.content
ticket = "I was charged twice for my subscription this month. Can you refund the duplicate?"
start = time.perf_counter()
result = cached_triage(ticket)
print(f"Cache miss: {time.perf_counter() - start:.3f}s")
print(result)
start = time.perf_counter()
result = cached_triage(ticket)
print(f"Cache hit: {time.perf_counter() - start:.3f}s")
print(result)Step 5: Route simple tickets to a faster model
Not every ticket needs a 70B parameter model. I added a lightweight router that sends refund and password requests to DeepSeek V3.2, and escalates bug reports to Llama 3.3 70B.
import time
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a support triage agent. Rules:
1. Classify the ticket into Refund, Bug, or Account.
2. Draft a 1-sentence first response.
3. Output only valid JSON with keys: category, reply."""
def route_and_run(ticket: str):
lowered = ticket.lower()
if any(word in lowered for word in ["refund", "charge", "payment", "password", "login"]):
model = "deepseek-v3.2"
else:
model = "llama-3.3-70b"
start = time.perf_counter()
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": ticket},
],
)
latency = time.perf_counter() - start
return model, latency, response.choices[0].message.content
tickets = [
"I forgot my password and cannot log in.",
"The export button throws a 500 error when I click it.",
]
for t in tickets:
model, latency, result = route_and_run(t)
print(f"Model: {model} | Latency: {latency:.3f}s")
print(result)
print()Run it
I tied everything together in a single script that checks the cache, picks the right model, and returns structured JSON. Here is the full agent.
import time
from functools import lru_cache
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a support triage agent. Rules:
1. Classify the ticket into Refund, Bug, or Account.
2. Draft a 1-sentence first response.
3. Output only valid JSON with keys: category, reply."""
@lru_cache(maxsize=128)
def cached_triage(ticket_text: str, model: str):
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": ticket_text},
],
)
return response.choices[0].message.content
def route_model(ticket: str) -> str:
lowered = ticket.lower()
if any(word in lowered for word in ["refund", "charge", "payment", "password", "login"]):
return "deepseek-v3.2"
return "llama-3.3-70b"
def main():
tickets = [
"I was charged twice this month. Help?",
"I was charged twice this month. Help?",
"The dashboard export button crashes every time I press it.",
]
for t in tickets:
model = route_model(t)
start = time.perf_counter()
result = cached_triage(t, model)
elapsed = time.perf_counter() - start
print(f"Ticket: {t}")
print(f"Model: {model}")
print(f"Result: {result}")
print()
if __name__ == "__main__":
main()Example output:
Ticket: I was charged twice this month. Help?
Model: deepseek-v3.2
Result: {"category": "Refund", "reply": "We have issued a refund for the duplicate charge; it should appear in 3-5 business days."}
Ticket: I was charged twice this month. Help?
Model: deepseek-v3.2
Result: {"category": "Refund", "reply": "We have issued a refund for the duplicate charge; it should appear in 3-5 business days."}
Ticket: The dashboard export button crashes every time I press it.
Model: llama-3.3-70b
Result: {"category": "Bug", "reply": "Thanks for reporting this. Our engineering team is investigating the 500 error on exports and will update you within 2 hours."}Next steps
Try wiring the agent to Oxlo.ai's vision endpoint so it can read screenshot attachments directly, or switch the cache to Redis so multiple workers share hits across your fleet. Both take about ten minutes and compound the latency gains.
Top comments (0)