Skip to content

Python Utilities

trulens_eval.utils.python

Utilities related to core python functionalities.

Attributes

Thunk module-attribute

Thunk = Callable[[], T]

A function that takes no arguments.

NoneType module-attribute

NoneType = NoneType

Alias for types.NoneType .

In python < 3.10, it is defined as type(None) instead.

Classes

Future

Bases: Generic[A], Future

Alias for concurrent.futures.Future.

In python < 3.9, a sublcass of concurrent.futures.Future with Generic[A] is used instead.

Queue

Bases: Generic[A], Queue

Alias for queue.Queue .

In python < 3.9, a sublcass of queue.Queue with Generic[A] is used instead.

EmptyType

Bases: type

A type that cannot be instantiated or subclassed.

OpaqueWrapper

Bases: Generic[T]

Wrap an object preventing all access.

Any access except to unwrap will result in an exception with the given message.

PARAMETER DESCRIPTION
obj

The object to wrap.

TYPE: T

e

The exception to raise when an attribute is accessed.

TYPE: Exception

Functions
unwrap
unwrap() -> T

Get the wrapped object back.

SingletonPerName

Bases: Generic[T]

Class for creating singleton instances except there being one instance max, there is one max per different name argument. If name is never given, reverts to normal singleton behaviour.

Functions
delete_singleton_by_name staticmethod
delete_singleton_by_name(name: str)

Delete the singleton instance with the given name. Can be used for testing to create another singleton.

delete_singleton
delete_singleton()

Delete the singleton instance. Can be used for testing to create another singleton.

Functions

class_name

class_name(obj: Union[Type, Any]) -> str

Get the class name of the given object or instance.

module_name

module_name(obj: Union[ModuleType, Type, Any]) -> str

Get the module name of the given module, class, or instance.

callable_name

callable_name(c: Callable)

Get the name of the given callable.

id_str

id_str(obj: Any) -> str

Get the id of the given object as a string in hex.

is_really_coroutinefunction

is_really_coroutinefunction(func) -> bool

Determine whether the given function is a coroutine function.

Warning

Inspect checkers for async functions do not work on openai clients, perhaps because they use @typing.overload. Because of that, we detect them by checking __wrapped__ attribute instead. Note that the inspect docs suggest they should be able to handle wrapped functions but perhaps they handle different type of wrapping? See https://docs.python.org/3/library/inspect.html#inspect.iscoroutinefunction . Another place they do not work is the decorator langchain uses to mark deprecated functions.

safe_signature

safe_signature(func_or_obj: Any)

Get the signature of the given function.

Sometimes signature fails for wrapped callables and in those cases we check for __call__ attribute and use that instead.

safe_hasattr

safe_hasattr(obj: Any, k: str) -> bool

Check if the given object has the given attribute.

Attempts to use static checks (see inspect.getattr_static) to avoid any side effects of attribute access (i.e. for properties).

safe_issubclass

safe_issubclass(cls: Type, parent: Type) -> bool

Check if the given class is a subclass of the given parent class.

code_line

code_line(func) -> Optional[str]

Get a string representation of the location of the given function func.

locals_except

locals_except(*exceptions)

Get caller's locals except for the named exceptions.

caller_frame

caller_frame(offset=0) -> 'frame'

Get the caller's (of this function) frame. See https://docs.python.org/3/reference/datamodel.html#frame-objects .

task_factory_with_stack

task_factory_with_stack(loop, coro, *args, **kwargs) -> Sequence['frame']

A task factory that annotates created tasks with stacks of their parents.

All of such annotated stacks can be retrieved with stack_with_tasks as one merged stack.

tru_new_event_loop

tru_new_event_loop()

Replacement for new_event_loop that sets the task factory to make tasks that copy the stack from their creators.

get_task_stack

get_task_stack(task: Task) -> Sequence['frame']

Get the annotated stack (if available) on the given task.

merge_stacks

merge_stacks(s1: Sequence['frame'], s2: Sequence['frame']) -> Sequence['frame']

Assuming s1 is a subset of s2, combine the two stacks in presumed call order.

stack_with_tasks

stack_with_tasks() -> Sequence['frame']

Get the current stack (not including this function) with frames reaching across Tasks.

get_all_local_in_call_stack

get_all_local_in_call_stack(key: str, func: Callable[[Callable], bool], offset: Optional[int] = 1, skip: Optional[Any] = None) -> Iterator[Any]

Find locals in call stack by name.

PARAMETER DESCRIPTION
key

The name of the local variable to look for.

TYPE: str

func

Recognizer of the function to find in the call stack.

TYPE: Callable[[Callable], bool]

offset

The number of top frames to skip.

TYPE: Optional[int] DEFAULT: 1

skip

A frame to skip as well.

TYPE: Optional[Any] DEFAULT: None

Note

offset is unreliable for skipping the intended frame when operating with async tasks. In those cases, the skip argument is more reliable.

RETURNS DESCRIPTION
Iterator[Any]

An iterator over the values of the local variable named key in the stack at all of the frames executing a function which func recognizes (returns True on) starting from the top of the stack except offset top frames.

Returns None if func does not recognize any function in the stack.

RAISES DESCRIPTION
RuntimeError

Raised if a function is recognized but does not have key in its locals.

This method works across threads as long as they are started using TP.

get_first_local_in_call_stack

get_first_local_in_call_stack(key: str, func: Callable[[Callable], bool], offset: Optional[int] = 1, skip: Optional[Any] = None) -> Optional[Any]

Get the value of the local variable named key in the stack at the nearest frame executing a function which func recognizes (returns True on) starting from the top of the stack except offset top frames. If skip frame is provided, it is skipped as well. Returns None if func does not recognize the correct function. Raises RuntimeError if a function is recognized but does not have key in its locals.

This method works across threads as long as they are started using the TP class above.

NOTE: offset is unreliable for skipping the intended frame when operating with async tasks. In those cases, the skip argument is more reliable.

wrap_awaitable

wrap_awaitable(awaitable: Awaitable[T], on_await: Optional[Callable[[], Any]] = None, on_done: Optional[Callable[[T], Any]] = None) -> Awaitable[T]

Wrap an awaitable in another awaitable that will call callbacks before and after the given awaitable finishes.

Note that the resulting awaitable needs to be awaited for the callback to eventually trigger.

PARAMETER DESCRIPTION
awaitable

The awaitable to wrap.

TYPE: Awaitable[T]

on_await

The callback to call when the wrapper awaitable is awaited but before the wrapped awaitable is awaited.

TYPE: Optional[Callable[[], Any]] DEFAULT: None

on_done

The callback to call with the result of the wrapped awaitable once it is ready.

TYPE: Optional[Callable[[T], Any]] DEFAULT: None

wrap_generator

wrap_generator(gen: Generator[T, None, None], on_iter: Optional[Callable[[], Any]] = None, on_next: Optional[Callable[[T], Any]] = None, on_done: Optional[Callable[[], Any]] = None) -> Generator[T, None, None]

Wrap a generator in another generator that will call callbacks at various points in the generation process.

PARAMETER DESCRIPTION
gen

The generator to wrap.

TYPE: Generator[T, None, None]

on_iter

The callback to call when the wrapper generator is created but before a first iteration is produced.

TYPE: Optional[Callable[[], Any]] DEFAULT: None

on_next

The callback to call with the result of each iteration of the wrapped generator.

TYPE: Optional[Callable[[T], Any]] DEFAULT: None

on_done

The callback to call when the wrapped generator is exhausted.

TYPE: Optional[Callable[[], Any]] DEFAULT: None

trulens_eval.utils.pyschema

Serialization of Python objects

In order to serialize (and optionally deserialize) python entities while still being able to inspect them in their serialized form, we employ several storage classes that mimic basic python entities:

Serializable representation Python entity
Class (python) class
Module (python) module
Obj (python) object
Function (python) function
Method (python) method

Classes

Class

Bases: SerialModel

A python class. Should be enough to deserialize the constructor. Also includes bases so that we can query subtyping relationships without deserializing the class first.

Functions
base_class
base_class() -> 'Class'

Get the deepest base class in the same module as this class.

Obj

Bases: SerialModel

An object that may or may not be loadable from its serialized form. Do not use for base types that don't have a class. Loadable if init_bindings is not None.

FunctionOrMethod

Bases: SerialModel

Functions
of_callable staticmethod
of_callable(c: Callable, loadable: bool = False) -> 'FunctionOrMethod'

Serialize the given callable. If loadable is set, tries to add enough info for the callable to be deserialized.

Method

Bases: FunctionOrMethod

A python method. A method belongs to some class in some module and must have a pre-bound self object. The location of the method is encoded in obj alongside self. If obj is Obj with init_bindings, this method should be deserializable.

Function

Bases: FunctionOrMethod

A python function. Could be a static method inside a class (not instance of the class).

WithClassInfo

Bases: BaseModel

Mixin to track class information to aid in querying serialized components without having to load them.

Functions
load staticmethod
load(obj, *args, **kwargs)

Deserialize/load this object using the class information in tru_class_info to lookup the actual class that will do the deserialization.

model_validate classmethod
model_validate(*args, **kwargs) -> Any

Deserialized a jsonized version of the app into the instance of the class it was serialized from.

Note

This process uses extra information stored in the jsonized object and handled by WithClassInfo.

Functions

is_noserio

is_noserio(obj)

Determines whether the given json object represents some non-serializable object. See noserio.

noserio

noserio(obj, **extra: Dict) -> dict

Create a json structure to represent a non-serializable object. Any additional keyword arguments are included.

safe_getattr

safe_getattr(obj: Any, k: str, get_prop: bool = True) -> Any

Try to get the attribute k of the given object. This may evaluate some code if the attribute is a property and may fail. In that case, an dict indicating so is returned.

If get_prop is False, will not return contents of properties (will raise ValueException).

clean_attributes

clean_attributes(obj, include_props: bool = False) -> Dict[str, Any]

Determine which attributes of the given object should be enumerated for storage and/or display in UI. Returns a dict of those attributes and their values.

For enumerating contents of objects that do not support utility classes like pydantic, we use this method to guess what should be enumerated when serializing/displaying.

If include_props is True, will produce attributes which are properties; otherwise those will be excluded.

trulens_eval.utils.threading

Threading Utilities

Classes

Thread

Bases: Thread

Thread that wraps target with stack/context tracking.

App components that do not use this thread class might not be properly tracked.

ThreadPoolExecutor

Bases: ThreadPoolExecutor

A ThreadPoolExecutor that keeps track of the stack prior to each thread's invocation.

Apps that do not use this thread pool might not be properly tracked.

TP

Bases: SingletonPerName

Manager of thread pools.

Singleton.

Attributes
MAX_THREADS class-attribute instance-attribute
MAX_THREADS: int = 128

Maximum number of threads to run concurrently.

DEBUG_TIMEOUT class-attribute instance-attribute
DEBUG_TIMEOUT: Optional[float] = 600.0

How long to wait (seconds) for any task before restarting it.

Functions

trulens_eval.utils.asynchro

Synchronization/Async Utilities

NOTE: we cannot name a module "async" as it is a python keyword.

Synchronous vs. Asynchronous

Some functions in trulens_eval come with asynchronous versions. Those use "async def" instead of "def" and typically start with the letter "a" in their name with the rest matching their synchronous version.

Due to how python handles such functions and how they are executed, it is relatively difficult to reshare code between the two versions. Asynchronous functions are executed by an async loop (see EventLoop). Python prevents any threads from having more than one running loop meaning one may not be able to create one to run some async code if one has already been created/running in the thread. The method sync here, used to convert an async computation into a sync computation, needs to create a new thread. The impact of this, whether overhead, or record info, is uncertain.

What should be Sync/Async?

Try to have all internals be async but for users we may expose sync versions via the sync method. If internals are async and don't need exposure, don't need to provide a synced version.

Attributes

MaybeAwaitable module-attribute

MaybeAwaitable = Union[T, Awaitable[T]]

Awaitable or not.

May be checked with isawaitable.

CallableMaybeAwaitable module-attribute

CallableMaybeAwaitable = Union[Callable[[A], B], Callable[[A], Awaitable[B]]]

Function or coroutine function.

May be checked with is_really_coroutinefunction.

CallableAwaitable module-attribute

CallableAwaitable = Callable[[A], Awaitable[B]]

Function that produces an awaitable / coroutine function.

ThunkMaybeAwaitable module-attribute

ThunkMaybeAwaitable = Union[Thunk[T], Thunk[Awaitable[T]]]

Thunk or coroutine thunk.

May be checked with is_really_coroutinefunction.

Classes

Functions

desync async

desync(func: CallableMaybeAwaitable[A, T], *args, **kwargs) -> T

Run the given function asynchronously with the given args. If it is not asynchronous, will run in thread. Note: this has to be marked async since in some cases we cannot tell ahead of time that func is asynchronous so we may end up running it to produce a coroutine object which we then need to run asynchronously.

sync

sync(func: CallableMaybeAwaitable[A, T], *args, **kwargs) -> T

Get result of calling function on the given args. If it is awaitable, will block until it is finished. Runs in a new thread in such cases.