How to Implement Custom Agent Routing in SelectorGroupChat: Complete Guide

Override the default LLM-based speaker selection in AutoGen's SelectorGroupChat by providing a Python callable to the selector_func parameter, which receives the full conversation history and returns the next agent name or None to delegate back to the model.

SelectorGroupChat is a multi-agent team implementation in the Microsoft AutoGen framework where agents take turns speaking based on a selection mechanism. While the default behavior uses a language model to choose the next speaker via a selector prompt, the architecture allows you to inject custom agent routing logic through a user-defined function. This gives you deterministic control over turn-taking while preserving the ability to fall back to LLM-based selection when needed.

Where Custom Routing Lives in the Source Code

The custom routing logic is implemented in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py.

The SelectorGroupChatManager class handles turn-taking through its select_speaker() method. At lines 62-71, the manager first checks whether a selector_func was supplied during construction. If present, the function is invoked with the current message history. The manager detects whether the callable is asynchronous using iscoroutinefunction and awaits it if necessary. When the function returns a non-None string, that name is validated against registered participants and returned directly as the next speaker. If the function returns None, the execution falls back to the built-in LLM-based selector at lines 79-101.

The public SelectorGroupChat component stores the selector_func in self._selector_func during initialization (lines 98-103) and passes it unchanged to the manager factory via _create_group_chat_manager_factory (lines 74-77).

The Selector Function Signature

The expected type for custom routing is defined as:

SelectorFuncType = Union[
    Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], str | None],
    Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[str | None]],
]

• Input: A sequence of BaseAgentEvent or BaseChatMessage objects representing the complete conversation history
• Output: The string name of the participant that should speak next, or None to trigger the default LLM-based selection

Implementing a Custom Selector Function

Synchronous Selector for Deterministic Routing

For simple round-robin or rule-based routing, implement a synchronous function that inspects the message history and returns the appropriate agent name.

from typing import Sequence
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage

def simple_selector(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
    """
    Alternate between two agents: if the last message came from AgentA,
    let AgentB speak next; otherwise let AgentA speak.
    """
    if not messages:                     # First turn → let LLM decide

        return None                      
    last = messages[-1]
    return "AgentB" if last.source == "AgentA" else "AgentA"

Wire the function into the team configuration:

from autogen_agentchat.teams import SelectorGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
import asyncio

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent_a = AssistantAgent("AgentA", model_client, description="First agent")
    agent_b = AssistantAgent("AgentB", model_client, description="Second agent")

    termination = TextMentionTermination("DONE")

    team = SelectorGroupChat(
        participants=[agent_a, agent_b],
        model_client=model_client,
        selector_func=simple_selector,          # Custom routing logic

        termination_condition=termination,
    )

    await Console(team.run_stream(task="Start the conversation."))
    
asyncio.run(main())

Asynchronous Selector for External API Integration

When routing decisions require external services or database lookups, use an async function. The manager automatically detects coroutines and awaits them.

import aiohttp
from typing import Sequence
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage

async def api_based_selector(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
    """Fetch next speaker from an external routing service."""
    async with aiohttp.ClientSession() as session:
        async with session.get("https://example.com/next-speaker") as resp:
            data = await resp.json()
            # Assume the service returns {"speaker": "AgentX"}

            return data.get("speaker")

Pass this async callable to selector_func exactly as you would a synchronous one. The select_speaker() method handles the await logic internally.

Validation and Fallback Behavior

The SelectorGroupChatManager enforces strict validation on custom routing returns. According to lines 71-75 in _selector_group_chat.py:

1. Existence Validation: The returned name must exist in self._participant_names. If you return an unregistered agent name, the manager raises a ValueError immediately.
2. Fallback Mechanism: Returning None from your selector function signals the manager to proceed with the default LLM-based selection logic. This allows hybrid approaches where you handle specific scenarios programmatically but delegate complex decisions to the model.
3. Repeated Speaker Handling: While the manager respects allow_repeated_speaker when using LLM selection, your custom function may return any valid name regardless of the previous speaker, including the same agent consecutively.

Combining Custom Routing with Candidate Filtering

For advanced scenarios, you can combine selector_func with candidate_func to first restrict the eligible pool, then apply custom selection logic. The manager first evaluates candidate_func to obtain the allowed subset of participants, then invokes selector_func (if provided) using that reduced list.

def candidate_filter(messages):
    """Only allow worker agents to be selected."""
    return [name for name in ["Worker1", "Worker2", "Manager"] if "Worker" in name]

def custom_selector(messages):
    """Custom logic applied only to filtered candidates."""
    # Implementation here receives only Worker1 and Worker2 as options

    return "Worker1"

team = SelectorGroupChat(
    participants=[worker1, worker2, manager],
    model_client=model_client,
    selector_func=custom_selector,
    candidate_func=candidate_filter,
    allow_repeated_speaker=False,
)

Summary

• Custom routing in SelectorGroupChat is implemented via the selector_func parameter, which accepts either synchronous or asynchronous callables.
• The function receives the full conversation history as Sequence[BaseAgentEvent | BaseChatMessage] and must return a valid participant name or None.
• Validation occurs in SelectorGroupChatManager.select_speaker() at lines 71-75, ensuring returned names exist in the participant registry.
• Async support is automatic—the manager detects coroutines with iscoroutinefunction and awaits them appropriately.
• Fallback to LLM happens when your function returns None, allowing hybrid human-coded and model-driven routing strategies.
• The implementation lives in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py with the public API exposed through SelectorGroupChat.__init__.

Frequently Asked Questions

What happens if my custom selector returns an invalid agent name?

The SelectorGroupChatManager validates the returned name against self._participant_names at lines 71-75 of _selector_group_chat.py. If the name is not registered in the team, the manager raises a ValueError immediately, preventing the conversation from proceeding with an undefined participant.

Can I mix custom routing with the default LLM-based selection?

Yes. Design your selector_func to return None for turns where you want the LLM to decide. When the function returns None, the manager falls back to the built-in selector logic (lines 79-101), allowing you to handle specific edge cases programmatically while delegating general routing to the model.

Does the custom selector work with async agents and external APIs?

Absolutely. The selector_func signature supports Awaitable[str | None] return types. The manager detects async functions using iscoroutinefunction (lines 94-95) and properly awaits them. This enables database queries, REST API calls, or other I/O-bound routing decisions without blocking the event loop.

How do I prevent the same agent from speaking twice in a row when using custom routing?

The allow_repeated_speaker parameter only affects the default LLM-based selector. When implementing selector_func, you must enforce anti-repetition logic manually by tracking the last speaker in the message history and selecting a different participant. The custom function has full control over turn-taking and is not restricted by the built-in repetition checks.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}