Most production LLM failures happen on edge cases, not happy paths. In this tutorial, I will build a high-accuracy research agent that answers complex software engineering questions by decomposing queries, generating structured candidates, and voting across multiple inference passes. The result is a workflow you can adapt to any domain where hallucinations are expensive.
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
- A terminal to run the script
Step 1: Pin the system prompt and baseline client
I always start by pinning the system prompt. Accuracy begins at the instruction layer, so I force the model to reason silently before answering and to cite sources. I also point the OpenAI SDK at Oxlo.ai's endpoint.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
SYSTEM_PROMPT = """You are a high-accuracy technical agent. Follow these rules strictly:
1. Analyze the user's question and identify core facts.
2. Reason step by step before answering.
3. Cite Python documentation or language behavior explicitly.
4. If uncertain, say so. Never hallucinate APIs or behavior."""
def ask_baseline(question: str) -> str:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": question},
],
temperature=0.2,
)
return response.choices[0].message.content
question = "When should I use asyncio.gather versus a queue with asyncio.create_task in Python?"
print(ask_baseline(question))
Step 2: Decompose the question into sub-questions
Next, I break the user's question into smaller, verifiable sub-questions. I use Qwen 3 32B with JSON mode because it handles reasoning well and reliably follows output schemas.
import json
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
def decompose(question: str) -> list[str]:
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": "Break the user's question into 1 to 3 specific, verifiable sub-questions. Return a JSON object with key 'sub_questions' containing a list of strings."},
{"role": "user", "content": question},
],
response_format={"type": "json_object"},
temperature=0.2,
)
data = json.loads(response.choices[0].message.content)
return data.get("sub_questions", [question])
question = "When should I use asyncio.gather versus a queue with asyncio.create_task in Python?"
sub_questions = decompose(question)
print(sub_questions)
Step 3: Generate structured candidates with JSON mode
For each sub-question, I generate a structured candidate answer. I use DeepSeek V3.2 with JSON mode to enforce a rigid schema that eliminates meandering prose.
import json
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
def generate_candidate(sub_question: str) -> dict:
schema = {
"reasoning": "string",
"answer": "string",
"confidence": "low|medium|high",
"caveats": "string"
}
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": f"You are a precise technical assistant. Answer the question and respond with a JSON object matching this schema: {json.dumps(schema)}. Be concise and accurate."},
{"role": "user", "content": sub_question},
],
response_format={"type": "json_object"},
temperature=0.6,
)
return json.loads(response.choices[0].message.content)
sub_question = "What are the memory trade-offs between asyncio.gather and manually managed asyncio queues?"
candidate = generate_candidate(sub_question)
print(json.dumps(candidate, indent=2))
Step 4: Vote with self-consistency across multiple passes
A single sample can be unlucky, so I run three independent generations and vote by semantic similarity. Because Oxlo.ai bills per request rather than per token, this multi-pass step stays predictable even when reasoning traces grow long.
import json
import os
from difflib import SequenceMatcher
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
def generate_candidate(sub_question: str) -> dict:
schema = {
"reasoning": "string",
"answer": "string",
"confidence": "low|medium|high",
"caveats": "string"
}
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": f"You are a precise technical assistant. Answer the question and respond with a JSON object matching this schema: {json.dumps(schema)}. Be concise and accurate."},
{"role": "user", "content": sub_question},
],
response_format={"type": "json_object"},
temperature=0.6,
)
return json.loads(response.choices[0].message.content)
def similarity(a: str, b: str) -> float:
return SequenceMatcher(None, a, b).ratio()
def self_consensus(sub_question: str, n: int = 3) -> dict:
candidates = [generate_candidate(sub_question) for _ in range(n)]
best = candidates[0]
best_score = 0
for c in candidates:
score = sum(similarity(c["answer"], other["answer"]) for other in candidates)
if score > best_score:
best_score = score
best = c
return best
sub_question = "What are the memory trade-offs between asyncio.gather and manually managed asyncio queues?"
winner = self_consensus(sub_question, n=3)
print("Winning candidate:", json.dumps(winner, indent=2))
Step 5: Critique and refine with a reasoning specialist
With a winning candidate selected, I feed it to Kimi K2.6 for a critique pass. A specialist reasoning model catches overstatements and missing edge cases that the initial generator missed.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
def critique_and_refine(question: str, candidate: dict) -> str:
critique_prompt = f"""You are a senior Python engineer reviewing a draft answer.
Original question: {question}
Draft reasoning: {candidate['reasoning']}
Draft answer: {candidate['answer']}
Draft caveats: {candidate.get('caveats', 'None')}
Instructions:
1. Identify any factual errors, missing edge cases, or overstatements.
2. Output a refined, final answer that preserves what is correct and fixes what is not.
3. Be concise."""
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": critique_prompt},
{"role": "user", "content": "Produce the final reviewed answer."},
],
temperature=0.2,
)
return response.choices[0].message.content
question = "When should I use asyncio.gather versus a queue with asyncio.create_task in Python?"
draft = {
"reasoning": "asyncio.gather waits for all tasks and returns results in order, while a manual queue allows dynamic producer-consumer patterns.",
"answer": "Use gather for fixed sets of coroutines; use a queue when tasks are produced dynamically or backpressure is needed.",
"caveats": "Gather keeps references to all tasks until done, which can hold memory.",
"confidence": "high"
}
final = critique_and_refine(question, draft)
print(final)
Step 6: Assemble the final response
Finally, I synthesize the verified sub-answers into a coherent response. I use Llama 3.3 70B for assembly because it follows the constraint not to introduce new facts.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
def assemble_response(question: str, sub_questions: list[str], answers: list[str]) -> str:
context = "\n\n".join([f"Q: {sq}\nA: {ans}" for sq, ans in zip(sub_questions, answers)])
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "Synthesize the following verified sub-answers into a single, coherent response to the user's original question. Do not add new facts."},
{"role": "user", "content": f"Original question: {question}\n\nVerified sub-answers:\n{context}"},
],
temperature=0.2,
)
return response.choices[0].message.content
question = "When should I use asyncio.gather versus a queue with asyncio.create_task in Python?"
sub_questions = [
"When is asyncio.gather the appropriate tool in Python?",
"When should I use a manual queue with asyncio.create_task instead?",
"What are the memory trade-offs between asyncio.gather and manually managed asyncio queues?"
]
answers = [
"Use asyncio.gather when you have a fixed, finite set of coroutines to run concurrently and you need all results before proceeding.",
"Use a manual queue with asyncio.create_task when tasks are generated dynamically, when you need producer-consumer backpressure, or when you want to process results as they complete rather than waiting for all of them.",
"asyncio.gather holds references to every task and its result until all complete, which can increase peak memory. A queue with bounded size limits in-flight tasks and can reduce peak memory at the cost of more complex code."
]
print(assemble_response(question, sub_questions, answers))
Run it
Save the full script as accuracy_agent.py, export your OXLO_API_KEY, and run it. The agent prints each step so you can observe the decomposition, consensus, and critique pipeline in action.
import json
import os
from difflib import SequenceMatcher
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
SYSTEM_PROMPT = """You are a high-accuracy technical agent. Follow these rules strictly:
1. Analyze the user's question and identify core facts.
2. Reason step by step before answering.
3. Cite Python documentation or language behavior explicitly.
4. If uncertain, say so. Never hallucinate APIs or behavior."""
def decompose(question: str) -> list[str]:
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": "Break the user's question into 1 to 3 specific, verifiable sub-questions. Return a JSON object with key 'sub_questions' containing a list of strings."},
{"role": "user", "content": question},
],
response_format={"type": "json_object"},
temperature=0.2,
)
data = json.loads(response.choices[0].message.content)
return data.get("sub_questions", [question])
def generate_candidate(sub_question: str) -> dict:
schema = {
"reasoning": "string",
"answer": "string",
"confidence": "low|medium|high",
"caveats": "string"
}
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": f"You are a precise technical assistant. Answer the question and respond with a JSON object matching this schema: {json.dumps(schema)}. Be concise and accurate."},
{"role": "user", "content": sub_question},
],
response_format={"type": "json_object"},
temperature=0.6,
)
return json.loads(response.choices[0].message.content)
def similarity(a: str, b: str) -> float:
return SequenceMatcher(None, a, b).ratio()
def self_consensus(sub_question: str, n: int = 3) -> dict:
candidates = [generate_candidate(sub_question) for _ in range(n)]
best = candidates[0]
best_score = 0
for c in candidates:
score = sum(similarity(c["answer"], other["answer"]) for other in candidates)
if score > best_score:
best_score = score
best = c
return best
def critique_and_refine(question: str, candidate: dict) -> str:
critique_prompt = f"""You are a senior Python engineer reviewing a draft answer.
Original question: {question}
Draft reasoning: {candidate['reasoning']}
Draft answer: {candidate['answer']}
Draft caveats: {candidate.get('caveats', 'None')}
Instructions:
1. Identify any factual errors, missing edge cases, or overstatements.
2. Output a refined, final answer that preserves what is correct and fixes what is not.
3. Be concise."""
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": critique_prompt},
{"role": "user", "content": "Produce the final reviewed answer."},
],
temperature=0.2,
)
return response.choices[0].message.content
def assemble_response(question: str, sub_questions: list[str], answers: list[str]) -> str:
context = "\n\n".join([f"Q: {sq}\nA: {ans}" for sq, ans in zip(sub_questions, answers)])
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "Synthesize the following verified sub-answers into a single, coherent response to the user's original question. Do not add new facts."},
{"role": "user", "content": f"Original question: {question}\n\nVerified sub-answers:\n{context}"},
],
temperature=0.2,
)
return response.choices[0].message.content
def research_agent(question: str) -> str:
print("Step 1: Decomposing question...")
sub_questions = decompose(question)
print(f" Sub-questions: {sub_questions}")
refined_answers = []
for sq in sub_questions:
print(f"Step 2: Self-consensus for: {sq[:60]}...")
winner = self_consensus(sq, n=3)
print("Step 3: Critiquing with Kimi K2.6...")
refined = critique_and_refine(question, winner)
refined_answers.append(refined)
print("Step 4: Assembling final response...")
return assemble_response(question, sub_questions, refined_answers)
if __name__ == "__main__":
question = "When should I use asyncio.gather versus a queue with asyncio.create_task in Python?"
print("\n=== FINAL ANSWER ===\n")
print(research_agent(question))
Example output:
Step 1: Decomposing question...
Sub-questions: ['When is asyncio.gather the appropriate tool in Python?', 'When should I use a manual queue with asyncio.create_task instead?', 'What are the memory trade-offs between asyncio.gather and manually managed asyncio queues?']
Step 2: Self-consensus for: When is asyncio.gather the appropriate tool in Pyt...
Step 3: Critiquing with Kimi K2.6...
Step 2: Self-consensus for: When should I use a manual queue with asyncio.crea...
Step 3: Critiquing with Kimi K2.6...
Step 2: Self-consensus for: What are the memory trade-offs between asyncio.gat...
Step 3: Critiquing with Kimi K2.6...
Step 4: Assembling final response...
=== FINAL ANSWER ===
Use asyncio.gather when you have a fixed collection of coroutines that you want to run concurrently and you need all results before moving on. It is the simplest and most readable tool for this job, but it stores every task reference and result until the last one finishes, which increases peak memory usage.
Use a manual queue paired with asyncio.create_task when work is generated dynamically, when you need backpressure to limit concurrency, or when you want to process results incrementally rather than waiting for the entire batch. This pattern adds code complexity but gives you explicit control over memory and scheduling.
Wrap-up
Wire in a vector database and use Oxlo.ai's BGE-Large embeddings endpoint to ground answers in your internal documentation. Alternatively, cache decomposition results in Redis since sub-questions often repeat across user sessions, saving requests and cutting latency.
Top comments (0)