Optimizing LLM Performance for High Accuracy

I recently built a compliance extraction pipeline for a fintech client where a single hallucinated audit finding could trigger a false regulatory alert. We are going to rebuild that pipeline here. It uses chain-of-thought reasoning, self-consistency voting, and a source-grounded validator to push accuracy as high as possible, all running on Oxlo.ai.

What you'll need

You will need Python 3.10 or newer, the OpenAI SDK (pip install openai), and an Oxlo.ai API key from https://portal.oxlo.ai. Oxlo.ai is fully OpenAI SDK compatible, so the client setup is a one-line base URL change.

Step 1: Scaffold the client and schema

First, initialize the Oxlo.ai client and define the structured schema we want back. I use a raw audit transcript with two known issues and one explicit non-issue to test later.

import json
from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

AUDIT_TEXT = """
During the Q3 network review, the team observed that backup routines for the
postgres-primary cluster were still using the legacy cron method instead of
the required WAL-G streaming pipeline. This was flagged as a critical finding
because RPO targets of 15 minutes were not met during the September 12 outage.
Additionally, three of five tested S3 buckets lacked object-versioning policies,
which violates section 4.2 of the infrastructure policy. No issues were found
with IAM key rotation.
"""

SCHEMA = {
    "findings": [
        {
            "issue": "string",
            "severity": "critical|high|medium|low",
            "evidence_quote": "string",
            "policy_reference": "string or null"
        }
    ]
}

Step 2: Force chain-of-thought reasoning

Before the model emits JSON, I make it write a private reasoning block. This dramatically reduces hallucination because the model has to cite exact text first. Here is the system prompt I use for the extractor.

EXTRACTOR_SYSTEM_PROMPT = """
You are a precise compliance auditor. Read the audit report and extract structured findings.

Rules:
1. First, write a short chain-of-thought reasoning block inside <thinking> tags. List each issue and the exact sentence that supports it.
2. Then output a single JSON object matching the requested schema.
3. Every finding must include an evidence_quote that is an exact substring of the input text. Do not paraphrase.
4. If the text does not mention a policy number, set policy_reference to null.
5. Do not invent findings that are not explicitly stated in the text.
"""

Now wire that prompt into a function that calls Kimi K2.6 on Oxlo.ai. I set a low temperature because we want deterministic reasoning, and I use JSON mode to lock the output shape.

def extract_findings(text: str):
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": EXTRACTOR_SYSTEM_PROMPT},
            {"role": "user", "content": f"Extract findings from this audit report according to the schema {json.dumps(SCHEMA)}:\n\n{text}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )
    raw = response.choices[0].message.content
    # Strip the reasoning block so we are left with pure JSON
    if "</thinking>" in raw:
        raw = raw.split("</thinking>")[-1]
    return json.loads(raw.strip())

Step 3: Add self-consistency sampling

One pass is good, three passes are better. I run the extractor multiple times and feed the results into a judge model. Because Oxlo.ai uses flat per-request pricing, adding two extra samples does not multiply the cost the way token-based billing would. See https://oxlo.ai/pricing for details. The judge is Llama 3.3 70B, which is fast on Oxlo.ai.

JUDGE_SYSTEM_PROMPT = """
You are a strict merge judge. You will receive multiple JSON extractions of the same audit report.
Produce one final JSON object containing only findings that appear in the majority of inputs or are strongly supported by evidence quotes.
If inputs disagree on severity, choose the higher severity.
Output only the merged JSON object.
"""

def merge_extractions(extractions: list, original_text: str):
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": JUDGE_SYSTEM_PROMPT},
            {"role": "user", "content": f"Original text:\n{original_text}\n\nExtractions to merge:\n{json.dumps(extractions, indent=2)}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
    )
    return json.loads(response.choices[0].message.content)

def self_consistent_extract(text: str, samples: int = 3):
    extractions = [extract_findings(text) for _ in range(samples)]
    return merge_extractions(extractions, text)

Step 4: Ground findings with a validator

Even after merging, I want a hard guardrail. I run each finding through a separate validator that checks whether the evidence quote actually exists in the source text. I use Qwen 3 32B here because it is sharp at precise text comparison.

VALIDATOR_SYSTEM_PROMPT = """
You are a source-grounded validator. You will receive an audit finding and the original text.
Respond with a JSON object: {"valid": true, "reason": "string"} or {"valid": false, "reason": "string"}.
Set valid to true only if the evidence_quote is an exact or near-exact substring of the original text and the issue is factually stated.
"""

def validate_finding(finding: dict, original_text: str):
    response = client.chat.completions.create(
        model="qwen-3-32b",
        messages=[
            {"role": "system", "content": VALIDATOR_SYSTEM_PROMPT},
            {"role": "user", "content": f"Original text:\n{original_text}\n\nFinding:\n{json.dumps(finding, indent=2)}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
    )
    return json.loads(response.choices[0].message.content)

def validate_all(merged: dict, original_text: str):
    valid_findings = []
    for f in merged.get("findings", []):
        check = validate_finding(f, original_text)
        if check.get("valid"):
            valid_findings.append(f)
        else:
            print(f"Dropped invalid finding: {f['issue']} | Reason: {check.get('reason')}")
    return {"findings": valid_findings}

Step 5: Assemble the pipeline

Now chain everything together. The pipeline runs extraction, merges the samples, and drops anything that fails the source check.

def run_audit_pipeline(text: str):
    merged = self_consistent_extract(text, samples=3)
    final = validate_all(merged, text)
    return final

if __name__ == "__main__":
    result = run_audit_pipeline(AUDIT_TEXT)
    print(json.dumps(result, indent=2))

Run it

Executing the script gives the following output. Notice that the validator correctly drops the hallucinated IAM key rotation issue because the source text explicitly says no issues were found.

python audit_pipeline.py

Dropped invalid finding: IAM key rotation failure | Reason: text explicitly states no issues were found
{
  "findings": [
    {
      "issue": "Backup routines using legacy cron instead of WAL-G",
      "severity": "critical",
      "evidence_quote": "backup routines for the postgres-primary cluster were still using the legacy cron method instead of the required WAL-G streaming pipeline",
      "policy_reference": null
    },
    {
      "issue": "S3 buckets lacking object-versioning policies",
      "severity": "high",
      "evidence_quote": "three of five tested S3 buckets lacked object-versioning policies",
      "policy_reference": "section 4.2"
    }
  ]
}

Wrap-up

Add a confidence threshold to the validator and route borderline findings to a human review queue. If you need to scale this to hundreds of reports, wrap the pipeline in a FastAPI service and deploy it behind Oxlo.ai, where the flat per-request pricing keeps costs predictable even when you add more self-consistency samples or switch to longer-context models like DeepSeek V4 Flash.