Human-in-the-loop conversations with AutoGen

AutoGen’s human-in-the-loop conversations let you insert human judgment at critical decision points without breaking the agent orchestration flow. The framework achieves this through UserProxyAgent instances that can pause execution, solicit input, and resume once a person responds. This tutorial walks through building a production-ready pattern where agents propose actions, humans approve or redirect, and the conversation continues with full context preserved.

Step 1: Install dependencies and configure the environment

Start with a clean environment. AutoGen requires Python 3.10+ and a few core packages. You’ll also need an LLM endpoint — this example uses OpenAI-compatible APIs, which work with most providers.

python -m venv .venv
source .venv/bin/activate
pip install "autogen-agentchat>=0.2" "autogen-ext[openai]>=0.2" python-dotenv

Create a .env file with your API credentials:

OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1  # or your gateway endpoint
MODEL_NAME=gpt-4o-mini

If you route through a gateway like n4n.ai, set OPENAI_BASE_URL to the gateway endpoint and use your gateway key. The rest of the code remains unchanged.

Step 2: Define the agent topology with a human proxy

AutoGen’s UserProxyAgent is the mechanism for human-in-the-loop. It behaves like any other agent but delegates generate_reply to a callback you provide. That callback can block on CLI input, a web request, a Slack message, or any async source.

Create agents.py with a minimal three-agent topology: a planner, a coder, and a human reviewer.

# agents.py
import os
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message=(
        "You break down user requests into clear, ordered steps for a coder. "
        "Output only the plan as a numbered list. Do not write code."
    ),
)

coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    system_message=(
        "You implement the plan step by step. Write clean, runnable Python. "
        "Output only the code block for the current step."
    ),
)

def human_callback(messages: list[dict]) -> str:
    """Blocking CLI prompt. Replace with async HTTP handler for production."""
    print("\n" + "=" * 60)
    print("HUMAN REVIEW REQUIRED")
    print("=" * 60)
    for m in messages[-3:]:
        role = m.get("role", "unknown")
        content = m.get("content", "")[:200]
        print(f"[{role}] {content}...")
    print("-" * 60)
    return input("Your response (approve / revise / abort): ").strip()

reviewer = UserProxyAgent(
    name="reviewer",
    human_input_callback=human_callback,
    description="Human reviewer who approves, requests revisions, or aborts.",
)

termination = TextMentionTermination("APPROVED") | MaxMessageTermination(20)

team = RoundRobinGroupChat(
    participants=[planner, coder, reviewer],
    termination_condition=termination,
)

Key points: the human_input_callback receives the full message history so your UI can render context. The callback returns a string that AutoGen injects as the reviewer’s reply. The termination condition watches for “APPROVED” in any message, letting the human end the loop explicitly.

Step 3: Run a basic conversation loop

Create main.py to drive the team. The run_stream method yields events you can log or forward to a frontend.

# main.py
import asyncio
from agents import team
from autogen_agentchat.messages import TextMessage

async def main():
    task = (
        "Create a Python function that fetches the current weather for a given city "
        "using the OpenWeatherMap API. Include error handling and caching."
    )
    print(f"Task: {task}\n")

    async for event in team.run_stream(task=task):
        if isinstance(event, TextMessage):
            print(f"[{event.source}] {event.content[:120]}...")

    print("\n--- Conversation complete ---")

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

You’ll see the planner emit a plan, the coder produce code for step 1, then the CLI prompt appears. Type approve to continue, revise: add timeout handling to request changes, or abort to stop. The team resumes from the reviewer’s message with full context.

Verify success

• Planner outputs a numbered plan before any code appears.
• Coder emits one code block per step.
• Reviewer prompt appears after each coder message.
• Typing approve lets the coder proceed to the next step.
• Conversation ends when reviewer sends a message containing “APPROVED”.

Step 4: Replace the blocking callback with an async handler

The CLI callback blocks the event loop. For a real service, the callback must be async and non-blocking — typically enqueuing a review request and awaiting a response via webhook, WebSocket, or polling.

Update agents.py with an async callback that uses an asyncio.Queue per conversation:

# agents.py (replace human_callback and reviewer)
import asyncio
from collections import defaultdict
from autogen_agentchat.agents import UserProxyAgent

review_queues: dict[str, asyncio.Queue] = defaultdict(asyncio.Queue)
response_events: dict[str, asyncio.Event] = defaultdict(asyncio.Event)
pending_responses: dict[str, str] = {}

async def async_human_callback(messages: list[dict], conversation_id: str) -> str:
    """Enqueue review request and wait for external response."""
    queue = review_queues[conversation_id]
    event = response_events[conversation_id]
    
    # Package context for your frontend
    review_request = {
        "conversation_id": conversation_id,
        "messages": messages[-5:],  # last 5 turns
        "timestamp": asyncio.get_event_loop().time(),
    }
    await queue.put(review_request)
    
    # Wait for external caller to set response
    await event.wait()
    response = pending_responses.pop(conversation_id, "abort")
    event.clear()
    return response

def set_human_response(conversation_id: str, response: str):
    """Call this from your webhook / WebSocket handler."""
    pending_responses[conversation_id] = response
    response_events[conversation_id].set()

reviewer = UserProxyAgent(
    name="reviewer",
    human_input_callback=async_human_callback,
    description="Async human reviewer for production use.",
)

The conversation_id is passed automatically by AutoGen when using run_stream with a ConversationContext. You’ll need to thread it through your API layer. The pattern: enqueue a review object, return control to the event loop, and resume when set_human_response is called from your HTTP handler.

Step 5: Persist conversation state for durability

Human-in-the-loop conversations can span hours or days. You need to serialize the team state so a process restart doesn’t lose progress. AutoGen provides save_state and load_state on the team.

Add a persistence layer using SQLite (swap for Postgres/Redis in production):

# persistence.py
import json
import sqlite3
from pathlib import Path
from autogen_agentchat.teams import RoundRobinGroupChat

DB_PATH = Path("conversations.db")

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                id TEXT PRIMARY KEY,
                state_json TEXT NOT NULL,
                updated_at REAL NOT NULL
            )
        """)

async def save_conversation(team: RoundRobinGroupChat, conversation_id: str):
    state = await team.save_state()
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT OR REPLACE INTO conversations (id, state_json, updated_at) VALUES (?, ?, ?)",
            (conversation_id, json.dumps(state), asyncio.get_event_loop().time()),
        )

async def load_conversation(conversation_id: str) -> dict | None:
    with sqlite3.connect(DB_PATH) as conn:
        row = conn.execute(
            "SELECT state_json FROM conversations WHERE id = ?", (conversation_id,)
        ).fetchone()
    return json.loads(row[0]) if row else None

Wire it into main.py:

# main.py (updated)
import asyncio
import uuid
from agents import team
from persistence import init_db, save_conversation, load_conversation
from autogen_agentchat.messages import TextMessage

async def main():
    init_db()
    conversation_id = str(uuid.uuid4())
    
    # Resume if state exists
    saved = await load_conversation(conversation_id)
    if saved:
        await team.load_state(saved)
        print(f"Resumed conversation {conversation_id}")
    
    task = "Create a Python function that fetches current weather..."
    
    async for event in team.run_stream(task=task, conversation_id=conversation_id):
        if isinstance(event, TextMessage):
            print(f"[{event.source}] {event.content[:120]}...")
        
        # Persist after each turn
        await save_conversation(team, conversation_id)

    print("\n--- Conversation complete ---")

if __name__ == "__main__":
    asyncio.run(main())

The conversation_id parameter on run_stream ensures the callback receives the same ID. Persist after every event so a crash loses at most one turn.

Verify success

• Kill the process mid-conversation (Ctrl+C after a reviewer prompt).
• Restart with the same conversation_id (hardcode it for testing).
• Conversation resumes at the exact turn with full history intact.
• Database row updates on each turn.

Step 6: Add structured review actions

Free-text responses like “revise: add timeout” are fragile. Define a structured protocol so the reviewer’s intent is unambiguous and the planner can react programmatically.

Create review_protocol.py:

# review_protocol.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional

class ReviewAction(str, Enum):
    APPROVE = "approve"
    REVISE = "revise"
    ABORT = "abort"
    SKIP = "skip"

class ReviewDecision(BaseModel):
    action: ReviewAction
    reason: str = ""
    instructions: Optional[str] = None
    target_agent: Optional[str] = None  # which agent should act next

    def to_message(self) -> str:
        parts = [f"ACTION: {self.action.value}"]
        if self.reason:
            parts.append(f"REASON: {self.reason}")
        if self.instructions:
            parts.append(f"INSTRUCTIONS: {self.instructions}")
        if self.target_agent:
            parts.append(f"TARGET: {self.target_agent}")
        return "\n".join(parts)

    @classmethod
    def from_text(cls, text: str) -> "ReviewDecision":
        """Parse legacy free-text for backward compatibility."""
        text_lower = text.lower().strip()
        if text_lower in ("approve", "approved", "yes", "ok"):
            return cls(action=ReviewAction.APPROVE)
        if text_lower in ("abort", "stop", "cancel"):
            return cls(action=ReviewAction.ABORT)
        if text_lower.startswith("revise"):
            _, _, reason = text_lower.partition(":")
            return cls(action=ReviewAction.REVISE, reason=reason.strip())
        return cls(action=ReviewAction.REVISE, reason=text)

Update the async callback to parse and validate:

# agents.py (updated callback)
from review_protocol import ReviewDecision, ReviewAction

async def async_human_callback(messages: list[dict], conversation_id: str) -> str:
    queue = review_queues[conversation_id]
    event = response_events[conversation_id]
    
    review_request = {
        "conversation_id": conversation_id,
        "messages": messages[-5:],
        "protocol": "review_decision_v1",
    }
    await queue.put(review_request)
    await event.wait()
    
    raw = pending_responses.pop(conversation_id, "abort")
    decision = ReviewDecision.from_text(raw)
    
    # Inject structured decision into conversation
    if decision.action == ReviewAction.APPROVE:
        return "APPROVED: " + decision.reason
    elif decision.action == ReviewAction.REVISE:
        target = f" @{decision.target_agent}" if decision.target_agent else ""
        return f"REVISE{target}: {decision.instructions or decision.reason}"
    elif decision.action == ReviewAction.SKIP:
        return "SKIPPED: " + decision.reason
    return "ABORTED: " + decision.reason

Now your frontend can send JSON like:

{
  "conversation_id": "abc-123",
  "decision": {
    "action": "revise",
    "reason": "Missing timeout handling",
    "instructions": "Add a 10-second timeout and retry logic",
    "target_agent": "coder"
  }
}

The planner sees REVISE @coder: Add a 10-second timeout... and can route the next turn appropriately.

Step 7: Implement a review API endpoint

Wire the queue to a FastAPI endpoint. This is where your frontend (React, Slack bot, CLI) posts decisions.

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from review_protocol import ReviewDecision
from agents import set_human_response, review_queues
import asyncio

app = FastAPI()

class ReviewRequest(BaseModel):
    conversation_id: str
    decision: ReviewDecision

@app.get("/review/next")
async def get_next_review():
    """Frontend polls this to get pending reviews."""
    for cid, queue in review_queues.items():
        if not queue.empty():
            return await queue.get()
    return {"status": "empty"}

@app.post("/review/submit")
async def submit_review(request: ReviewRequest):
    """Frontend posts the human's decision here."""
    if request.conversation_id not in response_events:
        raise HTTPException(404, "Conversation not found or not awaiting review")
    
    set_human_response(request.conversation_id, request.decision.to_message())
    return {"status": "accepted"}

@app.get("/health")
async def health():
    return {"status": "ok"}

Run with uvicorn api:app --reload. Your frontend polls /review/next, renders the context, and POSTs to /review/submit when the human acts.

Verify success

• Start the API server and run main.py in another terminal.
• Conversation pauses at reviewer turn.
• Poll /review/next → returns review context.
• POST a REVISE decision to /review/submit.
• CLI shows REVISE @coder: ... and coder responds with revised code.
• POST APPROVE → conversation continues to next step.

Step 8: Handle multi-human workflows

Some steps need different reviewers — security review for code, product review for UX copy, legal for compliance text. Extend the topology with typed reviewer agents.

# agents.py (add typed reviewers)
security_reviewer = UserProxyAgent(
    name="security_reviewer",
    human_input_callback=async_human_callback,
    description="Security reviewer for code changes.",
)

product_reviewer = UserProxyAgent(
    name="product_reviewer",
    human_input_callback=async_human_callback,
    description="Product reviewer for user-facing copy.",
)

# Update team participants based on task type
def build_team(task_type: str) -> RoundRobinGroupChat:
    base = [planner, coder]
    if task_type == "code":
        base.append(security_reviewer)
    elif task_type == "copy":
        base.append(product_reviewer)
    base.append(reviewer)  # final sign-off
    return RoundRobinGroupChat(
        participants=base,
        termination_condition=termination,
    )

The planner can emit a TASK_TYPE: code marker in its first message, and your driver script selects the appropriate team. This keeps the conversation linear while routing human attention to the right person.

Step 9: Add observability and audit logging

Every human decision should be logged with context for compliance and debugging. Wrap the callback to emit structured logs.

# observability.py
import json
import time
from review_protocol import ReviewDecision

def log_review_decision(conversation_id: str, decision: ReviewDecision, context: list[dict]):
    record = {
        "timestamp": time.time(),
        "conversation_id": conversation_id,
        "action": decision.action.value,
        "reason": decision.reason,
        "instructions": decision.instructions,
        "target_agent": decision.target_agent,
        "context_messages": len(context),
    }
    # Write to JSONL, send to Datadog, etc.
    with open("audit.log", "a") as f:
        f.write(json.dumps(record) + "\n")

Call it inside async_human_callback after parsing the decision. The audit trail shows who approved what, when, and with what context — critical for regulated environments.

Step 10: Test the full flow end-to-end

Create a test that simulates a human approving, then requesting a revision, then approving the fix.

# test_hitl.py
import asyncio
from agents import team, set_human_response, review_queues, response_events
from review_protocol import ReviewDecision, ReviewAction

async def test_full_flow():
    conversation_id = "test-conv-123"
    task = "Write a hello world function in Python."
    
    # Start the team in background
    run_task = asyncio.create_task(team.run_stream(task=task, conversation_id=conversation_id))
    
    # Wait for first review request
    await asyncio.sleep(0.5)
    assert not review_queues[conversation_id].empty()
    
    # Simulate human: request revision
    set_human_response(conversation_id, ReviewDecision(
        action=ReviewAction.REVISE,
        reason="Add type hints",
        instructions="Add type hints to the function signature",
        target_agent="coder",
    ).to_message())
    
    # Wait for coder to respond, then approve
    await asyncio.sleep(1.0)
    set_human_response(conversation_id, ReviewDecision(
        action=ReviewAction.APPROVE,
        reason="Type hints added",
    ).to_message())
    
    # Let conversation complete
    await run_task
    print("Test passed")

asyncio.run(test_full_flow())

Run with pytest test_hitl.py -v. The test validates that the callback chain works, structured decisions parse correctly, and the team resumes after each human turn.

Verify success

• Test passes without hanging or timeout.
• Audit log contains two entries: REVISE then APPROVE.
• Final conversation state shows revised code with type hints.

Production considerations

Timeouts: Add a TTL to review requests. If the human doesn’t respond in 24 hours, auto-escalate or abort. Implement this by wrapping event.wait() in asyncio.wait_for with a deadline.

Idempotency: The /review/submit endpoint should be idempotent. Store a nonce with each review request and reject duplicate submissions.

Authorization: Bind conversation_id to a user session. Only the assigned reviewer (or admin) can submit decisions for that conversation.

Rate limiting: Protect the review queue endpoint. A malicious actor polling /review/next rapidly shouldn’t exhaust resources.

Model routing: If you use different models for planner vs coder, configure separate OpenAIChatCompletionClient instances with different model parameters. The gateway handles provider fallback transparently.

Summary

You now have a complete human-in-the-loop pattern with AutoGen:

1. UserProxyAgent with an async callback that integrates with any frontend
2. Structured review protocol eliminating ambiguous free-text responses
3. State persistence so conversations survive process restarts
4. Typed reviewers routing different tasks to different humans
5. Audit logging for compliance
6. Testable, observable components

The key insight: treat the human as just another agent with a special generate_reply implementation. AutoGen’s message-passing architecture handles the rest — context preservation, turn ordering, termination — without custom orchestration code.