Skip to content

Tru Basic App


Bases: App

Instantiates a Basic app that makes little assumptions.

Assumes input text and output text.

def custom_application(prompt: str) -> str:
    return "a response"

from trulens_eval import TruBasicApp
# f_lang_match, f_qa_relevance, f_qs_relevance are feedback functions
tru_recorder = TruBasicApp(custom_application, 
    app_id="Custom Application v1",
    feedbacks=[f_lang_match, f_qa_relevance, f_qs_relevance])

# Basic app works by turning your callable into an app
# This app is accessbile with the `app` attribute in the recorder
with tru_recorder as recording:

tru_record = recording.records[0]

See Feedback Functions for instantiating feedback functions.


A str to str callable.

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


A TruWrapperApp instance. If not provided, text_to_text must be provided.

TYPE: Optional[TruWrapperApp] DEFAULT: None


Additional arguments to pass to App and AppDefinition

TYPE: dict DEFAULT: {}


app_id instance-attribute

app_id: AppID = app_id

Unique identifier for this app.

tags instance-attribute

tags: Tags = tags

Tags for the app.

metadata instance-attribute

metadata: Metadata = metadata

Metadata for the app.

feedback_definitions class-attribute instance-attribute

feedback_definitions: Sequence[FeedbackDefinition] = []

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.

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.

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.


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.

tru class-attribute instance-attribute

tru: Optional[Tru] = Field(default=None, exclude=True)

Workspace manager.

If this is not povided, a singleton Tru will be made (if not already) and used.

db class-attribute instance-attribute

db: Optional[DB] = Field(default=None, exclude=True)

Database interface.

If this is not provided, a singleton SQLAlchemyDB 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)

Sequnces 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: Queue[Record] = Field(exclude=True, default_factory=lambda: Queue(maxsize=1024))

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 instance-attribute

app: TruWrapperApp

The app to be instrumented.

root_callable class-attribute

root_callable: FunctionOrMethod = Field(default_factory=lambda: of_callable(_call))

The root callable to be instrumented.

This is the method that will be called by the main_input method.



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

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


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

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


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.


on_new_record(func) -> Iterable[RecordingContext]

Called at the start of record creation.

See WithInstrumentCallbacks.on_new_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.

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.


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.


This is an experimental feature with ongoing work.


The json serialized app.



The app to continue the session with.



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.


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


Gets a list of all of the loadable apps.


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() -> None

Wait for all feedbacks functions to complete.

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_acall async

main_acall(human: str) -> str

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


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.


json(*args, **kwargs)

Create a json string representation of this app.

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 mananger 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 mananger instead.


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(cost: Cost = mod_schema.Cost(), perf: Perf =, ts: datetime =, 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() -> Iterable[Tuple[Lens, ComponentView]]

Iteration over instrumented components and their categories.


print_instrumented() -> None

Print the instrumented components and methods.


format_instrumented_methods() -> str

Build a string containing a listing of instrumented methods.


print_instrumented_methods() -> None

Print instrumented methods.


print_instrumented_components() -> None

Print instrumented components and their categories.