autogen_core.components#

The autogen_core.components module provides building blocks for creating single agents

class autogen_core.components.ClosureAgent(description: str, closure: Callable[[AgentRuntime, AgentId, T, MessageContext], Awaitable[Any]])[source]#

Bases: Agent

property id: AgentId#: ID of the agent.

load_state(state: Mapping[str, Any]) → None[source]#

Load in the state of the agent obtained from save_state.

Parameters:: state (Mapping[str, Any]) – State of the agent. Must be JSON serializable.

property metadata: AgentMetadata#: Metadata of the agent.

async on_message(message: Any, ctx: MessageContext) → Any[source]#

Message handler for the agent. This should only be called by the runtime, not by other agents.

Parameters:

message (Any) – Received message. Type is one of the types in subscriptions.
ctx (MessageContext) – Context of the message.

Returns:

Any – Response to the message. Can be None.

Raises:

asyncio.CancelledError – If the message was cancelled.
CantHandleException – If the agent cannot handle the message.

async classmethod register(runtime: AgentRuntime, type: str, closure: Callable[[AgentRuntime, AgentId, T, MessageContext], Awaitable[Any]], *, description: str = '', subscriptions: Callable[[], list[Subscription] | Awaitable[list[Subscription]]] | None = None) → AgentType[source]#

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

property runtime: AgentRuntime#

save_state() → Mapping[str, Any][source]#: Save the state of the agent. The result must be JSON serializable.

class autogen_core.components.DefaultSubscription(topic_type: str = 'default', agent_type: str | None = None)[source]#

Bases: TypeSubscription

The default subscription is designed to be a sensible default for applications that only need global scope for agents.

This topic by default uses the “default” topic type and attempts to detect the agent type to use based on the instantiation context.

Example

await runtime.register("MyAgent", agent_factory, lambda: [DefaultSubscription()])

Parameters:

topic_type (str, optional) – The topic type to subscribe to. Defaults to “default”.
agent_type (str, optional) – The agent type to use for the subscription. Defaults to None, in which case it will attempt to detect the agent type based on the instantiation context.

class autogen_core.components.DefaultTopicId(type: str = 'default', source: str | None = None)[source]#

Bases: TopicId

DefaultTopicId provides a sensible default for the topic_id and source fields of a TopicId.

If created in the context of a message handler, the source will be set to the agent_id of the message handler, otherwise it will be set to “default”.

Parameters:

type (str, optional) – Topic type to publish message to. Defaults to “default”.
source (str | None, optional) – Topic source to publish message to. If None, the source will be set to the agent_id of the message handler if in the context of a message handler, otherwise it will be set to “default”. Defaults to None.

class autogen_core.components.FunctionCall(id: 'str', arguments: 'str', name: 'str')[source]#

Bases: object

arguments: str#

id: str#

name: str#

class autogen_core.components.Image(image: Image)[source]#

Bases: object

property data_uri: str#

classmethod from_base64(base64_str: str) → Image[source]#

classmethod from_file(file_path: Path) → Image[source]#

classmethod from_pil(pil_image: Image) → Image[source]#

classmethod from_uri(uri: str) → Image[source]#

async classmethod from_url(url: str) → Image[source]#

to_base64() → str[source]#

to_openai_format(detail: Literal['auto', 'low', 'high'] = 'auto') → ChatCompletionContentPartImageParam[source]#

class autogen_core.components.RoutedAgent(description: str)[source]#

Bases: BaseAgent

A base class for agents that route messages to handlers based on the type of the message and optional matching functions.

To create a routed agent, subclass this class and add message handlers as methods decorated with either event() or rpc() decorator.

Example:

from autogen_core.base import MessageContext
from autogen_core.components import RoutedAgent, event, rpc
# Assume Message, MessageWithContent, and Response are defined elsewhere.


class MyAgent(RoutedAgent):
    def __init__(self):
        super().__init__("MyAgent")

    @event
    async def handle_event_message(self, message: Message, ctx: MessageContext) -> None:
        self.publish_message(MessageWithContent("event handled"), ctx.topic_id)

    @rpc(match=lambda message, ctx: message.content == "special")
    async def handle_special_rpc_message(self, message: MessageWithContent, ctx: MessageContext) -> Response:
        return Response()

internal_extra_handles_types: ClassVar[List[Tuple[Type[Any], List[MessageSerializer[Any]]]]] = []#

internal_unbound_subscriptions_list: ClassVar[List[UnboundSubscription]] = []#

async on_message(message: Any, ctx: MessageContext) → Any | None[source]#: Handle a message by routing it to the appropriate message handler. Do not override this method in subclasses. Instead, add message handlers as methods decorated with either the event() or rpc() decorator.

async on_unhandled_message(message: Any, ctx: MessageContext) → None[source]#: Called when a message is received that does not have a matching message handler. The default implementation logs an info message.

class autogen_core.components.TypeRoutedAgent(description: str)[source]#

Bases: RoutedAgent

Deprecated. Use RoutedAgent instead.

internal_extra_handles_types: ClassVar[List[Tuple[Type[Any], List[MessageSerializer[Any]]]]] = []#

internal_unbound_subscriptions_list: ClassVar[List[UnboundSubscription]] = []#

class autogen_core.components.TypeSubscription(topic_type: str, agent_type: str)[source]#

Bases: Subscription

This subscription matches on topics based on the type and maps to agents using the source of the topic as the agent key.

This subscription causes each source to have its own agent instance.

Example

subscription = TypeSubscription(topic_type="t1", agent_type="a1")

In this case:

A topic_id with type t1 and source s1 will be handled by an agent of type a1 with key s1
A topic_id with type t1 and source s2 will be handled by an agent of type a1 with key s2.

Parameters:

topic_type (str) – Topic type to match against
agent_type (str) – Agent type to handle this subscription

property agent_type: str#

property id: str#

Get the ID of the subscription.

Implementations should return a unique ID for the subscription. Usually this is a UUID.

Returns:: str – ID of the subscription.

is_match(topic_id: TopicId) → bool[source]#

Check if a given topic_id matches the subscription.

Parameters:: topic_id (TopicId) – TopicId to check.
Returns:: bool – True if the topic_id matches the subscription, False otherwise.

map_to_agent(topic_id: TopicId) → AgentId[source]#

Map a topic_id to an agent. Should only be called if is_match returns True for the given topic_id.

Parameters:: topic_id (TopicId) – TopicId to map.
Returns:: AgentId – ID of the agent that should handle the topic_id.
Raises:: CantHandleException – If the subscription cannot handle the topic_id.

property topic_type: str#

autogen_core.components.default_subscription(cls: Type[BaseAgentType] | None = None) → Callable[[Type[BaseAgentType]], Type[BaseAgentType]] | Type[BaseAgentType][source]#

autogen_core.components.event(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, None]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, None]]], MessageHandler[AgentT, ReceivesT, None]] | MessageHandler[AgentT, ReceivesT, None][source]#

Decorator for event message handlers.

Add this decorator to methods in a RoutedAgent class that are intended to handle event messages. These methods must have a specific signature that needs to be followed for it to be valid:

The method must be an async method.
The method must be decorated with the @message_handler decorator.
The method must have exactly 3 arguments:
1. self
2. message: The event message to be handled, this must be type-hinted with the message type that it is intended to handle.
3. ctx: A autogen_core.base.MessageContext object.
The method must return None.

Handlers can handle more than one message type by accepting a Union of the message types.

Parameters:

func – The function to be decorated.
strict – If True, the handler will raise an exception if the message type is not in the target types. If False, it will log a warning instead.
match – A function that takes the message and the context as arguments and returns a boolean. This is used for secondary routing after the message type. For handlers addressing the same message type, the match function is applied in alphabetical order of the handlers and the first matching handler will be called while the rest are skipped. If None, the first handler in alphabetical order matching the same message type will be called.

autogen_core.components.message_handler(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][source]#

Decorator for generic message handlers.

Add this decorator to methods in a RoutedAgent class that are intended to handle both event and RPC messages. These methods must have a specific signature that needs to be followed for it to be valid:

The method must be an async method.
The method must be decorated with the @message_handler decorator.
The method must have exactly 3 arguments:
1. self
2. message: The message to be handled, this must be type-hinted with the message type that it is intended to handle.
3. ctx: A autogen_core.base.MessageContext object.
The method must be type hinted with what message types it can return as a response, or it can return None if it does not return anything.

Handlers can handle more than one message type by accepting a Union of the message types. It can also return more than one message type by returning a Union of the message types.

Parameters:

func – The function to be decorated.
strict – If True, the handler will raise an exception if the message type or return type is not in the target types. If False, it will log a warning instead.
match – A function that takes the message and the context as arguments and returns a boolean. This is used for secondary routing after the message type. For handlers addressing the same message type, the match function is applied in alphabetical order of the handlers and the first matching handler will be called while the rest are skipped. If None, the first handler in alphabetical order matching the same message type will be called.

autogen_core.components.rpc(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][source]#

Decorator for RPC message handlers.

Add this decorator to methods in a RoutedAgent class that are intended to handle RPC messages. These methods must have a specific signature that needs to be followed for it to be valid:

The method must be an async method.
The method must be decorated with the @message_handler decorator.
The method must have exactly 3 arguments:
1. self
2. message: The message to be handled, this must be type-hinted with the message type that it is intended to handle.
3. ctx: A autogen_core.base.MessageContext object.
The method must be type hinted with what message types it can return as a response, or it can return None if it does not return anything.

Handlers can handle more than one message type by accepting a Union of the message types. It can also return more than one message type by returning a Union of the message types.

Parameters:

func – The function to be decorated.
strict – If True, the handler will raise an exception if the message type or return type is not in the target types. If False, it will log a warning instead.
match – A function that takes the message and the context as arguments and returns a boolean. This is used for secondary routing after the message type. For handlers addressing the same message type, the match function is applied in alphabetical order of the handlers and the first matching handler will be called while the rest are skipped. If None, the first handler in alphabetical order matching the same message type will be called.

autogen_core.components.type_subscription(topic_type: str) → Callable[[Type[BaseAgentType]], Type[BaseAgentType]][source]#