AutoGen human-in-the-loop: pausing agents for review

Shipping autonomous agents without a checkpoint is how you get silent data corruption. The autogen pause agent for human review pattern lets you halt execution before a risky action, collect sign-off, and resume deterministically. Below is an end-to-end setup that pauses an AutoGen agent before it writes a file; you can adapt the same gate to API calls, database mutations, or deploys.

Step 1: Install AutoGen and configure the LLM endpoint

Pin the package so a breaking release doesn’t surprise you:

pip install pyautogen==0.2.32

AutoGen talks to any OpenAI-compatible chat completion API. Define a config_list entry with your model, key, and base URL. If you point it at a gateway like n4n.ai, the single endpoint covers 240+ models and fails over automatically when a provider is rate-limited, so your autogen pause agent for human review loop doesn’t die mid-conversation.

import os
import autogen

llm_config = {
    "config_list": [{
        "model": "gpt-4o",
        "api_key": os.environ["N4N_API_KEY"],
        "base_url": "https://api.n4n.ai/v1",
    }],
    "temperature": 0,
    "timeout": 30,
}

Use temperature=0 for review workflows. You want the assistant to propose the same structured call every time, not improvise. The timeout guards against a hung upstream.

Step 2: Define the assistant’s action contract

The assistant must propose the risky action through a typed function call, not free text. AutoGen derives the JSON schema from the Python signature and docstring, so write both precisely.

def write_file(filename: str, content: str) -> str:
    """Write content to a file at filename. Returns a status string."""
    # Implementation injected by the user proxy; see Step 3.
    raise NotImplementedError

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message=(
        "You are a file-writing assistant. "
        "When asked to write a file, call write_file(filename, content). "
        "Never describe the write in plain text. Never execute shell commands."
    ),
)

The stub write_file exists only so the schema is visible to the LLM. The real implementation lives in the user proxy’s function_map, which is where we insert the human gate. Keeping the side effect out of the assistant’s process boundary is what makes the autogen pause agent for human review pattern safe.

Step 3: Implement the human review gate

This is the core. The user proxy owns the executable function. We block inside it until a human approves, then perform the side effect or reject.

def write_file_with_review(filename: str, content: str) -> str:
    # autogen pause agent for human review: halt at the side-effect boundary.
    prompt = (
        f"Agent requests write to '{filename}':\n"
        f"--- content ---\n{content}\n--- end ---\n"
        "Approve write? (y/n): "
    )
    decision = input(prompt).strip().lower()
    if decision != "y":
        return "REJECTED: human did not approve the write."
    with open(filename, "w") as f:
        f.write(content)
    return f"OK: wrote {len(content)} bytes to {filename}."

user = autogen.UserProxyAgent(
    name="human_reviewer",
    human_input_mode="NEVER",  # we manage the pause inside the function
    code_execution_config=False,
    function_map={"write_file": write_file_with_review},
)

Setting human_input_mode="NEVER" stops AutoGen from prompting on every assistant message. The only pause happens exactly where the world changes. That is auditable and avoids operator fatigue. If you need to review the assistant’s plan before it even calls the function, register a reply hook instead—but for side effects, the function-level gate is cleaner.

Step 4: Run the conversation and handle rejection

Start the chat from the user proxy. The assistant will emit a write_file call; the proxy executes the wrapped function, which blocks for input.

if __name__ == "__main__":
    user.initiate_chat(
        assistant,
        message="Write hello.txt containing the word 'hi'.",
    )

Sample terminal flow on approval:

Agent requests write to 'hello.txt':
--- content ---
hi
--- end ---
Approve write? (y/n): y

The assistant receives "OK: wrote 2 bytes to hello.txt." and can terminate. On rejection (n), it gets "REJECTED: human did not approve the write." and should apologize or ask for a different task. The autogen pause agent for human review loop is synchronous here; the agent thread is parked until the human answers.

Step 5: Verify the pause works

Verification must be concrete, not hand-wavy.

1. Run the script and answer n. Assert the target file does not exist on disk.
2. Run again and answer y. Assert the file exists with the exact content.
3. Capture the assistant’s final message; it must echo the status string returned by the gate.

A pytest that mocks input proves the gate holds without a human in the loop:

import os
from your_module import write_file_with_review

def test_reject_write(tmp_path, monkeypatch):
    monkeypatch.setattr("builtins.input", lambda _: "n")
    result = write_file_with_review(str(tmp_path / "x.txt"), "data")
    assert "REJECTED" in result
    assert not (tmp_path / "x.txt").exists()

def test_approve_write(tmp_path, monkeypatch):
    monkeypatch.setattr("builtins.input", lambda _: "y")
    target = tmp_path / "x.txt"
    result = write_file_with_review(str(target), "data")
    assert "OK" in result
    assert target.read_text() == "data"

If both tests pass, your pause is real. Ship it behind a flag first.

Alternative: human_input_mode for ad-hoc pauses

AutoGen ships human_input_mode="ALWAYS" on UserProxyAgent. It prints every agent message and asks for input. Do not use this in production—it trains operators to blindly hit enter and defeats the purpose of structured review. The explicit function-level gate above is the disciplined version of the same idea.

In a GroupChat, inject a dedicated UserProxyAgent with human_input_mode="TERMINATE" so it only interrupts when the conversation tries to end. That complements, but does not replace, the per-action pause.

Production considerations

Async and services. input() blocks the event loop. In a FastAPI or Celery worker, serialize the pending filename and content to Redis, return a review URL, and resume the agent when the human clicks approve. The agent task should sleep or yield, not block a thread pool.

Timeouts. If no human responds in 10 minutes, auto-reject. Orphaned agent turns waste tokens and leave locks held. Use a TTL on the review record.

Audit trail. Log the filename, a SHA-256 of content, the decision, and the reviewer identity to your SIEM. The pause is worthless if you can’t prove what was approved six months later.

Model routing and caching. When you call the LLM through a gateway, set cache_control hints on long system messages. n4n.ai forwards those hints, so repeated reviews of the same instructions cost fewer tokens. Use per-token metering to spot agents that retry rejected actions in loops.

Error isolation. Wrap the function body in try/except. A crash inside the gate should return REJECTED plus the exception, not hang the conversation. Never let the human-review function raise into the AutoGen scheduler.

The autogen pause agent for human review pattern is not about slowing agents down. It is about making their rare, irreversible steps reversible. Put the pause at the boundary where the world changes, keep the rest of the loop autonomous, and you get speed without sacrificing control.