Building a Self-Correcting AI Agent with Reflection Loops in Python

Deploying Large Language Models (LLMs) into production workflows reveals a fundamental challenge: language models produce incorrect answers. They do not do this occasionally; they do it regularly. Whether it is a subtle logical hallucination, a malformed JSON structure, or a violation of business constraints, probabilistic models are inherently prone to drift. When automating critical tasks, relying on a single raw LLM call is a recipe for runtime failures.

To build production-grade systems, developers must bridge the gap between probabilistic model outputs and deterministic software expectations. This is where agentic design patterns come in. Among the most practical and lightweight of these patterns is the Reflection Loop. A reflection loop is a control flow pattern where an agent runs a task, evaluates the output against a specific quality bar, and decides whether to retry with constructive critique.

When building these loops, developers often use the n1n.ai API gateway to access high-speed, stable endpoints for leading models like Claude 3.5 Sonnet, OpenAI o3, and DeepSeek-V3. This unified access allows developers to build multi-model reflection pipelines where cheap models generate initial drafts and highly capable reasoning models perform the critique.

The Core Mechanics of Reflection

At its heart, a reflection loop operates on a simple cognitive asymmetry: critique is computationally and semantically easier than generation.

When an LLM is asked to generate a complex JSON structure that adheres to a strict schema while simultaneously analyzing a raw text document, it must allocate its attention across multiple tasks: parsing the input, reasoning over the facts, structuring the JSON, and maintaining syntax validity. It is highly common for the model to miss edge cases or drop required fields during this single-pass execution.

However, if you present the same LLM with its own output and ask, "What is wrong with this JSON based on the following schema?", the model's attention is focused entirely on evaluation. This asymmetric capability makes reflection loops highly effective. The model can easily spot errors in a second pass that it completely missed in the first.

A standard reflection loop consists of three main components:

1. The Generator: The LLM prompt responsible for executing the primary task.
2. The Validator: A deterministic or semantic checker that evaluates the generator's output.
3. The Critiquer: An LLM call (often using a different, more analytical prompt or model) that explains why the validation failed and provides actionable feedback for the next generation attempt.

Implementing a Reflection Agent in Python

Let's build a clean, reusable reflection agent in Python. We will write a class that wraps our API calls and orchestrates the generation, validation, and critique phases.

To ensure maximum flexibility and avoid vendor lock-in, we will utilize the unified API from n1n.ai to route our requests. This allows us to easily switch between providers and benchmark different models.

First, let's look at the core implementation of the agent:

import json
from typing import Any, Callable
import httpx

MAX_ITERATIONS = 4

def call_llm(prompt: str, system: str = "") -> str:
    # In production, route through a unified aggregator like n1n.ai
    # to access Claude, DeepSeek, or OpenAI models via a single API key
    response = httpx.post(
        "https://api.example-llm.com/v1/messages",
        headers={"x-api-key": "YOUR_KEY", "Content-Type": "application/json"},
        json={
            "model": "your-model",
            "max_tokens": 1024,
            "system": system,
            "messages": [{"role": "user", "content": prompt}],
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["content"][0]["text"]

class ReflectionAgent:
    def __init__(
        self,
        task_prompt: str,
        critique_prompt: str,
        validator: Callable[[str], tuple[bool, str]],
    ):
        self.task_prompt = task_prompt
        self.critique_prompt = critique_prompt
        self.validator = validator

    def run(self, user_input: str) -> dict[str, Any]:
        history: list[dict] = []
        attempt = 0

        while attempt < MAX_ITERATIONS:
            attempt += 1
            context = f"User input: {user_input}"
            if history:
                last = history[-1]
                context += (
                    f"\n\nPrevious attempt:\n{last['output']}"
                    f"\nCritique:\n{last['critique']}"
                )

            # 1. Generate step
            output = call_llm(f"{self.task_prompt}\n\n{context}")

            # 2. Deterministic validation step
            ok, critique = self.validator(output)

            history.append(
                {"attempt": attempt, "output": output, "critique": critique, "passed": ok}
            )

            if ok:
                return {
                    "success": True,
                    "output": output,
                    "iterations": attempt,
                    "history": history
                }

            # 3. Semantic critique step (only run if validator fails)
            llm_critique = call_llm(
                f"{self.critique_prompt}\n\nOutput to critique:\n{output}",
                system="Be specific about what is wrong. Do not repeat the corrected version.",
            )
            history[-1]["critique"] = llm_critique

        return {
            "success": False,
            "output": history[-1]["output"],
            "iterations": attempt,
            "history": history
        }

How the Agent State Flow Works

1. State Tracking: The history list stores the trace of every attempt, including the raw output, the validation result, and the LLM critique.
2. Context Accumulation: In subsequent attempts, the agent appends the previous output and the critique to the prompt. This forces the model to acknowledge its previous mistakes and focus on correcting them.
3. Early Exit: The loop exits immediately as soon as the validator returns True, preventing unnecessary token usage and latency.

Integrating Deterministic Schema Validation

Malformed JSON is the most common failure mode in structured LLM output tasks. While you can ask an LLM to critique its own JSON formatting, doing so is highly inefficient. A deterministic validator (like a JSON parser or a schema validator) is faster, 100% reliable, and costs zero tokens.

We can use the jsonschema library to enforce strict structure. By isolating the validator from the LLM logic, we can unit-test it independently.

import jsonschema

EXPECTED_SCHEMA = {
    "type": "object",
    "required": ["summary", "severity", "cve_ids"],
    "properties": {
        "summary": {"type": "string", "minLength": 10},
        "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
        "cve_ids": {
            "type": "array",
            "items": {"type": "string", "pattern": "^CVE-\\d{4}-\\d+$"},
        },
    },
}

def validate_security_report(text: str) -> tuple[bool, str]:
    clean = text.strip()
    # Strip markdown code blocks if the LLM wrapped the JSON
    if clean.startswith("```json"):
        clean = "\n".join(clean.split("\n")[1:])
    if clean.startswith("```"):
        clean = "\n".join(clean.split("\n")[1:])
    if clean.endswith("```"):
        clean = "\n".join(clean.split("\n")[:-1])
    clean = clean.strip()

    try:
        data = json.loads(clean)
    except json.JSONDecodeError as exc:
        return False, f"Invalid JSON: \{exc\}"

    try:
        jsonschema.validate(data, EXPECTED_SCHEMA)
    except jsonschema.ValidationError as exc:
        return False, f"Schema violation: \{exc.message\}"

    return True, "OK"

Here is how we wire the validator into our ReflectionAgent:

agent = ReflectionAgent(
    task_prompt=(
        "Analyze the following security advisory and return a JSON object "
        "with keys 'summary' (string), 'severity' (low/medium/high/critical), "
        "and 'cve_ids' (array of CVE strings). Return only the JSON, no prose."
    ),
    critique_prompt=(
        "Review this JSON output for accuracy, completeness, and schema compliance. "
        "List each specific violation and explain why it fails."
    ),
    validator=validate_security_report,
)

result = agent.run(
    "CVE-2024-3094: backdoor found in XZ Utils 5.6.0 and 5.6.1 affecting liblzma, "
    "allowing remote code execution on affected Linux distributions."
)
print(json.dumps(result, indent=2))

By placing the deterministic schema check before the LLM critique, we catch basic syntax errors immediately. If the JSON is invalid, the error message returned to the generator is precise (e.g., Invalid JSON: Expecting ',' delimiter: line 4 column 5). The LLM critique step is reserved for semantic problems that a schema cannot express, such as verifying if the extracted CVE IDs actually match the text.

Production Guardrails: Token Budgets and Logging

Running reflection loops in production without strict boundaries will lead to runaway latency and API bills. If a model gets stuck in an error loop, it can consume thousands of tokens in seconds.

To prevent this, you must enforce three operational rules:

1. Hard Iteration Limits

Never set the maximum iteration limit (MAX_ITERATIONS) higher than 4 or 5. If an LLM cannot correct its output within four iterations, the issue is almost certainly a fundamental flaw in the prompt, the model's reasoning capacity, or the validator's rules. Continuing the loop will only waste tokens.

2. Token and Cost Budgets

Implement a budget guardrail that calculates the running cost of the loop. If the accumulated cost exceeds a specific threshold, the agent must abort and raise an exception.

COST_PER_1K_TOKENS = 0.003  # Adjust based on model pricing

class BudgetedReflectionAgent(ReflectionAgent):
    def __init__(self, *args, max_cost_usd: float = 0.10, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_cost_usd = max_cost_usd
        self._total_cost = 0.0

    def _check_budget(self, prompt: str) -> None:
        # Simple heuristic for token estimation (1 word ≈ 1.3 tokens)
        # In production, read actual usage tokens from the API response headers/body
        estimated_tokens = len(prompt.split()) * 1.3
        projected_cost = (estimated_tokens / 1000) * COST_PER_1K_TOKENS

        if self._total_cost + projected_cost > self.max_cost_usd:
            raise RuntimeError(
                f"Budget exceeded: $\{self._total_cost:.4f\} spent, "
                f"limit $\{self.max_cost_usd\}"
            )
        self._total_cost += projected_cost

3. Structured Logging

Log every single iteration to a centralized database (like SQLite or PostgreSQL). This allows you to track failure rates, identify which validations fail most frequently, and optimize your prompts accordingly.

Here is a simple SQLite logging schema for reflection runs:

CREATE TABLE reflection_logs (
    task_id TEXT,
    attempt INTEGER,
    model TEXT,
    input_tokens INTEGER,
    output_tokens INTEGER,
    passed BOOLEAN,
    critique_text TEXT,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);

Comparing Validation Strategies

Choosing the right validation strategy involves balancing latency, cost, and accuracy. The table below outlines the trade-offs of each approach:

By routing calls through the n1n.ai unified API, you can easily implement a hybrid approach: use a fast, cost-effective model like DeepSeek-V3 for the initial generation and a deterministic schema validator for the first line of defense, falling back to a Claude 3.5 Sonnet critique only when complex logical errors occur.

Common Failure Modes and Mitigations

While reflection loops are powerful, they fail in predictable ways if not designed carefully.

Failure Mode 1: Self-Reinforcing Errors

During iteration 2, the model often accepts a false assumption it made in iteration 1, treating its own previous output as ground truth.

• Mitigation: Instruct the model in the system prompt: "Do not treat facts or assumptions from your previous output as verified. Re-derive all conclusions directly from the original user input."

Failure Mode 2: Vague Critique Prompts

If you ask an LLM, "Is this output correct?", it will almost always respond with a generic "Yes, the output is correct."

• Mitigation: Force the critique prompt to be adversarial. Use prompts like: "List every specific constraint from the system prompt that this output violates. If there are no violations, write 'PASSED'. Otherwise, write a bulleted list of specific failures."

Failure Mode 3: Subjective Success Criteria

Reflection loops require clear, objective validation criteria. If you try to run a reflection loop on subjective qualities like "tone," "persuasiveness," or "brand alignment," the loop will fluctuate endlessly without converging.

• Mitigation: Reserve reflection loops for structured, verifiable tasks (e.g., code generation, data extraction, math, logical constraints). For subjective tasks, route the output to a human reviewer.

Security Hardening for Agentic Workflows

When an agent is allowed to execute tools (such as database queries or API calls) based on its self-corrected output, reflection loops must be combined with strict sandboxing.

1. Least-Privilege Execution: Ensure the credentials used by the agent can only access the specific resources required for the task.
2. Output Sanitization: Never execute code generated by an LLM directly on a host machine. Use isolated Docker containers or WebAssembly runtimes.
3. Transaction Rollbacks: For database writes, run the operations inside a transaction block. If the final validation step fails after all iterations, roll back the transaction to prevent corrupted states.

Conclusion

Implementing reflection loops is one of the most effective ways to increase the reliability of your LLM applications without the complexity of fine-tuning. By combining deterministic validators with semantic LLM critiques, you can catch formatting and logical errors before they reach your users.

For developers building high-performance agents, managing access to multiple LLM APIs can become complex. Using a unified aggregator like n1n.ai simplifies this process, providing a single endpoint to access, benchmark, and route requests across the world's leading language models.

Get a free API key at n1n.ai.