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 
219
220
221
222
223
224
225
226
227
228
229
230
231
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
233
234
235
236
237
238
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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 
53
54
55
56
57
58
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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 
171
172
173
174
175
176
177
178
179
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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

Bases: BaseModel

A conversation containing 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 
364
365
366
367
368
369
370
371
372
373
374
375
376
377
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 
391
392
393
394
395
396
397
398
def as_list(
    self, default_role: Optional[MessageRole] = USER_ROLE
) -> List[Dict[str, str]]:
    """Return a list of dict representation of the conversation."""
    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 
346
347
348
349
350
351
352
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 
379
380
381
382
383
384
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 
400
401
402
403
404
405
def make_copy(self):
    """Make a copy of the conversation."""
    return Conversation(
        system_messages=self.system_messages.copy(),
        messages=self.messages.copy(),
    )

materialize 

materialize() -> None

Materialize the messages in the conversation.

Source code in src/appl/core/message.py 
354
355
356
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 
386
387
388
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 
358
359
360
361
362
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 
185
186
187
188
189
190
191
192
193
194
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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 
266
267
268
269
270
271
272
273
274
275
276
277
278
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
280
281
282
283
284
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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 
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 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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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, 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 
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
133
134
135
136
137
138
139
140
141
142
143
144
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(contents=[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 
 93
 94
 95
 96
 97
 98
 99
100
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 
146
147
148
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 
80
81
82
83
84
85
86
87
88
89
90
91
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[MessageRole],
    content: StrOrImg,
    *args: Any,
    **kwargs: Any
) -> BaseMessage

Create a message with role, content and extra arguments.

Source code in src/appl/core/message.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def as_message(
    role: Optional[MessageRole],
    content: StrOrImg,
    *args: Any,
    **kwargs: Any,
) -> BaseMessage:
    """Create a message with role, content and extra arguments."""
    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]
    if isinstance(content, Image):
        content = ContentList(contents=[content])  # type: ignore
    return cls(content=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 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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

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 
33
34
35
36
37
38
39
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 
27
28
29
30
def get_role_color(role: MessageRole) -> Optional[str]:
    """Get the color of the message based on the role."""
    color_dict = configs.getattrs("settings.messages.colors", {})
    return color_dict.get(role.type, None)