Skip to content

trulens.core.app

trulens.core.app

Attributes

Classes

ComponentView

Bases: ABC

Views of common app component types for sorting them and displaying them in some unified manner in the UI. Operates on components serialized into json dicts representing various components, not the components themselves.

Functions
of_json classmethod 
of_json(json: JSON) -> 'ComponentView'

Sort the given json into the appropriate component view type.

class_is abstractmethod staticmethod 
class_is(cls_obj: Class) -> bool

Determine whether the given class representation cls is of the type to be viewed as this component type.

unsorted_parameters 
unsorted_parameters(
    skip: Set[str],
) -> Dict[str, JSON_BASES_T]

All basic parameters not organized by other accessors.

innermost_base staticmethod 
innermost_base(
    bases: Optional[Sequence[Class]] = None,
    among_modules=set(
        ["langchain", "llama_index", "trulens"]
    ),
) -> Optional[str]

Given a sequence of classes, return the first one which comes from one of the among_modules. You can use this to determine where ultimately the encoded class comes from in terms of langchain, llama_index, or trulens even in cases they extend each other's classes. Returns None if no module from among_modules is named in bases.

TrulensComponent

Bases: ComponentView

Components provided in trulens.

Functions
unsorted_parameters 
unsorted_parameters(
    skip: Set[str],
) -> Dict[str, JSON_BASES_T]

All basic parameters not organized by other accessors.

innermost_base staticmethod 
innermost_base(
    bases: Optional[Sequence[Class]] = None,
    among_modules=set(
        ["langchain", "llama_index", "trulens"]
    ),
) -> Optional[str]

Given a sequence of classes, return the first one which comes from one of the among_modules. You can use this to determine where ultimately the encoded class comes from in terms of langchain, llama_index, or trulens even in cases they extend each other's classes. Returns None if no module from among_modules is named in bases.

RecordingContext

Manager of the creation of records from record calls.

An instance of this class is produced when using an App as a context manager, i.e.:

Example 
app = ...  # your app
truapp: TruChain = TruChain(app, ...) # recorder for LangChain apps

with truapp as recorder:
    app.invoke(...) # use your app

recorder: RecordingContext

Each instance of this class produces a record for every "root" instrumented method called. Root method here means the first instrumented method in a call stack. Note that there may be more than one of these contexts in play at the same time due to:

  • More than one wrapper of the same app.
  • More than one context manager ("with" statement) surrounding calls to the same app.
  • Calls to "with_record" on methods that themselves contain recording.
  • Calls to apps that use trulens internally to track records in any of the supported ways.
  • Combinations of the above.
Attributes
calls instance-attribute 
calls: Dict[CallID, RecordAppCall] = {}

A record (in terms of its RecordAppCall) in process of being created.

Storing as a map as we want to override calls with the same id which may happen due to methods producing awaitables or generators. These result in calls before the awaitables are awaited and then get updated after the result is ready.

records instance-attribute 
records: List[Record] = []

Completed records.

lock instance-attribute 
lock: Lock = Lock()

Lock blocking access to calls and records when adding calls or finishing a record.

token instance-attribute 
token: Optional[Token] = None

Token for context management.

app instance-attribute 
app: WithInstrumentCallbacks = app

App for which we are recording.

record_metadata instance-attribute 
record_metadata = record_metadata

Metadata to attach to all records produced in this context.

Functions
get 
get() -> Record

Get the single record only if there was exactly one. Otherwise throw an error.

add_call 
add_call(call: RecordAppCall)

Add the given call to the currently tracked call list.

finish_record 
finish_record(
    calls_to_record: Callable[
        [List[RecordAppCall], Metadata, Optional[Record]],
        Record,
    ],
    existing_record: Optional[Record] = None,
)

Run the given function to build a record from the tracked calls and any pre-specified metadata.

App

Bases: AppDefinition, WithInstrumentCallbacks, Hashable

Base app recorder type.

Non-serialized fields here while the serialized ones are defined in AppDefinition.

This class is abstract. Use one of these concrete subclasses as appropriate: - TruLlama for LlamaIndex apps. - TruChain for LangChain apps. - TruRails for NeMo Guardrails apps. - TruVirtual for recording information about invocations of apps without access to those apps. - TruCustomApp for custom apps. These need to be decorated to have appropriate data recorded. - TruBasicApp for apps defined solely by a string-to-string method.

Attributes
tru_class_info instance-attribute 
tru_class_info: Class

Class information of this pydantic object for use in deserialization.

Using this odd key to not pollute attribute names in whatever class we mix this into. Should be the same as CLASS_INFO.

app_id class-attribute instance-attribute 
app_id: AppID = Field(frozen=True)

Unique identifier for this app.

Computed deterministically from app_name and app_version. Leaving it here for it to be dumped when serializing. Also making it read-only as it should not be changed after creation.

app_name instance-attribute 
app_name: AppName

Name for this app. Default is "default_app".

app_version instance-attribute 
app_version: AppVersion

Version tag for this app. Default is "base".

tags instance-attribute 
tags: Tags = tags

Tags for the app.

metadata instance-attribute 
metadata: Metadata

Metadata for the app.

feedback_definitions class-attribute instance-attribute 
feedback_definitions: Sequence[FeedbackDefinitionID] = []

Feedback functions to evaluate on each record.

feedback_mode class-attribute instance-attribute 
feedback_mode: FeedbackMode = WITH_APP_THREAD

How to evaluate feedback functions upon producing a record.

record_ingest_mode instance-attribute 
record_ingest_mode: RecordIngestMode = record_ingest_mode

Mode of records ingestion.

root_class instance-attribute 
root_class: Class

Class of the main instrumented object.

Ideally this would be a ClassVar but since we want to check this without instantiating the subclass of AppDefinition that would define it, we cannot use ClassVar.

root_callable class-attribute 
root_callable: FunctionOrMethod

App's main method.

This is to be filled in by subclass.

initial_app_loader_dump class-attribute instance-attribute 
initial_app_loader_dump: Optional[SerialBytes] = None

Serialization of a function that loads an app.

Dump is of the initial app state before any invocations. This can be used to create a new session.

Warning

Experimental work in progress.

app_extra_json instance-attribute 
app_extra_json: JSON

Info to store about the app and to display in dashboard.

This can be used even if app itself cannot be serialized. app_extra_json, then, can stand in place for whatever data the user might want to keep track of about the app.

feedbacks class-attribute instance-attribute 
feedbacks: List[Feedback] = Field(
    exclude=True, default_factory=list
)

Feedback functions to evaluate on each record.

connector class-attribute instance-attribute 
connector: DBConnector = Field(
    default_factory=lambda: connector, exclude=True
)

Database connector.

If this is not provided, a DefaultDBConnector will be made (if not already) and used.

instrument class-attribute instance-attribute 
instrument: Optional[Instrument] = Field(None, exclude=True)

Instrumentation class.

This is needed for serialization as it tells us which objects we want to be included in the json representation of this app.

recording_contexts class-attribute instance-attribute 
recording_contexts: ContextVar[RecordingContext] = Field(
    None, exclude=True
)

Sequences of records produced by the this class used as a context manager are stored in a RecordingContext.

Using a context var so that context managers can be nested.

instrumented_methods class-attribute instance-attribute 
instrumented_methods: Dict[int, Dict[Callable, Lens]] = (
    Field(exclude=True, default_factory=dict)
)

Mapping of instrumented methods (by id(.) of owner object and the function) to their path in this app.

records_with_pending_feedback_results class-attribute instance-attribute 
records_with_pending_feedback_results: BlockingSet[
    Record
] = Field(exclude=True, default_factory=BlockingSet)

Records produced by this app which might have yet to finish feedback runs.

manage_pending_feedback_results_thread class-attribute instance-attribute 
manage_pending_feedback_results_thread: Optional[Thread] = (
    Field(exclude=True, default=None)
)

Thread for manager of pending feedback results queue.

See _manage_pending_feedback_results.

selector_check_warning class-attribute instance-attribute 
selector_check_warning: bool = False

Issue warnings when selectors are not found in the app with a placeholder record.

If False, constructor will raise an error instead.

selector_nocheck class-attribute instance-attribute 
selector_nocheck: bool = False

Ignore selector checks entirely.

This may be necessary if the expected record content cannot be determined before it is produced.

app class-attribute instance-attribute 
app: Any = app

The app to be recorded.

Functions
__rich_repr__ 
__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

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.

continue_session staticmethod 
continue_session(
    app_definition_json: JSON, app: Any
) -> AppDefinition

Instantiate the given app with the given state app_definition_json.

Warning

This is an experimental feature with ongoing work.

PARAMETER DESCRIPTION
app_definition_json

The json serialized app.

TYPE: JSON

app

The app to continue the session with.

TYPE: Any

RETURNS DESCRIPTION
AppDefinition

A new AppDefinition instance with the given app and the given app_definition_json state.
new_session staticmethod 
new_session(
    app_definition_json: JSON,
    initial_app_loader: Optional[Callable] = None,
) -> AppDefinition

Create an app instance at the start of a session.

Warning

This is an experimental feature with ongoing work.

Create a copy of the json serialized app with the enclosed app being initialized to its initial state before any records are produced (i.e. blank memory).

get_loadable_apps staticmethod 
get_loadable_apps()

Gets a list of all of the loadable apps.

Warning

This is an experimental feature with ongoing work.

This is those that have initial_app_loader_dump set.

select_inputs classmethod 
select_inputs() -> Lens

Get the path to the main app's call inputs.

select_outputs classmethod 
select_outputs() -> Lens

Get the path to the main app's call outputs.

wait_for_feedback_results 
wait_for_feedback_results(
    feedback_timeout: Optional[float] = None,
) -> List[Record]

Wait for all feedbacks functions to complete.

PARAMETER DESCRIPTION
feedback_timeout

Timeout in seconds for waiting for feedback results for each feedback function. Note that this is not the total timeout for this entire blocking call.

TYPE: Optional[float] DEFAULT: None

RETURNS DESCRIPTION
List[Record]

A list of records that have been waited on. Note a record will be included even if a feedback computation for it failed or timed out.

This applies to all feedbacks on all records produced by this app. This call will block until finished and if new records are produced while this is running, it will include them.

select_context classmethod 
select_context(app: Optional[Any] = None) -> Lens

Try to find retriever components in the given app and return a lens to access the retrieved contexts that would appear in a record were these components to execute.

main_call 
main_call(human: str) -> str

If available, a single text to a single text invocation of this app.

main_acall async 
main_acall(human: str) -> str

If available, a single text to a single text invocation of this app.

main_input 
main_input(
    func: Callable, sig: Signature, bindings: BoundArguments
) -> JSON

Determine the main input string for the given function func with signature sig if it is to be called with the given bindings bindings.

PARAMETER DESCRIPTION
func

The main function we are targetting in this determination.

TYPE: Callable

sig

The signature of the above.

TYPE: Signature

bindings

The arguments to be passed to the function.

TYPE: BoundArguments

RETURNS DESCRIPTION
JSON

The main input string.
main_output 
main_output(
    func: Callable,
    sig: Signature,
    bindings: BoundArguments,
    ret: Any,
) -> JSON

Determine the main out string for the given function func with signature sig after it is called with the given bindings and has returned ret.

on_method_instrumented 
on_method_instrumented(
    obj: object, func: Callable, path: Lens
)

Called by instrumentation system for every function requested to be instrumented by this app.

get_methods_for_func 
get_methods_for_func(
    func: Callable,
) -> Iterable[Tuple[int, Callable, Lens]]

Get the methods (rather the inner functions) matching the given func and the path of each.

See WithInstrumentCallbacks.get_methods_for_func.

get_method_path 
get_method_path(obj: object, func: Callable) -> Lens

Get the path of the instrumented function method relative to this app.

json 
json(*args, **kwargs)

Create a json string representation of this app.

on_new_record 
on_new_record(func) -> Iterable[RecordingContext]

Called at the start of record creation.

See WithInstrumentCallbacks.on_new_record.

on_add_record 
on_add_record(
    ctx: RecordingContext,
    func: Callable,
    sig: Signature,
    bindings: BoundArguments,
    ret: Any,
    error: Any,
    perf: Perf,
    cost: Cost,
    existing_record: Optional[Record] = None,
) -> Record

Called by instrumented methods if they use _new_record to construct a record call list.

See WithInstrumentCallbacks.on_add_record.

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

Call the given async func with the given *args and **kwargs while recording, producing func results. The record of the computation is available through other means like the database or dashboard. If you need a record of this execution immediately, you can use awith_record or the App as a context manager instead.

with_ async 
with_(func: Callable[[A], T], *args, **kwargs) -> T

Call the given async func with the given *args and **kwargs while recording, producing func results. The record of the computation is available through other means like the database or dashboard. If you need a record of this execution immediately, you can use awith_record or the App as a context manager instead.

with_record 
with_record(
    func: Callable[[A], T],
    *args,
    record_metadata: JSON = None,
    **kwargs
) -> Tuple[T, Record]

Call the given func with the given *args and **kwargs, producing its results as well as a record of the execution.

awith_record async 
awith_record(
    func: Callable[[A], Awaitable[T]],
    *args,
    record_metadata: JSON = None,
    **kwargs
) -> Tuple[T, Record]

Call the given func with the given *args and **kwargs, producing its results as well as a record of the execution.

dummy_record 
dummy_record(
    cost: Cost = mod_base_schema.Cost(),
    perf: Perf = mod_base_schema.Perf.now(),
    ts: datetime = datetime.datetime.now(),
    main_input: str = "main_input are strings.",
    main_output: str = "main_output are strings.",
    main_error: str = "main_error are strings.",
    meta: Dict = {"metakey": "meta are dicts"},
    tags: str = "tags are strings",
) -> Record

Create a dummy record with some of the expected structure without actually invoking the app.

The record is a guess of what an actual record might look like but will be missing information that can only be determined after a call is made.

All args are Record fields except these:

- `record_id` is generated using the default id naming schema.
- `app_id` is taken from this recorder.
- `calls` field is constructed based on instrumented methods.
instrumented 
instrumented() -> Iterable[Tuple[Lens, ComponentView]]

Iteration over instrumented components and their categories.

print_instrumented 
print_instrumented() -> None

Print the instrumented components and methods.

format_instrumented_methods 
format_instrumented_methods() -> str

Build a string containing a listing of instrumented methods.

print_instrumented_methods 
print_instrumented_methods() -> None

Print instrumented methods.

print_instrumented_components 
print_instrumented_components() -> None

Print instrumented components and their categories.

Functions

instrumented_component_views 

instrumented_component_views(
    obj: object,
) -> Iterable[Tuple[Lens, ComponentView]]

Iterate over contents of obj that are annotated with the CLASS_INFO attribute/key. Returns triples with the accessor/selector, the Class object instantiated from CLASS_INFO, and the annotated object itself.