AgentGroupChat Class

An agent chat that supports multi-turn interactions.

Note: This class is experimental and may change in the future.

Initialize a new instance of AgentGroupChat.

Inheritance

AgentChat

AgentGroupChat

Constructor

AgentGroupChat(agents: list[Agent] | None = None, termination_strategy: TerminationStrategy | None = None, selection_strategy: SelectionStrategy | None = None)

Parameters

Name	Description
agents	The agents to add to the group chat. Default value: None
termination_strategy	The termination strategy to use. Default value: None
selection_strategy	The selection strategy Default value: None

Methods

add_agent	Add an agent to the group chat.
invoke	Invoke the agent chat asynchronously. Handles both group interactions and single agent interactions based on the provided arguments.
invoke_single_turn	Invoke the agent chat for a single turn.
invoke_stream	Invoke the agent chat stream asynchronously. Handles both group interactions and single agent interactions based on the provided arguments.
invoke_stream_single_turn	Invoke the agent chat for a single turn.
model_post_init	We need to both initialize private attributes and call the user-defined model_post_init method.

add_agent

Add an agent to the group chat.

add_agent(agent: Agent) -> None

Parameters

Name	Description
agent Required	The agent to add.

invoke

Invoke the agent chat asynchronously.

Handles both group interactions and single agent interactions based on the provided arguments.

async invoke(agent: Agent | None = None, is_joining: bool = True) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent	The agent to invoke. If not provided, the method processes all agents in the chat. Default value: None
is_joining	Controls whether the agent joins the chat. Defaults to True. Default value: True

invoke_single_turn

Invoke the agent chat for a single turn.

async invoke_single_turn(agent: Agent) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent Required	The agent to invoke.

invoke_stream

Invoke the agent chat stream asynchronously.

Handles both group interactions and single agent interactions based on the provided arguments.

async invoke_stream(agent: Agent | None = None, is_joining: bool = True) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent	The agent to invoke. If not provided, the method processes all agents in the chat. Default value: None
is_joining	Controls whether the agent joins the chat. Defaults to True. Default value: True

invoke_stream_single_turn

Invoke the agent chat for a single turn.

async invoke_stream_single_turn(agent: Agent) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent Required	The agent to invoke.

model_post_init

We need to both initialize private attributes and call the user-defined model_post_init method.

model_post_init(context: Any, /) -> None

Positional-Only Parameters

Name	Description
context Required

Attributes

model_computed_fields

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_assignment': True}

model_fields

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.fields from Pydantic V1.

model_fields: ClassVar[Dict[str, FieldInfo]] = {'agent_channels': FieldInfo(annotation=dict[str, AgentChannel], required=False, default_factory=dict), 'agent_ids': FieldInfo(annotation=set[str], required=True), 'agents': FieldInfo(annotation=list[Agent], required=False, default_factory=list), 'broadcast_queue': FieldInfo(annotation=BroadcastQueue, required=False, default_factory=BroadcastQueue), 'channel_map': FieldInfo(annotation=dict[Agent, str], required=False, default_factory=dict), 'history': FieldInfo(annotation=ChatHistory, required=False, default_factory=ChatHistory), 'is_complete': FieldInfo(annotation=bool, required=False, default=False), 'selection_strategy': FieldInfo(annotation=SelectionStrategy, required=False, default_factory=SequentialSelectionStrategy), 'termination_strategy': FieldInfo(annotation=TerminationStrategy, required=False, default_factory=DefaultTerminationStrategy)}

agent_channels

agent_channels: dict[str, AgentChannel]

agent_ids

agent_ids: set[str]

agents

agents: list[Agent]

broadcast_queue

broadcast_queue: BroadcastQueue

channel_map

channel_map: dict[Agent, str]

history

history: ChatHistory

is_complete

is_complete: bool

is_experimental

is_experimental = True

selection_strategy

selection_strategy: SelectionStrategy

termination_strategy

termination_strategy: TerminationStrategy