Skip to content

message

ChatMessage

Bases: BaseModel

Represents a chat message with role, content, and metadata.

Attributes:

Name Type Description
role MessageRole

Message role (system/user/assistant/function)
name Optional[str]

Optional name associated with the message
content Optional[Any]

Main content of the message
reasoning_content Optional[Any]

Internal reasoning information
tool_calls Optional[List[ChatTool]]

List of tools called in this message
additional_kwargs dict

Extra metadata dictionary
time_created datetime

Timestamp of message creation
Source code in rm_gallery/core/model/message.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class ChatMessage(BaseModel):
    """
    Represents a chat message with role, content, and metadata.

    Attributes:
        role: Message role (system/user/assistant/function)
        name: Optional name associated with the message
        content: Main content of the message
        reasoning_content: Internal reasoning information
        tool_calls: List of tools called in this message
        additional_kwargs: Extra metadata dictionary
        time_created: Timestamp of message creation
    """

    role: MessageRole = Field(default=MessageRole.USER)
    name: Optional[str] = Field(default=None)
    content: Optional[Any] = Field(default="")
    reasoning_content: Optional[Any] = Field(default="")
    tool_calls: Optional[List[ChatTool]] = Field(default=None)
    additional_kwargs: dict = Field(default_factory=dict)
    time_created: datetime = Field(
        default_factory=datetime.now,
        description="Timestamp marking the message creation time",
    )

    def __str__(self) -> str:
        """Returns formatted string representation with timestamp and role."""
        return f"{self.time_created.strftime('%Y-%m-%d %H:%M:%S')} {self.role.value}: {self.content}"

    def __add__(self, other: Any) -> "ChatMessage":
        """
        Concatenates message content with another message delta.

        Args:
            other: Message to merge with current one

        Returns:
            New ChatMessage instance with merged content

        Raises:
            TypeError: If other is not None or ChatMessage
        """
        if other is None:
            return self
        elif isinstance(other, ChatMessage):
            return self.__class__(
                role=self.role,
                name=self.name,
                content=self.content + (other.content if other.content else ""),
                tool_calls=other.tool_calls,
                additional_kwargs=other.additional_kwargs,
            )
        else:
            raise TypeError(
                'unsupported operand type(s) for +: "'
                f"{self.__class__.__name__}"
                f'" and "{other.__class__.__name__}"'
            )

    @staticmethod
    def convert_from_strings(messages: List[str], system_message: str) -> str:
        """
        Converts string list to structured ChatMessage list for debugging.

        Args:
            messages: List of alternating user/assistant messages
            system_message: Initial system message content

        Returns:
            List of structured ChatMessage objects
        """
        result_messages = [
            ChatMessage(role=MessageRole.SYSTEM, content=system_message),
        ]

        toggle_roles = [MessageRole.USER, MessageRole.ASSISTANT]
        for index, msg in enumerate(messages):
            result_messages.append(
                ChatMessage(role=toggle_roles[index % 2], content=msg)
            )

        return result_messages

    @staticmethod
    def convert_to_strings(messages: List["ChatMessage"]) -> Tuple[List[str], str]:
        """
        Converts structured ChatMessages to plain strings for debugging.

        Args:
            messages: List of ChatMessage objects

        Returns:
            Tuple containing:
            - List of non-system messages
            - Extracted system message content
        """
        vanilla_messages = []
        system_message = ""

        for index, msg in enumerate(messages):
            if msg.role == MessageRole.SYSTEM:
                system_message += msg.content
            else:
                vanilla_messages.append(msg.content)

        return vanilla_messages, system_message

__add__(other)

Concatenates message content with another message delta.

Parameters:

Name Type Description Default
other Any

Message to merge with current one

 required

Returns:

Type Description
ChatMessage

New ChatMessage instance with merged content

Raises:

Type Description
TypeError

If other is not None or ChatMessage
Source code in rm_gallery/core/model/message.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __add__(self, other: Any) -> "ChatMessage":
    """
    Concatenates message content with another message delta.

    Args:
        other: Message to merge with current one

    Returns:
        New ChatMessage instance with merged content

    Raises:
        TypeError: If other is not None or ChatMessage
    """
    if other is None:
        return self
    elif isinstance(other, ChatMessage):
        return self.__class__(
            role=self.role,
            name=self.name,
            content=self.content + (other.content if other.content else ""),
            tool_calls=other.tool_calls,
            additional_kwargs=other.additional_kwargs,
        )
    else:
        raise TypeError(
            'unsupported operand type(s) for +: "'
            f"{self.__class__.__name__}"
            f'" and "{other.__class__.__name__}"'
        )

__str__()

Returns formatted string representation with timestamp and role.

Source code in rm_gallery/core/model/message.py 
57
58
59
def __str__(self) -> str:
    """Returns formatted string representation with timestamp and role."""
    return f"{self.time_created.strftime('%Y-%m-%d %H:%M:%S')} {self.role.value}: {self.content}"

convert_from_strings(messages, system_message) staticmethod

Converts string list to structured ChatMessage list for debugging.

Parameters:

Name Type Description Default
messages List[str]

List of alternating user/assistant messages

 required
system_message str

Initial system message content

 required

Returns:

Type Description
str

List of structured ChatMessage objects
Source code in rm_gallery/core/model/message.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@staticmethod
def convert_from_strings(messages: List[str], system_message: str) -> str:
    """
    Converts string list to structured ChatMessage list for debugging.

    Args:
        messages: List of alternating user/assistant messages
        system_message: Initial system message content

    Returns:
        List of structured ChatMessage objects
    """
    result_messages = [
        ChatMessage(role=MessageRole.SYSTEM, content=system_message),
    ]

    toggle_roles = [MessageRole.USER, MessageRole.ASSISTANT]
    for index, msg in enumerate(messages):
        result_messages.append(
            ChatMessage(role=toggle_roles[index % 2], content=msg)
        )

    return result_messages

convert_to_strings(messages) staticmethod

Converts structured ChatMessages to plain strings for debugging.

Parameters:

Name Type Description Default
messages List[ChatMessage]

List of ChatMessage objects

 required

Returns:

Type Description
List[str]

Tuple containing:
str
  • List of non-system messages
Tuple[List[str], str]
  • Extracted system message content
Source code in rm_gallery/core/model/message.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@staticmethod
def convert_to_strings(messages: List["ChatMessage"]) -> Tuple[List[str], str]:
    """
    Converts structured ChatMessages to plain strings for debugging.

    Args:
        messages: List of ChatMessage objects

    Returns:
        Tuple containing:
        - List of non-system messages
        - Extracted system message content
    """
    vanilla_messages = []
    system_message = ""

    for index, msg in enumerate(messages):
        if msg.role == MessageRole.SYSTEM:
            system_message += msg.content
        else:
            vanilla_messages.append(msg.content)

    return vanilla_messages, system_message

ChatResponse

Bases: BaseModel

Represents a chat response with message and metadata.

Attributes:

Name Type Description
message ChatMessage

Main chat message content
raw Optional[dict]

Raw response dictionary from API
delta Optional[ChatMessage]

Incremental update message
error_message Optional[str]

Error description if any
additional_kwargs dict

Extra metadata dictionary
Source code in rm_gallery/core/model/message.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
class ChatResponse(BaseModel):
    """
    Represents a chat response with message and metadata.

    Attributes:
        message: Main chat message content
        raw: Raw response dictionary from API
        delta: Incremental update message
        error_message: Error description if any
        additional_kwargs: Extra metadata dictionary
    """

    message: ChatMessage
    raw: Optional[dict] = None
    delta: Optional[ChatMessage] = None
    error_message: Optional[str] = None
    additional_kwargs: dict = Field(
        default_factory=dict
    )  # other information like token usage or log probs.

    def __str__(self):
        """Returns error message if present, otherwise string representation of main message."""
        if self.error_message:
            return f"Errors: {self.error_message}"
        else:
            return str(self.message)

    def __add__(self, other: Any) -> "ChatResponse":
        """
        Combines response with another response delta.

        Args:
            other: Response to merge with current one

        Returns:
            New ChatResponse instance with merged content

        Raises:
            TypeError: If other is not None or ChatResponse
        """
        if other is None:
            return self
        elif isinstance(other, ChatResponse):
            return self.__class__(
                message=self.message + other.message,
                raw=other.raw,
                delta=other.message,
                error_message=other.error_message,
                additional_kwargs=other.additional_kwargs,
            )
        else:
            raise TypeError(
                'unsupported operand type(s) for +: "'
                f"{self.__class__.__name__}"
                f'" and "{other.__class__.__name__}"'
            )

__add__(other)

Combines response with another response delta.

Parameters:

Name Type Description Default
other Any

Response to merge with current one

 required

Returns:

Type Description
ChatResponse

New ChatResponse instance with merged content

Raises:

Type Description
TypeError

If other is not None or ChatResponse
Source code in rm_gallery/core/model/message.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def __add__(self, other: Any) -> "ChatResponse":
    """
    Combines response with another response delta.

    Args:
        other: Response to merge with current one

    Returns:
        New ChatResponse instance with merged content

    Raises:
        TypeError: If other is not None or ChatResponse
    """
    if other is None:
        return self
    elif isinstance(other, ChatResponse):
        return self.__class__(
            message=self.message + other.message,
            raw=other.raw,
            delta=other.message,
            error_message=other.error_message,
            additional_kwargs=other.additional_kwargs,
        )
    else:
        raise TypeError(
            'unsupported operand type(s) for +: "'
            f"{self.__class__.__name__}"
            f'" and "{other.__class__.__name__}"'
        )

__str__()

Returns error message if present, otherwise string representation of main message.

Source code in rm_gallery/core/model/message.py 
160
161
162
163
164
165
def __str__(self):
    """Returns error message if present, otherwise string representation of main message."""
    if self.error_message:
        return f"Errors: {self.error_message}"
    else:
        return str(self.message)

ChatTool

Bases: BaseModel

Represents a chat tool call with function metadata.

Source code in rm_gallery/core/model/message.py 
24
25
26
27
28
29
class ChatTool(BaseModel):
    """Represents a chat tool call with function metadata."""

    id: str
    function: Tool
    type: Literal["function"]

MessageRole

Bases: str, Enum

Message role.

Source code in rm_gallery/core/model/message.py 
 8
 9
10
11
12
13
14
class MessageRole(str, Enum):
    """Message role."""

    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    FUNCTION = "function"

Tool

Bases: BaseModel

Represents a function tool with name and arguments.

Source code in rm_gallery/core/model/message.py 
17
18
19
20
21
class Tool(BaseModel):
    """Represents a function tool with name and arguments."""

    arguments: str
    name: str

format_messages(messages)

Formats chat messages into XML-style string representation.

Parameters:

Name Type Description Default
messages List[ChatMessage]

List of ChatMessage objects to format

 required

Returns:

Type Description
str

String with messages wrapped in role-specific tags
Source code in rm_gallery/core/model/message.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
def format_messages(messages: List[ChatMessage]) -> str:
    """
    Formats chat messages into XML-style string representation.

    Args:
        messages: List of ChatMessage objects to format

    Returns:
        String with messages wrapped in role-specific tags
    """
    return "\n".join(
        [f"<{message.role}>{message.content}</{message.role}>" for message in messages]
    )