AgentGroupChat Class

  • Reference

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