Skip to content

response

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.

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

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]

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)

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)