AutoGen’s human-in-the-loop patterns work well in notebooks, but production introduces requirements the tutorials skip: durable state, audit trails, approval timeouts, and horizontal scaling. This guide walks through the engineering decisions you’ll face when moving from prototype to a system that handles real traffic and real compliance requirements.
The production gap
In development, user_proxy.initiate_chat() blocks until a human responds. In production, that blocking call becomes a liability — it ties up a worker process, prevents horizontal scaling, and loses context if the pod restarts. The fix is separating the agent loop from the human interaction surface.
Start by modeling the conversation as a state machine with explicit states: AGENT_TURN, AWAITING_HUMAN, HUMAN_RESPONDED, COMPLETE, FAILED. Persist every transition. This lets you recover from crashes, replay for debugging, and implement approval SLAs.
# conversation_state.py
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
class ConversationState(Enum):
AGENT_TURN = "agent_turn"
AWAITING_HUMAN = "awaiting_human"
HUMAN_RESPONDED = "human_responded"
COMPLETE = "complete"
FAILED = "failed"
@dataclass
class HumanInLoopConversation:
conversation_id: str
state: ConversationState = ConversationState.AGENT_TURN
messages: list[dict] = field(default_factory=list)
pending_approval: Optional[dict] = None
approval_requested_at: Optional[datetime] = None
approval_expires_at: Optional[datetime] = None
metadata: dict = field(default_factory=dict)
version: int = 0 # optimistic locking
def to_json(self) -> str:
return json.dumps({
"conversation_id": self.conversation_id,
"state": self.state.value,
"messages": self.messages,
"pending_approval": self.pending_approval,
"approval_requested_at": self.approval_requested_at.isoformat() if self.approval_requested_at else None,
"approval_expires_at": self.approval_expires_at.isoformat() if self.approval_expires_at else None,
"metadata": self.metadata,
"version": self.version,
})
@classmethod
def from_json(cls, data: str) -> "HumanInLoopConversation":
d = json.loads(data)
conv = cls(
conversation_id=d["conversation_id"],
state=ConversationState(d["state"]),
messages=d["messages"],
pending_approval=d["pending_approval"],
metadata=d["metadata"],
version=d["version"],
)
if d["approval_requested_at"]:
conv.approval_requested_at = datetime.fromisoformat(d["approval_requested_at"])
if d["approval_expires_at"]:
conv.approval_expires_at = datetime.fromisoformat(d["approval_expires_at"])
return convStore this in Redis with a TTL matching your longest approval window, or in Postgres if you need durability across redeploys. The version field enables optimistic locking — critical when the agent worker and the approval API might race on the same conversation.
Decoupling the agent loop from approval
Replace the blocking initiate_chat with an event-driven pattern. The agent worker runs until it needs human input, then emits an approval_requested event and exits. A separate approval service handles the human interaction, then emits approval_received to resume the agent.
# agent_worker.py
from conversation_state import HumanInLoopConversation, ConversationState
import redis
import json
from autogen import AssistantAgent, UserProxyAgent
from typing import Callable
class ProductionAgentWorker:
def __init__(
self,
redis_client: redis.Redis,
approval_queue: "ApprovalQueue",
llm_config: dict,
):
self.redis = redis_client
self.approval_queue = approval_queue
self.assistant = AssistantAgent(
name="assistant",
llm_config=llm_config,
system_message="You are a helpful assistant. Request approval for high-risk actions.",
)
self.user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER", # we handle human input externally
code_execution_config=False,
)
def process_conversation(self, conversation_id: str) -> None:
# Load with optimistic lock
for attempt in range(3):
raw = self.redis.get(f"conversation:{conversation_id}")
if not raw:
raise ValueError(f"Conversation {conversation_id} not found")
conv = HumanInLoopConversation.from_json(raw)
if conv.state != ConversationState.AGENT_TURN:
return # not our turn
# Run agent turn
new_messages = self._run_agent_turn(conv.messages)
conv.messages.extend(new_messages)
conv.version += 1
# Check if agent requested approval
last_msg = new_messages[-1] if new_messages else None
if self._requires_approval(last_msg):
conv.state = ConversationState.AWAITING_HUMAN
conv.pending_approval = self._extract_approval_request(last_msg)
conv.approval_requested_at = datetime.utcnow()
conv.approval_expires_at = datetime.utcnow() + timedelta(hours=24)
# Save and enqueue approval request
if self._try_save(conv):
self.approval_queue.enqueue(conv)
return
# Check for completion
if self._is_complete(last_msg):
conv.state = ConversationState.COMPLETE
self._try_save(conv)
return
# Continue agent loop
if self._try_save(conv):
continue # next iteration
# Optimistic lock failed, retry
raise RuntimeError("Max retries exceeded")
def _try_save(self, conv: HumanInLoopConversation) -> bool:
"""Optimistic lock save. Returns True on success."""
key = f"conversation:{conv.conversation_id}"
# Lua script for atomic check-and-set
script = """
local current = redis.call('GET', KEYS[1])
if not current then return 0 end
local data = cjson.decode(current)
if data.version ~= tonumber(ARGV[1]) then return 0 end
data.version = data.version + 1
redis.call('SET', KEYS[1], cjson.encode(data))
return 1
"""
result = self.redis.eval(script, 1, key, conv.version - 1, conv.to_json())
return result == 1The approval queue can be a simple Redis list, or a proper message broker (RabbitMQ, Kafka) if you need durability guarantees and dead-letter handling. The key insight: the agent worker is stateless and short-lived. It processes one turn, persists, and exits. This enables horizontal scaling behind a queue consumer.
Building the approval surface
Your approval API needs to handle: presenting the request to a human, capturing the response, enforcing expiration, and resuming the agent. Keep it separate from the agent worker — different scaling profile, different auth model, different deployment cadence.
# approval_api.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from conversation_state import HumanInLoopConversation, ConversationState
from agent_worker import ProductionAgentWorker
import redis
from datetime import datetime
app = FastAPI()
redis_client = redis.Redis(decode_responses=True)
class ApprovalResponse(BaseModel):
approved: bool
feedback: str | None = None
reviewer_id: str
@app.get("/approvals/{conversation_id}")
async def get_pending_approval(conversation_id: str):
raw = redis_client.get(f"conversation:{conversation_id}")
if not raw:
raise HTTPException(404, "Conversation not found")
conv = HumanInLoopConversation.from_json(raw)
if conv.state != ConversationState.AWAITING_HUMAN:
raise HTTPException(400, "No pending approval")
if conv.approval_expires_at and datetime.utcnow() > conv.approval_expires_at:
conv.state = ConversationState.FAILED
conv.metadata["failure_reason"] = "approval_expired"
redis_client.set(f"conversation:{conversation_id}", conv.to_json())
raise HTTPException(410, "Approval expired")
return {
"conversation_id": conversation_id,
"pending_approval": conv.pending_approval,
"context": conv.messages[-5:], # last 5 messages for context
"expires_at": conv.approval_expires_at.isoformat() if conv.approval_expires_at else None,
}
@app.post("/approvals/{conversation_id}")
async def submit_approval(
conversation_id: str,
response: ApprovalResponse,
background_tasks: BackgroundTasks,
):
raw = redis_client.get(f"conversation:{conversation_id}")
if not raw:
raise HTTPException(404, "Conversation not found")
conv = HumanInLoopConversation.from_json(raw)
if conv.state != ConversationState.AWAITING_HUMAN:
raise HTTPException(400, "No pending approval")
# Record the human response
conv.messages.append({
"role": "user",
"content": response.feedback or ("Approved" if response.approved else "Rejected"),
"metadata": {
"approval_response": True,
"approved": response.approved,
"reviewer_id": response.reviewer_id,
"responded_at": datetime.utcnow().isoformat(),
}
})
conv.pending_approval = None
conv.state = ConversationState.HUMAN_RESPONDED
conv.version += 1
# Optimistic save
key = f"conversation:{conversation_id}"
script = """
local current = redis.call('GET', KEYS[1])
if not current then return 0 end
local data = cjson.decode(current)
if data.version ~= tonumber(ARGV[1]) then return 0 end
redis.call('SET', KEYS[1], ARGV[2])
return 1
"""
saved = redis_client.eval(script, 1, key, conv.version - 1, conv.to_json())
if not saved:
raise HTTPException(409, "Concurrent modification, please retry")
# Resume agent worker asynchronously
background_tasks.add_task(resume_agent, conversation_id)
return {"status": "accepted", "conversation_id": conversation_id}
async def resume_agent(conversation_id: str):
# In production, push to a queue consumed by agent workers
# For simplicity, direct call here
worker = ProductionAgentWorker(redis_client, None, {})
worker.process_conversation(conversation_id)Observability you’ll actually use
AutoGen’s built-in logging captures model calls, but production needs correlation IDs, latency percentiles, and approval funnel metrics. Instrument three things: conversation lifecycle, approval SLA, and token spend per conversation.
# observability.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from functools import wraps
import time
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
)
tracer = trace.get_tracer(__name__)
def trace_conversation_turn(conversation_id: str):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
with tracer.start_as_current_span(
"agent_turn",
attributes={
"conversation_id": conversation_id,
"service.name": "autogen-worker",
},
) as span:
start = time.perf_counter()
try:
result = func(*args, **kwargs)
span.set_attribute("turn.success", True)
return result
except Exception as e:
span.set_attribute("turn.success", False)
span.record_exception(e)
raise
finally:
span.set_attribute("turn.duration_ms", (time.perf_counter() - start) * 1000)
return wrapper
return decorator
# Usage in agent_worker.py
@trace_conversation_turn(conversation_id)
def _run_agent_turn(self, messages: list[dict]) -> list[dict]:
# ... existing logicFor approval SLAs, emit a metric when entering AWAITING_HUMAN and when leaving it. Alert on p95 > 4 hours (or your SLA). Track approval rate vs rejection rate — a sudden spike in rejections often means the agent’s approval prompts are confusing.
# Approval SLA breach rate
rate(approval_duration_seconds_bucket{le="14400"}[5m])
/ rate(approval_duration_seconds_count[5m]) < 0.95
# Token cost per conversation (requires token counting in your LLM client)
histogram_quantile(0.95,
rate(llm_token_usage_total{conversation_id!=""}[5m])
)Scaling the worker pool
The agent worker is CPU-bound during model inference but I/O-bound waiting for Redis and the LLM API. Run multiple workers per CPU core (2-4x) since most time is network wait. Use a work-stealing queue (Redis BLMOVE or a proper broker) so slow conversations don’t block fast ones.
# worker_pool.py
import multiprocessing
import signal
import sys
from agent_worker import ProductionAgentWorker
import redis
class WorkerPool:
def __init__(self, num_workers: int, redis_url: str, approval_queue):
self.num_workers = num_workers
self.redis_url = redis_url
self.approval_queue = approval_queue
self.processes: list[multiprocessing.Process] = []
self._shutdown = False
def start(self):
for i in range(self.num_workers):
p = multiprocessing.Process(target=self._worker_loop, args=(i,))
p.start()
self.processes.append(p)
def _worker_loop(self, worker_id: int):
redis_client = redis.from_url(self.redis_url, decode_responses=True)
worker = ProductionAgentWorker(redis_client, self.approval_queue, {})
def handle_shutdown(signum, frame):
nonlocal running
running = False
signal.signal(signal.SIGTERM, handle_shutdown)
running = True
while running and not self._shutdown:
# Blocking pop with timeout for graceful shutdown
conversation_id = self.approval_queue.dequeue(timeout=5)
if conversation_id:
try:
worker.process_conversation(conversation_id)
except Exception as e:
# Log, emit metric, maybe dead-letter
pass
def shutdown(self):
self._shutdown = True
for p in self.processes:
p.join(timeout=30)
if p.is_alive():
p.terminate()Common pitfalls and tradeoffs
Pitfall: Treating approval as a synchronous RPC. If your approval API calls the agent worker directly, you’ve coupled their lifecycles. Use a queue. The approval API should return 202 Accepted immediately.
Pitfall: Losing context on resume. The agent needs the full conversation history, not just the last message. Persist everything. If context window is a concern, implement a summarization step as a separate agent turn before resuming.
Pitfall: No idempotency on approval submission. Humans double-click. The optimistic lock in the approval endpoint handles this, but your frontend should also disable the button after first click.
Tradeoff: Redis vs Postgres for state. Redis gives you sub-millisecond latency and built-in TTL for expiration. Postgres gives you durability, ad-hoc queries, and easier audit trails. Use Redis for hot conversations (active approvals), archive to Postgres on completion. A hybrid approach works well: write to both, read from Redis, fall back to Postgres on cache miss.
Tradeoff: Long-running vs short-lived workers. Long-running workers keep model connections warm but accumulate memory leaks. Short-lived workers (one conversation per process) are cleaner but pay connection overhead. For most teams, a middle ground — workers that process 50-100 conversations then recycle — balances both.
Pitfall: Ignoring provider failures mid-conversation. If your LLM provider returns 5xx or hits rate limits mid-turn, the conversation state must remain consistent. The worker should catch provider errors, leave state as AGENT_TURN, and re-queue with exponential backoff. If you’re using a gateway that handles automatic fallback across 240+ models, this becomes simpler — the gateway absorbs provider failures before they reach your worker.
Deployment checklist
Before shipping:
- Load test the approval funnel — simulate 1000 concurrent conversations with 30% approval rate. Verify p99 latency stays under your SLA.
- Chaos test state recovery — kill workers mid-turn, verify conversations resume correctly from Redis.
- Verify audit completeness — every state transition, every human decision, every model call should be traceable by conversation_id.
- Test approval expiration — ensure expired approvals transition to
FAILEDand trigger notifications. - Capacity plan token budgets — set per-conversation and per-tenant token limits, enforce in the worker before each model call.
The pattern scales. The same state machine, queue, and worker architecture handles code execution approvals, financial transaction reviews, content moderation escalations, and multi-step planning with human checkpoints. The investment in durable state and decoupled approval pays off every time you add a new human-in-the-loop workflow without rewriting the orchestration layer.