Self-correcting AutoGen agents that debug their own code

AutoGen’s code-executing agents become dramatically more useful when they can detect their own failures and retry with corrections. This guide walks through building an autogen agent self debugging code execution loop that catches exceptions, parses tracebacks, and feeds context back to the model for another attempt. You’ll end up with a reusable pattern that handles the most common failure modes without human intervention.

The core loop: execute, observe, correct

The fundamental pattern is straightforward: the agent writes code, a sandbox executes it, and the result — stdout, stderr, or exception traceback — returns to the agent as context for the next turn. AutoGen’s UserProxyAgent with code_execution_config handles the execution side. The challenge is structuring the conversation so the model actually uses the error information.

from autogen import UserProxyAgent, AssistantAgent
from autogen.agentchat.contrib.capabilities import transform_messages

assistant = AssistantAgent(
    name="coder",
    system_message=(
        "You write Python code to solve problems. "
        "When code fails, you receive the full traceback. "
        "Analyze the error, fix the code, and resend the complete corrected script. "
        "Only output the code block — no explanations."
    ),
    llm_config={"model": "gpt-4o"},
)

user_proxy = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False,  # set True for isolation
        "last_n_messages": 3,
    },
)

The last_n_messages parameter matters — it controls how much conversation history the executor sends back. Too little and the model loses context; too much and you burn tokens on irrelevant turns. Three messages (the task, the code, the error) is usually the sweet spot.

Structuring the task for self-correction

The initial prompt must explicitly tell the agent what failure looks like and how to respond. Vague instructions like “fix any errors” produce meandering attempts. Be specific:

TASK = """
Write a script that:
1. Reads data from 'input.csv'
2. Computes the rolling 7-day average of the 'value' column
3. Writes results to 'output.csv'

Requirements:
- Handle missing file gracefully
- Handle missing 'value' column gracefully
- If any error occurs, the executor will return the traceback.
- You must then output the ENTIRE corrected script in a single code block.
- Do not output partial fixes or explanations.
"""

The key phrase is “output the ENTIRE corrected script.” Without it, models often send diffs or fragments that the executor can’t run.

Parsing and categorizing failures

Not all errors deserve the same response. A FileNotFoundError means the environment setup is wrong — the agent should check paths or create test data. A KeyError on a column name means the data schema differs from expectations. A SyntaxError means the model generated invalid Python. Categorizing these lets you route to different recovery strategies or escalate to a human.

import traceback
from enum import Enum

class ErrorCategory(Enum):
    ENVIRONMENT = "environment"      # missing files, perms, network
    DATA = "data"                    # schema mismatch, corrupt data
    LOGIC = "logic"                  # algorithmic bugs, off-by-one
    SYNTAX = "syntax"                # invalid Python
    DEPENDENCY = "dependency"        # missing import, version conflict
    UNKNOWN = "unknown"

def categorize_error(exc: BaseException, tb: str) -> ErrorCategory:
    if isinstance(exc, (FileNotFoundError, PermissionError, ConnectionError)):
        return ErrorCategory.ENVIRONMENT
    if isinstance(exc, (KeyError, ValueError, IndexError)) and "column" in tb.lower():
        return ErrorCategory.DATA
    if isinstance(exc, SyntaxError):
        return ErrorCategory.SYNTAX
    if isinstance(exc, (ImportError, ModuleNotFoundError)):
        return ErrorCategory.DEPENDENCY
    if isinstance(exc, (TypeError, AttributeError, ZeroDivisionError)):
        return ErrorCategory.LOGIC
    return ErrorCategory.UNKNOWN

You can hook this into the agent’s reply logic by subclassing UserProxyAgent and overriding process_last_received_message or by using a custom code_execution_config callback. The category then drives whether to retry automatically, modify the environment, or escalate.

Limiting retries and preventing loops

Agents can get stuck in infinite correction cycles — fixing one bug introduces another, or the model hallucinates a fix that doesn’t address the root cause. Hard limits are essential.

MAX_RETRIES = 4

class SelfCorrectingExecutor(UserProxyAgent):
    def __init__(self, *args, max_retries=MAX_RETRIES, **kwargs):
        super().__init__(*args, **kwargs)
        self._retry_count = 0
        self._max_retries = max_retries
        self._last_error = None

    def initiate_chat(self, *args, **kwargs):
        self._retry_count = 0
        return super().initiate_chat(*args, **kwargs)

    def process_last_received_message(self, *args, **kwargs):
        # Called after code execution completes
        last_msg = self.chat_messages[self._get_agent_key(args[0])][-1]
        content = last_msg.get("content", "")
        
        if "Traceback (most recent call last):" in content:
            self._retry_count += 1
            self._last_error = content
            if self._retry_count >= self._max_retries:
                # Inject a final message that stops the loop
                return {
                    "content": f"MAX RETRIES ({self._max_retries}) EXCEEDED. Last error:\n{content}\n"
                               "Return a final summary of what was attempted and why it failed.",
                    "terminate": True,
                }
            # Inject error context for the next attempt
            return {
                "content": f"Attempt {self._retry_count} failed with error:\n{content}\n"
                           "Analyze the traceback and output the COMPLETE corrected script.",
                "terminate": False,
            }
        return super().process_last_received_message(*args, **kwargs)

This subclass tracks attempts and injects structured feedback. The terminate: True flag ends the conversation cleanly rather than letting it drift.

Providing environment context upfront

Many “self-debugging” failures are actually environment mismatches. The agent assumes a file exists at a path, or a package version, or a column name — and the assumption is wrong. Give the agent a snapshot of reality before it writes the first line.

def build_environment_context(work_dir: str) -> str:
    import os
    import subprocess
    import sys
    
    context = ["## Environment Snapshot"]
    
    # Python version and key packages
    context.append(f"Python: {sys.version.split()[0]}")
    try:
        import pandas, numpy, requests
        context.append(f"pandas: {pandas.__version__}, numpy: {numpy.__version__}, requests: {requests.__version__}")
    except ImportError as e:
        context.append(f"Missing imports: {e}")
    
    # Files in work directory
    files = os.listdir(work_dir) if os.path.exists(work_dir) else []
    context.append(f"Files in {work_dir}: {files or '(empty)'}")
    
    # First few lines of any CSV/JSON files
    for f in files:
        if f.endswith(('.csv', '.json', '.txt')):
            path = os.path.join(work_dir, f)
            try:
                with open(path) as fp:
                    preview = ''.join([next(fp) for _ in range(5)])
                context.append(f"--- {f} (first 5 lines) ---\n{preview}")
            except Exception:
                pass
    
    return "\n".join(context)

Pass this as part of the initial system message or first user message. It eliminates an entire class of FileNotFoundError and KeyError retries.

Handling partial successes

Sometimes code runs without crashing but produces wrong output — empty DataFrame, all NaN values, wrong shape. The executor sees exit code 0 and considers it success. You need a validation step.

def validate_output(result: dict, expected_shape: tuple = None, required_columns: list = None) -> tuple[bool, str]:
    """
    Returns (is_valid, error_message).
    Checks stdout for printed DataFrame info or reads output files.
    """
    stdout = result.get("stdout", "")
    stderr = result.get("stderr", "")
    
    if stderr:
        return False, f"stderr not empty: {stderr[:500]}"
    
    # If the script prints the result (common pattern), parse it
    if "shape" in stdout.lower() or "columns" in stdout.lower():
        # Heuristic: look for pandas-like output
        pass  # implement based on your output format
    
    # If output file expected, read and validate
    # This requires knowing the output path — pass it in task context
    
    return True, ""

Integrate this in your executor subclass. If validation fails, treat it like an exception and trigger a retry with the validation error as context.

Common pitfalls and tradeoffs

Docker vs. local execution. use_docker=True gives isolation but adds latency (container startup) and complexity (volume mounts, network). For development, local is faster. For production with untrusted code, Docker or a dedicated sandbox (gVisor, Firecracker) is non-negotiable. The work_dir must be mounted correctly — a frequent source of “file not found” errors that look like agent failures.

Token growth. Each retry adds the previous code, the error, and the new code to context. After 3-4 retries you’re sending 10k+ tokens per turn. Mitigations: truncate old tracebacks to the last 50 lines, summarize earlier attempts, or reset the conversation with a “here’s what we’ve tried” summary prompt.

Model capability floor. GPT-4o and Claude 3.5 Sonnet handle this loop reliably. Smaller models (Llama 3.1 8B, Phi-3) often fail to parse tracebacks correctly or produce syntactically invalid fixes. If you’re routing to smaller models for cost, expect lower self-correction success rates and plan for human escalation sooner.

Stateful vs. stateless retries. The pattern above is stateless — each attempt is a fresh script. For long-running processes (training loops, streaming), you need stateful correction: the agent modifies a running notebook or process. That’s a different architecture (Jupyter kernel + agent) with its own failure modes.

Production hardening

When this runs unattended, you need observability and guardrails.

import json
import time
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class ExecutionRecord:
    task_id: str
    attempt: int
    timestamp: str
    code: str
    stdout: str
    stderr: str
    exit_code: int
    error_category: str
    duration_ms: int
    success: bool

class ObservableExecutor(SelfCorrectingExecutor):
    def __init__(self, *args, audit_log_path: str = "execution_audit.jsonl", **kwargs):
        super().__init__(*args, **kwargs)
        self.audit_log_path = audit_log_path
    
    def _log_attempt(self, record: ExecutionRecord):
        with open(self.audit_log_path, "a") as f:
            f.write(json.dumps(asdict(record)) + "\n")
    
    def process_last_received_message(self, *args, **kwargs):
        start = time.time()
        result = super().process_last_received_message(*args, **kwargs)
        duration = int((time.time() - start) * 1000)
        
        # Extract execution result from the last message
        # (implementation depends on your message structure)
        
        return result

Log every attempt with structured fields. This lets you query “which tasks failed on dependency errors” or “average retries per task type” — essential for improving the system.

Rate limits and provider failures. If your LLM calls hit rate limits mid-loop, the agent appears to hang or fail silently. Route through a gateway that handles automatic fallback across providers — n4n.ai does this with a single OpenAI-compatible endpoint across 240+ models, including automatic retry on provider degradation. This keeps the correction loop moving even when one provider is unhealthy.

When to escalate

Define clear escalation criteria. My rule: if the same error category repeats twice, or if MAX_RETRIES is hit, or if the error category is ENVIRONMENT (which usually means infrastructure, not code), stop and alert.

ESCALATION_RULES = {
    ErrorCategory.ENVIRONMENT: 1,      # escalate immediately
    ErrorCategory.DEPENDENCY: 2,       # maybe auto-install once, then escalate
    ErrorCategory.DATA: 2,             # schema issues often need human clarification
    ErrorCategory.LOGIC: 4,            # give the model more chances
    ErrorCategory.SYNTAX: 3,           # usually fixable
    ErrorCategory.UNKNOWN: 2,
}

Wire this into your executor to automatically create a ticket, send a Slack message, or queue for human review.

Putting it together

The complete flow: environment snapshot → initial task → execute → categorize result → validate output → retry with context → log → escalate if needed. Each piece is simple; the reliability comes from composing them with clear boundaries and observability.

Start with the basic UserProxyAgent + AssistantAgent pair, add the retry counter, then layer on categorization, validation, and audit logging. You’ll catch 80% of the value with the first two additions. The rest is operational maturity.

The pattern generalizes beyond AutoGen — any code-generation loop with execution feedback benefits from the same structure: explicit failure contracts, categorized errors, bounded retries, and audit trails. Build it once, reuse it across agents.