Teams running LLMs in production need a safe way to test new model releases before cutting over traffic. In this tutorial we will build a versioned inference router that calls multiple Oxlo.ai models, compares their outputs, and recommends a winner. The final script gives you canary deployments and A/B tests without external infrastructure.
What you'll need
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
Step 1: Scaffold the version manifest and generator
First we define a manifest that maps semantic tags to Oxlo.ai model IDs. This keeps application code decoupled from provider-specific model strings so swaps only require a one-line change.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
MANIFEST = {
"stable": "llama-3.3-70b",
"candidate": "qwen-3-32b",
"fallback": "deepseek-v3.2"
}
def generate(tag: str, system_prompt: str, user_message: str, temperature: float = 0.7):
model = MANIFEST.get(tag, MANIFEST["fallback"])
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
temperature=temperature,
)
return {
"tag": tag,
"model": model,
"content": response.choices[0].message.content,
}Step 2: Add automatic failover
If a model is temporarily unavailable, we catch the error and fall back to the fallback tag. Oxlo.ai hosts over 45 models with no cold starts on popular ones, so switching is fast.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
MANIFEST = {
"stable": "llama-3.3-70b",
"candidate": "qwen-3-32b",
"fallback": "deepseek-v3.2"
}
def generate_with_failover(tag: str, system_prompt: str, user_message: str, temperature: float = 0.7):
model = MANIFEST.get(tag, MANIFEST["fallback"])
try:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
temperature=temperature,
)
return {
"tag": tag,
"model": model,
"content": response.choices[0].message.content,
}
except Exception:
if tag == "fallback":
raise
return generate_with_failover("fallback", system_prompt, user_message, temperature)Step 3: Build the side-by-side comparison harness
Before promoting a model from candidate to stable, we run the same prompt against both versions and collect the results in a structured dictionary.
def compare_versions(system_prompt: str, user_message: str):
return {
"stable": generate_with_failover("stable", system_prompt, user_message),
"candidate": generate_with_failover("candidate", system_prompt, user_message),
}Step 4: Define the evaluator agent system prompt
I use a separate judge to score outputs. Because Oxlo.ai uses flat per-request pricing (https://oxlo.ai/pricing), adding this evaluation step stays cheap even when we pass long contexts to the judge.
EVALUATOR_PROMPT = """You are a model-version evaluator. Two responses are provided for the same user prompt.
Task:
1. Score each response from 1 to 10 on accuracy, clarity, and safety.
2. State which response is better and why.
3. Recommend whether the candidate model should replace the stable model.
Respond in JSON with keys: stable_score, candidate_score, winner, reasoning, recommend_promotion."""
def evaluate(stable_output: str, candidate_output: str, user_message: str):
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
payload = (
f"User prompt: {user_message}\n\n"
f"Stable response:\n{stable_output}\n\n"
f"Candidate response:\n{candidate_output}"
)
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": EVALUATOR_PROMPT},
{"role": "user", "content": payload},
],
response_format={"type": "json_object"},
)
return response.choices[0].message.contentStep 5: Wire the canary pipeline
This final helper orchestrates the comparison and the judge into a single verdict. I run the stable and candidate versions through Oxlo.ai, then pass both outputs to the evaluator.
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def run_canary(system_prompt: str, user_message: str):
outputs = compare_versions(system_prompt, user_message)
judge_text = evaluate(
outputs["stable"]["content"],
outputs["candidate"]["content"],
user_message,
)
return {
"stable": outputs["stable"],
"candidate": outputs["candidate"],
"verdict": json.loads(judge_text),
}Run it
Test with a support scenario that mixes policy logic and tone. The script calls Oxlo.ai three times: once for stable, once for candidate, and once for the judge.
SYSTEM_PROMPT = "You are a helpful support agent. Refund requests under $50 are approved automatically. Explain your decision in one sentence."
if __name__ == "__main__":
result = run_canary(
SYSTEM_PROMPT,
"I want a refund for my $29 charging cable. It arrived frayed."
)
print("Stable:", result["stable"]["content"])
print("Candidate:", result["candidate"]["content"])
print("Verdict:", json.dumps(result["verdict"], indent=2))Example output:
Stable: Your refund for the $29 charging cable has been approved because it arrived damaged and falls under our automatic under-$50 refund policy.
Candidate: I've approved your $29 refund since damaged items under $50 qualify for automatic approval.
Verdict: {
"stable_score": 8,
"candidate_score": 9,
"winner": "candidate",
"reasoning": "Both answers are accurate, but the candidate is more concise while preserving all required details.",
"recommend_promotion": true
}Next steps
Hook this script into your CI pipeline so every manifest change triggers a canary test before merge. You can also store results in SQLite and track win rates across a prompt regression suite to automate model promotions.
Top comments (0)