Skip to content

role_changer

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

AIRole 

AIRole(name: Optional[str] = None, **kwargs: Any)

Bases: RoleChanger

Change the role of the prompts to assistant.

Parameters:

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

    The name of the assistant role. Defaults to None.

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

    The keyword arguments to pass to the Role

Source code in src/appl/role_changer.py 
62
63
64
65
66
67
68
69
70
def __init__(self, name: Optional[str] = None, **kwargs: Any):
    """Initialize the AIRole object.

    Args:
        name: The name of the assistant role. Defaults to None.
        **kwargs: The keyword arguments to pass to the Role
    """
    role = MessageRole(ASSISTANT, name=name)
    super().__init__(role=role, **kwargs)

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

BaseServer

Bases: ABC

The base class for all servers.

Servers are responsible for communicating with the underlying model.

model_name abstractmethod property 

model_name: str

The name of the model used by the server.

close abstractmethod 

close()

Close the server.

Source code in src/appl/core/server.py 
176
177
178
179
@abstractmethod
def close(self):
    """Close the server."""
    raise NotImplementedError

create 

create(
    args: GenArgs, gen_id: str, **kwargs: Any
) -> CompletionResponse

Create a CompletionResponse from the model with given arguments.

Parameters:

  • args (GenArgs) –

    The arguments for generating the response

  • gen_id (str) –

    The ID of the generation

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

    Additional keyword arguments

Returns: The response from the model.

Source code in src/appl/core/server.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def create(self, args: GenArgs, gen_id: str, **kwargs: Any) -> CompletionResponse:
    """Create a CompletionResponse from the model with given arguments.

    Args:
        args: The arguments for generating the response
        gen_id: The ID of the generation
        **kwargs: Additional keyword arguments
    Returns:
        The response from the model.
    """
    log_llm_call_args = configs.getattrs("settings.logging.display.llm_call_args")
    log_llm_usage = configs.getattrs("settings.logging.display.llm_usage")
    log_llm_response = configs.getattrs("settings.logging.display.llm_response")

    create_args = self._get_create_args(args, **kwargs)
    if log_llm_call_args:
        logger.info(f"Call generation [{gen_id}] with args: {create_args}")

    results = self._create(gen_id=gen_id, **create_args)
    if log_llm_response:
        logger.info(f"Generation [{gen_id}] results: {results}")
    if results.usage and log_llm_usage:
        logger.info(f"Generation [{gen_id}] token usage: {results.usage}")
    if results.cost:
        if "mock_response" in create_args:
            if configs.getattrs("settings.logging.display.llm_cost"):
                logger.info(
                    f"Mock response, estimated cost for real request: {results.cost:.4f}"
                )
        else:
            _update_cost(
                self.model_name,
                results.cost,
                getattr(self, "_cost_currency", "USD"),
            )

    dump_args = create_args.copy()
    for k, v in dump_args.items():
        if k in ["response_format", "response_model"]:
            if isinstance(v, type) and issubclass(v, BaseModel):
                dump_args[k] = json.dumps(v.model_json_schema(), indent=4)

    def trace_gen_response(response: CompletionResponse) -> None:
        add_to_trace(
            GenerationResponseEvent(name=gen_id, args=dump_args, ret=str(response))
        )

    results.register_post_finish_callback(trace_gen_response)
    return results

BaseTool 

BaseTool(func: Callable, **predefined: Any)

Bases: BaseModel, ABC

The base class for a Tool.

Source code in src/appl/core/tool.py 
40
41
42
43
44
45
46
47
48
49
50
def __init__(self, func: Callable, **predefined: Any):
    """Create a tool from a function."""
    name = func.__name__
    sig = inspect.signature(func)
    doc = func.__doc__
    super().__init__(name=name, **self.parse_data(sig, doc, predefined))
    self._predefined = predefined
    self._func = func
    self.__name__ = name
    self.__signature__ = sig  # type: ignore
    self.__doc__ = doc  # overwrite the doc string

examples class-attribute instance-attribute 

examples: List[str] = Field(
    [], description="The examples of the Tool"
)

The examples of the Tool.

info class-attribute instance-attribute 

info: Dict = Field(
    {}, description="Additional information of the Tool"
)

Additional information of the Tool.

long_desc class-attribute instance-attribute 

long_desc: str = Field(
    "", description="The long description of the Tool"
)

The long description of the Tool.

name class-attribute instance-attribute 

name: str = Field(..., description='The name of the Tool')

The name of the Tool.

openai_schema property 

openai_schema: dict

Get the OpenAI schema of the tool.

params class-attribute instance-attribute 

params: type[BaseModel] = Field(
    ..., description="The parameters of the Tool"
)

The parameters of the Tool.

raises class-attribute instance-attribute 

raises: List[Dict[str, Optional[str]]] = Field(
    [], description="The exceptions raised by the Tool"
)

The exceptions raised by the Tool.

returns class-attribute instance-attribute 

returns: type[BaseModel] = Field(
    ..., description="The return of the Tool"
)

The return of the Tool.

short_desc class-attribute instance-attribute 

short_desc: str = Field(
    "", description="The short description of the Tool"
)

The short description of the Tool.

parse_data classmethod 

parse_data(
    sig: Signature,
    docstring: Optional[str],
    predefined: Dict[str, Any],
) -> Dict[str, Any]

Parse data from the signature and docstring of a function.

Source code in src/appl/core/tool.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def parse_data(
    cls, sig: Signature, docstring: Optional[str], predefined: Dict[str, Any]
) -> Dict[str, Any]:
    """Parse data from the signature and docstring of a function."""
    doc = parse(docstring or "")
    data: Dict[str, Any] = {
        "short_desc": doc.short_description or "",
        "long_desc": doc.long_description or "",
    }

    # build params
    params = {}
    doc_param = {p.arg_name: p for p in doc.params}
    for name, param in sig.parameters.items():
        anno = param.annotation
        default = param.default

        if default is param.empty:
            default = ...  # required
        if name in doc_param:
            # fill in desc for the param
            default = Field(default, description=doc_param[name].description)
            # fill in type annotation if not annotated in the function
            if (anno is param.empty) and (doc_param[name].type_name is not None):
                # use type annotation from docstring
                anno = doc_param[name].type_name
        # replace empty annotation with Any
        if anno is param.empty:
            anno = Any
        if name not in predefined:
            params[name] = (anno, default)
    data["params"] = create_model("parameters", **params)  # type: ignore

    # build returns
    anno = sig.return_annotation
    if anno is sig.empty:
        if (doc.returns is not None) and (doc.returns.type_name is not None):
            # use type annotation from docstring
            anno = doc.returns.type_name
        else:
            anno = Any
    default = ...  # required
    if doc.returns is not None:
        # fill in desc for the return
        default = Field(..., description=doc.returns.description)
    data["returns"] = create_model("returns", returns=(anno, default))

    # build raises
    data["raises"] = [
        {"type": exc.type_name, "desc": exc.description} for exc in doc.raises
    ]

    # build examples
    data["examples"] = doc.examples
    return data

to_str 

to_str() -> str

Represent the tool as a string.

Source code in src/appl/core/tool.py 
121
122
123
124
125
def to_str(self) -> str:
    """Represent the tool as a string."""
    s = f"def {self.name}{self.__signature__}:\n"
    s += f'    """{self.__doc__}"""'
    return s

BracketedDefinition 

BracketedDefinition(
    name: Optional[String] = None,
    desc: Optional[String] = None,
    *,
    sep: String = ": ",
    details: Any = None,
    fstr: Optional[str] = None,
    var_name: Optional[str] = None
)

Bases: Definition

A Definition that is formatted with square brackets.

Parameters:

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

    The name of the definition.

  • desc (Optional[String], default: None ) –

    A description of the definition.

  • sep (String, default: ': ' ) –

    The separator between the name and description.

  • details (Any, default: None ) –

    Additional details about the definition.

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

    The format string for the definition.

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

    The name of the variable that the definition is stored in.

Source code in src/appl/core/promptable/definition.py 
21
22
23
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 __init__(
    self,
    name: Optional[String] = None,
    desc: Optional[String] = None,
    *,
    sep: String = ": ",
    details: Any = None,
    fstr: Optional[str] = None,
    var_name: Optional[str] = None,
):
    """Initialize the Definition with the given name and description.

    Args:
        name: The name of the definition.
        desc: A description of the definition.
        sep: The separator between the name and description.
        details: Additional details about the definition.
        fstr: The format string for the definition.
        var_name: The name of the variable that the definition is stored in.
    """
    self.name = name or self.name or self.__doc__
    if self.name is None:
        raise ValueError("Name must be provided for Definition.")

    if desc is not None:
        self.desc = desc
    self.sep = sep
    self.details = details
    if fstr is not None:
        self.fstr = fstr
    self.var_name = var_name or self.name

    self._forks.append(self)

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

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.

CompletionResponse

Bases: BaseModel

A class wrapping the response from the LLM model.

For a streaming response, it tracks the chunks of the response and builds the complete response when the streaming is finished.

chunks class-attribute instance-attribute 

chunks: List[Union[ModelResponse, ChatCompletionChunk]] = (
    Field(
        [],
        description="The chunks of the response when streaming",
    )
)

The chunks of the response when streaming.

complete_response property 

complete_response: Union[ModelResponse, ChatCompletion]

The complete response from the model. This will block until the response is finished.

cost class-attribute instance-attribute 

cost: Optional[float] = Field(
    None, description="The cost of the completion"
)

The cost of the completion.

is_finished class-attribute instance-attribute 

is_finished: bool = Field(
    False,
    description="Whether the response stream is finished",
)

Whether the response stream is finished.

is_stream class-attribute instance-attribute 

is_stream: bool = Field(
    False, description="Whether the response is a stream"
)

Whether the response is a stream.

message class-attribute instance-attribute 

message: Optional[str] = Field(
    None,
    description="The top-choice message from the completion",
)

The top-choice message from the completion.

post_finish_callbacks class-attribute instance-attribute 

post_finish_callbacks: List[Callable] = Field(
    [], description="The post finish callbacks"
)

The post finish callbacks.

raw_response class-attribute instance-attribute 

raw_response: Any = Field(
    None, description="The raw response from the model"
)

The raw response from the model.

response_model class-attribute instance-attribute 

response_model: Any = Field(
    None,
    description="The BaseModel's subclass specifying the response format.",
)

The BaseModel's subclass specifying the response format.

response_obj class-attribute instance-attribute 

response_obj: Any = Field(
    None,
    description="The response object of response model, could be a stream",
)

The response object of response model, could be a stream.

results property 

results: Any

The results of the response.

Returns:

  • message ( str ) –

    The message if the response is a text completion.

  • tool_calls ( List[ToolCall] ) –

    The tool calls if the response is a list of tool calls.

  • response_obj ( Any ) –

    The object if the response is a response object.

tool_calls class-attribute instance-attribute 

tool_calls: List[ToolCall] = Field(
    [], description="The tool calls"
)

The tool calls.

type property 

type: ResponseType

The type of the response.

usage class-attribute instance-attribute 

usage: Optional[CompletionUsage] = Field(
    None, description="The usage of the completion"
)

The usage of the completion.

format_stream 

format_stream()

Format the stream response as a text generator.

Source code in src/appl/core/response.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def format_stream(self):
    """Format the stream response as a text generator."""
    suffix = ""
    for chunk in iter(self):
        # chunk: Union[ModelResponse, ChatCompletionChunk]
        delta: Union[Delta, ChoiceDelta] = chunk.choices[0].delta  # type: ignore

        if delta is not None:
            if delta.content is not None:
                yield delta.content
            elif getattr(delta, "tool_calls", None):
                f: Union[Function, ChoiceDeltaToolCallFunction] = delta.tool_calls[
                    0
                ].function  # type: ignore
                if f.name is not None:
                    if suffix:
                        yield f"{suffix}, "
                    yield f"{f.name}("
                    suffix = ")"
                if f.arguments is not None:
                    yield f.arguments
    yield suffix

register_post_finish_callback 

register_post_finish_callback(callback: Callable) -> None

Register a post finish callback.

The callback will be called after the response is finished.

Source code in src/appl/core/response.py 
179
180
181
182
183
184
185
186
187
def register_post_finish_callback(self, callback: Callable) -> None:
    """Register a post finish callback.

    The callback will be called after the response is finished.
    """
    if self.is_finished:
        callback(self)
    else:
        self.post_finish_callbacks.append(callback)

set_response_obj 

set_response_obj(response_obj: Any) -> None

Set the response object.

Source code in src/appl/core/response.py 
76
77
78
def set_response_obj(self, response_obj: Any) -> None:
    """Set the response object."""
    self.response_obj = response_obj

streaming 

streaming(
    display: bool = True, title: str = "APPL Streaming"
) -> CompletionResponse

Stream the response object and finish the response.

Source code in src/appl/core/response.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def streaming(
    self, display: bool = True, title: str = "APPL Streaming"
) -> "CompletionResponse":
    """Stream the response object and finish the response."""
    if not self.is_stream:
        raise ValueError("Cannot iterate over non-streaming response")
    if self.is_finished:
        return self

    if self.response_obj is not None:
        target = self.response_obj
    else:
        target = self.format_stream()
    if display:
        refresh_interval = configs.getattrs(
            "settings.logging.display.stream_interval", 1.0
        )
        start_time = time.time()

        def panel(
            content: str, iter_index: Optional[int] = None, truncate: bool = False
        ) -> Panel:
            style = "magenta"
            display_title = title
            if iter_index is not None:
                time_elapsed = time.time() - start_time
                avg_iters_per_sec = (iter_index + 1) / time_elapsed
                stream_info = (
                    f"[{time_elapsed:.3f}s ({avg_iters_per_sec:.2f} it/s)]"
                )
                display_title += f" - {stream_info}"
            return make_panel(
                content, title=display_title, style=style, truncate=truncate
            )

        with Live(
            panel("Waiting for Response ..."),
            refresh_per_second=refresh_interval,
            # vertical_overflow="visible", # manually display the tail lines instead
        ) as live:
            content = ""
            for i, chunk in enumerate(iter(target)):
                if isinstance(chunk, BaseModel):
                    content = json.dumps(chunk.model_dump(), indent=2)
                else:
                    content += str(chunk)
                live.update(panel(content, i, truncate=True))
                # live.refresh() # might be too frequent
            # display untruncated content at the end
            live.update(panel(content, i))
            live.refresh()
    else:
        for chunk in iter(target):
            pass
    if self.response_obj is not None:
        self.set_response_obj(chunk)
    return self

Compositor 

Compositor(
    sep: Optional[str] = None,
    indexing: Union[Indexing, Optional[str]] = None,
    indent: Optional[Union[str, int]] = None,
    new_indent: Optional[Union[str, int]] = None,
    is_inline: Optional[bool] = None,
    role: Optional[MessageRole] = None,
    _ctx: Optional[PromptContext] = None,
)

Bases: PrinterModifier

The contextual compositor of the prompts.

This class represents a contextual compositor that modifies the behavior of the printer. It provides various options for customizing the output format of the prompts within this context manager.

Attributes:

  • _sep (Optional[str]) –

    The class default separator string is None, indicating inherit from the parent.

  • _indexing (Indexing) –

    The class default indexing mode is empty indexing.

  • _inc_indent (str) –

    The class default indentation string is empty string.

  • _new_indent (Optional[str]) –

    The class default new indentation string is None, indicating not overwrite the indent.

  • _is_inline (bool) –

    The class default inline flag is False.

  • _new_role (Optional[MessageRole]) –

    The class default role of the modifier is None, indicating not overwrite the role.

Parameters:

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

    The separator string. Defaults to use the class default.

  • indexing (Union[Indexing, Optional[str]], default: None ) –

    The indexing mode. Defaults to use the class default.

  • indent (Optional[Union[str, int]], default: None ) –

    The indentation string. Defaults to use the class default.

  • new_indent (Optional[Union[str, int]], default: None ) –

    The new indentation string. Defaults to use the class default.

  • is_inline (Optional[bool], default: None ) –

    Flag indicating if the modifier is inline. Defaults to use the class default.

  • role (Optional[MessageRole], default: None ) –

    The role of the modifier. Defaults to use the class default.

  • _ctx (Optional[PromptContext], default: None ) –

    The prompt context filled automatically by the APPL function.

Source code in src/appl/core/modifiers.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __init__(
    self,
    sep: Optional[str] = None,
    indexing: Union[Indexing, Optional[str]] = None,
    indent: Optional[Union[str, int]] = None,
    new_indent: Optional[Union[str, int]] = None,
    is_inline: Optional[bool] = None,
    role: Optional[MessageRole] = None,
    _ctx: Optional[PromptContext] = None,
):
    """Initialize the Compositor object.

    Args:
        sep:
            The separator string. Defaults to use the class default.
        indexing:
            The indexing mode. Defaults to use the class default.
        indent:
            The indentation string. Defaults to use the class default.
        new_indent:
            The new indentation string. Defaults to use the class default.
        is_inline:
            Flag indicating if the modifier is inline. Defaults to use the class default.
        role:
            The role of the modifier. Defaults to use the class default.
        _ctx: The prompt context filled automatically by the APPL function.
    """
    super().__init__(_ctx)
    if sep is not None:
        self._sep = sep
    if indexing is not None:
        if isinstance(indexing, str):
            indexing = Indexing(indexing)
        self._indexing = indexing
    else:
        if self._indexing is None:
            raise ValueError("Indexing must be provided.")
        self._indexing = copy.copy(self._indexing)
        # copy to avoid changing the class default
    if indent is not None:
        if isinstance(indent, int):
            indent = " " * indent
        self._inc_indent = indent
    if new_indent is not None:
        if isinstance(new_indent, int):
            new_indent = " " * new_indent
        self._new_indent = new_indent
    if is_inline is not None:
        self._is_inline = is_inline
    if role is not None:
        self._new_role = role

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

Definition 

Definition(
    name: Optional[String] = None,
    desc: Optional[String] = None,
    *,
    sep: String = ": ",
    details: Any = None,
    fstr: Optional[str] = None,
    var_name: Optional[str] = None
)

Bases: Promptable, Formattable

Represent a definition of a concept.

Attributes:

  • fstr (str) –

    The format string for the definition.

  • name (Optional[String]) –

    The name of the definition.

  • desc (String) –

    A description of the definition.

  • _forks (List[Definition]) –

    A list of all instances of this class.

Parameters:

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

    The name of the definition.

  • desc (Optional[String], default: None ) –

    A description of the definition.

  • sep (String, default: ': ' ) –

    The separator between the name and description.

  • details (Any, default: None ) –

    Additional details about the definition.

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

    The format string for the definition.

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

    The name of the variable that the definition is stored in.

Source code in src/appl/core/promptable/definition.py 
21
22
23
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 __init__(
    self,
    name: Optional[String] = None,
    desc: Optional[String] = None,
    *,
    sep: String = ": ",
    details: Any = None,
    fstr: Optional[str] = None,
    var_name: Optional[str] = None,
):
    """Initialize the Definition with the given name and description.

    Args:
        name: The name of the definition.
        desc: A description of the definition.
        sep: The separator between the name and description.
        details: Additional details about the definition.
        fstr: The format string for the definition.
        var_name: The name of the variable that the definition is stored in.
    """
    self.name = name or self.name or self.__doc__
    if self.name is None:
        raise ValueError("Name must be provided for Definition.")

    if desc is not None:
        self.desc = desc
    self.sep = sep
    self.details = details
    if fstr is not None:
        self.fstr = fstr
    self.var_name = var_name or self.name

    self._forks.append(self)

FormatterMeta

Bases: ABCMeta

Metaclass for classes that can be formatted.

FutureValue

Bases: ABC

Represents a value that may not be ready yet.

val property 

val

The value of the future.

GenArgs

Bases: BaseModel

Common arguments for generating a response from a model.

preprocess 

preprocess(
    convert_func: Callable, is_openai: bool = False
) -> dict

Convert the GenArgs into a dictionary for creating the response.

Source code in src/appl/core/server.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
def preprocess(self, convert_func: Callable, is_openai: bool = False) -> dict:
    """Convert the GenArgs into a dictionary for creating the response."""
    # build dict, filter out the None values
    args = self.model_dump(exclude_none=True)

    # messages
    args["messages"] = convert_func(self.messages)

    # format the tool
    tools = self.tools
    tool_format = args.pop("tool_format")
    if len(tools):
        if tool_format == "auto":
            tool_format = "openai" if is_openai else "str"
        formatted_tools = []
        for tool in tools:
            tool_str: Any = None
            if tool_format == "openai":
                tool_str = tool.openai_schema
            else:  # TODO: supports more formats
                tool_str = str(tool)
            formatted_tools.append(tool_str)
        args["tools"] = formatted_tools
    else:
        args.pop("tools", None)
    return args

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)

Indexing 

Indexing(
    method: Optional[str] = None,
    ind: int = 0,
    prefix: str = "",
    suffix: Optional[str] = None,
)

The indexing method for the printer.

Source code in src/appl/core/printer.py 
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(
    self,
    method: Optional[str] = None,
    ind: int = 0,
    prefix: str = "",
    suffix: Optional[str] = None,
):
    """Initialize the indexing method."""
    self._method = method
    self._ind = ind
    self._prefix = prefix
    self._suffix = suffix

get_index 

get_index(ind: Optional[int] = None) -> str

Get the index string for the current or given index.

Source code in src/appl/core/printer.py 
57
58
59
60
61
62
63
64
def get_index(self, ind: Optional[int] = None) -> str:
    """Get the index string for the current or given index."""
    if ind is None:
        ind = self._ind
        self._ind += 1
    if ind < 0:
        raise ValueError("Indexing method does not support negative indexing.")
    return self._get_index(ind)

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

PromptFunc 

PromptFunc(
    func: Callable,
    ctx_method: str = "new",
    comp: Optional[Compositor] = None,
    default_return: Optional[Literal["prompt"]] = None,
    include_docstring: bool = False,
    new_ctx_func: Callable = PromptContext,
)

A wrapper for an APPL function, can be called as a normal function.

The function contains a prompt context, which could be same as or copied from its caller function, or created from scratch, or resumed from the last run.

Parameters:

  • func (Callable) –

    the function being wrapped

  • ctx_method (str, default: 'new' ) –

    the method to deal with the child context, available methods includes:

    • (default) "new" or "new_ctx": create a brand new context.
    • "copy" or "copy_ctx": copy from the parent's context, the change will not affect the parent's context.
    • "same" or "same_ctx": use the same context as the parent's, the change will affect the parent's context.
    • "resume" or "resume_ctx": resume its own context from the last run. For the first run, it will copy the parent's context.
  • comp (Compositor, default: None ) –

    the default compositor to be used. Defaults to None.

  • default_return (str, default: None ) –

    The default return value, "prompt" means return the prompt within the function. Defaults to None.

  • include_docstring (bool, default: False ) –

    set to True to include the triple-quoted docstring in the prompt. Defaults to False.

  • new_ctx_func (Callable, default: PromptContext ) –

    the function to create a new context. Defaults to PromptContext.

Source code in src/appl/core/function.py 
23
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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    func: Callable,
    ctx_method: str = "new",
    comp: Optional[Compositor] = None,
    default_return: Optional[Literal["prompt"]] = None,
    include_docstring: bool = False,
    new_ctx_func: Callable = PromptContext,
    # default_sep: Optional[str] = None,
    # ? set the default printer behavior for the prompt function?
):
    """Initialize the PromptFunc.

    Args:
        func (Callable): the function being wrapped
        ctx_method (str):
            the method to deal with the child context, available methods includes:

            - (default) "new" or "new_ctx": create a brand new context.
            - "copy" or "copy_ctx":
                copy from the parent's context, the change will not
                affect the parent's context.
            - "same" or "same_ctx":
                use the same context as the parent's, the change will
                affect the parent's context.
            - "resume" or "resume_ctx":
                resume its own context from the last run.
                For the first run, it will copy the parent's context.

        comp (Compositor, optional):
            the default compositor to be used. Defaults to None.
        default_return (str, optional):
            The default return value, "prompt" means return the prompt within
            the function. Defaults to None.
        include_docstring (bool, optional):
            set to True to include the triple-quoted docstring in the prompt.
            Defaults to False.
        new_ctx_func (Callable, optional):
            the function to create a new context. Defaults to PromptContext.
    """
    self._func = appl_compile(func)
    self._signature = inspect.signature(func)
    self._doc = func.__doc__
    self._name = func.__name__
    self._qualname = func.__qualname__
    self._default_ctx_method = self._process_ctx_method(ctx_method)
    self._default_compositor = comp
    if default_return is not None and default_return != "prompt":
        raise NotImplementedError("Only support default_return='prompt' now.")
    self._default_return = default_return
    self._include_docstring = include_docstring
    self._new_ctx_func = new_ctx_func
    self._persist_ctx: Optional[PromptContext] = None
    self._reset_context_func: Optional[Callable[[], None]] = None

compiled_func property 

compiled_func

The compiled function.

PromptPrinter 

PromptPrinter(
    states: Optional[List[PrinterState]] = None,
    is_newline: bool = True,
)

A class to print prompt records as conversation.

The printer maintains a stack of printer states about the current role, separator, indexing, and indentation.

Source code in src/appl/core/printer.py 
171
172
173
174
175
176
177
178
def __init__(
    self, states: Optional[List[PrinterState]] = None, is_newline: bool = True
) -> None:
    """Initialize the prompt printer."""
    if states is None:
        states = [PrinterState()]
    self._states = states
    self._is_newline = is_newline

states property 

states: List[PrinterState]

The stack of printer states.

pop 

pop() -> None

Pop the last printer state from the stack.

Source code in src/appl/core/printer.py 
189
190
191
192
193
def pop(self) -> None:
    """Pop the last printer state from the stack."""
    if len(self._states) == 1:
        raise ValueError("Cannot pop the first state.")
    self._states.pop()

push 

push(data: PrinterPush) -> None

Push a new printer state to the stack.

Source code in src/appl/core/printer.py 
185
186
187
def push(self, data: PrinterPush) -> None:
    """Push a new printer state to the stack."""
    self._push(**data.__dict__)

PromptRecords 

PromptRecords()

A class represents a list of prompt records.

Source code in src/appl/core/printer.py 
124
125
126
def __init__(self) -> None:
    """Initialize the prompt records."""
    self._records: List[RecordType] = []

records property 

records: List[RecordType]

The list of records.

as_convo 

as_convo() -> Conversation

Convert the prompt records to a conversation.

Source code in src/appl/core/printer.py 
133
134
135
def as_convo(self) -> Conversation:
    """Convert the prompt records to a conversation."""
    return PromptPrinter()(self)

copy 

copy() -> 'PromptRecords'

Copy the prompt records.

Source code in src/appl/core/printer.py 
156
157
158
def copy(self) -> "PromptRecords":
    """Copy the prompt records."""
    return copy.deepcopy(self)

extend 

extend(record: 'PromptRecords') -> None

Extend the prompt records with another prompt records.

Source code in src/appl/core/printer.py 
152
153
154
def extend(self, record: "PromptRecords") -> None:
    """Extend the prompt records with another prompt records."""
    self._records.extend(record._records)

record 

record(record: Union[str, RecordType]) -> None

Record a string, message, image, printer push or printer pop.

Source code in src/appl/core/printer.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def record(self, record: Union[str, RecordType]) -> None:
    """Record a string, message, image, printer push or printer pop."""
    if isinstance(record, str):  # compatible to str
        record = StringFuture(record)
    if (
        isinstance(record, StringFuture)
        or isinstance(record, Image)
        or isinstance(record, BaseMessage)
        or isinstance(record, PrinterPush)
        or isinstance(record, PrinterPop)
    ):
        self._records.append(record)
    else:
        raise ValueError("Can only record Message, PrinterPush or PrinterPop")

Promptable

Bases: ABC

Interface for objects that can be converted to a prompt string.

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.

RoleChanger 

RoleChanger(
    role: Optional[MessageRole] = None,
    _ctx: Optional[PromptContext] = None,
)

Bases: PrinterModifier

The contextual role changer of the prompts.

Parameters:

  • role (Optional[MessageRole], default: None ) –

    The new role of the prompts. Defaults to None.

  • _ctx (Optional[PromptContext], default: None ) –

    The prompt context filled automatically by the APPL function.

Source code in src/appl/role_changer.py 
12
13
14
15
16
17
18
19
20
21
22
23
def __init__(
    self, role: Optional[MessageRole] = None, _ctx: Optional[PromptContext] = None
):
    """Initialize the RoleChanger object.

    Args:
        role: The new role of the prompts. Defaults to None.
        _ctx: The prompt context filled automatically by the APPL function.
    """
    super().__init__(_ctx)
    if role is not None:
        self._new_role = role

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

SystemRole 

SystemRole(name: Optional[str] = None, **kwargs: Any)

Bases: RoleChanger

Change the role of the prompts to system.

Parameters:

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

    The name of the system role. Defaults to None.

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

    The keyword arguments to pass to the RoleChanger constructor.

Source code in src/appl/role_changer.py 
34
35
36
37
38
39
40
41
42
def __init__(self, name: Optional[str] = None, **kwargs: Any):
    """Initialize the SystemRole object.

    Args:
        name: The name of the system role. Defaults to None.
        **kwargs: The keyword arguments to pass to the RoleChanger constructor.
    """
    role = MessageRole(SYSTEM, name=name)
    super().__init__(role=role, **kwargs)

Tool 

Tool(
    func: Callable,
    use_short_desc: bool = False,
    **kwargs: Any
)

Bases: BaseTool

The Tool class that can be called by LLMs.

Parameters:

  • func (Callable) –

    The function to create the tool from.

  • use_short_desc (bool, default: False ) –

    Whether to use the short description instead of the full description.

  • kwargs (Any, default: {} ) –

    Additional arguments for the tool.

Source code in src/appl/core/tool.py 
152
153
154
155
156
157
158
159
160
161
162
def __init__(self, func: Callable, use_short_desc: bool = False, **kwargs: Any):
    """Create a tool from a function.

    Args:
        func: The function to create the tool from.
        use_short_desc:
            Whether to use the short description instead of the full description.
        kwargs: Additional arguments for the tool.
    """
    super().__init__(func=func, **kwargs)
    self._use_short_desc = use_short_desc

examples class-attribute instance-attribute 

examples: List[str] = Field(
    [], description="The examples of the Tool"
)

The examples of the Tool.

info class-attribute instance-attribute 

info: Dict = Field(
    {}, description="Additional information of the Tool"
)

Additional information of the Tool.

long_desc class-attribute instance-attribute 

long_desc: str = Field(
    "", description="The long description of the Tool"
)

The long description of the Tool.

name class-attribute instance-attribute 

name: str = Field(..., description='The name of the Tool')

The name of the Tool.

openai_schema property 

openai_schema: dict

Get the OpenAI schema of the tool.

params class-attribute instance-attribute 

params: type[BaseModel] = Field(
    ..., description="The parameters of the Tool"
)

The parameters of the Tool.

raises class-attribute instance-attribute 

raises: List[Dict[str, Optional[str]]] = Field(
    [], description="The exceptions raised by the Tool"
)

The exceptions raised by the Tool.

returns class-attribute instance-attribute 

returns: type[BaseModel] = Field(
    ..., description="The return of the Tool"
)

The return of the Tool.

short_desc class-attribute instance-attribute 

short_desc: str = Field(
    "", description="The short description of the Tool"
)

The short description of the Tool.

parse_data classmethod 

parse_data(
    sig: Signature,
    docstring: Optional[str],
    predefined: Dict[str, Any],
) -> Dict[str, Any]

Parse data from the signature and docstring of a function.

Source code in src/appl/core/tool.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def parse_data(
    cls, sig: Signature, docstring: Optional[str], predefined: Dict[str, Any]
) -> Dict[str, Any]:
    """Parse data from the signature and docstring of a function."""
    doc = parse(docstring or "")
    data: Dict[str, Any] = {
        "short_desc": doc.short_description or "",
        "long_desc": doc.long_description or "",
    }

    # build params
    params = {}
    doc_param = {p.arg_name: p for p in doc.params}
    for name, param in sig.parameters.items():
        anno = param.annotation
        default = param.default

        if default is param.empty:
            default = ...  # required
        if name in doc_param:
            # fill in desc for the param
            default = Field(default, description=doc_param[name].description)
            # fill in type annotation if not annotated in the function
            if (anno is param.empty) and (doc_param[name].type_name is not None):
                # use type annotation from docstring
                anno = doc_param[name].type_name
        # replace empty annotation with Any
        if anno is param.empty:
            anno = Any
        if name not in predefined:
            params[name] = (anno, default)
    data["params"] = create_model("parameters", **params)  # type: ignore

    # build returns
    anno = sig.return_annotation
    if anno is sig.empty:
        if (doc.returns is not None) and (doc.returns.type_name is not None):
            # use type annotation from docstring
            anno = doc.returns.type_name
        else:
            anno = Any
    default = ...  # required
    if doc.returns is not None:
        # fill in desc for the return
        default = Field(..., description=doc.returns.description)
    data["returns"] = create_model("returns", returns=(anno, default))

    # build raises
    data["raises"] = [
        {"type": exc.type_name, "desc": exc.description} for exc in doc.raises
    ]

    # build examples
    data["examples"] = doc.examples
    return data

to_str 

to_str() -> str

Represent the tool as a string.

Source code in src/appl/core/tool.py 
121
122
123
124
125
def to_str(self) -> str:
    """Represent the tool as a string."""
    s = f"def {self.name}{self.__signature__}:\n"
    s += f'    """{self.__doc__}"""'
    return s

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

ToolRole 

ToolRole(name: Optional[str] = None, **kwargs: Any)

Bases: RoleChanger

Change the role of the prompts to tool.

Parameters:

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

    The name of the tool role. Defaults to None.

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

    The keyword arguments to pass to the RoleChanger constructor.

Source code in src/appl/role_changer.py 
76
77
78
79
80
81
82
83
84
def __init__(self, name: Optional[str] = None, **kwargs: Any):
    """Initialize the ToolRole object.

    Args:
        name: The name of the tool role. Defaults to None.
        **kwargs: The keyword arguments to pass to the RoleChanger constructor.
    """
    role = MessageRole(TOOL, name=name)
    super().__init__(role=role, **kwargs)

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

UserRole 

UserRole(name: Optional[str] = None, **kwargs: Any)

Bases: RoleChanger

Change the role of the prompts to user.

Parameters:

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

    The name of the user role. Defaults to None.

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

    The keyword arguments to pass to the RoleChanger constructor.

Source code in src/appl/role_changer.py 
48
49
50
51
52
53
54
55
56
def __init__(self, name: Optional[str] = None, **kwargs: Any):
    """Initialize the UserRole object.

    Args:
        name: The name of the user role. Defaults to None.
        **kwargs: The keyword arguments to pass to the RoleChanger constructor.
    """
    role = MessageRole(USER, name=name)
    super().__init__(role=role, **kwargs)

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)