types

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.

String module-attribute

String: TypeAlias = Union[StringFuture, str]

String is a type alias for StringFuture or str.

TOOL_ROLE module-attribute

TOOL_ROLE = MessageRole(TOOL)

The tool role with name not specified.

USER_ROLE module-attribute

USER_ROLE = MessageRole(USER)

The user role with name not specified.

Audio

Audio(data: str, format: Literal['mp3', 'wav'])

Bases: ContentPart

Represent an audio in the message.

Source code in src/appl/core/types/content.py

def __init__(self, data: str, format: Literal["mp3", "wav"]) -> None:
    """Initialize the audio with the base64 encoded data and format."""
    super().__init__(type=ContentPartType.AUDIO, data=data, format=format)

from_file classmethod

from_file(file: PathLike) -> Audio

Construct an audio prompt from an audio file.

Source code in src/appl/core/types/content.py

@classmethod
def from_file(cls, file: PathLike) -> "Audio":
    """Construct an audio prompt from an audio file."""
    ext = Path(file).suffix.lower()[1:]
    if ext not in ("mp3", "wav"):
        raise ValueError(f"Unsupported audio format: {ext}")
    with open(file, "rb") as f:
        audio_data = f.read()
        base64_encoded = base64.b64encode(audio_data).decode("utf-8")
    return cls(data=base64_encoded, format=ext)  # type: ignore

from_url classmethod

from_url(url: str, format: Literal['mp3', 'wav']) -> Audio

Construct an audio prompt from an audio URL.

Source code in src/appl/core/types/content.py

@classmethod
def from_url(cls, url: str, format: Literal["mp3", "wav"]) -> "Audio":
    """Construct an audio prompt from an audio URL."""
    audio_data = urlopen(url).read()
    base64_encoded = base64.b64encode(audio_data).decode("utf-8")
    return cls(data=base64_encoded, format=format)

get_dict

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py

def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

CallFuture

CallFuture(
    func: Callable[..., R],
    *args: Any,
    executor_type: ExecutorType = ExecutorType.GENERAL_THREAD_POOL,
    lazy_eval: bool = False,
    **kwargs: Any
)

Bases: FutureValue, Generic[R]

Represent a function call that may not be ready yet.

Parameters:

• func (Callable[..., R]) –

  The function to call.

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

  The arguments of the function.

• executor_type (ExecutorType, default: GENERAL_THREAD_POOL ) –

  The type of the executor to run the call.

• 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

def __init__(
    self,
    func: Callable[..., R],
    *args: Any,
    executor_type: ExecutorType = ExecutorType.GENERAL_THREAD_POOL,
    lazy_eval: bool = False,
    **kwargs: Any,
):
    """Initialize the CallFuture.

    Args:
        func: The function to call.
        *args: The arguments of the function.
        executor_type: The type of the executor to run the call.
        lazy_eval: Whether to delay the start of the call until needed.
        **kwargs: The keyword arguments of the function.
    """
    self._executor_type = executor_type
    self._executor = global_executors.get_executor(executor_type)
    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: 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

def cancel(self) -> bool:
    """Cancel the call."""
    # Attempt to cancel the call
    res = self.future.cancel()
    if res and self._executor_type in [
        ExecutorType.NEW_THREAD,
        ExecutorType.NEW_PROCESS,
    ]:
        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

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) -> R

Get the result of the call.

Source code in src/appl/core/types/futures.py

def result(self, timeout: Optional[float] = None) -> R:
    """Get the result of the call."""
    # This will block until the result is available
    res = self.future.result(timeout)
    if self._executor_type in [ExecutorType.NEW_THREAD, ExecutorType.NEW_PROCESS]:
        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

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

name instance-attribute

name: str

The name of the event.

parent_func class-attribute instance-attribute

parent_func: Optional[str] = None

The name of the parent function.

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

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

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

Bases: BaseModel

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

Source code in src/appl/core/types/content.py

def __init__(
    self, contents: Union[Iterable[ContentPart], Iterator[Dict[str, Any]]]
) -> None:
    """Initialize the content list with a list of contents."""
    res: List[ContentPart] = []
    for c in contents:
        if isinstance(c, ContentPart):
            res.append(c)
        elif isinstance(c, dict):
            if "type" not in c:
                raise ValueError(f"Invalid content: {c}")
            t = ContentPartType(c["type"])
            cls = CONTENT_PART_CLASS_MAP[t]
            kwargs = c[t.value]
            res.append(cls(**kwargs) if isinstance(kwargs, dict) else cls(kwargs))
        else:
            raise ValueError(f"Invalid content: {c}")
    super().__init__(contents=res)

append

append(content: ContentPart) -> None

Append a content to the list.

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

Source code in src/appl/core/types/content.py

def append(self, content: ContentPart) -> None:
    """Append a content to the list.

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

extend

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

Extend the list with multiple contents.

Source code in src/appl/core/types/content.py

def extend(self, contents: list[ContentPart]) -> None:
    """Extend the list with multiple contents."""
    for content in contents:
        self.append(content)

get_contents

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

Return the contents as a list of dictionaries.

Source code in src/appl/core/types/content.py

def get_contents(self) -> List[Dict[str, Any]]:
    """Return the contents as a list of dictionaries."""
    return [c.get_dict() for c in self.contents]

ContentPart

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

Bases: BaseModel, ABC

Represent a part of the message content.

Source code in src/appl/core/types/content.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the content part."""
    super().__init__(*args, **kwargs)

get_dict

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py

def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

ContentPartType

Bases: Enum

The types of the content part.

ExecutorNotFoundError

ExecutorNotFoundError(message='Executor not found')

Bases: Exception

Custom exception to indicate the executor is not found.

Source code in src/appl/core/types/executor.py

def __init__(self, message="Executor not found"):
    """Initialize the exception."""
    super().__init__(message)

ExecutorType

Bases: str, Enum

The type of the executor.

FunctionCallEvent

Bases: TraceEventBase

A class representing a function call event.

args instance-attribute

args: Dict

The arguments of the function call.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

name instance-attribute

name: str

The name of the event.

parent_func class-attribute instance-attribute

parent_func: Optional[str] = None

The name of the parent function.

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

name instance-attribute

name: str

The name of the event.

ret class-attribute instance-attribute

ret: Any = None

The return value of the function.

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

name instance-attribute

name: str

The name of the event.

parent_func class-attribute instance-attribute

parent_func: Optional[str] = None

The name of the parent function.

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the event.

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.

GitInfo dataclass

GitInfo(
    user: Optional[str] = None,
    email: Optional[str] = None,
    branch: Optional[str] = None,
    commit_hash: Optional[str] = None,
)

Git information.

GlobalExecutors

The global executors.

get_executor

get_executor(
    executor_type: ExecutorType,
) -> Union[ThreadPoolExecutor, ProcessPoolExecutor]

Get the executor of a given type.

Source code in src/appl/core/types/executor.py

def get_executor(
    self, executor_type: ExecutorType
) -> Union[ThreadPoolExecutor, ProcessPoolExecutor]:
    """Get the executor of a given type."""
    if executor_type == ExecutorType.LLM_THREAD_POOL:
        if self._llm_thread_executor is None:
            raise ExecutorNotFoundError("LLM thread executor is not set")
        return self._llm_thread_executor
    elif executor_type == ExecutorType.GENERAL_THREAD_POOL:
        if self._general_thread_executor is None:
            raise ExecutorNotFoundError("General thread executor is not set")
        return self._general_thread_executor
    elif executor_type == ExecutorType.GENERAL_PROCESS_POOL:
        if self._general_process_executor is None:
            raise ExecutorNotFoundError("General process executor is not set")
        return self._general_process_executor
    else:
        return self._create_executor(executor_type)

set_executor

set_executor(
    executor_type: ExecutorType,
    executor: Optional[
        Union[ThreadPoolExecutor, ProcessPoolExecutor]
    ],
) -> None

Set the global executors.

Source code in src/appl/core/types/executor.py

def set_executor(
    self,
    executor_type: ExecutorType,
    executor: Optional[Union[ThreadPoolExecutor, ProcessPoolExecutor]],
) -> None:
    """Set the global executors."""
    if executor_type == ExecutorType.LLM_THREAD_POOL:
        if not isinstance(executor, ThreadPoolExecutor):
            raise ValueError("LLM thread executor must be a ThreadPoolExecutor")
        self._llm_thread_executor = executor
    elif executor_type == ExecutorType.GENERAL_THREAD_POOL:
        if not isinstance(executor, ThreadPoolExecutor):
            raise ValueError("General thread executor must be a ThreadPoolExecutor")
        self._general_thread_executor = executor
    elif executor_type == ExecutorType.GENERAL_PROCESS_POOL:
        if not isinstance(executor, ProcessPoolExecutor):
            raise ValueError(
                "General process executor must be a ProcessPoolExecutor"
            )
        self._general_process_executor = executor
    else:
        raise ValueError(f"Cannot set executor of type: {executor_type}")

Image

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

Bases: ContentPart

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

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__(type=ContentPartType.IMAGE, 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

@classmethod
def from_file(cls, file: PathLike, detail: Optional[str] = None) -> "Image":
    """Construct an image prompt from an image file."""
    image = PIL.Image.open(Path(file))
    return cls.from_image(image, detail)  # type: ignore

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

@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)

get_dict

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py

def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

MessageRole

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

Bases: BaseModel

The role of the message owner.

Parameters:

• type (Optional[Union[str, MessageRoleType]], 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

def __init__(
    self,
    type: Optional[Union[str, MessageRoleType]] = None,
    name: Optional[str] = None,
):
    """Initialize the MessageRole object.

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

is_assistant property

is_assistant: bool

Whether the role is an assistant role.

is_system property

is_system: bool

Whether the role is a system role.

is_tool property

is_tool: bool

Whether the role is a tool role.

is_user property

is_user: bool

Whether the role is a user role.

get_dict

get_dict() -> Dict[str, Any]

Get the role as a dictionary.

Source code in src/appl/core/types/role.py

def get_dict(self) -> Dict[str, Any]:
    """Get the role as a dictionary."""
    data = {"role": self.type}
    if self.name:
        data["name"] = self.name
    return data

MessageRoleType

Bases: str, Enum

The type of the role.

MetaData dataclass

MetaData(
    appl_version: str,
    cwd: str,
    run_cmd: str,
    git_info: GitInfo,
    exec_file_path: str,
    exec_file_basename: str,
    start_time: str,
    dotenvs: List[str],
    appl_config_files: List[str],
    log_file: Optional[str] = None,
    trace_file: Optional[str] = None,
)

Metadata for the run.

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

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

@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

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

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

def serialize(self) -> str:
    """Serialize the StringFuture."""
    return str(self)

TextContent

TextContent(text: String)

Bases: ContentPart

Represent a text in the message.

Source code in src/appl/core/types/content.py

def __init__(self, text: String) -> None:
    """Initialize the text content."""
    super().__init__(type=ContentPartType.TEXT, text=text)

get_dict

get_dict() -> Dict[str, Any]

Return a dict representation of the content part.

Source code in src/appl/core/types/content.py

def get_dict(self) -> Dict[str, Any]:
    """Return a dict representation of the content part."""
    return {"type": self.type.value, self.type.value: self._get_data()}

TraceEngineBase

Bases: ABC

A base class for trace engines.

events abstractmethod 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 abstractmethod property

trace_nodes: Dict[str, TraceNode]

The dictionary of trace nodes in the trace.

append abstractmethod

append(event: TraceEventBase) -> None

Append an event to the trace.

Source code in src/appl/core/types/trace.py

@abstractmethod
def append(self, event: TraceEventBase) -> None:
    """Append an event to the trace."""
    raise NotImplementedError

find_cache abstractmethod

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

Find a completion result in the cache.

Parameters:

• name (str) –

  The name of the completion.

• args (Dict[str, Any]) –

  The arguments of the completion.

Returns:

• Any –

  The completion result if found, otherwise None.

Source code in src/appl/core/types/trace.py

@abstractmethod
def find_cache(self, name: str, args: Dict[str, Any]) -> 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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data of the 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.

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.

metadata class-attribute instance-attribute

metadata: Optional[Dict] = None

The meta data 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 abstractmethod

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> Any

Print the trace.

Source code in src/appl/core/types/trace.py

@abstractmethod
def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> Any:
    """Print the trace."""
    raise NotImplementedError

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

def is_string(s: Any) -> bool:
    """Check if the object is a StringFuture or str."""
    return isinstance(s, StringFuture) or isinstance(s, str)

is_thread_executor

is_thread_executor(executor_type: ExecutorType) -> bool

Check if the executor is a thread executor.

Source code in src/appl/core/types/executor.py

def is_thread_executor(executor_type: ExecutorType) -> bool:
    """Check if the executor is a thread executor."""
    return "thread" in executor_type.value