trulens.apps.custom¶

trulens.apps.custom ¶

Custom class application

This wrapper is the most flexible option for instrumenting an application, and can be used to instrument any custom python class.

Example:

Consider a mock question-answering app with a context retriever component coded
up as two classes in two python, `CustomApp` and `CustomRetriever`:

### `custom_app.py`

```python
from trulens.apps.custom import instrument
from custom_retriever import CustomRetriever


class CustomApp:
    # NOTE: No restriction on this class.

    def __init__(self):
        self.retriever = CustomRetriever()

    @instrument
    def retrieve_chunks(self, data):
        return self.retriever.retrieve_chunks(data)

    @instrument
    def respond_to_query(self, input):
        chunks = self.retrieve_chunks(input) output = f"The answer to {input} is
        probably {chunks[0]} or something ..." return output
```

### `custom_retriever.py`

```python
from trulens.apps.custom import instrument

class CustomRetriever:
    # NOTE: No restriction on this class either.

    @instrument
    def retrieve_chunks(self, data):
        return [
            f"Relevant chunk: {data.upper()}", f"Relevant chunk: {data[::-1]}"
        ]
```

The core tool for instrumenting these classes is the @instrument decorator. TruLens needs to be aware of two high-level concepts to usefully monitor the app: components and methods used by components. The instrument must decorate each method that the user wishes to track.

The owner classes of any decorated method is then viewed as an app component. In this example, case CustomApp and CustomRetriever are components.

Example:

### `example.py`

```python
from custom_app import CustomApp
from trulens.apps.custom import TruCustomApp

custom_app = CustomApp()

# Normal app Usage:
response = custom_app.respond_to_query("What is the capital of Indonesia?")

# Wrapping app with `TruCustomApp`:
tru_recorder = TruCustomApp(ca)

# Tracked usage:
with tru_recorder:
    custom_app.respond_to_query, input="What is the capital of Indonesia?")
```

`TruCustomApp` constructor arguments are like in those higher-level

apps as well including the feedback functions, metadata, etc.

Instrumenting 3rd party classes¶

In cases you do not have access to a class to make the necessary decorations for tracking, you can instead use one of the static methods of instrument, for example, the alternative for making sure the custom retriever gets instrumented is via:

# custom_app.py`:

from trulens.apps.custom import instrument
from some_package.from custom_retriever import CustomRetriever

instrument.method(CustomRetriever, "retrieve_chunks")

# ... rest of the custom class follows ...

API Usage Tracking¶

Uses of python libraries for common LLMs like OpenAI are tracked in custom class apps.

Covered LLM Libraries¶

Official OpenAI python package (https://github.com/openai/openai-python).

Huggingface¶

Uses of huggingface inference APIs are tracked as long as requests are made through the requests class's post method to the URL https://api-inference.huggingface.co .

Limitations¶

Tracked (instrumented) components must be accessible through other tracked components. Specifically, an app cannot have a custom class that is not instrumented but that contains an instrumented class. The inner instrumented class will not be found by trulens.
All tracked components are categorized as "Custom" (as opposed to Template, LLM, etc.). That is, there is no categorization available for custom components. They will all show up as "uncategorized" in the dashboard.
Non json-like contents of components (that themselves are not components) are not recorded or available in dashboard. This can be alleviated to some extent with the app_extra_json argument to TruCustomClass as it allows one to specify in the form of json additional information to store alongside the component hierarchy. Json-like (json bases like string, int, and containers like sequences and dicts are included).

What can go wrong¶

If a with_record or awith_record call does not encounter any instrumented method, it will raise an error. You can check which methods are instrumented using App.print_instrumented. You may have forgotten to decorate relevant methods with @instrument.

app.print_instrumented()

### output example:
Components:
        TruCustomApp (Other) at 0x171bd3380 with path *.__app__
        CustomApp (Custom) at 0x12114b820 with path *.__app__.app
        CustomLLM (Custom) at 0x12114be50 with path *.__app__.app.llm
        CustomMemory (Custom) at 0x12114bf40 with path *.__app__.app.memory
        CustomRetriever (Custom) at 0x12114bd60 with path *.__app__.app.retriever
        CustomTemplate (Custom) at 0x12114bf10 with path *.__app__.app.template

Methods:
Object at 0x12114b820:
        <function CustomApp.retrieve_chunks at 0x299132ca0> with path *.__app__.app
        <function CustomApp.respond_to_query at 0x299132d30> with path *.__app__.app
        <function CustomApp.arespond_to_query at 0x299132dc0> with path *.__app__.app
Object at 0x12114be50:
        <function CustomLLM.generate at 0x299106b80> with path *.__app__.app.llm
Object at 0x12114bf40:
        <function CustomMemory.remember at 0x299132670> with path *.__app__.app.memory
Object at 0x12114bd60:
        <function CustomRetriever.retrieve_chunks at 0x299132790> with path *.__app__.app.retriever
Object at 0x12114bf10:
        <function CustomTemplate.fill at 0x299132a60> with path *.__app__.app.template

If an instrumented / decorated method's owner object cannot be found when traversing your custom class, you will get a warning. This may be ok in the end but may be indicative of a problem. Specifically, note the "Tracked" limitation above. You can also use the app_extra_json argument to App / TruCustomApp to provide a structure to stand in place for (or augment) the data produced by walking over instrumented components to make sure this hierarchy contains the owner of each instrumented method.

The owner-not-found error looks like this:

Function <function CustomRetriever.retrieve_chunks at 0x177935d30> was not found during instrumentation walk. Make sure it is accessible by traversing app <custom_app.CustomApp object at 0x112a005b0> or provide a bound method for it as TruCustomApp constructor argument `methods_to_instrument`.
Function <function CustomTemplate.fill at 0x1779474c0> was not found during instrumentation walk. Make sure it is accessible by traversing app <custom_app.CustomApp object at 0x112a005b0> or provide a bound method for it as TruCustomApp constructor argument `methods_to_instrument`.
Function <function CustomLLM.generate at 0x1779471f0> was not found during instrumentation walk. Make sure it is accessible by traversing app <custom_app.CustomApp object at 0x112a005b0> or provide a bound method for it as TruCustomApp constructor argument `methods_to_instrument`.

Subsequent attempts at with_record/awith_record may result in the "Empty record" exception.

Usage tracking not tracking. We presently have limited coverage over which APIs we track and make some assumptions with regards to accessible APIs through lower-level interfaces. Specifically, we only instrument the requests module's post method for the lower level tracking. Please file an issue on github with your use cases so we can work out a more complete solution as needed.

Classes¶

TruCustomApp ¶

Bases: App

This recorder is the most flexible option for instrumenting an application, and can be used to instrument any custom python class.

Track any custom app using methods decorated with @instrument, or whose methods are instrumented after the fact by instrument.method.

Example: "Using the @instrument decorator"

```python
from trulens.core import instrument

class CustomApp:

    def __init__(self):
        self.retriever = CustomRetriever()
        self.llm = CustomLLM()
        self.template = CustomTemplate(
            "The answer to {question} is probably {answer} or something ..."
        )

    @instrument
    def retrieve_chunks(self, data):
        return self.retriever.retrieve_chunks(data)

    @instrument
    def respond_to_query(self, input):
        chunks = self.retrieve_chunks(input)
        answer = self.llm.generate(",".join(chunks))
        output = self.template.fill(question=input, answer=answer)

        return output

ca = CustomApp()
```

Example: "Using instrument.method"

```python
from trulens.core import instrument

class CustomApp:

    def __init__(self):
        self.retriever = CustomRetriever()
        self.llm = CustomLLM()
        self.template = CustomTemplate(
            "The answer to {question} is probably {answer} or something ..."
        )

    def retrieve_chunks(self, data):
        return self.retriever.retrieve_chunks(data)

    def respond_to_query(self, input):
        chunks = self.retrieve_chunks(input)
        answer = self.llm.generate(",".join(chunks))
        output = self.template.fill(question=input, answer=answer)

        return output

custom_app = CustomApp()

instrument.method(CustomApp, "retrieve_chunks")
```

Once a method is tracked, its arguments and returns are available to be used in feedback functions. This is done by using the Select class to select the arguments and returns of the method.

Doing so follows the structure:

For args: Select.RecordCalls.<method_name>.args.<arg_name>
For returns: Select.RecordCalls.<method_name>.rets.<ret_name>

Example: "Defining feedback functions with instrumented methods"

```python
f_context_relevance = (
    Feedback(provider.context_relevance_with_cot_reasons, name = "Context Relevance")
    .on(Select.RecordCalls.retrieve_chunks.args.query) # refers to the query arg of CustomApp's retrieve_chunks method
    .on(Select.RecordCalls.retrieve_chunks.rets.collect())
    .aggregate(np.mean)
    )
```

Last, the TruCustomApp recorder can wrap our custom application, and provide logging and evaluation upon its use.

Example: "Using the TruCustomApp recorder"

```python
from trulens.apps.custom import TruCustomApp

tru_recorder = TruCustomApp(custom_app,
    app_name="Custom Application",
    app_version="base",
    feedbacks=[f_context_relevance])

with tru_recorder as recording:
    custom_app.respond_to_query("What is the capital of Indonesia?")
```

See [Feedback
Functions](https://www.trulens.org/trulens/api/feedback/) for
instantiating feedback functions.

PARAMETER	DESCRIPTION
`app`	Any class. TYPE: `Any`
`**kwargs`	Additional arguments to pass to App and AppDefinition TYPE: `Any` DEFAULT: `{}`

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.

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.

functions_to_instrument `class-attribute` ¶

functions_to_instrument: Set[Callable] = set()

Methods marked as needing instrumentation.

These are checked to make sure the object walk finds them. If not, a message is shown to let user know how to let the TruCustomApp constructor know where these methods are.

main_method_loaded `class-attribute` `instance-attribute` ¶

main_method_loaded: Optional[Callable] = Field(
    None, exclude=True
)

Main method of the custom app.

main_method `class-attribute` `instance-attribute` ¶

main_method: Optional[Function] = None

Serialized version of the main method.

Functions¶

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_method_path ¶

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

Get the path of the instrumented function method relative to 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.

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.

__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_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.

json ¶

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

instrument ¶

Bases: instrument

Decorator for marking methods to be instrumented in custom classes that are wrapped by TruCustomApp.

Functions¶

methods `classmethod` ¶

methods(of_cls: type, names: Iterable[str]) -> None

Add the class with methods named names, its module, and the named methods to the Default instrumentation walk filters.

__set_name__ ¶

__set_name__(cls: type, name: str)

For use as method decorator.

trulens.apps.custom¶