Implementation:Microsoft Autogen SequentialRoutedAgent

Key	Value
id	Microsoft_Autogen_SequentialRoutedAgent
source	Microsoft_Autogen
category	Core Agent

Overview

Description

The SequentialRoutedAgent is a specialized subclass of autogen_core.RoutedAgent that enforces sequential processing of specific message types using a First-In-First-Out (FIFO) lock. This ensures that messages of designated types are processed in the exact order they are received, preventing race conditions in distributed multi-agent systems.

The class implements:

• FIFOLock: A custom lock ensuring coroutines acquire the lock in request order
• Selective Sequential Processing: Only specified message types use the FIFO lock
• Concurrent Processing: Non-sequential message types are processed normally without locking
• Order Preservation: Critical for maintaining conversation coherence in group chats

This is essential for group chat scenarios where message order matters. For example, when multiple agents respond simultaneously, their responses must be processed in the order they were published to maintain a coherent conversation thread.

Usage

SequentialRoutedAgent is primarily used as a base class for agents that participate in distributed group chat systems, particularly ChatAgentContainer. It ensures that group chat control messages (start, request, response, etc.) are processed sequentially even when multiple messages arrive concurrently.

Key use cases:

• Agent containers in group chat teams
• Any agent that needs to maintain strict message ordering
• Distributed systems where race conditions on message handling would cause issues

The selective locking approach means only critical message types incur synchronization overhead, while other messages can be processed concurrently for better performance.

Code Reference

Source Location

• Repository: https://github.com/microsoft/autogen
• File Path: /tmp/kapso_repo_2mr4n2g4/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_sequential_routed_agent.py
• Lines: 1-73

Signature

class FIFOLock:
    """A lock that ensures coroutines acquire the lock in the order they request it."""

    def __init__(self) -> None:
        self._queue = asyncio.Queue[asyncio.Event]()
        self._locked = False

    async def acquire(self) -> None:
        ...

    def release(self) -> None:
        ...


class SequentialRoutedAgent(RoutedAgent):
    """A subclass of RoutedAgent that ensures messages of certain types
    are processed sequentially using a FIFO lock."""

    def __init__(self, description: str, sequential_message_types: Sequence[type[Any]]) -> None:
        ...

    async def on_message_impl(self, message: Any, ctx: MessageContext) -> Any | None:
        ...

Import

from autogen_agentchat.teams._group_chat import SequentialRoutedAgent

I/O Contract

FIFOLock

Method	Parameters	Returns	Description
__init__	None	None	Initialize the FIFO lock with empty queue
acquire	None	None (async)	Acquire lock in FIFO order; waits if lock is held
release	None	None	Release lock; grants to next waiting coroutine

SequentialRoutedAgent

Parameter	Type	Required	Description
description	str	Yes	Description of the agent (passed to RoutedAgent)
sequential_message_types	Sequence[type[Any]]	Yes	Message types that require sequential processing with FIFO lock

Message Processing Behavior

Message Type	Processing Mode	Description
In sequential_message_types	Sequential (with FIFO lock)	Processed in exact arrival order, one at a time
Not in sequential_message_types	Concurrent (no lock)	Processed immediately without waiting

Usage Examples

Basic Usage

from autogen_agentchat.teams._group_chat import SequentialRoutedAgent
from autogen_core import MessageContext


class MyGroupChatAgent(SequentialRoutedAgent):
    def __init__(self):
        super().__init__(
            description="My agent",
            sequential_message_types=[
                GroupChatStart,
                GroupChatRequestPublish,
                GroupChatReset
            ]
        )

    async def handle_message(self, message: Any, ctx: MessageContext):
        # If message type is in sequential_message_types,
        # this will be called with FIFO lock held
        # Otherwise, called without lock
        pass

ChatAgentContainer Example

from autogen_agentchat.teams._group_chat import SequentialRoutedAgent
from autogen_agentchat.teams._group_chat._events import (
    GroupChatStart,
    GroupChatRequestPublish,
    GroupChatReset,
    GroupChatAgentResponse,
    GroupChatTeamResponse
)


class ChatAgentContainer(SequentialRoutedAgent):
    def __init__(self, parent_topic_type: str, output_topic_type: str, agent, message_factory):
        # Ensure these message types are processed sequentially
        super().__init__(
            description=agent.description,
            sequential_message_types=[
                GroupChatStart,           # Start messages must be ordered
                GroupChatRequestPublish,  # Request processing must be sequential
                GroupChatReset,           # Reset must happen in order
                GroupChatAgentResponse,   # Agent responses must maintain order
                GroupChatTeamResponse     # Team responses must maintain order
            ]
        )
        self._agent = agent
        # ... rest of initialization ...

FIFO Lock Behavior

import asyncio


async def demonstrate_fifo_lock():
    from autogen_agentchat.teams._group_chat._sequential_routed_agent import FIFOLock

    lock = FIFOLock()
    results = []

    async def worker(id: int, delay: float):
        # Workers request lock in order: 1, 2, 3
        # They will acquire in same order regardless of processing time
        await lock.acquire()
        try:
            await asyncio.sleep(delay)
            results.append(id)
        finally:
            lock.release()

    # Start workers - they request lock in order
    await asyncio.gather(
        worker(1, 0.1),  # Requests first, completes slow
        worker(2, 0.01), # Requests second, would complete fast
        worker(3, 0.01)  # Requests third, would complete fast
    )

    # Results are [1, 2, 3] - FIFO order preserved
    assert results == [1, 2, 3]

Selective Sequential Processing

from autogen_agentchat.teams._group_chat import SequentialRoutedAgent
from autogen_core import event


class MyAgent(SequentialRoutedAgent):
    def __init__(self):
        super().__init__(
            description="Selective agent",
            sequential_message_types=[CriticalMessage]
        )

    @event
    async def handle_critical(self, message: CriticalMessage, ctx):
        # This uses FIFO lock - processed sequentially
        await process_critically(message)

    @event
    async def handle_normal(self, message: NormalMessage, ctx):
        # This does NOT use FIFO lock - processed concurrently
        await process_normally(message)


async def test_selective_processing():
    agent = MyAgent()

    # These are processed concurrently (no lock)
    await asyncio.gather(
        agent.handle_normal(NormalMessage(), ctx),
        agent.handle_normal(NormalMessage(), ctx),
        agent.handle_normal(NormalMessage(), ctx)
    )

    # These are processed sequentially (FIFO lock)
    await asyncio.gather(
        agent.handle_critical(CriticalMessage(), ctx),
        agent.handle_critical(CriticalMessage(), ctx),
        agent.handle_critical(CriticalMessage(), ctx)
    )

Exception Safety

from autogen_agentchat.teams._group_chat import SequentialRoutedAgent


class SafeAgent(SequentialRoutedAgent):
    async def on_message_impl(self, message, ctx):
        if any(isinstance(message, t) for t in self._sequential_message_types):
            # Lock is acquired
            await self._fifo_lock.acquire()
            try:
                # Even if this raises, lock is released
                return await super().on_message_impl(message, ctx)
            finally:
                # CRITICAL: Lock is always released
                self._fifo_lock.release()
        else:
            # No lock needed
            return await super().on_message_impl(message, ctx)


async def test_exception_safety():
    agent = SafeAgent(
        description="test",
        sequential_message_types=[ErrorMessage]
    )

    try:
        # Even if this raises exception...
        await agent.on_message_impl(ErrorMessage(), ctx)
    except Exception:
        pass

    # ...the lock is still released for next message
    await agent.on_message_impl(ErrorMessage(), ctx)  # Won't deadlock

Performance Comparison

import time


async def benchmark_sequential_vs_concurrent():
    class TestAgent(SequentialRoutedAgent):
        def __init__(self, sequential: bool):
            types = [TestMessage] if sequential else []
            super().__init__(description="test", sequential_message_types=types)

        async def handle(self, message, ctx):
            await asyncio.sleep(0.01)  # Simulate work

    # Sequential processing
    agent_seq = TestAgent(sequential=True)
    start = time.time()
    await asyncio.gather(*[
        agent_seq.on_message_impl(TestMessage(), ctx)
        for _ in range(10)
    ])
    sequential_time = time.time() - start
    print(f"Sequential: {sequential_time:.2f}s")

    # Concurrent processing
    agent_conc = TestAgent(sequential=False)
    start = time.time()
    await asyncio.gather(*[
        agent_conc.on_message_impl(TestMessage(), ctx)
        for _ in range(10)
    ])
    concurrent_time = time.time() - start
    print(f"Concurrent: {concurrent_time:.2f}s")

    # Sequential is ~10x slower (processes one at a time)
    # Concurrent is ~10x faster (processes all in parallel)

Custom Sequential Types

from pydantic import BaseModel


class OrderedMessage(BaseModel):
    """Message that must be processed in order."""
    sequence_number: int
    content: str


class UnorderedMessage(BaseModel):
    """Message that can be processed concurrently."""
    content: str


class CustomAgent(SequentialRoutedAgent):
    def __init__(self):
        super().__init__(
            description="Custom agent",
            sequential_message_types=[
                OrderedMessage,  # Must be sequential
                # UnorderedMessage not included - will be concurrent
            ]
        )

    @event
    async def handle_ordered(self, message: OrderedMessage, ctx):
        # Guaranteed to process in arrival order
        print(f"Processing ordered message {message.sequence_number}")

    @event
    async def handle_unordered(self, message: UnorderedMessage, ctx):
        # Can process concurrently
        print(f"Processing unordered message")

Related Pages

• Microsoft_Autogen_ChatAgentContainer - Primary user of SequentialRoutedAgent
• Microsoft_Autogen_RoutedAgent - Base class from autogen_core
• Microsoft_Autogen_GroupChat_Events - Events that require sequential processing
• Microsoft_Autogen_BaseGroupChat - Group chat orchestration system
• Microsoft_Autogen_MessageContext - Context passed to message handlers
• Microsoft_Autogen_RoundRobinGroupChat - Group chat using sequential agents
• Microsoft_Autogen_SelectorGroupChat - Group chat using sequential agents
• Microsoft_Autogen_MagenticOneGroupChat - Group chat using sequential agents