"""
本模块定义了用于代理间通信的各种消息类型。
每种消息类型都继承自 BaseChatMessage 类或 BaseAgentEvent 类，
并包含与所发送消息类型相关的特定字段。
"""
from abc import ABC, abstractmethod
from datetime import datetime, timezone
from typing import Any, Dict, Generic, List, Literal, Mapping, Optional, Type, TypeVar
from autogen_core import Component, ComponentBase, FunctionCall, Image
from autogen_core.code_executor import CodeBlock, CodeResult
from autogen_core.memory import MemoryContent
from autogen_core.models import (
FunctionExecutionResult,
LLMMessage,
RequestUsage,
UserMessage,
)
from autogen_core.utils import schema_to_pydantic_model
from pydantic import BaseModel, Field, computed_field
from typing_extensions import Annotated, Self
[文档]
class BaseMessage(BaseModel, ABC):
"""AgentChat 中所有消息类型的抽象基类。
.. warning::
如需创建新的消息类型，请不要直接继承此类。
应继承 :class:`BaseChatMessage` 或 :class:`BaseAgentEvent`
以明确消息类型的用途。
"""
[文档]
@abstractmethod
def to_text(self) -> str:
"""将消息内容转换为纯字符串表示形式
可在控制台中渲染并由用户或条件检查。
这不用于为模型创建纯文本内容。
对于:class:`BaseChatMessage`类型，请改用:meth:`to_model_text`。"""
...
[文档]
def dump(self) -> Mapping[str, Any]:
"""将消息转换为可JSON序列化的字典。
默认实现使用Pydantic模型的
:meth:`model_dump`方法将消息转换为字典。
如需自定义序列化过程或向输出添加额外字段，请重写此方法。
"""
return self.model_dump()
[文档]
@classmethod
def load(cls, data: Mapping[str, Any]) -> Self:
"""从可JSON序列化的数据字典创建消息。
默认实现使用Pydantic模型的
:meth:`model_validate`方法从数据创建消息。
如需自定义反序列化过程或向输入数据添加额外字段，请重写此方法。"""
return cls.model_validate(data)
[文档]
class BaseChatMessage(BaseMessage, ABC):
"""聊天消息的抽象基类。
.. note::
如需创建用于智能体间通信的新消息类型，请继承此类，
或者当内容类型是 Pydantic BaseModel 的子类时，
直接使用 :class:`StructuredMessage`。
此类用于聊天对话中智能体之间传递的消息。
智能体应通过模型处理消息内容，
并返回另一个 :class:`BaseChatMessage` 作为响应。
"""
source: str
"""发送此消息的代理名称。"""
models_usage: RequestUsage | None = None
"""生成此消息时产生的模型客户端使用情况。"""
metadata: Dict[str, str] = {}
"""关于此消息的附加元数据。"""
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
"""消息创建的时间。"""
[文档]
@abstractmethod
def to_model_text(self) -> str:
"""将消息内容转换为纯文本表示形式。
这用于为模型创建纯文本内容。
此方法不用于在控制台渲染消息。为此，请使用
:meth:`~BaseMessage.to_text`。
此方法与 :meth:`to_model_message` 的区别在于：
本方法用于为模型客户端构造消息的部分内容，
而 :meth:`to_model_message` 用于为模型客户端创建完整消息。
"""
...
[文档]
@abstractmethod
def to_model_message(self) -> UserMessage:
"""将消息内容转换为 :class:`~autogen_core.models.UserMessage`
以便与模型客户端一起使用，例如 :class:`~autogen_core.models.ChatCompletionClient`。
"""
...
[文档]
class BaseTextChatMessage(BaseChatMessage, ABC):
"""所有纯文本 :class:`BaseChatMessage` 类型的基类。
它实现了 :meth:`to_text`、:meth:`to_model_text`
和 :meth:`to_model_message` 方法。
如果您的消息内容类型是字符串，请继承此类。
"""
content: str
"""消息的内容。"""
[文档]
def to_text(self) -> str:
return self.content
[文档]
def to_model_text(self) -> str:
return self.content
[文档]
def to_model_message(self) -> UserMessage:
return UserMessage(content=self.content, source=self.source)
[文档]
class BaseAgentEvent(BaseMessage, ABC):
"""代理事件的基础类。
.. note::
如果你想创建一个新的消息类型用于向用户和应用程序发送可观察事件信号，
请继承这个类。
代理事件用于将代理和团队产生的动作与思考信号发送给用户和应用程序。
它们不用于代理间通信，也不期望被其他代理处理。
如果你想提供自定义的内容渲染方式，应该重写 :meth:`to_text` 方法。
"""
source: str
"""发送此消息的代理名称。"""
models_usage: RequestUsage | None = None
"""生成此消息时产生的模型客户端使用量。"""
metadata: Dict[str, str] = {}
"""关于消息的额外元数据。"""
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
"""消息创建的时间。"""
StructuredContentType = TypeVar("StructuredContentType", bound=BaseModel, covariant=True)
"""结构化内容类型的类型变量。"""
[文档]
class StructuredMessage(BaseChatMessage, Generic[StructuredContentType]):
"""一个内容类型未指定的 :class:`BaseChatMessage` 类型。
要创建新的结构化消息类型，需将内容类型指定为
`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}
.. 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="agent",
format_string="Hello, {text} {number}!",
)
print(message.to_text()) # Hello, agent 42!
"""
content: StructuredContentType
"""消息的内容。必须是 `Pydantic BaseModel <https://docs.pydantic.dev/latest/concepts/models/>`_ 的子类。"""
format_string: Optional[str] = None
"""(实验性) 一个可选格式字符串，用于将内容渲染为人类可读格式。
该格式字符串可以使用内容模型的字段作为占位符。
例如，如果内容模型有一个字段 `name`，你可以在格式字符串中使用
`{name}` 来包含该字段的值。
该格式字符串在 :meth:`to_text` 方法中用于创建消息的
人类可读表示形式。
此设置是实验性的，未来会发生变化。
"""
@computed_field
def type(self) -> str:
return self.__class__.__name__
[文档]
def to_text(self) -> str:
if self.format_string is not None:
return self.format_string.format(**self.content.model_dump())
else:
return self.content.model_dump_json()
[文档]
def to_model_text(self) -> str:
if self.format_string is not None:
return self.format_string.format(**self.content.model_dump())
else:
return self.content.model_dump_json()
[文档]
def to_model_message(self) -> UserMessage:
return UserMessage(
content=self.content.model_dump_json(),
source=self.source,
)
class StructureMessageConfig(BaseModel):
"""结构化输出的声明式配置。"""
json_schema: Dict[str, Any]
format_string: Optional[str] = None
content_model_name: str
class StructuredMessageFactory(ComponentBase[StructureMessageConfig], Component[StructureMessageConfig]):
""":meta private:
一个从 Pydantic 模型或 JSON 模式创建结构化聊天消息的组件。
该组件帮助生成内容由 Pydantic 模型定义的强类型聊天消息。
适用于需要验证、格式化和序列化消息结构的声明式工作流。
可以直接使用 `BaseModel` 子类初始化该组件，或从配置对象（如从磁盘或数据库加载）动态创建。
### 示例 1: 从 Pydantic 模型创建
.. code-block:: python
from pydantic import BaseModel
from autogen_agentchat.messages import StructuredMessageFactory
class TestContent(BaseModel):
field1: str
field2: int
format_string = "This is a string {field1} and this is an int {field2}"
sm_component = StructuredMessageFactory(input_model=TestContent, format_string=format_string)
message = sm_component.StructuredMessage(
source="test_agent", content=TestContent(field1="Hello", field2=42), format_string=format_string
)
print(message.to_model_text()) # Output: This is a string Hello and this is an int 42
config = sm_component.dump_component()
s_m_dyn = StructuredMessageFactory.load_component(config)
message = s_m_dyn.StructuredMessage(
source="test_agent",
content=s_m_dyn.ContentModel(field1="dyn agent", field2=43),
format_string=s_m_dyn.format_string,
)
print(type(message)) # StructuredMessage[GeneratedModel]
print(message.to_model_text()) # Output: This is a string dyn agent and this is an int 43
属性:
component_config_schema (StructureMessageConfig): 定义该组件的配置结构。
component_provider_override (str): 在外部工具中引用该组件的路径。
component_type (str): 用于分类的标识符（如 "structured_message"）。
异常:
ValueError: 如果既未提供 `json_schema` 也未提供 `input_model`。
参数:
json_schema (Optional[str]): 用于动态创建 Pydantic 模型的 JSON 模式。
input_model (Optional[Type[BaseModel]]): 定义预期消息结构的 `BaseModel` 子类。
format_string (Optional[str]): 将内容渲染为人类可读格式的可选字符串。
content_model_name (Optional[str]): 生成的 Pydantic 模型的可选名称。
"""
component_config_schema = StructureMessageConfig
component_provider_override = "autogen_agentchat.messages.StructuredMessageFactory"
component_type = "structured_message"
def __init__(
self,
json_schema: Optional[Dict[str, Any]] = None,
input_model: Optional[Type[BaseModel]] = None,
format_string: Optional[str] = None,
content_model_name: Optional[str] = None,
) -> None:
self.format_string = format_string
if json_schema:
self.ContentModel = schema_to_pydantic_model(
json_schema, model_name=content_model_name or "GeneratedContentModel"
)
elif input_model:
self.ContentModel = input_model
else:
raise ValueError("Either `json_schema` or `input_model` must be provided.")
self.StructuredMessage = StructuredMessage[self.ContentModel] # type: ignore[name-defined]
def _to_config(self) -> StructureMessageConfig:
return StructureMessageConfig(
json_schema=self.ContentModel.model_json_schema(),
format_string=self.format_string,
content_model_name=self.ContentModel.__name__,
)
@classmethod
def _from_config(cls, config: StructureMessageConfig) -> "StructuredMessageFactory":
return cls(
json_schema=config.json_schema,
format_string=config.format_string,
content_model_name=config.content_model_name,
)
[文档]
class TextMessage(BaseTextChatMessage):
"""仅包含字符串内容的文本消息。"""
type: Literal["TextMessage"] = "TextMessage"
[文档]
class MultiModalMessage(BaseChatMessage):
"""多模态消息。"""
content: List[str | Image]
"""消息的内容。"""
type: Literal["MultiModalMessage"] = "MultiModalMessage"
[文档]
def to_model_text(self, image_placeholder: str | None = "[image]") -> str:
"""将消息内容转换为纯字符串表示形式。
如果存在图像，默认情况下将被替换为图像占位符，
否则当设置为 None 时将是 base64 字符串。
"""
text = ""
for c in self.content:
if isinstance(c, str):
text += c
elif isinstance(c, Image):
if image_placeholder is not None:
text += f" {image_placeholder}"
else:
text += f" {c.to_base64()}"
return text
[文档]
def to_text(self, iterm: bool = False) -> str:
result: List[str] = []
for c in self.content:
if isinstance(c, str):
result.append(c)
else:
if iterm:
# iTerm2 image rendering protocol: https://iterm2.com/documentation-images.html
image_data = c.to_base64()
result.append(f"\033]1337;File=inline=1:{image_data}\a\n")
else:
result.append("<image>")
return "\n".join(result)
[文档]
def to_model_message(self) -> UserMessage:
return UserMessage(content=self.content, source=self.source)
[文档]
class StopMessage(BaseTextChatMessage):
"""请求停止对话的消息。"""
type: Literal["StopMessage"] = "StopMessage"
[文档]
class HandoffMessage(BaseTextChatMessage):
"""请求将对话转交给其他代理的消息。"""
target: str
"""要交接的目标代理的名称。"""
context: List[LLMMessage] = []
"""要传递给目标代理的模型上下文。"""
type: Literal["HandoffMessage"] = "HandoffMessage"
[文档]
class CodeGenerationEvent(BaseAgentEvent):
"""表示代码生成事件的事件。"""
retry_attempt: int
"Retry number, 0 means first generation"
content: str
"The complete content as string."
code_blocks: List[CodeBlock]
"List of code blocks present in content"
type: Literal["CodeGenerationEvent"] = "CodeGenerationEvent"
[文档]
def to_text(self) -> str:
return self.content
[文档]
class CodeExecutionEvent(BaseAgentEvent):
"""表示代码执行事件的事件。"""
retry_attempt: int
"Retry number, 0 means first execution"
result: CodeResult
"Code Execution Result"
type: Literal["CodeExecutionEvent"] = "CodeExecutionEvent"
[文档]
def to_text(self) -> str:
return self.result.output
[文档]
class MemoryQueryEvent(BaseAgentEvent):
"""表示内存查询结果的事件。"""
content: List[MemoryContent]
"""内存查询结果。"""
type: Literal["MemoryQueryEvent"] = "MemoryQueryEvent"
[文档]
def to_text(self) -> str:
return str(self.content)
[文档]
class ModelClientStreamingChunkEvent(BaseAgentEvent):
"""表示模型客户端在流模式下输出的文本块的事件。"""
content: str
"""来自模型客户端的字符串片段。"""
type: Literal["ModelClientStreamingChunkEvent"] = "ModelClientStreamingChunkEvent"
[文档]
def to_text(self) -> str:
return self.content
[文档]
class ThoughtEvent(BaseAgentEvent):
"""表示模型思考过程的事件。
用于传递推理模型生成的推理标记，
或函数调用生成的额外文本内容。"""
content: str
"""模型的思考过程。"""
type: Literal["ThoughtEvent"] = "ThoughtEvent"
[文档]
def to_text(self) -> str:
return self.content
[文档]
class SelectSpeakerEvent(BaseAgentEvent):
"""表示已为对话选择发言者的事件。"""
content: List[str]
"""所选发言者的名称。"""
type: Literal["SelectSpeakerEvent"] = "SelectSpeakerEvent"
[文档]
def to_text(self) -> str:
return str(self.content)
class SelectorEvent(BaseAgentEvent):
"""从`SelectorGroupChat`发出的事件。"""
content: str
"""事件的内容。"""
type: Literal["SelectorEvent"] = "SelectorEvent"
def to_text(self) -> str:
return str(self.content)
class MessageFactory:
""":meta private:
用于从可JSON序列化的字典创建消息的工厂。
这对于从JSON数据反序列化消息很有用。
"""
def __init__(self) -> None:
self._message_types: Dict[str, type[BaseAgentEvent | BaseChatMessage]] = {}
# Register all message types.
self._message_types[TextMessage.__name__] = TextMessage
self._message_types[MultiModalMessage.__name__] = MultiModalMessage
self._message_types[StopMessage.__name__] = StopMessage
self._message_types[ToolCallSummaryMessage.__name__] = ToolCallSummaryMessage
self._message_types[HandoffMessage.__name__] = HandoffMessage
self._message_types[ToolCallRequestEvent.__name__] = ToolCallRequestEvent
self._message_types[ToolCallExecutionEvent.__name__] = ToolCallExecutionEvent
self._message_types[MemoryQueryEvent.__name__] = MemoryQueryEvent
self._message_types[UserInputRequestedEvent.__name__] = UserInputRequestedEvent
self._message_types[ModelClientStreamingChunkEvent.__name__] = ModelClientStreamingChunkEvent
self._message_types[ThoughtEvent.__name__] = ThoughtEvent
self._message_types[SelectSpeakerEvent.__name__] = SelectSpeakerEvent
self._message_types[CodeGenerationEvent.__name__] = CodeGenerationEvent
self._message_types[CodeExecutionEvent.__name__] = CodeExecutionEvent
def is_registered(self, message_type: type[BaseAgentEvent | BaseChatMessage]) -> bool:
"""检查消息类型是否已在工厂中注册。"""
# Get the class name of the message type.
class_name = message_type.__name__
# Check if the class name is already registered.
return class_name in self._message_types
def register(self, message_type: type[BaseAgentEvent | BaseChatMessage]) -> None:
"""向工厂注册一个新的消息类型。"""
if self.is_registered(message_type):
raise ValueError(f"Message type {message_type} is already registered.")
if not issubclass(message_type, BaseChatMessage) and not issubclass(message_type, BaseAgentEvent):
raise ValueError(f"Message type {message_type} must be a subclass of BaseChatMessage or BaseAgentEvent.")
# Get the class name of the
class_name = message_type.__name__
# Check if the class name is already registered.
# Register the message type.
self._message_types[class_name] = message_type
def create(self, data: Mapping[str, Any]) -> BaseAgentEvent | BaseChatMessage:
"""从可JSON序列化的数据字典创建消息。"""
# Get the type of the message from the dictionary.
message_type = data.get("type")
if message_type is None:
raise ValueError("Field 'type' is required in the message data to recover the message type.")
if message_type not in self._message_types:
raise ValueError(f"Unknown message type: {message_type}")
if not isinstance(message_type, str):
raise ValueError(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.
assert issubclass(message_class, BaseChatMessage) or issubclass(message_class, BaseAgentEvent)
return message_class.load(data)
ChatMessage = Annotated[
TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage,
Field(discriminator="type"),
]
"""所有 :class:`BaseChatMessage` 内置具体子类的联合类型。
不包括 :class:`StructuredMessage` 类型。"""
AgentEvent = Annotated[
ToolCallRequestEvent
| ToolCallExecutionEvent
| MemoryQueryEvent
| UserInputRequestedEvent
| ModelClientStreamingChunkEvent
| ThoughtEvent
| SelectSpeakerEvent
| CodeGenerationEvent
| CodeExecutionEvent,
Field(discriminator="type"),
]
"""所有 :class:`BaseAgentEvent` 内置具体子类的联合类型。"""
__all__ = [
"AgentEvent",
"BaseMessage",
"ChatMessage",
"BaseChatMessage",
"BaseAgentEvent",
"BaseTextChatMessage",
"StructuredContentType",
"StructuredMessage",
"StructuredMessageFactory",
"HandoffMessage",
"MultiModalMessage",
"StopMessage",
"TextMessage",
"ToolCallExecutionEvent",
"ToolCallRequestEvent",
"ToolCallSummaryMessage",
"MemoryQueryEvent",
"UserInputRequestedEvent",
"ModelClientStreamingChunkEvent",
"ThoughtEvent",
"SelectSpeakerEvent",
"MessageFactory",
"CodeGenerationEvent",
"CodeExecutionEvent",
]
