# Building a Self-Correcting AI Agent with Reflection Loops in Python > Source: > Published: 2026-08-23 10:03:11+00:00 Language models produce wrong answers. Not occasionally — regularly. When you deploy an LLM to automate tasks, you need a way to catch and fix those errors without human intervention. Reflection loops are one practical answer: the model checks its own output, flags problems, and retries until it meets a quality bar you define in code. A reflection loop is a control flow pattern where an agent runs a task, evaluates the result, then decides whether to retry. The simplest form is a two-step cycle: *generate* and *critique*. The critique step is usually a second LLM call with a different prompt, but it can also be a deterministic check — a JSON parser, a unit test runner, or a schema validator. The key insight is that critique is easier than generation. A language model will often miss edge cases in a first pass but correctly identify them when asked “what is wrong with this output?” This asymmetry is what makes the pattern work at all. Here is a Python class that wraps any LLM call in a reflection loop: ``` python import json from typing import Any, Callable import httpx MAX_ITERATIONS = 4 def call_llm(prompt: str, system: str = "") -> str: 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']}" ) output = call_llm(f"{self.task_prompt}\n\n{context}") 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} 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} ``` The `validator` is a plain Python callable that returns `(passed: bool, message: str)` . Keeping it separate from the LLM logic means you can unit-test it independently and swap it without touching the agent loop. Malformed or schema-violating JSON is the most common failure mode in structured LLM output tasks. Here is a concrete validator that enforces a fixed schema: ``` python 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() if clean.startswith("``` "): clean = "\n".join(clean.split("\n")[1:]) if clean.endswith(" ```"): clean = "\n".join(clean.split("\n")[:-1]) 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" 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)) ``` The deterministic schema check runs *before* the LLM critique. If JSON parsing fails, the error message is already precise and actionable. The LLM critique step is reserved for semantic problems that a schema cannot express. Reflection loops without a hard cap on iterations will generate runaway costs in production. Three rules to enforce: **1. Hard iteration limit.** Never exceed 4–5 cycles per request. If the model has not produced valid output after four attempts, the problem is almost always the prompt, not a fixable runtime error. **2. Track token spend across iterations.** Add a budget guard: ``` COST_PER_1K_TOKENS = 0.003 # adjust to your model's 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: # Real usage: read token counts from the API response 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. Log every iteration to a structured store.** A simple SQLite table with `(task_id, attempt, input_tokens, passed, critique_text)` is enough to identify which task types consistently fail and need prompt improvements. The pattern fails in predictable ways: **Self-reinforcing errors.** The model accepts a wrong assumption from iteration 1 in iteration 2. Fix: add to the system prompt — *"Do not treat facts from the previous output as verified. Re-derive from the original input."* **Vague critique prompts.** Asking “is this correct?” returns “yes” more often than not. Ask “list each specific constraint the output violates and explain why.” **No ground truth.** Reflection only works when the validator has a clear, programmatic success criterion. Subjective tasks — tone, persuasiveness, brand alignment — need human review, not another LLM iteration. Security-sensitive use cases, where the agent takes real actions based on its output, need sandboxing on top of reflection. Tool calls should be scoped to least-privilege, outputs sanitized before use, and every action logged with rollback support. We maintain [free hardening checklists for agentic LLM deployments](https://ayinedjimi-consultants.fr/checklists) covering tool isolation, output validation, and incident recovery. Reflection loops improve output reliability without fine-tuning or swapping models. The pattern has three components: a generation step, a deterministic validator, and an LLM critique that feeds the next attempt. Keep the iteration count low (≤4), track costs per request, and log every iteration. The goal is not to make the model infallible — it is to catch errors programmatically before they propagate downstream. Start with the schema validator version above. It handles 80% of structured output failures and requires zero prompt engineering on the critique side. Add LLM self-critique only when deterministic checks cannot express the quality bar you need. *I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.*