Most coding agents fail on complex tasks because they jump straight to implementation without planning. In this tutorial, I will walk you through building a multi-step coding agent on Oxlo.ai that analyzes requirements, generates a design plan, writes Python code, and reviews its own output for logic errors. The finished pipeline handles concurrency, error handling, and edge cases better than a single-shot prompt.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK installed with
pip install openai
Step 1: Initialize the Oxlo.ai client
I always verify the endpoint and my API key before building logic. This snippet connects to Oxlo.ai and runs a quick sanity check with Llama 3.3 70B.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[{"role": "user", "content": "Say hello"}],
)
print("Oxlo.ai connection OK:", response.choices[0].message.content)Step 2: Define the system prompt
A strong system prompt keeps the model focused on architecture and edge cases rather than superficial syntax. Store this in a constant so you can iterate quickly.
SYSTEM_PROMPT = """You are a principal software engineer who writes production-grade Python.
Follow these rules on every task:
1. Analyze requirements for concurrency, error handling, and edge cases before writing code.
2. Use type hints, docstrings, and the standard library unless there is a compelling reason to import third-party packages.
3. Prefer composition over inheritance. Keep functions small and single-purpose.
4. For concurrent code, explicitly state which primitives protect shared state.
5. After generating code, list any assumptions or potential risks."""Step 3: Decompose the task into a plan
Before writing code, the agent needs to reason about concurrency, data structures, and failure modes. I use Kimi K2.6 here because its long context window handles detailed requirements without truncation.
def plan_task(client, requirements):
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Create a detailed implementation plan for the following requirements. Do not write code yet. Outline modules, classes, and concurrency approach.\n\nRequirements:\n{requirements}"},
],
temperature=0.2,
)
return response.choices[0].message.content
requirements = (
"Implement a thread-safe priority task queue in Python. "
"It must support rate limiting per worker, exponential backoff for failed tasks, "
"and graceful shutdown that waits for in-flight tasks. Include unit tests."
)
plan = plan_task(client, requirements)
print("=== PLAN ===")
print(plan)Step 4: Generate the implementation
With a plan in hand, we switch to DeepSeek V3.2, which is tuned for coding and reasoning. Passing the plan into the user message keeps the model grounded in the design.
def generate_code(client, plan):
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Implementation plan:\n{plan}\n\nNow write the complete Python implementation. Include all classes, methods, and unit tests in a single file."},
],
temperature=0.2,
)
return response.choices[0].message.content
code = generate_code(client, plan)
print("=== CODE ===")
print(code)Step 5: Self-review for bugs and race conditions
Complex code needs a second look. I send the generated code back to Kimi K2.6 alongside the original plan and ask it to flag race conditions, missing error handling, and type safety issues.
def review_code(client, plan, code):
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Implementation plan:\n{plan}\n\nGenerated code:\n{code}\n\nReview the code. Flag any race conditions, missing error handling, type safety issues, or deviations from the plan. Be specific."},
],
temperature=0.2,
)
return response.choices[0].message.content
review = review_code(client, plan, code)
print("=== REVIEW ===")
print(review)Step 6: Assemble the complete agent
Now wire the three stages into a single pipeline. The final script reads a task description, runs the planner, coder, and reviewer, then prints the final code and critique.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)
SYSTEM_PROMPT = """You are a principal software engineer who writes production-grade Python.
Follow these rules on every task:
1. Analyze requirements for concurrency, error handling, and edge cases before writing code.
2. Use type hints, docstrings, and the standard library unless there is a compelling reason to import third-party packages.
3. Prefer composition over inheritance. Keep functions small and single-purpose.
4. For concurrent code, explicitly state which primitives protect shared state.
5. After generating code, list any assumptions or potential risks."""
def plan_task(client, requirements):
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Create a detailed implementation plan for the following requirements. Do not write code yet. Outline modules, classes, and concurrency approach.\n\nRequirements:\n{requirements}"},
],
temperature=0.2,
)
return response.choices[0].message.content
def generate_code(client, plan):
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Implementation plan:\n{plan}\n\nNow write the complete Python implementation. Include all classes, methods, and unit tests in a single file."},
],
temperature=0.2,
)
return response.choices[0].message.content
def review_code(client, plan, code):
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Implementation plan:\n{plan}\n\nGenerated code:\n{code}\n\nReview the code. Flag any race conditions, missing error handling, type safety issues, or deviations from the plan. Be specific."},
],
temperature=0.2,
)
return response.choices[0].message.content
def main():
requirements = (
"Implement a thread-safe priority task queue in Python. "
"It must support rate limiting per worker, exponential backoff for failed tasks, "
"and graceful shutdown that waits for in-flight tasks. Include unit tests."
)
print("Planning...")
plan = plan_task(client, requirements)
print("Coding...")
code = generate_code(client, plan)
print("Reviewing...")
review = review_code(client, plan, code)
print("\n=== FINAL CODE ===\n")
print(code)
print("\n=== REVIEW ===\n")
print(review)
if __name__ == "__main__":
main()Run it
Save the script as complex_coder.py, set your API key, and run it with a challenging prompt. Here is the exact command and the output I received when I asked for a thread-safe priority queue with rate limiting.
export OXLO_API_KEY="sk-oxlo.ai-..."
python complex_coder.pyPlanning...
Coding...
Reviewing...
=== FINAL CODE ===
import threading
import time
import unittest
from queue import PriorityQueue, Empty
from typing import Callable, Any, Optional
class Task:
"""Represents a unit of work with priority and retry state."""
def __init__(
self,
fn: Callable,
args: tuple = (),
kwargs: Optional[dict] = None,
priority: int = 5,
):
self.fn = fn
self.args = args
self.kwargs = kwargs or {}
self.priority = priority
self.attempts = 0
self.lock = threading.Lock()
def run(self) -> Any:
return self.fn(*self.args, **self.kwargs)
def __lt__(self, other: "Task") -> bool:
return self.priority < other.priority
class PriorityTaskQueue:
"""Thread-safe priority queue with rate limiting and graceful shutdown."""
def __init__(self, max_workers: int = 4, rate_limit: float = 10.0):
self.max_workers = max_workers
self.rate_limit = rate_limit
self._queue: PriorityQueue[Task] = PriorityQueue()
self._workers: list[threading.Thread] = []
self._shutdown = False
self._active_tasks = 0
self._lock = threading.Lock()
self._condition = threading.Condition(self._lock)
def submit(self, task: Task) -> None:
with self._lock:
if self._shutdown:
raise RuntimeError("Queue is shut down")
self._queue.put(task)
def _worker_loop(self) -> None:
while True:
with self._lock:
if self._shutdown and self._queue.empty() and self._active_tasks == 0:
break
try:
task = self._queue.get(timeout=0.5)
except Empty:
continue
with self._lock:
self._active_tasks += 1
try:
if task.attempts > 0:
backoff = min(2 ** task.attempts, 60)
time.sleep(backoff)
task.run()
except Exception:
with self._lock:
task.attempts += 1
if task.attempts < 3:
self._queue.put(task)
finally:
with self._lock:
self._active_tasks -= 1
self._condition.notify_all()
time.sleep(1.0 / self.rate_limit)
def start(self) -> None:
for _ in range(self.max_workers):
t = threading.Thread(target=self._worker_loop, daemon=True)
t.start()
self._workers.append(t)
def shutdown(self, wait: bool = True) -> None:
with self._lock:
self._shutdown = True
if wait:
for t in self._workers:
t.join()
class TestPriorityTaskQueue(unittest.TestCase):
def test_basic_submit_and_run(self) -> None:
q = PriorityTaskQueue(max_workers=1, rate_limit=100.0)
results: list[int] = []
q.submit(Task(fn=lambda: results.append(1), priority=1))
q.start()
time.sleep(0.2)
q.shutdown()
self.assertEqual(results, [1])
if __name__ == "__main__":
unittest.main()
=== REVIEW ===
1. Race condition: task.attempts is incremented inside self._lock in the exception handler, but Task.attempts is also read in _worker_loop outside the queue lock. Use task.lock when reading task.attempts in the worker.
2. Graceful shutdown: The shutdown logic waits for the queue to empty, but if a task is requeued after failure, shutdown may hang until backoff expires. Consider a shutdown timeout.
3. Type safety: PriorityQueue requires a total ordering. If priorities are equal, Python falls back to comparing the Task objects themselves. If two tasks have the same priority and different callables, this can raise a TypeError. Add a tie-breaker such as a monotonic sequence number.Wrap-up
Two concrete next steps. First, add a file writer so the agent emits code to disk, then run pytest on the output and feed any failures back into the agent for a fix pass. Second, try swapping in qwen-3-32b for the planning step or llama-3.3-70b for the review step on Oxlo.ai. Because Oxlo.ai uses request-based pricing, you can send long system prompts and extensive context without watching input tokens drive up cost. See https://oxlo.ai/pricing for details.
Top comments (0)