Skip to content

core

ASSISTANT_ROLE module-attribute 

ASSISTANT_ROLE = MessageRole(ASSISTANT)

The assistant role with name not specified.

MaybeOneOrMany module-attribute 

MaybeOneOrMany = Union[_T, Sequence[_T], None]

A type that can be either a single item, a sequence of items, or None.

OneOrMany module-attribute 

OneOrMany = Union[_T, Sequence[_T]]

A type that can be either a single item or a sequence of items.

SYSTEM_ROLE module-attribute 

SYSTEM_ROLE = MessageRole(SYSTEM)

The system role with name not specified.

StrOrImg module-attribute 

StrOrImg = Union[String, Image]

A type that can be either a string or an image.

String module-attribute 

String = 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 
205
206
207
208
209
210
211
212
213
214
215
216
217
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
219
220
221
222
223
224
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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 
39
40
41
42
43
44
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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}")

CallFuture 

CallFuture(
    func: Callable,
    *args: Any,
    use_process: bool = False,
    lazy_eval: bool = False,
    **kwargs: Any
)

Bases: FutureValue

Represent a function call that may not be ready yet.

Parameters:

  • func (Callable) –

    The function to call.

  • *args (Any, default: () ) –

    The arguments of the function.

  • use_process (bool, default: False ) –

    Whether to use a process pool executor.

  • lazy_eval (bool, default: False ) –

    Whether to delay the start of the call until needed.

  • **kwargs (Any, default: {} ) –

    The keyword arguments of the function.

Source code in src/appl/core/types/futures.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    func: Callable,
    *args: Any,
    use_process: bool = False,
    lazy_eval: bool = False,
    **kwargs: Any,
):
    """Initialize the CallFuture.

    Args:
        func: The function to call.
        *args: The arguments of the function.
        use_process: Whether to use a process pool executor.
        lazy_eval: Whether to delay the start of the call until needed.
        **kwargs: The keyword arguments of the function.
    """
    # ? maybe use a global executor from the config, or use thread-level executor if running in multi-threading.
    self._executor = (
        ProcessPoolExecutor(max_workers=1)
        if use_process
        else ThreadPoolExecutor(
            max_workers=1, thread_name_prefix=threading.current_thread().name
        )
    )
    self._submit_fn = lambda: self._executor.submit(func, *args, **kwargs)
    self._submitted = False
    self._info = func.__name__
    # self._debug = False
    # if self._debug:
    #     # arg and kwargs might contains future objects
    #     args_list = [f"{arg}" for arg in args] + [
    #         f"{k}={v!r}" for k, v in kwargs.items()
    #     ]
    #     args_str = ", ".join(args_list)
    #     self._info += f"({args_str})"
    if not lazy_eval:
        # delay the start of the call until needed
        self._submit()

future property 

future

The future object of the call.

val property 

val

The value of the future.

cancel 

cancel() -> bool

Cancel the call.

Source code in src/appl/core/types/futures.py 
89
90
91
92
93
94
95
def cancel(self) -> bool:
    """Cancel the call."""
    # Attempt to cancel the call
    res = self.future.cancel()
    if res:
        self._executor.shutdown()  # the executor is not needed anymore
    return res

done 

done() -> bool

Check if the call has completed.

Source code in src/appl/core/types/futures.py 
 97
 98
 99
100
def done(self) -> bool:
    """Check if the call has completed."""
    # Check if the future has completed
    return self.future.done()

result 

result(timeout: Optional[float] = None) -> Any

Get the result of the call.

Source code in src/appl/core/types/futures.py 
82
83
84
85
86
87
def result(self, timeout: Optional[float] = None) -> Any:
    """Get the result of the call."""
    # This will block until the result is available
    res = self.future.result(timeout)
    self._executor.shutdown()  # the executor is not needed anymore
    return res

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 
157
158
159
160
161
162
163
164
165
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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}")

CmpStringFuture 

CmpStringFuture(
    a: StringFuture,
    b: StringFuture,
    op: Callable[[str, str], bool],
)

Bases: FutureValue

Represent a comparison between a StringFuture and another value.

Source code in src/appl/core/types/futures.py 
116
117
118
119
120
121
122
def __init__(
    self, a: "StringFuture", b: "StringFuture", op: Callable[[str, str], bool]
):
    """Initialize the CmpStringFuture."""
    self._a = a
    self._b = b
    self._op = op

val property 

val

The value of the future.

ContentList

Bases: BaseModel

Represent a list of contents containing text and images.

append 

append(content: StrOrImg) -> 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 
76
77
78
79
80
81
82
83
84
def append(self, content: StrOrImg) -> 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[StrOrImg]) -> None

Extend the list with multiple contents.

Source code in src/appl/core/types/content.py 
86
87
88
89
def extend(self, contents: list[StrOrImg]) -> 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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_contents(self) -> List[Dict[str, Any]]:
    """Return the contents as a list of dictionaries."""

    def get_dict(content):
        if isinstance(content, Image):
            image_args = {"url": content.url}
            if content.detail:
                image_args["detail"] = content.detail
            return {"type": "image_url", "image_url": image_args}
        return {"type": "text", "text": str(content)}

    return [get_dict(c) for c in self.contents]

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 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
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 
373
374
375
376
377
378
379
380
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 
332
333
334
335
336
337
338
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 
365
366
367
368
369
370
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 
382
383
384
385
386
387
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 
340
341
342
def materialize(self) -> None:
    """Materialize the messages in the conversation."""
    str(self)

set_system_messages 

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

Set the system messages.

Source code in src/appl/core/message.py 
344
345
346
347
348
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.

Image 

Image(url: str, detail: Optional[str] = None)

Bases: BaseModel

Represent an image in the message.

See the guide for more information about the detail level.

Source code in src/appl/core/types/content.py 
24
25
26
27
28
29
30
def __init__(self, url: str, detail: Optional[str] = None) -> None:
    """Initialize the image with the URL and detail level.

    See [the guide](https://platform.openai.com/docs/guides/vision/low-or-high-fidelity-image-understanding)
    for more information about the detail level.
    """
    super().__init__(url=url, detail=detail)

from_file classmethod 

from_file(
    file: PathLike, detail: Optional[str] = None
) -> Image

Construct an image prompt from an image file.

Source code in src/appl/core/types/content.py 
43
44
45
46
47
@classmethod
def from_file(cls, file: PathLike, detail: Optional[str] = None) -> "Image":
    """Construct an image prompt from an image file."""
    image = PIL.Image.open(file)
    return cls.from_image(image, detail)

from_image classmethod 

from_image(
    image: ImageFile, detail: Optional[str] = None
) -> Image

Construct an image prompt from a PIL ImageFile.

Source code in src/appl/core/types/content.py 
32
33
34
35
36
37
38
39
40
41
@classmethod
def from_image(cls, image: ImageFile, detail: Optional[str] = None) -> "Image":
    """Construct an image prompt from a PIL ImageFile."""
    buffered = BytesIO()
    # Save the image to the buffer in PNG format
    image.save(buffered, format="PNG")
    # Get the byte data from the buffer
    img_byte = buffered.getvalue()
    img_base64 = base64.b64encode(img_byte).decode("utf-8")
    return cls(url=f"data:image/png;base64,{img_base64}", detail=detail)

MessageRole 

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

Bases: BaseModel

The role of the message owner.

Parameters:

  • type (Optional[str], default: None ) –

    The type of the role.

  • name (Optional[str], default: None ) –

    An optional name for the role, differentiate between roles of the same type."

Source code in src/appl/core/types/role.py 
16
17
18
19
20
21
22
23
def __init__(self, type: Optional[str] = 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."
    """
    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 
45
46
47
48
49
50
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

ResponseType

Bases: str, Enum

The type of generation response.

IMAGE class-attribute instance-attribute 

IMAGE = 'image'

An image.

OBJECT class-attribute instance-attribute 

OBJECT = 'obj'

An instance of a response model.

TEXT class-attribute instance-attribute 

TEXT = 'text'

A text completion.

TOOL_CALL class-attribute instance-attribute 

TOOL_CALL = 'tool_calls'

A list of tool calls.

UNFINISHED class-attribute instance-attribute 

UNFINISHED = 'unfinished'

The response is not finished.

StringFuture 

StringFuture(content: Any = '', set_value: bool = False)

Bases: FutureValue, BaseModel

StringFuture is a string that may not be ready yet.

Source code in src/appl/core/types/futures.py 
136
137
138
139
140
141
142
143
144
def __init__(self, content: Any = "", set_value: bool = False):
    """Initialize the StringFuture."""
    if set_value:
        if not isinstance(content, List):
            raise ValueError("Cannot set value to non-list.")
        s = content
    else:
        s = [content]
    super().__init__(s=s)

val property 

val

The value of the future.

from_list classmethod 

from_list(content: List[Any]) -> StringFuture

Create a StringFuture from a list of content.

Source code in src/appl/core/types/futures.py 
146
147
148
149
@classmethod
def from_list(cls, content: List[Any]) -> "StringFuture":
    """Create a StringFuture from a list of content."""
    return cls(content, set_value=True)

join 

join(iterable: Iterable[StringFuture]) -> StringFuture

Concatenate any number of strings.

The StringFuture whose method is called is inserted in between each given StringFuture. The result is returned as a new StringFuture.

Source code in src/appl/core/types/futures.py 
163
164
165
166
167
168
169
170
171
172
173
174
def join(self, iterable: Iterable["StringFuture"]) -> "StringFuture":
    """Concatenate any number of strings.

    The StringFuture whose method is called is inserted in between each
    given StringFuture. The result is returned as a new StringFuture.
    """
    result = []
    for i, x in enumerate(iterable):
        if i != 0:
            result.append(self)
        result.append(x)
    return StringFuture.from_list(result)

materialized 

materialized() -> StringFuture

Materialize the StringFuture.

Source code in src/appl/core/types/futures.py 
154
155
156
157
def materialized(self) -> "StringFuture":
    """Materialize the StringFuture."""
    self.s = [self._collapse()]
    return self

serialize 

serialize() -> str

Serialize the StringFuture.

Source code in src/appl/core/types/futures.py 
159
160
161
def serialize(self) -> str:
    """Serialize the StringFuture."""
    return str(self)

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 
171
172
173
174
175
176
177
178
179
180
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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}")

ToolCall

Bases: BaseModel

The class representing a tool call.

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 
197
198
199
200
201
202
203
204
205
@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 
207
208
209
210
211
212
213
214
@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 
186
187
188
189
190
191
192
193
194
195
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 
252
253
254
255
256
257
258
259
260
261
262
263
264
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
266
267
268
269
270
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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 
186
187
188
189
190
191
192
193
194
195
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
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 
119
120
121
122
123
124
125
126
127
128
129
130
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 
79
80
81
82
83
84
85
86
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 
132
133
134
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 
66
67
68
69
70
71
72
73
74
75
76
77
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 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def as_message(
    role: Optional[MessageRole],
    content: StrOrImg,
    *args: Any,
    **kwargs: Any,
) -> BaseMessage:
    """Create a message with role, content and extra arguments."""
    role_type = 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 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
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

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 
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
54
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 
19
20
21
22
23
24
25
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 
20
21
22
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 
13
14
15
16
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)

is_string 

is_string(s: Any) -> bool

Check if the object is a StringFuture or str.

Source code in src/appl/core/types/futures.py 
256
257
258
def is_string(s: Any) -> bool:
    """Check if the object is a StringFuture or str."""
    return isinstance(s, StringFuture) or isinstance(s, str)

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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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 
14
15
16
17
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)