Skip to content

message

AIMessage 

AIMessage(
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    tool_calls: Optional[List[ToolCall]] = None,
    **kwargs: Any
)

Bases: BaseMessage

An assistant message in the conversation.

Source code in src/appl/core/message.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
def __init__(
    self,
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    tool_calls: Optional[List[ToolCall]] = None,
    **kwargs: Any,
) -> None:
    """Create an assistant message with content and extra arguments."""
    if tool_calls is None:
        tool_calls = []
    super().__init__(content=content, role=role, tool_calls=tool_calls, **kwargs)
    self.validate_role(ASSISTANT_ROLE)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(
    default_role: Optional[MessageRole] = None,
) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
248
249
250
251
252
253
def get_dict(self, default_role: Optional[MessageRole] = None) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    data = super().get_dict(default_role)
    if len(self.tool_calls):
        data["tool_calls"] = [call.get_dict() for call in self.tool_calls]
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

BaseMessage 

BaseMessage(content: Any = None, *args: Any, **kwargs: Any)

Bases: BaseModel, ABC

The base class for messages.

Provides a more flexible way to create a message.

Source code in src/appl/core/message.py 
67
68
69
70
71
72
def __init__(self, content: Any = None, *args: Any, **kwargs: Any) -> None:
    """Create a message with content and extra arguments.

    Provides a more flexible way to create a message.
    """
    super().__init__(content=content, *args, **kwargs)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(
    default_role: Optional[MessageRole] = None,
) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_dict(self, default_role: Optional[MessageRole] = None) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    # materialize the content using str()
    role = self.role or default_role
    if role is None:
        raise ValueError("Role or default role must be provided.")
    if role.type is None:
        if default_role and default_role.type:
            role = MessageRole(type=default_role.type, name=role.name)
        else:
            raise ValueError("Role type must be provided.")
    data = {"content": self.get_content(as_str=True), **role.get_dict()}
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

ChatMessage 

ChatMessage(
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any
)

Bases: BaseMessage

A message in the chat conversation.

Source code in src/appl/core/message.py 
186
187
188
189
190
191
192
193
194
def __init__(
    self,
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any,
) -> None:
    """Create a chat message with content and extra arguments."""
    super().__init__(content=content, role=role, **kwargs)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(
    default_role: Optional[MessageRole] = None,
) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_dict(self, default_role: Optional[MessageRole] = None) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    # materialize the content using str()
    role = self.role or default_role
    if role is None:
        raise ValueError("Role or default role must be provided.")
    if role.type is None:
        if default_role and default_role.type:
            role = MessageRole(type=default_role.type, name=role.name)
        else:
            raise ValueError("Role type must be provided.")
    data = {"content": self.get_content(as_str=True), **role.get_dict()}
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

Conversation 

Conversation(
    messages: Optional[
        Union[Iterable[BaseMessage], Iterable[Dict]]
    ] = None,
    *,
    system_messages: Optional[
        Iterable[SystemMessage]
    ] = None
)

Bases: BaseModel

A conversation containing messages.

Source code in src/appl/core/message.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def __init__(
    self,
    messages: Optional[Union[Iterable[BaseMessage], Iterable[Dict]]] = None,
    *,
    system_messages: Optional[Iterable[SystemMessage]] = None,
) -> None:
    """Create a conversation with messages and system messages."""
    if messages is None:
        messages = []
    elif isinstance(messages, Iterable):
        messages = [convert_to_message(m) for m in messages]
    else:
        raise ValueError(f"Invalid messages: {messages}")
    if system_messages is None:
        system_messages = [m for m in messages if isinstance(m, SystemMessage)]
        messages = [m for m in messages if not isinstance(m, SystemMessage)]
    else:
        for m in messages:
            if isinstance(m, SystemMessage):
                raise ValueError("System messages found in both arguments.")
    super().__init__(
        messages=messages,
        system_messages=system_messages,
    )

has_message_role property 

has_message_role: bool

Whether the conversation has message roles.

append 

append(message: Message) -> None

Append a message to the conversation.

Source code in src/appl/core/message.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def append(self, message: Message) -> None:
    """Append a message to the conversation."""
    if message.is_system:
        if len(self.messages):
            # NOTE: Now allow appending system message after other messages
            # raise ValueError("Cannot append system message after other messages.")

            # Added a warning instead
            logger.warning(
                "Modifying system message after other types of messages."
            )
        self.system_messages.append(message)  # type: ignore
    else:
        self.messages.append(message)

as_list 

as_list(
    default_role: Optional[MessageRole] = USER_ROLE,
) -> List[Dict[str, str]]

Return a list of dict representation of the conversation.

Source code in src/appl/core/message.py 
454
455
456
457
458
459
460
461
462
def as_list(
    self, default_role: Optional[MessageRole] = USER_ROLE
) -> List[Dict[str, str]]:
    """Return a list of dict representation of the conversation."""
    # TODO: rename to `to_list`
    self.collapse()
    res = [m.get_dict() for m in self.system_messages]
    res += [m.get_dict(default_role) for m in self.messages]
    return res

collapse 

collapse() -> 'Conversation'

Collapse the messages in the conversation.

Source code in src/appl/core/message.py 
410
411
412
413
414
415
416
def collapse(self) -> "Conversation":
    """Collapse the messages in the conversation."""
    self.system_messages = collapse_messages(self.system_messages)
    if len(self.system_messages) > 1:
        raise ValueError("System messages cannot be fully collapsed.")
    self.messages = collapse_messages(self.messages)
    return self

extend 

extend(other: 'Conversation') -> None

Extend the conversation with another conversation.

Source code in src/appl/core/message.py 
443
444
445
446
447
448
def extend(self, other: "Conversation") -> None:
    """Extend the conversation with another conversation."""
    for sys_m in other.system_messages:
        self.append(sys_m)
    for m in other.messages:
        self.append(m)

make_copy 

make_copy()

Make a copy of the conversation.

Source code in src/appl/core/message.py 
464
465
466
467
468
469
def make_copy(self):
    """Make a copy of the conversation."""
    return Conversation(
        system_messages=self.system_messages.copy(),
        messages=self.messages.copy(),  # type: ignore
    )

materialize 

materialize() -> None

Materialize the messages in the conversation.

Source code in src/appl/core/message.py 
418
419
420
def materialize(self) -> None:
    """Materialize the messages in the conversation."""
    str(self)

pop 

pop() -> BaseMessage

Pop the last message from the conversation.

Source code in src/appl/core/message.py 
450
451
452
def pop(self) -> BaseMessage:
    """Pop the last message from the conversation."""
    return self.messages.pop()

set_system_messages 

set_system_messages(messages: List[SystemMessage]) -> None

Set the system messages.

Source code in src/appl/core/message.py 
422
423
424
425
426
def set_system_messages(self, messages: List[SystemMessage]) -> None:
    """Set the system messages."""
    if len(self.system_messages):
        logger.warning("Overwriting system message.")
    self.system_messages = messages

SystemMessage 

SystemMessage(
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any
)

Bases: BaseMessage

A system message in the conversation.

Source code in src/appl/core/message.py 
200
201
202
203
204
205
206
207
208
209
def __init__(
    self,
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any,
) -> None:
    """Create a system message with content and extra arguments."""
    super().__init__(content=content, role=role, **kwargs)
    self.validate_role(SYSTEM_ROLE)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(
    default_role: Optional[MessageRole] = None,
) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_dict(self, default_role: Optional[MessageRole] = None) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    # materialize the content using str()
    role = self.role or default_role
    if role is None:
        raise ValueError("Role or default role must be provided.")
    if role.type is None:
        if default_role and default_role.type:
            role = MessageRole(type=default_role.type, name=role.name)
        else:
            raise ValueError("Role type must be provided.")
    data = {"content": self.get_content(as_str=True), **role.get_dict()}
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

ToolMessage 

ToolMessage(
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    tool_call_id: str = "",
    **kwargs: Any
)

Bases: BaseMessage

A tool message in the conversation.

Source code in src/appl/core/message.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
def __init__(
    self,
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    tool_call_id: str = "",
    **kwargs: Any,
) -> None:
    """Create a tool message with content and extra arguments."""
    super().__init__(
        content=content, role=role, tool_call_id=tool_call_id, **kwargs
    )
    self.validate_role(TOOL_ROLE)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(*args: Any, **kwargs: Any) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
295
296
297
298
299
def get_dict(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    data = super().get_dict(*args, **kwargs)
    data["tool_call_id"] = self.tool_call_id
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

UserMessage 

UserMessage(
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any
)

Bases: BaseMessage

A user message in the conversation.

Source code in src/appl/core/message.py 
215
216
217
218
219
220
221
222
223
224
def __init__(
    self,
    content: Any = None,
    *,
    role: Optional[MessageRole] = None,
    **kwargs: Any,
) -> None:
    """Create a user message with content and extra arguments."""
    super().__init__(content=content, role=role, **kwargs)
    self.validate_role(USER_ROLE)

is_ai property 

is_ai: bool

Whether the message is an assistant message.

is_system property 

is_system: bool

Whether the message is a system message.

is_tool property 

is_tool: bool

Whether the message is a tool message.

is_user property 

is_user: bool

Whether the message is a user message.

get_content 

get_content(as_str: bool = False) -> Any

Get the content of the message.

Materialize the content if it is a FutureValue.

Source code in src/appl/core/message.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_content(self, as_str: bool = False) -> Any:
    """Get the content of the message.

    Materialize the content if it is a FutureValue.
    """
    content = self.content
    if content is not None:
        if isinstance(content, ContentList):
            return content.get_contents()  # return a list of dict
        if isinstance(content, Dict):
            return content  # not change the content
        if isinstance(content, FutureValue):
            # materialize the content
            content = content.val
        if as_str:  # not apply to ContentList
            content = str(content)
    return content

get_dict 

get_dict(
    default_role: Optional[MessageRole] = None,
) -> Dict[str, Any]

Return a dict representation of the message.

Source code in src/appl/core/message.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_dict(self, default_role: Optional[MessageRole] = None) -> Dict[str, Any]:
    """Return a dict representation of the message."""
    # materialize the content using str()
    role = self.role or default_role
    if role is None:
        raise ValueError("Role or default role must be provided.")
    if role.type is None:
        if default_role and default_role.type:
            role = MessageRole(type=default_role.type, name=role.name)
        else:
            raise ValueError("Role type must be provided.")
    data = {"content": self.get_content(as_str=True), **role.get_dict()}
    return data

merge 

merge(other: 'BaseMessage') -> Optional['Message']

Merge the message with another message.

Source code in src/appl/core/message.py 
148
149
150
151
152
153
154
155
156
157
158
159
def merge(self: "Message", other: "BaseMessage") -> Optional["Message"]:
    """Merge the message with another message."""
    if self.should_merge(other):
        # merge the content
        res = self.model_copy()
        if isinstance(other.content, ContentList) and not isinstance(
            res.content, ContentList
        ):
            res.content = ContentList([TextContent(res.content)])
        res.content += other.content
        return res
    return None

should_merge 

should_merge(other: 'BaseMessage') -> bool

Whether the message should be merged with the other message.

Source code in src/appl/core/message.py 
107
108
109
110
111
112
113
114
def should_merge(self, other: "BaseMessage") -> bool:
    """Whether the message should be merged with the other message."""
    if self.is_tool or other.is_tool:
        # not merge tool messages
        return False
    if self.content is None or other.content is None:
        return False
    return self.role == other.role

str_with_default_role 

str_with_default_role(
    default_role: Optional[MessageRole] = None,
) -> str

Return the string representation of the message with default role.

Source code in src/appl/core/message.py 
161
162
163
def str_with_default_role(self, default_role: Optional[MessageRole] = None) -> str:
    """Return the string representation of the message with default role."""
    return self._get_colored_content(self.role or default_role)

validate_role 

validate_role(target_role: MessageRole) -> None

Validate the role of the message, fill the role if not provided.

Source code in src/appl/core/message.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def validate_role(self, target_role: MessageRole) -> None:
    """Validate the role of the message, fill the role if not provided."""
    target_type = target_role.type
    if target_type is None:
        raise ValueError("Target role type must be provided.")
    if self.role is None:
        self.role = target_role
    elif self.role.type is None:
        # fill the role type as the target type
        self.role = MessageRole(type=target_type, name=self.role.name)
    elif self.role.type != target_type:
        raise ValueError(f"Invalid role for {target_type} message: {self.role}")

as_message 

as_message(
    role: Optional[Union[MessageRole, str]],
    content: Union[
        ContentPart,
        String,
        Dict,
        Iterable[Dict[str, Any]],
        None,
    ] = None,
    *args: Any,
    **kwargs: Any
) -> BaseMessage

Create a message with role, content and extra arguments.

Source code in src/appl/core/message.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def as_message(
    role: Optional[Union[MessageRole, str]],
    content: Union[ContentPart, String, Dict, Iterable[Dict[str, Any]], None] = None,
    *args: Any,
    **kwargs: Any,
) -> BaseMessage:
    """Create a message with role, content and extra arguments."""
    if isinstance(role, str):
        role = MessageRole(type=role)
    role_type = MessageRoleType(role.type) if role else None
    if role_type not in MESSAGE_CLASS_DICT:
        raise ValueError(f"Unknown role: {role}")
    cls = MESSAGE_CLASS_DICT[role_type]
    processed_content: Union[ContentList, String, Dict, None] = None
    if content is None:
        if role_type != MessageRoleType.ASSISTANT or "tool_calls" not in kwargs:
            raise ValueError("Content must be provided for non-tool-calls messages.")
        processed_content = content
    elif isinstance(content, (StringFuture, str, dict)):
        processed_content = content
    elif isinstance(content, ContentPart):
        processed_content = ContentList(contents=[content])
    elif isinstance(content, Iterable):
        processed_content = ContentList(contents=content)  # type: ignore # mypy failed
    else:
        raise ValueError(f"Invalid content: {content}")
    return cls(content=processed_content, role=role, *args, **kwargs)

collapse_messages 

collapse_messages(messages: List[Message]) -> List[Message]

Collapse a list of the messages by merging the messages with the same sender.

Source code in src/appl/core/message.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def collapse_messages(messages: List[Message]) -> List[Message]:
    """Collapse a list of the messages by merging the messages with the same sender."""
    res = []
    msg: Optional[Message] = None
    for m in messages:
        if msg is None:
            msg = m
        else:
            if (tmp := msg.merge(m)) is not None:
                # merge success, update the msg
                msg = tmp
            else:
                # merge failed, append the old message to the list
                res.append(msg)
                # a new message starts
                msg = m
    if msg is not None:
        res.append(msg)
    return res

convert_to_message 

convert_to_message(
    m: Union[BaseMessage, Dict]
) -> BaseMessage

Convert a message in different formats to a BaseMessage.

Source code in src/appl/core/message.py 
340
341
342
343
344
345
346
347
348
def convert_to_message(
    m: Union[BaseMessage, Dict],
) -> BaseMessage:
    """Convert a message in different formats to a BaseMessage."""
    if isinstance(m, BaseMessage):
        return m
    if isinstance(m, Dict):
        return as_message(**m)
    raise ValueError(f"Invalid message: {m}")

get_colored_role_text 

get_colored_role_text(
    role: Optional[MessageRole], content: str
) -> str

Get the colored text based on the role.

Source code in src/appl/core/message.py 
47
48
49
50
51
52
53
def get_colored_role_text(role: Optional[MessageRole], content: str) -> str:
    """Get the colored text based on the role."""
    if role:
        color = get_role_color(role)
        if color in COLORS:
            return colored(content, color)  # type: ignore
    return content

get_role_color 

get_role_color(role: MessageRole) -> Optional[str]

Get the color of the message based on the role.

Source code in src/appl/core/message.py 
39
40
41
42
43
44
def get_role_color(role: MessageRole) -> Optional[str]:
    """Get the color of the message based on the role."""
    colors = global_vars.configs.settings.messages.colors
    if role.type is None:
        raise ValueError("Cannot get color for role is None")
    return getattr(colors, role.type, None)