AgentChat Class

A base class chat interface for agents.

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

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Constructor

AgentChat(*, broadcast_queue: BroadcastQueue = None, agent_channels: dict[str, AgentChannel] = None, channel_map: dict[Agent, str] = None, history: ChatHistory = None)

Keyword-Only Parameters

Name	Description
broadcast_queue Required
agent_channels Required
channel_map Required
history Required

Methods

add_chat_message	Add a chat message.
add_chat_messages	Add chat messages.
clear_activity_signal	Clear the activity signal.
get_chat_messages	Get chat messages asynchronously.
get_messages_in_descending_order	Get messages in descending order asynchronously.
invoke	Invoke the agent asynchronously.
invoke_agent	Invoke an agent asynchronously.
invoke_agent_stream	Invoke an agent stream asynchronously.
model_post_init	This function is meant to behave like a BaseModel method to initialise private attributes. It takes context as an argument since that's what pydantic-core passes when calling it.
reset	Reset the agent chat.
set_activity_or_throw	Set the activity signal or throw an exception if another agent is active.

add_chat_message

Add a chat message.

async add_chat_message(message: str | ChatMessageContent) -> None

Parameters

Name	Description
message Required

add_chat_messages

Add chat messages.

async add_chat_messages(messages: list[ChatMessageContent]) -> None

Parameters

Name	Description
messages Required

clear_activity_signal

Clear the activity signal.

clear_activity_signal()

get_chat_messages

Get chat messages asynchronously.

async get_chat_messages(agent: Agent | None = None) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent	Default value: None

get_messages_in_descending_order

Get messages in descending order asynchronously.

async get_messages_in_descending_order() -> AsyncIterable[ChatMessageContent]

invoke

Invoke the agent asynchronously.

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

Parameters

Name	Description
agent	Default value: None
is_joining	Default value: True

invoke_agent

Invoke an agent asynchronously.

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

Parameters

Name	Description
agent Required

invoke_agent_stream

Invoke an agent stream asynchronously.

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

Parameters

Name	Description
agent Required

model_post_init

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

model_post_init(context: Any, /) -> None

Positional-Only Parameters

Name	Description
context Required

Parameters

Name	Description
self Required	The BaseModel instance.
context Required	The context.

reset

Reset the agent chat.

async reset() -> None

set_activity_or_throw

Set the activity signal or throw an exception if another agent is active.

set_activity_or_throw()

Attributes

is_active

Indicates whether the agent is currently active.

agent_channels

agent_channels: dict[str, AgentChannel]

broadcast_queue

broadcast_queue: BroadcastQueue

channel_map

channel_map: dict[Agent, str]

history

history: ChatHistory

is_experimental

is_experimental = True

stage_status

stage_status = 'experimental'