Skip to content

trace

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.

CompletionRequestEvent

Bases: TraceEventBase

A class representing a completion request event.

name instance-attribute 

name: str

The name of the event.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

CompletionResponseEvent

Bases: TraceEventBase

A class representing a completion response event.

args instance-attribute 

args: Dict

The arguments of the completion call.

cost instance-attribute 

cost: Optional[float]

The api cost of the completion call.

name instance-attribute 

name: str

The name of the event.

ret instance-attribute 

ret: Any

The return value of the completion call.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

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]

FunctionCallEvent

Bases: TraceEventBase

A class representing a function call event.

args instance-attribute 

args: Dict

The arguments of the function call.

name instance-attribute 

name: str

The name of the event.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

FunctionReturnEvent

Bases: TraceEventBase

A class representing a function return event.

name instance-attribute 

name: str

The name of the event.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

FutureValue

Bases: ABC

Represents a value that may not be ready yet.

val property 

val

The value of the future.

GenerationInitEvent

Bases: TraceEventBase

A class representing a generation init event.

name instance-attribute 

name: str

The name of the event.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

GenerationResponseEvent

Bases: TraceEventBase

A class representing a generation response event.

args instance-attribute 

args: Dict

The arguments of the generation call.

name instance-attribute 

name: str

The name of the event.

ret instance-attribute 

ret: Any

The return value of the generation call.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

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)

TraceEngineBase

Bases: ABC

A base class for trace engines.

events property 

events: List[TraceEventBase]

The list of events in the trace.

min_timestamp cached property 

min_timestamp: float

The minimum time stamp of the events in the trace.

trace_nodes property 

trace_nodes: Dict[str, TraceNode]

The dictionary of trace nodes in the trace.

append 

append(event: TraceEventBase) -> None

Append an event to the trace.

Source code in src/appl/core/trace.py 
126
127
128
129
@abstractmethod
def append(self, event: TraceEventBase) -> None:
    """Append an event to the trace."""
    raise NotImplementedError

find_cache 

find_cache(name: str, args: Dict) -> Any

Find a completion result in the cache.

Parameters:

  • name (str) –

    The name of the completion.

  • args (Dict) –

    The arguments of the completion.

Returns:

  • Any –

    The completion result if found, otherwise None.

Source code in src/appl/core/trace.py 
131
132
133
134
135
136
137
138
139
140
141
142
@abstractmethod
def find_cache(self, name: str, args: Dict) -> Any:
    """Find a completion result in the cache.

    Args:
        name: The name of the completion.
        args: The arguments of the completion.

    Returns:
        The completion result if found, otherwise None.
    """
    raise NotImplementedError

TraceEventBase

Bases: BaseModel

A base class for trace events.

name instance-attribute 

name: str

The name of the event.

time_stamp class-attribute instance-attribute 

time_stamp: float = None

The time stamp of the event.

TraceNode

Bases: BaseModel

The node of a trace tree containing information about trace events.

args class-attribute instance-attribute 

args: Optional[Dict] = None

The arguments of the trace node.

children class-attribute instance-attribute 

children: List[TraceNode] = []

The children of the trace node.

end_time class-attribute instance-attribute 

end_time: float = 0.0

The end time of the trace node.

info class-attribute instance-attribute 

info: Dict = {}

The extra information of the trace node.

name instance-attribute 

name: str

The name of the trace node.

parent class-attribute instance-attribute 

parent: Optional[TraceNode] = None

The parent of the trace node.

ret class-attribute instance-attribute 

ret: Any = None

The return value of the trace node.

runtime property 

runtime: float

The runtime of the trace node.

start_time class-attribute instance-attribute 

start_time: float = 0.0

The start time of the trace node.

type instance-attribute 

type: str

The type of the trace node.

TracePrinterBase

Bases: ABC

A base class for trace printers.

print 

print(
    trace: TraceEngineBase,
    meta_data: Optional[Configs] = None,
) -> Any

Print the trace.

Source code in src/appl/core/trace.py 
169
170
171
172
@abstractmethod
def print(self, trace: TraceEngineBase, meta_data: Optional[Configs] = None) -> Any:
    """Print the trace."""
    raise NotImplementedError

add_to_trace 

add_to_trace(event: TraceEventBase) -> None

Add an event to the trace.

Source code in src/appl/core/trace.py 
100
101
102
103
def add_to_trace(event: TraceEventBase) -> None:
    """Add an event to the trace."""
    if global_vars.trace_engine:
        global_vars.trace_engine.append(event)

find_in_cache 

find_in_cache(
    name: str,
    args: Dict,
    cache: Optional[TraceEngineBase] = None,
) -> Any

Find a completion result in the cache.

Parameters:

  • name (str) –

    The name of the completion.

  • args (Dict) –

    The arguments of the completion.

  • cache (Optional[TraceEngineBase], default: None ) –

    The cache to search in. Defaults to the global resume cache.

Returns:

  • Any –

    The completion result if found, otherwise None.

Source code in src/appl/core/trace.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def find_in_cache(
    name: str, args: Dict, cache: Optional[TraceEngineBase] = None
) -> Any:
    """Find a completion result in the cache.

    Args:
        name: The name of the completion.
        args: The arguments of the completion.
        cache: The cache to search in. Defaults to the global resume cache.

    Returns:
        The completion result if found, otherwise None.
    """
    if cache is None:
        if "resume_cache" in global_vars:
            cache = global_vars.resume_cache
    if cache is not None:
        return cache.find_cache(name, args)
    return None

is_string 

is_string(s: Any) -> bool

Check if the object is a StringFuture or str.

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