Human review checkpoints in AutoGen GroupChat

Adding a human review checkpoint to an autogen groupchat human review checkpoint workflow lets you pause multi-agent execution, inspect intermediate state, and decide whether to continue, revise, or abort. This pattern is essential when agents generate code, make external API calls, or produce output that requires compliance review before proceeding. Below is a complete, runnable implementation you can adapt to your own GroupChat pipelines.

Step 1: Define the checkpoint contract

Before wiring any AutoGen components, decide what the human reviewer sees and what actions they can take. A minimal checkpoint payload includes the current message history, the agent that produced the last message, and a structured decision enum.

# checkpoint.py
from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
from autogen import Agent, ConversableAgent

class CheckpointDecision(Enum):
    APPROVE = "approve"          # Continue normal execution
    REVISE = "revise"            # Send feedback to the last agent
    ABORT = "abort"              # Stop the GroupChat entirely

@dataclass
class CheckpointPayload:
    messages: List[dict]         # Full conversation history so far
    last_speaker: str            # Name of the agent that just spoke
    turn_number: int             # Monotonic counter for debugging
    context: dict                # Arbitrary domain data (e.g., PR diff, SQL query)

@dataclass
class CheckpointResult:
    decision: CheckpointDecision
    feedback: Optional[str] = None   # Required when decision == REVISE
    metadata: Optional[dict] = None  # Optional audit trail

This contract keeps the review UI decoupled from the agent logic. The context field carries domain-specific artifacts — code diffs, SQL, API responses — so reviewers don’t need to reconstruct state from raw messages.

Step 2: Build a reusable checkpoint agent

Create a dedicated HumanCheckpointAgent that inherits from ConversableAgent but overrides generate_reply to block until a reviewer responds. The agent does not call an LLM; it delegates to an external callback you provide (CLI, web UI, Slack bot, etc.).

# human_checkpoint_agent.py
from typing import Callable, Awaitable
from autogen import ConversableAgent
from checkpoint import CheckpointPayload, CheckpointResult, CheckpointDecision

class HumanCheckpointAgent(ConversableAgent):
    def __init__(
        self,
        name: str,
        review_callback: Callable[[CheckpointPayload], Awaitable[CheckpointResult]],
        **kwargs
    ):
        super().__init__(name=name, **kwargs)
        self._review_callback = review_callback
        self._turn_counter = 0

    async def generate_reply(self, *args, **kwargs) -> str:
        # Build payload from current chat history
        payload = CheckpointPayload(
            messages=self.chat_messages[self],
            last_speaker=self._last_speaker_name(),
            turn_number=self._turn_counter,
            context=self._extract_context()
        )
        self._turn_counter += 1

        result = await self._review_callback(payload)

        if result.decision == CheckpointDecision.APPROVE:
            return "CHECKPOINT_APPROVED"   # Special token recognized by GroupChat manager
        elif result.decision == CheckpointDecision.REVISE:
            # Inject feedback as a user message so the last agent can react
            self.receive(result.feedback or "Please revise.", self, request_reply=True)
            return "CHECKPOINT_REVISE"
        else:  # ABORT
            raise RuntimeError("Human reviewer aborted the workflow")

    def _last_speaker_name(self) -> str:
        # The most recent message in *this* agent's view that wasn't from itself
        for msg in reversed(self.chat_messages[self]):
            if msg["name"] != self.name:
                return msg["name"]
        return "unknown"

    def _extract_context(self) -> dict:
        # Override in subclasses to pull domain artifacts from the last message
        return {}

The special return tokens CHECKPOINT_APPROVED and CHECKPOINT_REVISE are consumed by a custom GroupChatManager in the next step. Raising RuntimeError on abort cleanly unwinds the async call stack.

Step 3: Extend GroupChatManager to honor checkpoint tokens

AutoGen’s default GroupChatManager treats every agent reply as a normal continuation. Subclass it to intercept the checkpoint tokens and control speaker selection accordingly.

# checkpoint_manager.py
from typing import List, Optional
from autogen import GroupChat, GroupChatManager, Agent
from human_checkpoint_agent import HumanCheckpointAgent

class CheckpointAwareManager(GroupChatManager):
    def __init__(self, groupchat: GroupChat, **kwargs):
        super().__init__(groupchat=groupchat, **kwargs)
        self._checkpoint_agent: Optional[HumanCheckpointAgent] = None
        self._awaiting_revision = False

    def _find_checkpoint_agent(self) -> Optional[HumanCheckpointAgent]:
        for agent in self.groupchat.agents:
            if isinstance(agent, HumanCheckpointAgent):
                return agent
        return None

    async def _select_speaker(self, last_speaker: Agent, selector: GroupChat) -> Agent:
        # If we just got a REVISE token, force the *previous* speaker to reply again
        if self._awaiting_revision:
            self._awaiting_revision = False
            return self._previous_speaker(last_speaker)

        speaker = await super()._select_speaker(last_speaker, selector)

        # If the selected speaker is the checkpoint agent, run it and interpret the token
        if isinstance(speaker, HumanCheckpointAgent):
            self._checkpoint_agent = speaker
            reply = await speaker.generate_reply()
            if reply == "CHECKPOINT_APPROVED":
                # Skip the checkpoint agent; move to next logical speaker
                return await super()._select_speaker(speaker, selector)
            elif reply == "CHECKPOINT_REVISE":
                self._awaiting_revision = True
                return self._previous_speaker(speaker)
        return speaker

    def _previous_speaker(self, current: Agent) -> Agent:
        # Walk chat history backward to find the last non-checkpoint agent
        for msg in reversed(self.groupchat.messages):
            candidate = next((a for a in self.groupchat.agents if a.name == msg["name"]), None)
            if candidate and not isinstance(candidate, HumanCheckpointAgent):
                return candidate
        return current  # Fallback (should not happen in well-formed flows)

Register this manager instead of the stock GroupChatManager when you construct the GroupChat.

Step 4: Wire a concrete review callback (CLI example)

For local development and CI pipelines, a blocking CLI callback is the fastest way to verify the loop works. Replace this with a webhook, Slack views.open, or internal dashboard in production.

# cli_review.py
import asyncio
import json
from checkpoint import CheckpointPayload, CheckpointResult, CheckpointDecision

async def cli_review_callback(payload: CheckpointPayload) -> CheckpointResult:
    print("\n" + "=" * 60)
    print(f"CHECKPOINT #{payload.turn_number}  |  Last speaker: {payload.last_speaker}")
    print("=" * 60)
    print("Recent messages:")
    for msg in payload.messages[-4:]:  # Show last 4 for context
        print(f"  [{msg['name']}] {msg['content'][:200]}")
    if payload.context:
        print("\nContext:")
        print(json.dumps(payload.context, indent=2, default=str))

    while True:
        choice = input("\nDecision [a]pprove / [r]evise / [x]abort: ").strip().lower()
        if choice == "a":
            return CheckpointResult(decision=CheckpointDecision.APPROVE)
        if choice == "r":
            feedback = input("Feedback for the agent: ").strip()
            return CheckpointResult(decision=CheckpointDecision.REVISE, feedback=feedback)
        if choice == "x":
            return CheckpointResult(decision=CheckpointDecision.ABORT)
        print("Invalid choice. Enter a, r, or x.")

Step 5: Assemble the full GroupChat with a checkpoint

Now compose the agents — your task agents plus the checkpoint agent — and start the chat. The example below uses a coder agent and a reviewer agent with a checkpoint after every coder turn.

# main.py
import asyncio
from autogen import UserProxyAgent, AssistantAgent, GroupChat
from human_checkpoint_agent import HumanCheckpointAgent
from checkpoint_manager import CheckpointAwareManager
from cli_review import cli_review_callback

# 1. Task agents
coder = AssistantAgent(
    name="coder",
    system_message="Write concise Python functions. Output only the code block.",
    llm_config={"config_list": [{"model": "gpt-4o-mini", "api_key": "YOUR_KEY"}]}
)

reviewer = AssistantAgent(
    name="reviewer",
    system_message="Review the code for correctness and style. Reply with 'APPROVED' or specific changes.",
    llm_config={"config_list": [{"model": "gpt-4o-mini", "api_key": "YOUR_KEY"}]}
)

# 2. Checkpoint agent inserted after coder
checkpoint = HumanCheckpointAgent(
    name="human_checkpoint",
    review_callback=cli_review_callback,
    human_input_mode="NEVER"   # We drive input via the callback
)

# 3. GroupChat with custom manager
groupchat = GroupChat(
    agents=[coder, checkpoint, reviewer],
    messages=[],
    max_round=12,
    speaker_selection_method="round_robin"  # Deterministic order for demo
)

manager = CheckpointAwareManager(groupchat=groupchat)

# 4. Kick off
user_proxy = UserProxyAgent(name="user", human_input_mode="NEVER")
asyncio.run(user_proxy.a_initiate_chat(
    manager,
    message="Write a function that parses ISO 8601 durations (e.g., 'PT1H30M') into timedelta."
))

Run python main.py. The coder produces code, the checkpoint pauses, you inspect and decide, and the loop continues until the reviewer says APPROVED or you abort.

Step 6: Verify success with automated tests

Manual CLI verification is fine for development, but CI needs a non-interactive path. Provide a MockReviewCallback that simulates approve/revise/abort sequences so you can assert final state.

# test_checkpoint.py
import asyncio
from unittest.mock import AsyncMock
from autogen import UserProxyAgent
from main import coder, reviewer, checkpoint, manager
from checkpoint import CheckpointResult, CheckpointDecision

async def test_approve_on_first_try():
    # Replace the real callback with a mock that always approves
    checkpoint._review_callback = AsyncMock(return_value=CheckpointResult(
        decision=CheckpointDecision.APPROVE
    ))

    user = UserProxyAgent(name="test_user", human_input_mode="NEVER")
    await user.a_initiate_chat(
        manager,
        message="Return the integer 42.",
        max_turns=3
    )

    # The final message should be from reviewer with APPROVED
    final_msg = manager.groupchat.messages[-1]
    assert final_msg["name"] == "reviewer"
    assert "APPROVED" in final_msg["content"]
    print("✓ test_approve_on_first_try passed")

async def test_revise_then_approve():
    calls = 0
    async def mock_callback(payload):
        nonlocal calls
        calls += 1
        if calls == 1:
            return CheckpointResult(decision=CheckpointDecision.REVISE, feedback="Add type hints")
        return CheckpointResult(decision=CheckpointDecision.APPROVE)

    checkpoint._review_callback = mock_callback

    user = UserProxyAgent(name="test_user", human_input_mode="NEVER")
    await user.a_initiate_chat(
        manager,
        message="Write a function add(a, b) -> int.",
        max_turns=6
    )

    # Should have gone: coder -> checkpoint(REVISE) -> coder(revised) -> checkpoint(APPROVE) -> reviewer
    assert calls == 2
    print("✓ test_revise_then_approve passed")

if __name__ == "__main__":
    asyncio.run(test_approve_on_first_try())
    asyncio.run(test_revise_then_approve())

Run pytest test_checkpoint.py -v. Both tests should pass in under 10 seconds with a fast model.

Step 7: Harden for production

The CLI callback and round-robin selection are teaching tools. Before shipping, address these gaps:

Step 8: Extend the pattern — multi-stage checkpoints

Real workflows often need different reviewers at different stages (security, legal, product). Parameterize the checkpoint agent with a stage label and route to the appropriate callback registry.

# staged_checkpoint.py
from human_checkpoint_agent import HumanCheckpointAgent
from checkpoint import CheckpointPayload, CheckpointResult

STAGE_CALLBACKS: dict[str, Callable[[CheckpointPayload], Awaitable[CheckpointResult]]] = {
    "security": security_review_callback,
    "legal": legal_review_callback,
    "product": product_review_callback,
}

class StagedCheckpointAgent(HumanCheckpointAgent):
    def __init__(self, stage: str, **kwargs):
        callback = STAGE_CALLBACKS.get(stage)
        if not callback:
            raise ValueError(f"No callback registered for stage '{stage}'")
        super().__init__(name=f"checkpoint_{stage}", review_callback=callback, **kwargs)
        self.stage = stage

Insert StagedCheckpointAgent("security") after the coder, StagedCheckpointAgent("legal") after the reviewer, etc. Each stage gets its own UI, SLA, and approver group.

You now have a production-ready pattern: a typed checkpoint contract, a reusable HumanCheckpointAgent, a CheckpointAwareManager that understands approve/revise/abort tokens, a CLI callback for local iteration, and automated tests that run in CI. Swap the callback for your internal review dashboard, add the hardening items in Step 7, and you have a human-in-the-loop gate that works reliably at scale.