AutoGen human-in-the-loop for high-stakes code execution

Wiring autogen human in the loop code execution safety into an agent that can mutate production databases or cloud infrastructure is not the same as running a toy notebook agent. The default AutoGen code executor will happily run whatever the assistant emits, and a single bad LLM suggestion becomes a real-world incident. This guide lays out an ordered path to put a human gate in front of every execution, sandbox the blast radius, and keep the loop usable.

1. Define the threat model before writing code

High-stakes means the executed code can cause irreversible side effects: writing to a primary datastore, deleting objects in object storage, calling paid third-party APIs, or changing IAM policies. If the worst-case output of a code block is “nothing happens” or “a local variable is wrong,” you do not need a human in the loop—you need tests. If the worst case is “production outage,” you need explicit approval and isolation.

AutoGen’s UserProxyAgent with human_input_mode="ALWAYS" is not sufficient by itself. It prompts for free-text input on every conversation turn, but it does not structurally separate “here is a code block” from “here is a chat message,” and it does not enforce that the human actually reviewed the diff. Build a gate that triggers specifically on code execution, not on conversation flow.

2. Intercept code blocks with a custom executor

AutoGen’s LocalCommandLineCodeExecutor is the default local runner. Subclass it to require synchronous human sign-off before any block runs. This keeps the approval tied to the exact code, not to a conversational prompt.

from autogen.coding import LocalCommandLineCodeExecutor

class GatedExecutor(LocalCommandLineCodeExecutor):
    def execute_code_blocks(self, code_blocks):
        for block in code_blocks:
            if block.language != "python":
                continue
            print(f"--- PROPOSED CODE ---\n{block.code}\n--------------------")
            ans = input("Approve execution? (yes/no): ").strip().lower()
            if ans != "yes":
                raise PermissionError("Human rejected code execution")
        return super().execute_code_blocks(code_blocks)

Wire it into the proxy:

from autogen import UserProxyAgent, AssistantAgent

user = UserProxyAgent(
    name="human_gate",
    code_execution_config={"executor": GatedExecutor(timeout=15)},
    human_input_mode="NEVER",  # we gate at exec, not at chat
)

assistant = AssistantAgent(
    name="coder",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

The tradeoff: this blocks the agent thread on input(). In a service context, replace input() with a request to an internal review queue (Slack, a web endpoint) that polls for a decision. The pattern stays identical.

3. Static pre-scan before the human ever sees it

Human attention is expensive. Run a cheap AST scan to reject obviously dangerous code before prompting a person. This reduces alert fatigue and catches reckless generation.

import ast

BANNED_ROOT_MODULES = {"os", "subprocess", "shutil", "sys", "socket"}

def scan_source(src: str) -> None:
    tree = ast.parse(src)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split(".")[0] in BANNED_ROOT_MODULES:
                    raise ValueError(f"Banned import: {alias.name}")
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.module.split(".")[0] in BANNED_ROOT_MODULES:
                raise ValueError(f"Banned import from: {node.module}")

If scan_source raises, abort the block and return the error to the assistant so it can rewrite. Do not show the human a rejection that a parser could have caught.

4. Sandbox the execution environment

Even approved code should run inside a container with no privileges, a read-only filesystem, and no network unless explicitly required. AutoGen supports passing docker options directly.

code_cfg = {
    "use_docker": {
        "image": "python:3.11-slim",
        "read_only": True,
        "network_mode": "none",
        "mem_limit": "256m",
        "cpu_quota": 50000,  # 0.5 CPU
    }
}
user = UserProxyAgent(
    name="human_gate",
    code_execution_config=code_cfg,
    human_input_mode="NEVER",
)

Pitfall: network_mode: none breaks code that needs to call your own internal APIs. For high-stakes workflows, prefer issuing short-lived credentials and mounting them as secrets rather than opening the network broadly. Docker on macOS has a filesystem performance penalty; accept it. Safety beats latency here.

5. Keep the model call itself reliable

The approval loop depends on the LLM generating code and summarizing results. If your primary model provider rate-limits the assistant mid-task, the human waits indefinitely. Route through an OpenAI-compatible gateway that fails over automatically. For example, n4n.ai exposes one OpenAI-compatible endpoint addressing 240+ models with automatic fallback when a provider is degraded; point AutoGen’s config_list at it so a vendor outage does not stall a production change request.

import os
config_list = [{
    "model": "gpt-4o",
    "base_url": "https://api.n4n.ai/v1",
    "api_key": os.environ["N4N_KEY"],
}]

This is a configuration change, not a code change. The same llm_config works with the gated executor above.

6. Log everything to JSONL for replay

A human approved a deletion three days ago and now finance wants to know why. You need an append-only audit log mapping code block → human decision → stdout/stderr.

import json, time

def log_decision(run_id, code, approved, output):
    with open("exec_audit.jsonl", "a") as f:
        f.write(json.dumps({
            "ts": time.time(),
            "run_id": run_id,
            "code": code,
            "approved": approved,
            "output": output,
        }) + "\n")

Call log_decision inside GatedExecutor.execute_code_blocks right after the human responds, and again after execution with the captured output. Store the log outside the container volume so a rogue block cannot tamper with it.

7. Bound time and memory, fail loud

Unbounded loops or accidental while True: will hang the executor. Set a hard timeout in the executor and catch the exception to return a structured error to the assistant.

try:
    result = super().execute_code_blocks(code_blocks)
except TimeoutError:
    return [{"code": b.code, "output": "TIMEOUT", "success": False} for b in code_blocks]

Also cap the size of captured stdout. If the assistant prints a 2 GB dataframe, you just ate your host memory. Truncate at 10k characters and mark it truncated.

8. Minimal end-to-end skeleton

from autogen import UserProxyAgent, AssistantAgent
from autogen.coding import LocalCommandLineCodeExecutor
import ast, os, json, time

BANNED = {"os", "subprocess", "shutil", "sys"}

class Gated(LocalCommandLineCodeExecutor):
    def execute_code_blocks(self, blocks):
        for b in blocks:
            if b.language != "python": continue
            ast.parse(b.code)  # syntax check
            for node in ast.walk(ast.parse(b.code)):
                # ... banned import scan ...
                pass
            print(b.code)
            if input("Run? ") != "yes":
                raise PermissionError("rejected")
            log_decision("run1", b.code, True, "")
        return super().execute_code_blocks(blocks)

user = UserProxyAgent("gate", code_execution_config={"executor": Gated(timeout=10)}, human_input_mode="NEVER")
assistant = AssistantAgent("coder", llm_config={"config_list": [{"model":"gpt-4o","api_key":os.environ["OPENAI_KEY"]}]})
user.initiate_chat(assistant, message="Generate a script to list stale S3 buckets but do NOT delete anything.")

This is intentionally small. In production, replace input() with your review service and add the docker config from step 4.

9. Common pitfalls and tradeoffs

Human fatigue. If the agent emits ten tiny blocks, a person will start typing “yes” without reading. Batch related blocks into a single approval when they share a logical step, and show a diff-style view rather than raw code.

Context bloat. Pasting full stdout back into the chat consumes tokens and can push the assistant into incoherence. Return only the last N lines plus a success flag.

False authority. A green checkmark from a human does not mean the code is correct, only that it was reviewed. Keep the human focused on side effects (“does this touch prod?”) not on algorithmic correctness.

Sandbox escape via dependencies. If you allow pip install inside the container, you have widened the attack surface considerably. Pin a fixed image with required packages pre-installed; forbid dynamic installs in the AST scan.

Latency vs. safety. Every gate adds seconds to minutes. For internal dev tooling this is fine. For a customer-facing agent, consider a two-tier system: low-risk read-only calls auto-approved in a strict sandbox, high-risk writes routed to the human gate.

The autogen human in the loop code execution safety pattern is not about trusting the model less; it is about engineering the boundary so that trust is explicit, logged, and contained. Build the gate at the executor level, sandbox hard, and keep the human in the loop only where the cost of a mistake is real.