Skip to content

core

ASSISTANT_ROLE module-attribute 

ASSISTANT_ROLE = MessageRole(ASSISTANT)

The assistant role with name not specified.

SYSTEM_ROLE module-attribute 

SYSTEM_ROLE = MessageRole(SYSTEM)

The system role with name not specified.

String module-attribute 

String: TypeAlias = Union[StringFuture, str]

String is a type alias for StringFuture or str.

TOOL_ROLE module-attribute 

TOOL_ROLE = MessageRole(TOOL)

The tool role with name not specified.

USER_ROLE module-attribute 

USER_ROLE = MessageRole(USER)

The user role with name not specified.

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}")

ContentList 

ContentList(
    contents: Union[
        Iterable[ContentPart], Iterator[Dict[str, Any]]
    ]
)

Bases: BaseModel

Represent a list of contents containing text, images and audio.

Source code in src/appl/core/types/content.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def __init__(
    self, contents: Union[Iterable[ContentPart], Iterator[Dict[str, Any]]]
) -> None:
    """Initialize the content list with a list of contents."""
    res: List[ContentPart] = []
    for c in contents:
        if isinstance(c, ContentPart):
            res.append(c)
        elif isinstance(c, dict):
            if "type" not in c:
                raise ValueError(f"Invalid content: {c}")
            t = ContentPartType(c["type"])
            cls = CONTENT_PART_CLASS_MAP[t]
            kwargs = c[t.value]
            res.append(cls(**kwargs) if isinstance(kwargs, dict) else cls(kwargs))
        else:
            raise ValueError(f"Invalid content: {c}")
    super().__init__(contents=res)

append 

append(content: ContentPart) -> None

Append a content to the list.

If the last content is a string, it will be concatenated with the new content.

Source code in src/appl/core/types/content.py 
207
208
209
210
211
212
213
214
215
def append(self, content: ContentPart) -> None:
    """Append a content to the list.

    If the last content is a string, it will be concatenated with the new content.
    """
    if is_string(content) and len(self.contents) and is_string(self.contents[-1]):
        self.contents[-1] += content  # type: ignore
    else:
        self.contents.append(content)

extend 

extend(contents: list[ContentPart]) -> None

Extend the list with multiple contents.

Source code in src/appl/core/types/content.py 
217
218
219
220
def extend(self, contents: list[ContentPart]) -> None:
    """Extend the list with multiple contents."""
    for content in contents:
        self.append(content)

get_contents 

get_contents() -> List[Dict[str, Any]]

Return the contents as a list of dictionaries.

Source code in src/appl/core/types/content.py 
222
223
224
def get_contents(self) -> List[Dict[str, Any]]:
    """Return the contents as a list of dictionaries."""
    return [c.get_dict() for c in self.contents]

ContentPart 

ContentPart(*args: Any, **kwargs: Any)

Bases: BaseModel, ABC

Represent a part of the message content.

Source code in src/appl/core/types/content.py 
41
42
43
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the content part."""
    super().__init__(*args, **kwargs)

get_dict 

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py 
49
50
51
def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

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

FutureValue

Bases: ABC

Represents a value that may not be ready yet.

val property 

val

The value of the future.

MessageRole 

MessageRole(
    type: Optional[Union[str, MessageRoleType]] = None,
    name: Optional[str] = None,
)

Bases: BaseModel

The role of the message owner.

Parameters:

Source code in src/appl/core/types/role.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    type: Optional[Union[str, MessageRoleType]] = None,
    name: Optional[str] = None,
):
    """Initialize the MessageRole object.

    Args:
        type: The type of the role.
        name: An optional name for the role, differentiate between roles of the same type."
    """
    if isinstance(type, MessageRoleType):
        type = type.value
    super().__init__(type=type, name=name)

is_assistant property 

is_assistant: bool

Whether the role is an assistant role.

is_system property 

is_system: bool

Whether the role is a system role.

is_tool property 

is_tool: bool

Whether the role is a tool role.

is_user property 

is_user: bool

Whether the role is a user role.

get_dict 

get_dict() -> Dict[str, Any]

Get the role as a dictionary.

Source code in src/appl/core/types/role.py 
57
58
59
60
61
62
def get_dict(self) -> Dict[str, Any]:
    """Get the role as a dictionary."""
    data = {"role": self.type}
    if self.name:
        data["name"] = self.name
    return data

MessageRoleType

Bases: str, Enum

The type of the role.

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}")

TextContent 

TextContent(text: String)

Bases: ContentPart

Represent a text in the message.

Source code in src/appl/core/types/content.py 
59
60
61
def __init__(self, text: String) -> None:
    """Initialize the text content."""
    super().__init__(type=ContentPartType.TEXT, text=text)

get_dict 

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py 
49
50
51
def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

ToolCall 

ToolCall(**kwargs: Any)

Bases: BaseModel

The class representing a tool call.

Source code in src/appl/core/tool.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
def __init__(self, **kwargs: Any):
    """Create a tool call from a dictionary."""
    if "type" in kwargs:
        if kwargs["type"] == "function":
            super().__init__(
                id=kwargs["id"],
                name=kwargs["function"]["name"],
                args=kwargs["function"]["arguments"],
            )
        else:
            raise ValueError(f"Invalid tool call type: {kwargs['type']}")
    else:
        super().__init__(**kwargs)

args class-attribute instance-attribute 

args: str = Field(
    ...,
    description="The arguments to call the function with.",
)

The arguments to call the function with.

id class-attribute instance-attribute 

id: str = Field(..., description="The ID of the tool call.")

The ID of the tool call.

name class-attribute instance-attribute 

name: str = Field(
    ..., description="The name of the function to call."
)

The name of the function to call.

from_dict classmethod 

from_dict(call: Dict) -> ToolCall

Create a ToolCall from a dictionary in the OpenAI format.

Source code in src/appl/core/tool.py 
251
252
253
254
255
256
257
258
259
@classmethod
def from_dict(cls, call: Dict) -> "ToolCall":
    """Create a ToolCall from a dictionary in the OpenAI format."""
    # throw error if incorrect format
    return cls(
        id=call["id"],
        name=call["function"]["name"],
        args=call["function"]["arguments"],
    )

from_openai_tool_call classmethod 

from_openai_tool_call(
    call: ChatCompletionMessageToolCall,
) -> ToolCall

Create a ToolCall from an OpenAI tool call.

Source code in src/appl/core/tool.py 
261
262
263
264
265
266
267
268
@classmethod
def from_openai_tool_call(cls, call: ChatCompletionMessageToolCall) -> "ToolCall":
    """Create a ToolCall from an OpenAI tool call."""
    return cls(
        id=call.id,
        name=call.function.name,
        args=call.function.arguments,
    )

get_dict 

get_dict()

Get the OpenAI format dictionary representation of the tool call.

Source code in src/appl/core/tool.py 
240
241
242
243
244
245
246
247
248
249
def get_dict(self):
    """Get the OpenAI format dictionary representation of the tool call."""
    return {
        "id": self.id,
        "type": "function",
        "function": {
            "name": self.name,
            "arguments": self.args,
        },
    }

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}")

dump_file 

dump_file(
    data: Any,
    file: str,
    mode: str = "w",
    ensure_folder_exists: bool = True,
    file_type: Optional[str] = None,
    *args: Any,
    **kwargs: Any
) -> None

Write the data to a file based on the file extension.

Source code in src/appl/core/io.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def dump_file(
    data: Any,
    file: str,
    mode: str = "w",
    ensure_folder_exists: bool = True,
    file_type: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    """Write the data to a file based on the file extension."""
    if file_type is None:
        file_type = get_ext(file)
    if ensure_folder_exists:
        makedirs(file)

    if file_type == ".json":
        dump_func: Callable = json.dump
    elif file_type in [".yaml", ".yml"]:
        dump_func = yaml.dump
    elif file_type == ".toml":
        dump_func = toml.dump
    elif file_type in PLAIN_TEXT_FILES:

        def dump_func(data, f, *args, **kwargs):
            f.write(data)

    else:
        raise ValueError(f"Unsupported file type {file_type}")
    with open(file, mode) as f:
        dump_func(data, f, *args, **kwargs)

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_ext 

get_ext(file: str) -> str

Get the extension of a file.

Source code in src/appl/core/io.py 
19
20
21
def get_ext(file: str) -> str:
    """Get the extension of a file."""
    return os.path.splitext(file)[1]

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)

load_file 

load_file(
    file: str,
    mode: str = "r",
    file_type: Optional[str] = None,
    open_kwargs: Optional[Dict[str, Any]] = None,
    *args: Any,
    **kwargs: Any
) -> Any

Load a file based on the file extension and return the data.

Source code in src/appl/core/io.py 
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
def load_file(
    file: str,
    mode: str = "r",
    file_type: Optional[str] = None,
    open_kwargs: Optional[Dict[str, Any]] = None,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Load a file based on the file extension and return the data."""
    if file_type is None:
        file_type = get_ext(file)
    if file_type == ".json":
        load_func: Callable = json.load
    elif file_type in [".yaml", ".yml"]:
        load_func = yaml.safe_load
    elif file_type == ".toml":
        load_func = toml.load
    # elif file_type == ".py":
    #     load_func = import_module
    elif file_type in PLAIN_TEXT_FILES:

        def load_func(f: IO[Any], *args: Any, **kwargs: Any) -> Any:
            return f.read()

    else:
        raise ValueError(f"Unsupported file type {file_type}")
    open_kwargs = open_kwargs or {}
    with open(file, mode, **open_kwargs) as f:
        return load_func(f, *args, **kwargs)

makedirs 

makedirs(file: str) -> None

Make the directory of the file if it does not exist.

Source code in src/appl/core/io.py 
13
14
15
16
def makedirs(file: str) -> None:
    """Make the directory of the file if it does not exist."""
    if folder := os.path.dirname(file):
        os.makedirs(folder, exist_ok=True)