Eric Zhu 025490a1bd
Use class hierarchy to organize AgentChat message types and introduce StructuredMessage type (#5998)
This PR refactored `AgentEvent` and `ChatMessage` union types to
abstract base classes. This allows for user-defined message types that
subclass one of the base classes to be used in AgentChat.

To support a unified interface for working with the messages, the base
classes added abstract methods for:
- Convert content to string
- Convert content to a `UserMessage` for model client
- Convert content for rendering in console.
- Dump into a dictionary
- Load and create a new instance from a dictionary

This way, all agents such as `AssistantAgent` and `SocietyOfMindAgent`
can utilize the unified interface to work with any built-in and
user-defined message type.

This PR also introduces a new message type, `StructuredMessage` for
AgentChat (Resolves #5131), which is a generic type that requires a
user-specified content type.

You can create a `StructuredMessage` as follow:

```python

class MessageType(BaseModel):
  data: str
  references: List[str]

message = StructuredMessage[MessageType](content=MessageType(data="data", references=["a", "b"]), source="user")

# message.content is of type `MessageType`. 
```

This PR addresses the receving side of this message type. To produce
this message type from `AssistantAgent`, the work continue in #5934.

Added unit tests to verify this message type works with agents and
teams.
2025-03-26 16:19:52 -07:00

45 lines
1.5 KiB
Python

from typing import List, Union
from autogen_core import FunctionCall, Image
from autogen_core.models import FunctionExecutionResult, LLMMessage, UserMessage
from pydantic import BaseModel
# Type aliases for convenience
_StructuredContent = BaseModel
_UserContent = Union[str, List[Union[str, Image]]]
_AssistantContent = Union[str, List[FunctionCall]]
_FunctionExecutionContent = List[FunctionExecutionResult]
_SystemContent = str
def content_to_str(
content: _UserContent | _AssistantContent | _FunctionExecutionContent | _SystemContent | _StructuredContent,
) -> str:
"""Convert the content of an LLMMessage to a string."""
if isinstance(content, str):
return content
elif isinstance(content, BaseModel):
return content.model_dump_json()
else:
result: List[str] = []
for c in content:
if isinstance(c, str):
result.append(c)
elif isinstance(c, Image):
result.append("<image>")
else:
result.append(str(c))
return "\n".join(result)
def remove_images(messages: List[LLMMessage]) -> List[LLMMessage]:
"""Remove images from a list of LLMMessages"""
str_messages: List[LLMMessage] = []
for message in messages:
if isinstance(message, UserMessage) and isinstance(message.content, list):
str_messages.append(UserMessage(content=content_to_str(message.content), source=message.source))
else:
str_messages.append(message)
return str_messages