I recently shipped an internal coding assistant that routes queries across multiple models, caches identical requests, and parallelizes tool calls. In this tutorial, we will rebuild that agent so you can drop it into your own stack on top of Oxlo.ai. It is meant for engineering teams who want to shave seconds off every LLM interaction without rewriting their inference stack.
What you'll need
- Python 3.10 or newer
pip install openai- An Oxlo.ai API key from https://portal.oxlo.ai
Step 1: Initialize the client and measure baseline latency
Set up the OpenAI-compatible client pointing at Oxlo.ai and log time-to-first-token via streaming. This baseline shows exactly how much latency we are working with before adding any optimizations.
import time
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def stream_baseline(prompt, model="deepseek-v3.2"):
start = time.perf_counter()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True,
)
first_token_time = None
chunks = []
for chunk in response:
if first_token_time is None and chunk.choices[0].delta.content:
first_token_time = time.perf_counter() - start
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
print(f"Model: {model}")
print(f"Time to first token: {first_token_time:.3f}s")
print(f"Response: {''.join(chunks)[:120]}...")
stream_baseline("Explain Python list comprehensions in one sentence.")
Step 2: Build a query complexity router
We will use Oxlo.ai's fast qwen-3-32b model to classify incoming questions into simple, coding, or deep_reasoning tiers. Mapping each tier to a different model prevents over-provisioning compute on trivial questions.
ROUTER_PROMPT = """You are a query classifier. Respond with exactly one word: simple, coding, or deep_reasoning.
simple: general knowledge or explanations.
coding: debugging, code generation, or review.
deep_reasoning: architecture, algorithms, or multi-step logic."""
def classify_query(user_message):
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": user_message},
],
max_tokens=10,
)
label = response.choices[0].message.content.strip().lower()
return label if label in {"simple", "coding", "deep_reasoning"} else "coding"
MODEL_MAP = {
"simple": "deepseek-v3.2",
"coding": "kimi-k2.6",
"deep_reasoning": "llama-3.3-70b",
}
query = "Design a distributed rate limiter with Redis and handle clock skew."
tier = classify_query(query)
print(f"Router selected: {tier} -> {MODEL_MAP[tier]}")
Step 3: Add request-level caching
Because Oxlo.ai charges a flat rate per request, duplicate prompts are pure overhead. We will add an in-memory cache keyed by a SHA-256 hash of the model and messages to eliminate redundant API calls entirely.
import hashlib
_request_cache = {}
def make_cache_key(model, messages):
payload = f"{model}:{messages}"
return hashlib.sha256(payload.encode()).hexdigest()
def cached_chat(model, messages, max_tokens=1024):
key = make_cache_key(model, tuple(str(m) for m in messages))
if key in _request_cache:
print("[cache hit]")
return _request_cache[key]
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
)
text = response.choices[0].message.content
_request_cache[key] = text
print("[cache miss]")
return text
Step 4: Speculative parallel tool generation
For coding tasks, we often need both an explanation and a code diff. Instead of two sequential round trips, we fire both requests in parallel using a thread pool. Oxlo.ai serves popular models without cold starts, so both streams begin immediately.
from concurrent.futures import ThreadPoolExecutor
def fetch_explanation(query, model):
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Explain the fix concisely."},
{"role": "user", "content": query},
],
)
return response.choices[0].message.content
def fetch_code(query, model):
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Output only the corrected code block."},
{"role": "user", "content": query},
],
)
return response.choices[0].message.content
def parallel_codegen(query, model):
with ThreadPoolExecutor(max_workers=2) as exe:
exp_future = exe.submit(fetch_explanation, query, model)
code_future = exe.submit(fetch_code, query, model)
return exp_future.result(), code_future.result()
explanation, code = parallel_codegen(
"Fix this Python function: def add(a,b): return a + b + 1",
"deepseek-v3.2"
)
print("Explanation:", explanation[:100])
print("Code:", code[:100])
Step 5: Assemble the optimized agent
Now we combine the router, cache, and parallel executor into one class. The system prompt forces structured output so downstream parsers do not break. We also expose a streaming option for the final aggregated response.
SYSTEM_PROMPT = """You are an optimized technical assistant.
1. Analyze the user's programming question.
2. Provide a concise explanation followed by a code block.
3. If debugging, reference common pitfalls.
4. Always wrap code in triple backticks with the language identifier."""
class OptimizedAgent:
def __init__(self):
self.client = client
self.model_map = MODEL_MAP
self.cache = _request_cache
def run(self, user_message, use_cache=True):
tier = classify_query(user_message)
model = self.model_map[tier]
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
cache_key = make_cache_key(model, messages)
if use_cache and cache_key in self.cache:
return {"tier": tier, "model": model, "cached": True, "response": self.cache[cache_key]}
if tier in ("coding", "deep_reasoning"):
explanation, code = parallel_codegen(user_message, model)
full = f"Explanation:\n{explanation}\n\nCode:\n{code}"
else:
response = self.client.chat.completions.create(
model=model,
messages=messages,
)
full = response.choices[0].message.content
self.cache[cache_key] = full
return {"tier": tier, "model": model, "cached": False, "response": full}
agent = OptimizedAgent()
result = agent.run("Write a Python decorator that retries HTTP requests with exponential backoff.")
print(result)
Run it
Pass a mix of simple and complex queries through the agent to watch the router and cache in action.
test_queries = [
"What is a Python dictionary?",
"Debug this recursive function that overflows: def fib(n): return fib(n-1) + fib(n-2)",
"Explain the trade-offs between Kafka and RabbitMQ for event streaming.",
]
for q in test_queries:
out = agent.run(q)
print(f"\nQuery: {q}")
print(f"Tier: {out['tier']} | Model: {out['model']} | Cached: {out['cached']}")
print(f"Response preview: {out['response'][:180]}...")
On the first pass you should see cache misses and the router picking deepseek-v3.2 for the dictionary question, kimi-k2.6 for the debugging task, and llama-3.3-70b for the architecture comparison. Running the same list a second time will return every result instantly from the cache.
Wrap-up
Two concrete next steps. First, persist the cache to Redis so multiple worker nodes share the same request deduplication layer. Second, replace the rule-based router with a few-shot classifier using embeddings from Oxlo.ai's BGE-Large endpoint to improve routing accuracy without adding latency.
Top comments (0)