Building a Deep Reasoning System

#engineering #oxlo #ai

We are going to build a deep reasoning research agent that decomposes complex technical questions into explicit sub-claims, critiques its own logic, and returns a structured, auditable answer. This is useful for engineering teams evaluating architecture trade-offs, reviewing incident retrospectives, or onboarding onto unfamiliar systems without reading hundreds of pages of documentation.

What you'll need

Python 3.10 or newer
The OpenAI SDK: pip install openai pydantic
An Oxlo.ai API key from https://portal.oxlo.ai

I recommend exporting the key as an environment variable so it never touches disk.

Step 1: Set Up the Oxlo.ai Client

I keep client initialization in one place so I never hardcode URLs in business logic. Oxlo.ai exposes a fully OpenAI-compatible endpoint, so the drop-in client is all we need.

from openai import OpenAI
import os

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ.get("OXLO_API_KEY")
)

Step 2: Write the System Prompt

The system prompt is the most important tunable in this stack. It forces the model to externalize its chain of thought before concluding, which makes debugging reasoning failures much easier. I also lock the output to JSON so downstream code never has to parse prose.

SYSTEM_PROMPT = """You are a deep reasoning engine. Your job is to answer complex technical questions through explicit, verifiable reasoning.

Follow these rules:
1. Break the user's question into 2 to 5 sub-questions or claims.
2. For each sub-question, provide a short analysis based on first principles.
3. If you lack certainty, state your confidence level and assumptions.
4. Before concluding, run a quick sanity check on your own logic.
5. Return your entire reasoning as a JSON object with two keys: "steps" (a list of strings) and "final_answer" (a string).

Be concise. Avoid speculation beyond what the reasoning supports."""

Step 3: Enforce Structure with Pydantic

I use Pydantic to define the contract between the LLM and the rest of the application. Oxlo.ai supports JSON mode, so passing response_format={"type": "json_object"} guarantees valid JSON that matches this schema.

import json
from typing import List
from pydantic import BaseModel, Field

class ReasoningOutput(BaseModel):
    steps: List[str] = Field(description="Chain of thought steps")
    final_answer: str = Field(description="Synthesized final answer")

def generate_reasoning(question: str) -> ReasoningOutput:
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )
    content = response.choices[0].message.content
    return ReasoningOutput(**json.loads(content))

Step 4: Add a Self-Critique Loop

Raw chain-of-thought helps, but a second pass catches lazy generalizations. I feed the draft output back to the same model with a reviewer identity and ask it to tighten the logic. Because Oxlo.ai uses flat per-request pricing, this extra round trip does not scale with token count, which keeps multi-step reasoning affordable.

CRITIQUE_PROMPT = """You are a logic reviewer. Review the following reasoning steps and final answer.

Identify any logical gaps, unstated assumptions, or alternative interpretations. Then produce an improved JSON object with the same schema: "steps" and "final_answer". Preserve only sound reasoning."""

def critique_and_refine(draft: ReasoningOutput) -> ReasoningOutput:
    payload = json.dumps(draft.model_dump(), indent=2)
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": CRITIQUE_PROMPT},
            {"role": "user", "content": payload},
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
    )
    data = json.loads(response.choices[0].message.content)
    return ReasoningOutput(**data)

Step 5: Wire the Full Pipeline

Finally, I connect both stages into a single function that prints intermediate output so I can trace the reasoning as it unfolds.

def deep_reason(question: str) -> str:
    print(f"Question: {question}\n")
    
    draft = generate_reasoning(question)
    print("Initial reasoning:")
    for step in draft.steps:
        print(f"  - {step}")
    print()
    
    refined = critique_and_refine(draft)
    print("Refined reasoning:")
    for step in refined.steps:
        print(f"  - {step}")
    print()
    
    return refined.final_answer

Run It

Save the complete script as reasoning_agent.py, set your environment variable, and run it.

export OXLO_API_KEY="sk-oxlo.ai-..."
python reasoning_agent.py

Here is the entry point I use for testing:

if __name__ == "__main__":
    question = (
        "I am building a distributed task queue. "
        "Should I use at-least-once delivery with idempotent workers, "
        "or exactly-once delivery with stronger coordination? "
        "Consider latency, operational complexity, and failure modes."
    )
    answer = deep_reason(question)
    print("Final Answer:")
    print(answer)

Example output:

Question: I am building a distributed task queue. Should I use at-least-once delivery with idempotent workers, or exactly-once delivery with stronger coordination? Consider latency, operational complexity, and failure modes.

Initial reasoning:
  - Sub-question 1: What are the latency implications of each model?
  - Sub-question 2: How do failure modes differ under network partitions?
  - Sub-question 3: What operational overhead does idempotency require versus distributed transactions?
  - Sanity check: At-least-once plus idempotency is the default in most large-scale systems, which suggests it is the simpler path.

Refined reasoning:
  - Latency: At-least-once requires only an ack, while exactly-once needs a consensus round or deduplication store lookup.
  - Failure modes: Exactly-once systems can stall if the coordinator fails; at-least-once systems continue but risk duplicate work.
  - Operational complexity: Idempotent workers push complexity to application code, which teams already own, rather than to infrastructure.
  - Sanity check: The claimed latency advantage holds only if the deduplication store is hot; otherwise it is a wash.

Final Answer:
Choose at-least-once delivery with idempotent workers unless you have a strict regulatory requirement for exactly-once semantics. The operational surface area is smaller, recovery from partitions is automatic, and latency remains predictable because you avoid distributed coordination on every enqueue.

Next Steps

I would extend this in two directions. First, add a retrieval step that feeds internal wiki pages or API docs into the context so the agent grounds its reasoning in your own systems instead of generic knowledge. Second, expose the intermediate steps through a streaming endpoint so a frontend can render the chain of thought as it arrives. Both are straightforward because Oxlo.ai supports streaming, function calling, and long context windows out of the box.