Speaker selection strategies in AutoGen GroupChat

Speaker selection is the backbone of any autogen groupchat speaker selection implementation. Get it wrong and your agents talk past each other, loop infinitely, or stall on the wrong task. Get it right and you unlock coherent multi-agent workflows that actually ship. This guide walks through the selection strategies AutoGen provides, when to reach for each, and how to compose them for real production workloads.

Understanding the selection interface

Every GroupChat instance accepts a speaker_selection_method parameter. This can be a string naming a built-in strategy, a callable that implements custom logic, or a list of transitions for explicit control. The method receives the current conversation state — message history, agent list, and the last speaker — and returns the next agent or None to terminate.

from autogen import GroupChat, Agent

groupchat = GroupChat(
    agents=[user_proxy, coder, reviewer, planner],
    messages=[],
    max_round=20,
    speaker_selection_method="auto"  # string, callable, or list
)

The method runs after every message. If it returns an agent not in the group, AutoGen raises. If it returns None, the chat ends. Keep this contract in mind — your custom selectors must honor it.

Built-in strategies and when to use them

AutoGen ships with three string-based selectors. Each has a distinct failure mode you should recognize before committing.

Round robin

Cycles through agents in list order. Predictable, deterministic, and rarely what you want for anything beyond toy demos.

speaker_selection_method="round_robin"

Pitfall: Forces every agent to speak every cycle. Your reviewer comments on the planner’s outline before code exists. Your coder writes implementation before requirements land. Use only when you genuinely need strict rotation — for example, a debate format where each persona must respond in turn.

Random

Picks uniformly from eligible agents. Useful for simulation or stress testing, harmful for production.

speaker_selection_method="random"

Pitfall: No semantic awareness. The documentation agent might speak after a bug report, or the planner might interrupt a debugging session. You’ll waste tokens and confuse context.

Auto (LLM-based)

The default. An LLM (by default the first agent’s LLM) reads the conversation and selects the next speaker. This is where most teams start and where many should stay — with guardrails.

speaker_selection_method="auto"

How it works: AutoGen constructs a prompt listing agents with their descriptions, the recent message history, and asks the LLM to pick the next speaker. The prompt template lives in autogen/agentchat/groupchat.py — read it. You’ll want to customize it.

Tradeoff: Flexible but opaque. The LLM can hallucinate agent names, pick the same agent repeatedly, or stall. Token cost scales with history length. For long conversations, truncate history or provide a summary agent.

Custom selector functions — the practical middle ground

A callable gives you deterministic control without hardcoding transitions. The signature:

def my_selector(last_speaker: Agent, agents: List[Agent], messages: List[Dict]) -> Agent | None:
    ...

Return the next agent or None to end. Here’s a pattern that handles 80% of real workflows: route by message content and role.

def content_aware_selector(last_speaker, agents, messages):
    if not messages:
        return agents[0]  # first agent starts
    
    last_msg = messages[-1]
    content = last_msg.get("content", "").lower()
    last_name = last_speaker.name if last_speaker else ""
    
    # Explicit handoffs via structured content
    if "handoff_to:" in content:
        target = content.split("handoff_to:")[1].split()[0].strip()
        for agent in agents:
            if agent.name == target:
                return agent
    
    # Role-based routing
    if last_name == "planner":
        return next(a for a in agents if a.name == "coder")
    if last_name == "coder":
        if "test" in content or "error" in content:
            return next(a for a in agents if a.name == "tester")
        return next(a for a in agents if a.name == "reviewer")
    if last_name == "reviewer":
        if "approve" in content or "lgtm" in content:
            return next(a for a in agents if a.name == "deployer")
        return next(a for a in agents if a.name == "coder")
    if last_name == "tester":
        if "pass" in content:
            return next(a for a in agents if a.name == "reviewer")
        return next(a for a in agents if a.name == "coder")
    
    # Default: planner kicks off new tasks
    return next(a for a in agents if a.name == "planner")

Why this works: Explicit handoffs in message content ("handoff_to: reviewer") let agents drive flow. Role-based fallbacks catch implicit transitions. The selector stays readable and testable.

Pitfall: Hardcoding agent names couples selector to team composition. Fix by reading from agent metadata:

def make_selector(role_map: Dict[str, str]):
    """role_map maps role -> agent name"""
    def selector(last_speaker, agents, messages):
        name_to_agent = {a.name: a for a in agents}
        # ... use role_map["coder"] instead of "coder"
    return selector

SpeakerTransition — explicit flow control

For workflows that must follow a strict graph (approval chains, compliance reviews), use a transition list. Each entry is (source, target) or (source, [targets]). AutoGen validates the graph at construction.

from autogen import SpeakerTransition

transitions = [
    SpeakerTransition("planner", ["coder", "architect"]),
    SpeakerTransition("coder", ["reviewer", "tester"]),
    SpeakerTransition("architect", ["coder"]),
    SpeakerTransition("reviewer", ["coder", "approver"]),
    SpeakerTransition("tester", ["coder", "reviewer"]),
    SpeakerTransition("approver", ["deployer", "planner"]),  # back to planner for next task
    SpeakerTransition("deployer", ["planner"]),
]

groupchat = GroupChat(
    agents=all_agents,
    messages=[],
    max_round=30,
    speaker_selection_method=transitions,
    allow_repeat_speaker=False  # prevents immediate self-loops
)

Key parameters:

• allow_repeat_speaker=False (default): Prevents an agent from speaking twice consecutively unless the transition explicitly allows it via a self-loop.
• allow_self_loop=False (default): Blocks SpeakerTransition("coder", "coder"). Set True if an agent must iterate internally.

Tradeoff: Transitions are auditable and debuggable but brittle. Adding a new agent requires updating the graph. Compose with a custom selector for dynamic segments:

def hybrid_selector(last_speaker, agents, messages):
    # First, check if a transition applies
    for trans in transitions:
        if trans.source == last_speaker.name:
            # If multiple targets, delegate to LLM or content
            if len(trans.targets) == 1:
                return name_to_agent[trans.targets[0]]
            # Multiple valid next speakers — use content awareness
            return content_aware_fallback(last_speaker, agents, messages)
    # No transition matched — fallback
    return content_aware_fallback(last_speaker, agents, messages)

LLM-based selection with custom prompts

When you need semantic routing but want control over the prompt, wrap the auto selector. AutoGen’s GroupChat uses select_speaker_auto internally — you can copy and modify its prompt.

from autogen.agentchat.groupchat import select_speaker_auto
from autogen import Agent

def prompted_selector(last_speaker, agents, messages):
    # Build a focused prompt
    agent_descriptions = "\n".join(
        f"- {a.name}: {a.system_message or 'No description'}" 
        for a in agents
    )
    
    recent = messages[-6:]  # last 6 messages only
    history = "\n".join(f"{m['name']}: {m['content'][:200]}" for m in recent)
    
    prompt = f"""You are a conversation router. Select the next speaker.

Agents:
{agent_descriptions}

Recent conversation:
{history}

Last speaker: {last_speaker.name if last_speaker else 'None'}

Rules:
1. Planner starts new tasks
2. Coder implements, reviewer reviews, tester tests
3. Only approver can approve for deployment
4. If last message contains "handoff_to:X", pick X
5. Return ONLY the agent name, nothing else.

Next speaker:"""
    
    # Use a cheap, fast model for routing
    router_llm = Agent(name="router", llm_config={"model": "gpt-4o-mini"})
    response = router_llm.generate_reply(messages=[{"role": "user", "content": prompt}])
    
    selected_name = response.strip()
    for agent in agents:
        if agent.name == selected_name:
            return agent
    
    # Fallback
    return agents[0]

Why a separate router agent: Keeps your main agents’ contexts clean. The router sees only what it needs. Use a small model (gpt-4o-mini, claude-3-haiku) — routing doesn’t need reasoning capacity.

Pitfall: The router can return invalid names. Always validate against the agent list and fallback. Log mismatches for debugging.

Hybrid strategy: transitions + LLM fallback

Production systems often need structure for the happy path and flexibility for exceptions. This pattern handles both:

def make_hybrid_selector(transitions, fallback_selector):
    trans_map = {}
    for t in transitions:
        trans_map.setdefault(t.source, []).extend(t.targets)
    
    def selector(last_speaker, agents, messages):
        if not last_speaker:
            return agents[0]
        
        # Explicit transition?
        valid_next = trans_map.get(last_speaker.name, [])
        if valid_next:
            # Filter to agents that exist
            valid_next = [n for n in valid_next if any(a.name == n for a in agents)]
            if len(valid_next) == 1:
                return next(a for a in agents if a.name == valid_next[0])
            # Multiple valid — use fallback to choose
            # Pass constrained agent list to fallback
            constrained_agents = [a for a in agents if a.name in valid_next]
            return fallback_selector(last_speaker, constrained_agents, messages)
        
        # No transition defined — full fallback
        return fallback_selector(last_speaker, agents, messages)
    
    return selector

Usage:

transitions = [
    SpeakerTransition("planner", ["coder"]),
    SpeakerTransition("coder", ["reviewer"]),
    SpeakerTransition("reviewer", ["coder", "approver"]),
]

selector = make_hybrid_selector(transitions, content_aware_selector)

groupchat = GroupChat(
    agents=all_agents,
    messages=[],
    max_round=25,
    speaker_selection_method=selector
)

This gives you auditable transitions for the core loop while letting the content-aware selector handle edge cases (errors, clarifications, parallel tracks).

Termination conditions

Speaker selection doesn’t exist in isolation. Pair it with termination logic or your chat runs until max_round.

def termination_check(messages):
    if not messages:
        return False
    last = messages[-1].get("content", "").lower()
    return any(kw in last for kw in ["deployed", "complete", "approved", "done"])

groupchat = GroupChat(
    agents=all_agents,
    messages=[],
    max_round=30,
    speaker_selection_method=selector,
    termination_condition=termination_check
)

Pitfall: Termination checks run after speaker selection. If your selector returns an agent but termination triggers, that agent never speaks. Order matters: put cheap checks in termination, expensive logic in selector.

Debugging speaker selection

When the chat stalls or loops, you need visibility. Add a wrapper that logs every decision:

def logged_selector(inner_selector):
    def wrapper(last_speaker, agents, messages):
        result = inner_selector(last_speaker, agents, messages)
        print(f"[SELECTOR] last={last_speaker.name if last_speaker else 'None'} "
              f"-> next={result.name if result else 'TERMINATE'} "
              f"round={len(messages)}")
        return result
    return wrapper

groupchat = GroupChat(
    agents=all_agents,
    messages=[],
    max_round=20,
    speaker_selection_method=logged_selector(hybrid_selector)
)

For production, swap print for structured logging with conversation ID, round number, and selector latency. This lets you trace loops in observability tools.

Common loop patterns to watch:

• Selector returns Agent A, Agent A’s reply triggers selector to return Agent A again (check allow_repeat_speaker)
• LLM selector hallucinates same agent repeatedly (add “do not repeat last speaker” to prompt)
• Transition graph has cycles without progress condition (add termination or progress tracking)

Testing selectors in isolation

Don’t test selectors only inside full GroupChat runs. Unit test the logic directly:

import pytest
from unittest.mock import MagicMock

def test_content_aware_selector_routes_coder_to_reviewer():
    agents = [
        MagicMock(name="planner"),
        MagicMock(name="coder"),
        MagicMock(name="reviewer"),
    ]
    messages = [{"role": "assistant", "name": "coder", "content": "Implementation complete"}]
    last_speaker = MagicMock(name="coder")
    
    next_agent = content_aware_selector(last_speaker, agents, messages)
    
    assert next_agent.name == "reviewer"

def test_selector_handles_handoff():
    agents = [MagicMock(name="coder"), MagicMock(name="tester")]
    messages = [{"role": "assistant", "name": "coder", "content": "handoff_to: tester"}]
    last_speaker = MagicMock(name="coder")
    
    next_agent = content_aware_selector(last_speaker, agents, messages)
    
    assert next_agent.name == "tester"

Mock agents with name attributes. Test edge cases: empty messages, unknown handoff targets, missing agents. This catches regressions when you add team members.

Performance considerations

Speaker selection runs every round. Keep it fast:

Rule of thumb: If your selector adds >100ms per round, you’ll feel it in 20-round conversations. Cache LLM router responses for repeated contexts. Truncate history passed to LLM selectors — 6-8 messages is usually sufficient.

n4n.ai note: When routing across 240+ models via a single endpoint, selector latency compounds with model routing latency. Keep selection logic local and fast; push model selection to the gateway layer.

Choosing your strategy — decision matrix

Migration path: Most teams start with "auto", hit a loop or cost issue, then migrate to hybrid. Plan for this. Write your custom selector first as a fallback, then promote it to primary once validated.

Common pitfalls summary

1. Forgetting allow_repeat_speaker=False — causes immediate self-loops when an agent’s reply triggers selection of the same agent.
2. LLM selector returning None or invalid name — always validate and fallback.
3. History growing unbounded in LLM selectors — truncate or summarize.
4. Hardcoding agent names in selectors — use role maps or agent metadata.
5. No termination condition — chat runs to max_round every time.
6. Selector logic depending on message format that agents don’t produce — define a handoff protocol and enforce it in agent system prompts.
7. Testing only in integration — unit test selectors with mocked agents and messages.

Putting it together: a production-ready template

from autogen import GroupChat, Agent, SpeakerTransition
from typing import List, Dict, Optional
import logging

logger = logging.getLogger(__name__)

def create_production_groupchat(
    agents: List[Agent],
    role_map: Dict[str, str],
    transitions: List[SpeakerTransition],
    max_rounds: int = 25
) -> GroupChat:
    """Factory for consistent GroupChat configuration."""
    
    name_to_agent = {a.name: a for a in agents}
    
    # Build transition map
    trans_map = {}
    for t in transitions:
        trans_map.setdefault(t.source, []).extend(t.targets)
    
    def selector(last_speaker: Optional[Agent], agents: List[Agent], messages: List[Dict]) -> Optional[Agent]:
        if not last_speaker:
            return name_to_agent[role_map["starter"]]
        
        # Check transitions
        valid_next_names = trans_map.get(last_speaker.name, [])
        if valid_next_names:
            valid_next = [name_to_agent[n] for n in valid_next_names if n in name_to_agent]
            if len(valid_next) == 1:
                return valid_next[0]
            # Multiple valid — route by content
            return route_by_content(last_speaker, valid_next, messages)
        
        # No transition — full content routing
        return route_by_content(last_speaker, agents, messages)
    
    def route_by_content(last_speaker, candidates, messages):
        last_msg = messages[-1].get("content", "") if messages else ""
        
        # Explicit handoff
        if "handoff_to:" in last_msg:
            target = last_msg.split("handoff_to:")[1].split()[0].strip()
            if target in name_to_agent and name_to_agent[target] in candidates:
                return name_to_agent[target]
        
        # Role-based heuristics
        last_role = get_role(last_speaker.name, role_map)
        if last_role == "planner":
            return pick(candidates, "coder", role_map)
        if last_role == "coder":
            if any(kw in last_msg.lower() for kw in ["test", "error", "fail", "bug"]):
                return pick(candidates, "tester", role_map)
            return pick(candidates, "reviewer", role_map)
        if last_role == "reviewer":
            if any(kw in last_msg.lower() for kw in ["approve", "lgtm", "looks good"]):
                return pick(candidates, "approver", role_map)
            return pick(candidates, "coder", role_map)
        if last_role == "tester":
            if "pass" in last_msg.lower():
                return pick(candidates, "reviewer", role_map)
            return pick(candidates, "coder", role_map)
        
        # Default
        return candidates[0]
    
    def get_role(agent_name, role_map):
        for role, name in role_map.items():
            if name == agent_name:
                return role
        return "unknown"
    
    def pick(candidates, role, role_map):
        target_name = role_map.get(role)
        if target_name:
            for c in candidates:
                if c.name == target_name:
                    return c
        return candidates[0]
    
    def termination(messages):
        if not messages:
            return False
        last = messages[-1].get("content", "").lower()
        return any(kw in last for kw in ["deployed", "complete", "approved", "done"])
    
    # Wrap for logging
    def logged_selector(last_speaker, agents, messages):
        result = selector(last_speaker, agents, messages)
        logger.info(
            "speaker_selected",
            extra={
                "last_speaker": last_speaker.name if last_speaker else None,
                "next_speaker": result.name if result else "TERMINATE",
                "round": len(messages),
                "conversation_id": getattr(messages[0], "conversation_id", "unknown") if messages else "unknown"
            }
        )
        return result
    
    return GroupChat(
        agents=agents,
        messages=[],
        max_round=max_rounds,
        speaker_selection_method=logged_selector,
        termination_condition=termination,
        allow_repeat_speaker=False
    )

Use this factory. Swap role_map and transitions per workflow. Keep the routing logic in one place. Test the selector independently. Log every decision. That’s how you ship autogen groupchat speaker selection that doesn’t wake you at 3 AM.