BedrockAgentChannel Class

An AgentChannel for a BedrockAgent that is based on a ChatHistory.

This channel allows Bedrock agents to interact with other types of agents in Semantic Kernel in an AgentGroupChat. However, since Bedrock agents require the chat history to alternate between user and agent messages, this channel will preprocess the chat history to ensure that it meets the requirements of the Bedrock agent. When an invalid pattern is detected, the channel will insert a placeholder user or assistant message to ensure that the chat history alternates between user and agent messages.

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

BedrockAgentChannel(*, messages: list[ChatMessageContent] = None, system_message: str | None = None)

Keyword-Only Parameters

Name	Description
messages Required
system_message Required

Methods

get_history	Retrieve the message history specific to this channel.
invoke	Perform a discrete incremental interaction between a single Agent and AgentChat.
invoke_stream	Perform a streaming interaction between a single Agent and AgentChat.
receive	Receive the conversation messages. Bedrock requires the chat history to alternate between user and agent messages. Thus, when receiving the history, the message sequence will be mutated by inserting empty agent or user messages as needed.
reset	Reset the channel state.

get_history

Retrieve the message history specific to this channel.

async get_history() -> AsyncIterable[ChatMessageContent]

Returns

Type	Description
	An async iterable of ChatMessageContent.

invoke

Perform a discrete incremental interaction between a single Agent and AgentChat.

async invoke(agent: Agent, **kwargs: Any) -> AsyncIterable[tuple[bool, ChatMessageContent]]

Parameters

Name	Description
agent Required	The agent to interact with.
kwargs Required	Additional keyword arguments.

Returns

Type	Description
	An async iterable of ChatMessageContent with a boolean indicating if the message should be visible external to the agent.

invoke_stream

Perform a streaming interaction between a single Agent and AgentChat.

async invoke_stream(agent: Agent, messages: list[ChatMessageContent], **kwargs: Any) -> AsyncIterable[ChatMessageContent]

Parameters

Name	Description
agent Required	The agent to interact with.
messages Required	The history of messages in the conversation.
kwargs Required	Additional keyword arguments.

Returns

Type	Description
	An async iterable of ChatMessageContent.

receive

Receive the conversation messages.

Bedrock requires the chat history to alternate between user and agent messages. Thus, when receiving the history, the message sequence will be mutated by inserting empty agent or user messages as needed.

async receive(history: list[ChatMessageContent]) -> None

Parameters

Name	Description
history Required	The history of messages in the conversation.

reset

Reset the channel state.

async reset() -> None

Attributes

MESSAGE_PLACEHOLDER

MESSAGE_PLACEHOLDER: ClassVar[str] = '[SILENCE]'

is_experimental

is_experimental = True

stage_status

stage_status = 'experimental'