AgentChat Class

A base class chat interface for agents.

Note: This class is 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.

Inheritance

KernelBaseModel

AgentChat

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: 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) -> AsyncGenerator[ChatMessageContent, None]

Parameters

Name	Description
agent	Default value: None

get_messages_in_descending_order

Get messages in descending order asynchronously.

async get_messages_in_descending_order()

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.

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), '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)}

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