"""This module defines various message types used for agent-to-agent communication.Each message type inherits either from the BaseChatMessage class or BaseAgentEventclass and includes specific fields relevant to the type of message being sent."""fromabcimportABC,abstractmethodfromtypingimportAny,Dict,Generic,List,Literal,Mapping,TypeVarfromautogen_coreimportFunctionCall,Imagefromautogen_core.memoryimportMemoryContentfromautogen_core.modelsimportFunctionExecutionResult,LLMMessage,RequestUsage,UserMessagefrompydanticimportBaseModel,Field,computed_fieldfromtyping_extensionsimportAnnotated,Self
[docs]classBaseMessage(BaseModel,ABC):"""Abstract base class for all message types in AgentChat. .. warning:: If you want to create a new message type, do not inherit from this class. Instead, inherit from :class:`BaseChatMessage` or :class:`BaseAgentEvent` to clarify the purpose of the message type. """
[docs]@abstractmethoddefto_text(self)->str:"""Convert the message content to a string-only representation that can be rendered in the console and inspected by the user or conditions. This is not used for creating text-only content for models. For :class:`BaseChatMessage` types, use :meth:`to_model_text` instead."""...
[docs]defdump(self)->Mapping[str,Any]:"""Convert the message to a JSON-serializable dictionary. The default implementation uses the Pydantic model's :meth:`model_dump` method to convert the message to a dictionary. Override this method if you want to customize the serialization process or add additional fields to the output. """returnself.model_dump()
[docs]@classmethoddefload(cls,data:Mapping[str,Any])->Self:"""Create a message from a dictionary of JSON-serializable data. The default implementation uses the Pydantic model's :meth:`model_validate` method to create the message from the data. Override this method if you want to customize the deserialization process or add additional fields to the input data."""returncls.model_validate(data)
[docs]classBaseChatMessage(BaseMessage,ABC):"""Abstract base class for chat messages. .. note:: If you want to create a new message type that is used for agent-to-agent communication, inherit from this class, or simply use :class:`StructuredMessage` if your content type is a subclass of Pydantic BaseModel. This class is used for messages that are sent between agents in a chat conversation. Agents are expected to process the content of the message using models and return a response as another :class:`BaseChatMessage`. """source:str"""The name of the agent that sent this message."""models_usage:RequestUsage|None=None"""The model client usage incurred when producing this message."""metadata:Dict[str,str]={}"""Additional metadata about the message."""
[docs]@abstractmethoddefto_model_text(self)->str:"""Convert the content of the message to text-only representation. This is used for creating text-only content for models. This is not used for rendering the message in console. For that, use :meth:`~BaseMessage.to_text`. The difference between this and :meth:`to_model_message` is that this is used to construct parts of the a message for the model client, while :meth:`to_model_message` is used to create a complete message for the model client. """...
[docs]@abstractmethoddefto_model_message(self)->UserMessage:"""Convert the message content to a :class:`~autogen_core.models.UserMessage` for use with model client, e.g., :class:`~autogen_core.models.ChatCompletionClient`."""...
[docs]classBaseTextChatMessage(BaseChatMessage,ABC):"""Base class for all text-only :class:`BaseChatMessage` types. It has implementations for :meth:`to_text`, :meth:`to_model_text`, and :meth:`to_model_message` methods. Inherit from this class if your message content type is a string. """content:str"""The content of the message."""
[docs]classBaseAgentEvent(BaseMessage,ABC):"""Base class for agent events. .. note:: If you want to create a new message type for signaling observable events to user and application, inherit from this class. Agent events are used to signal actions and thoughts produced by agents and teams to user and applications. They are not used for agent-to-agent communication and are not expected to be processed by other agents. You should override the :meth:`to_text` method if you want to provide a custom rendering of the content. """source:str"""The name of the agent that sent this message."""models_usage:RequestUsage|None=None"""The model client usage incurred when producing this message."""metadata:Dict[str,str]={}"""Additional metadata about the message."""
StructuredContentType=TypeVar("StructuredContentType",bound=BaseModel,covariant=True)"""Type variable for structured content types."""
[docs]classStructuredMessage(BaseChatMessage,Generic[StructuredContentType]):"""A :class:`BaseChatMessage` type with an unspecified content type. To create a new structured message type, specify the content type as a subclass of `Pydantic BaseModel <https://docs.pydantic.dev/latest/concepts/models/>`_. .. code-block:: python from pydantic import BaseModel from autogen_agentchat.messages import StructuredMessage class MyMessageContent(BaseModel): text: str number: int message = StructuredMessage[MyMessageContent]( content=MyMessageContent(text="Hello", number=42), source="agent1", ) print(message.to_text()) # {"text": "Hello", "number": 42} """content:StructuredContentType"""The content of the message. Must be a subclass of `Pydantic BaseModel <https://docs.pydantic.dev/latest/concepts/models/>`_."""@computed_fielddeftype(self)->str:returnself.__class__.__name__
[docs]classTextMessage(BaseTextChatMessage):"""A text message with string-only content."""type:Literal["TextMessage"]="TextMessage"
[docs]classMultiModalMessage(BaseChatMessage):"""A multimodal message."""content:List[str|Image]"""The content of the message."""type:Literal["MultiModalMessage"]="MultiModalMessage"
[docs]defto_model_text(self,image_placeholder:str|None="[image]")->str:"""Convert the content of the message to a string-only representation. If an image is present, it will be replaced with the image placeholder by default, otherwise it will be a base64 string when set to None. """text=""forcinself.content:ifisinstance(c,str):text+=celifisinstance(c,Image):ifimage_placeholderisnotNone:text+=f" {image_placeholder}"else:text+=f" {c.to_base64()}"returntext
[docs]classStopMessage(BaseTextChatMessage):"""A message requesting stop of a conversation."""type:Literal["StopMessage"]="StopMessage"
[docs]classHandoffMessage(BaseTextChatMessage):"""A message requesting handoff of a conversation to another agent."""target:str"""The name of the target agent to handoff to."""context:List[LLMMessage]=[]"""The model context to be passed to the target agent."""type:Literal["HandoffMessage"]="HandoffMessage"
[docs]classToolCallSummaryMessage(BaseTextChatMessage):"""A message signaling the summary of tool call results."""type:Literal["ToolCallSummaryMessage"]="ToolCallSummaryMessage"
[docs]classToolCallRequestEvent(BaseAgentEvent):"""An event signaling a request to use tools."""content:List[FunctionCall]"""The tool calls."""type:Literal["ToolCallRequestEvent"]="ToolCallRequestEvent"
[docs]classUserInputRequestedEvent(BaseAgentEvent):"""An event signaling a that the user proxy has requested user input. Published prior to invoking the input callback."""request_id:str"""Identifier for the user input request."""content:Literal[""]="""""Empty content for compat with consumers expecting a content field."""type:Literal["UserInputRequestedEvent"]="UserInputRequestedEvent"
[docs]classModelClientStreamingChunkEvent(BaseAgentEvent):"""An event signaling a text output chunk from a model client in streaming mode."""content:str"""A string chunk from the model client."""type:Literal["ModelClientStreamingChunkEvent"]="ModelClientStreamingChunkEvent"
[docs]classThoughtEvent(BaseAgentEvent):"""An event signaling the thought process of a model. It is used to communicate the reasoning tokens generated by a reasoning model, or the extra text content generated by a function call."""content:str"""The thought process of the model."""type:Literal["ThoughtEvent"]="ThoughtEvent"
classMessageFactory:""":meta private: A factory for creating messages from JSON-serializable dictionaries. This is useful for deserializing messages from JSON data. """def__init__(self)->None:self._message_types:Dict[str,type[BaseAgentEvent|BaseChatMessage]]={}# Register all message types.self._message_types[TextMessage.__name__]=TextMessageself._message_types[MultiModalMessage.__name__]=MultiModalMessageself._message_types[StopMessage.__name__]=StopMessageself._message_types[ToolCallSummaryMessage.__name__]=ToolCallSummaryMessageself._message_types[HandoffMessage.__name__]=HandoffMessageself._message_types[ToolCallRequestEvent.__name__]=ToolCallRequestEventself._message_types[ToolCallExecutionEvent.__name__]=ToolCallExecutionEventself._message_types[MemoryQueryEvent.__name__]=MemoryQueryEventself._message_types[UserInputRequestedEvent.__name__]=UserInputRequestedEventself._message_types[ModelClientStreamingChunkEvent.__name__]=ModelClientStreamingChunkEventself._message_types[ThoughtEvent.__name__]=ThoughtEventdefis_registered(self,message_type:type[BaseAgentEvent|BaseChatMessage])->bool:"""Check if a message type is registered with the factory."""# Get the class name of the message type.class_name=message_type.__name__# Check if the class name is already registered.returnclass_nameinself._message_typesdefregister(self,message_type:type[BaseAgentEvent|BaseChatMessage])->None:"""Register a new message type with the factory."""ifself.is_registered(message_type):raiseValueError(f"Message type {message_type} is already registered.")ifnotissubclass(message_type,BaseChatMessage)andnotissubclass(message_type,BaseAgentEvent):raiseValueError(f"Message type {message_type} must be a subclass of BaseChatMessage or BaseAgentEvent.")# Get the class name of theclass_name=message_type.__name__# Check if the class name is already registered.# Register the message type.self._message_types[class_name]=message_typedefcreate(self,data:Mapping[str,Any])->BaseAgentEvent|BaseChatMessage:"""Create a message from a dictionary of JSON-serializable data."""# Get the type of the message from the dictionary.message_type=data.get("type")ifmessage_typeisNone:raiseValueError("Field 'type' is required in the message data to recover the message type.")ifmessage_typenotinself._message_types:raiseValueError(f"Unknown message type: {message_type}")ifnotisinstance(message_type,str):raiseValueError(f"Message type must be a string, got {type(message_type)}")# Get the class for the message type.message_class=self._message_types[message_type]# Create an instance of the message class.assertissubclass(message_class,BaseChatMessage)orissubclass(message_class,BaseAgentEvent)returnmessage_class.load(data)ChatMessage=Annotated[TextMessage|MultiModalMessage|StopMessage|ToolCallSummaryMessage|HandoffMessage,Field(discriminator="type")]"""The union type of all built-in concrete subclasses of :class:`BaseChatMessage`.It does not include :class:`StructuredMessage` types."""AgentEvent=Annotated[ToolCallRequestEvent|ToolCallExecutionEvent|MemoryQueryEvent|UserInputRequestedEvent|ModelClientStreamingChunkEvent|ThoughtEvent,Field(discriminator="type"),]"""The union type of all built-in concrete subclasses of :class:`BaseAgentEvent`."""__all__=["AgentEvent","BaseMessage","ChatMessage","BaseChatMessage","BaseAgentEvent","BaseTextChatMessage","StructuredContentType","StructuredMessage","HandoffMessage","MultiModalMessage","StopMessage","TextMessage","ToolCallExecutionEvent","ToolCallRequestEvent","ToolCallSummaryMessage","MemoryQueryEvent","UserInputRequestedEvent","ModelClientStreamingChunkEvent","ThoughtEvent","MessageFactory",]