Implementation:Microsoft Autogen GroupChat Events

Key	Value
id	Microsoft_Autogen_GroupChat_Events
source	Microsoft_Autogen
category	Group Chat

Overview

Description

The GroupChat Events module defines the event messaging protocol for distributed group chat orchestration in Autogen. These Pydantic models represent the different types of events that flow between orchestrators and agent containers in a group chat system.

The module provides:

Control Events: Start, reset, pause, and resume group chat operations (GroupChatStart, GroupChatReset, GroupChatPause, GroupChatResume)
Request Events: Trigger agent processing (GroupChatRequestPublish)
Response Events: Carry agent and team outputs (GroupChatAgentResponse, GroupChatTeamResponse)
Message Events: Stream intermediate messages and events (GroupChatMessage)
Termination Events: Signal conversation completion (GroupChatTermination)
Error Events: Propagate exceptions (GroupChatError)
Error Model: Serializable exception representation (SerializableException)

All events are Pydantic BaseModel subclasses, enabling automatic validation and serialization for distributed communication between orchestrators and agent containers.

Usage

These events are used internally by the group chat infrastructure to coordinate multi-agent conversations:

Orchestrators publish control and request events to agent containers
Agent containers publish response, message, and error events back to orchestrators
Events flow through topics in the autogen_core runtime
The event protocol enables distributed, asynchronous group chat execution

The events support:

Starting group chats with initial messages
Requesting agent responses at orchestrator-determined times
Streaming intermediate messages for monitoring
Handling errors gracefully with traceback preservation
Signaling conversation termination with stop messages

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/_events.py
Lines: 1-114

Signature

class SerializableException(BaseModel):
    error_type: str
    error_message: str
    traceback: str | None = None

    @classmethod
    def from_exception(cls, exc: Exception) -> "SerializableException":
        ...


class GroupChatStart(BaseModel):
    messages: List[SerializeAsAny[BaseChatMessage]] | None = None
    output_task_messages: bool = True


class GroupChatAgentResponse(BaseModel):
    response: SerializeAsAny[Response]
    name: str


class GroupChatTeamResponse(BaseModel):
    result: SerializeAsAny[TaskResult]
    name: str


class GroupChatRequestPublish(BaseModel):
    ...


class GroupChatMessage(BaseModel):
    message: SerializeAsAny[BaseAgentEvent | BaseChatMessage]


class GroupChatTermination(BaseModel):
    message: StopMessage
    error: SerializableException | None = None


class GroupChatReset(BaseModel):
    ...


class GroupChatPause(BaseModel):
    ...


class GroupChatResume(BaseModel):
    ...


class GroupChatError(BaseModel):
    error: SerializableException

Import

from autogen_agentchat.teams._group_chat._events import (
    SerializableException,
    GroupChatStart,
    GroupChatAgentResponse,
    GroupChatTeamResponse,
    GroupChatRequestPublish,
    GroupChatMessage,
    GroupChatTermination,
    GroupChatReset,
    GroupChatPause,
    GroupChatResume,
    GroupChatError
)

I/O Contract

SerializableException

Field	Type	Description
error_type	str	The type name of the exception (e.g., "ValueError")
error_message	str	The error message describing what went wrong
traceback	None	Full traceback string if available

Methods:

from_exception(exc: Exception): Create SerializableException from a Python exception

Control Events

Event	Fields	Description
GroupChatStart	None, output_task_messages: bool	Start group chat with optional initial messages
GroupChatReset	None	Reset all agents in the group chat
GroupChatPause	None	Pause the group chat execution
GroupChatResume	None	Resume paused group chat execution

Request Events

Event	Fields	Description
GroupChatRequestPublish	None	Request agent/team to process buffered messages and publish response

Response Events

Event	Fields	Description
GroupChatAgentResponse	response: Response, name: str	Response from a ChatAgent with the agent's name
GroupChatTeamResponse	result: TaskResult, name: str	Result from a Team with the team's name

Message Events

Event	Fields	Description
GroupChatMessage	BaseChatMessage	Intermediate message or event for logging/streaming

Termination Events

Event	Fields	Description
GroupChatTermination	None	Group chat termination with stop message and optional error

Error Events

Event	Fields	Description
GroupChatError	error: SerializableException	Error that occurred during group chat processing

Usage Examples

Starting a Group Chat

from autogen_agentchat.teams._group_chat._events import GroupChatStart
from autogen_agentchat.messages import TextMessage


# Start with initial messages
start_event = GroupChatStart(
    messages=[
        TextMessage(content="Hello team!", source="user"),
        TextMessage(content="What's the task?", source="user")
    ],
    output_task_messages=True
)

# Start without initial messages
start_event = GroupChatStart(messages=None)

Agent Response Flow

from autogen_agentchat.teams._group_chat._events import (
    GroupChatRequestPublish,
    GroupChatAgentResponse
)
from autogen_agentchat.base import Response
from autogen_agentchat.messages import TextMessage


# Orchestrator publishes request
request = GroupChatRequestPublish()
# Agent container receives request, processes, and publishes response

# Agent container creates response
response = GroupChatAgentResponse(
    response=Response(
        chat_message=TextMessage(content="Task completed", source="assistant")
    ),
    name="assistant"
)
# Orchestrator receives response

Team Response Flow

from autogen_agentchat.teams._group_chat._events import GroupChatTeamResponse
from autogen_agentchat.base import TaskResult
from autogen_agentchat.messages import TextMessage


# Team container creates response
team_result = TaskResult(
    messages=[
        TextMessage(content="Analysis complete", source="analyst"),
        TextMessage(content="Report generated", source="writer")
    ],
    stop_reason="Task completed"
)

team_response = GroupChatTeamResponse(
    result=team_result,
    name="analysis_team"
)
# Orchestrator receives team response

Error Handling

from autogen_agentchat.teams._group_chat._events import (
    SerializableException,
    GroupChatError
)


# Agent encounters error
try:
    result = await agent.on_messages(messages, cancellation_token)
except ValueError as e:
    # Convert to serializable exception
    serializable_error = SerializableException.from_exception(e)

    # Publish error event
    error_event = GroupChatError(error=serializable_error)

    # Print error details
    print(str(serializable_error))
    # Output: ValueError: Invalid input
    #         Traceback:
    #         ...full traceback...

Message Streaming

from autogen_agentchat.teams._group_chat._events import GroupChatMessage
from autogen_agentchat.messages import TextMessage, ToolCallMessage


# Stream intermediate messages during agent processing
async for event in agent.on_messages_stream(messages, cancellation_token):
    if isinstance(event, Response):
        # Final response
        break
    else:
        # Intermediate event/message - wrap and publish
        message_event = GroupChatMessage(message=event)
        await publish_message(message_event, output_topic)

# Examples of streamed messages:
# - ToolCallMessage when agent calls a tool
# - ToolCallResultMessage with tool results
# - TextMessage with partial responses
# - Custom BaseAgentEvent instances

Termination Events

from autogen_agentchat.teams._group_chat._events import GroupChatTermination
from autogen_agentchat.messages import StopMessage


# Normal termination
termination = GroupChatTermination(
    message=StopMessage(content="Task completed successfully", source="orchestrator"),
    error=None
)

# Termination due to error
try:
    # ... group chat execution ...
    pass
except Exception as e:
    error = SerializableException.from_exception(e)
    termination = GroupChatTermination(
        message=StopMessage(content="Task failed", source="orchestrator"),
        error=error
    )

Lifecycle Management

from autogen_agentchat.teams._group_chat._events import (
    GroupChatReset,
    GroupChatPause,
    GroupChatResume
)


async def lifecycle_management():
    # Reset all agents
    reset_event = GroupChatReset()
    await publish_to_all_containers(reset_event)

    # Pause execution (e.g., for human review)
    pause_event = GroupChatPause()
    await publish_to_all_containers(pause_event)

    # Resume execution
    resume_event = GroupChatResume()
    await publish_to_all_containers(resume_event)

Custom Exception Serialization

from autogen_agentchat.teams._group_chat._events import SerializableException


class CustomAgentError(Exception):
    """Custom exception for agent errors."""
    pass


try:
    raise CustomAgentError("Agent failed to process input")
except CustomAgentError as e:
    # Serialize custom exception
    serializable = SerializableException.from_exception(e)

    assert serializable.error_type == "CustomAgentError"
    assert serializable.error_message == "Agent failed to process input"
    assert serializable.traceback is not None

    # Serialize to dict for transmission
    error_dict = serializable.model_dump()

    # Deserialize on receiving end
    restored = SerializableException.model_validate(error_dict)

Related Pages

Microsoft_Autogen_ChatAgentContainer - Container that handles these events
Microsoft_Autogen_BaseGroupChat - Orchestrator that publishes these events
Microsoft_Autogen_Response - Response object wrapped in GroupChatAgentResponse
Microsoft_Autogen_TaskResult - Result object wrapped in GroupChatTeamResponse
Microsoft_Autogen_BaseChatMessage - Base class for messages in events
Microsoft_Autogen_BaseAgentEvent - Base class for agent events
Microsoft_Autogen_StopMessage - Stop message in termination events
Microsoft_Autogen_RoundRobinGroupChat - Group chat that uses these events
Microsoft_Autogen_SelectorGroupChat - Group chat that uses these events
Microsoft_Autogen_MagenticOneGroupChat - Group chat that uses these events

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment