Tru Custom App

Custom class Apps

This wrapper covers apps that are not based on one of the high-level frameworks such as langchain or llama-index. We instead assume that some python class or classes implements an app which has similar functionality to LLM apps coded in the high-level frameworks in that it generally processes text queries to produce text outputs while making intermediate queries to things like LLMs, vector DBs, and similar.

Example Usage

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

from trulens_eval.tru_custom_app 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

from trulens_eval.tru_custom_app 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 method (actually class, but details are not important here). 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 watch (for it to show up on the dashboard). In the example, all of the functionalities are decorated. Additionally, the owner classes of any decorated method is viewed as an app component. In this case CustomApp and CustomRetriever are components.

Following the instrumentation, the app can be used with or without tracking:

example.py

from custom_app import CustomApp from trulens_eval.tru_custom_app
import TruCustomApp

ca = CustomApp()

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

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

# Wrapped Usage: must use the general `with_record` (or `awith_record`) method:
response, record = ta.with_record(
    ca.respond_to_query, input="What is the capital of Indonesia?"
)

The with_record use above returns both the response of the app normally produces as well as the record of the app as is the case with the higher-level wrappers. TruCustomApp constructor arguments are like in those higher-level apps as well including the feedback functions, metadata, etc.

Instrumenting 3^rd 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 alterative for making sure the custom retriever gets instrumented is via:

# custom_app.py`:

from trulens_eval.tru_custom_app import instrument
from somepackage.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:
Custom of trulens_eval.app component: *.__app__.app
Custom of trulens_eval.app component: *.__app__.app.llm
Custom of trulens_eval.app component: *.__app__.app.retriever
Custom of trulens_eval.app component: *.__app__.app.template

Methods:
Object at 0x1064c6d30:
        <function CustomApp.retrieve_chunks at 0x14c6c5700> with path *.__app__.app
        <function CustomApp.respond_to_query at 0x14c6c5790> with path *.__app__.app
Object at 0x1064c6dc0:
        <function CustomLLM.generate at 0x14c6c51f0> with path *.__app__.app.llm
Object at 0x1064c6f70:
        <function CustomRetriever.retrieve_chunks at 0x14c6b5c10> with path *.__app__.app.retriever
Object at 0x106272100:
        <function CustomTemplate.fill at 0x14c6c54c0> 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.

TruCustomApp

Bases: App

Instantiates a Custom App that can be tracked as long as methods are decorated with @instrument.

Usage:

from trulens_eval 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()
from trulens_eval import TruCustomApp
# f_lang_match, f_qa_relevance, f_qs_relevance are feedback functions
tru_recorder = TruCustomApp(ca, 
    app_id="Custom Application v1",
    feedbacks=[f_lang_match, f_qa_relevance, f_qs_relevance])

question = "What is the capital of Indonesia?"

# Normal Usage:
response_normal = ca.respond_to_query(question)

# Instrumented Usage:
with tru_recorder as recording:
    ca.respond_to_query(question)

tru_record = recording.records[0]

# To add record metadata 
with tru_recorder as recording:
    recording.record_metadata="this is metadata for all records in this context that follow this line"
    ca.respond_to_query("What is llama 2?")
    recording.record_metadata="this is different metadata for all records in this context that follow this line"
    ca.respond_to_query("Where do I download llama 2?")

See Feedback Functions for instantiating feedback functions.

Parameters:

Name	Type	Description	Default
app	Any	Any class	required

Source code in trulens_eval/trulens_eval/tru_custom_app.py

class TruCustomApp(App):
    """Instantiates a Custom App that can be tracked as long as methods are decorated with @instrument.

        **Usage:**

        ```
        from trulens_eval 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()
        from trulens_eval import TruCustomApp
        # f_lang_match, f_qa_relevance, f_qs_relevance are feedback functions
        tru_recorder = TruCustomApp(ca, 
            app_id="Custom Application v1",
            feedbacks=[f_lang_match, f_qa_relevance, f_qs_relevance])

        question = "What is the capital of Indonesia?"

        # Normal Usage:
        response_normal = ca.respond_to_query(question)

        # Instrumented Usage:
        with tru_recorder as recording:
            ca.respond_to_query(question)

        tru_record = recording.records[0]

        # To add record metadata 
        with tru_recorder as recording:
            recording.record_metadata="this is metadata for all records in this context that follow this line"
            ca.respond_to_query("What is llama 2?")
            recording.record_metadata="this is different metadata for all records in this context that follow this line"
            ca.respond_to_query("Where do I download llama 2?")

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

        Args:
            app (Any): Any class
    """
    app: Any

    root_callable: ClassVar[FunctionOrMethod] = Field(None)

    # 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.
    functions_to_instrument: ClassVar[Set[Callable]] = set([])

    main_method: Optional[Callable] = Field(exclude=True)
    main_async_method: Optional[Callable] = Field(exclude=True)

    def __init__(self, app: Any, methods_to_instrument=None, **kwargs):
        """
        Wrap a custom class for recording.

        Arguments:
        - app: Any -- the custom app object being wrapped.
        - More args in App
        - More args in AppDefinition
        - More args in WithClassInfo
        """

        kwargs['app'] = app
        kwargs['root_class'] = Class.of_object(app)

        kwargs['instrument'] = Instrument(
            app=self  # App mixes in WithInstrumentCallbacks
        )

        super().__init__(**kwargs)

        methods_to_instrument = methods_to_instrument or dict()

        # The rest of this code instruments methods explicitly passed to
        # constructor as needing instrumentation and checks that methods
        # decorated with @instrument or passed explicitly belong to some
        # component as per serialized version of this app. If they are not,
        # placeholders are made in `app_extra_json` so that subsequent
        # serialization looks like the components exist.
        json = self.dict()

        for m, path in methods_to_instrument.items():
            method_name = m.__name__

            full_path = JSONPath().app + path

            self.instrument.instrument_method(
                method_name=method_name, obj=m.__self__, query=full_path
            )

            # TODO: DEDUP with next condition

            # Check whether the path/location of the method is in json serialization and
            # if not, add a placeholder to app_extra_json.
            try:
                next(full_path(json))

                print(
                    f"{UNICODE_CHECK} Added method {m.__name__} under component at path {full_path}"
                )

            except Exception:
                logger.warning(
                    f"App has no component at path {full_path} . "
                    f"Specify the component with the `app_extra_json` argument to TruCustomApp constructor. "
                    f"Creating a placeholder there for now."
                )

                path.set(
                    self.app_extra_json, {
                        PLACEHOLDER:
                            "I was automatically added to `app_extra_json` because there was nothing here to refer to an instrumented method owner.",
                        m.__name__:
                            f"Placeholder for method {m.__name__}."
                    }
                )

        # Check that any functions marked with `TruCustomApp.instrument` has been
        # instrumented as a method under some object.
        for f in TruCustomApp.functions_to_instrument:
            obj_ids_methods_and_full_paths = list(self._get_methods_for_func(f))

            if len(obj_ids_methods_and_full_paths) == 0:
                logger.warning(
                    f"Function {f} was not found during instrumentation walk. "
                    f"Make sure it is accessible by traversing app {app} "
                    f"or provide a bound method for it as TruCustomApp constructor argument `methods_to_instrument`."
                )

            else:
                for obj_id, m, full_path in obj_ids_methods_and_full_paths:
                    try:
                        next(full_path(json))

                    except Exception as e:
                        logger.warning(
                            f"App has no component owner of instrumented method {m} at path {full_path}. "
                            f"Specify the component with the `app_extra_json` argument to TruCustomApp constructor. "
                            f"Creating a placeholder there for now."
                        )

                        path.set(
                            self.app_extra_json, {
                                PLACEHOLDER:
                                    "I was automatically added to `app_extra_json` because there was nothing here to refer to an instrumented method owner.",
                                m.__name__:
                                    f"Placeholder for method {m.__name__}."
                            }
                        )

        # DB stuff and checks:
        self.post_init()

    def __getattr__(self, __name: str) -> Any:
        # A message for cases where a user calls something that the wrapped
        # app has but we do not wrap yet.

        if hasattr(self.app, __name):
            return RuntimeError(
                f"TruCustomApp has no attribute {__name} but the wrapped app ({type(self.app)}) does. ",
                f"If you are calling a {type(self.app)} method, retrieve it from that app instead of from `TruCustomApp`. "
            )
        else:
            raise RuntimeError(
                f"TruCustomApp nor wrapped app have attribute named {__name}."
            )

    def main_call(self, human: str):
        if self.main_method is None:
            raise RuntimeError("`main_method` was not specified so we do not know how to run this app.")

        sig = signature(self.main_method)
        bindings = sig.bind(self.app, human) # self.app is app's "self"

        return self.with_(self.main_method, *bindings.args, **bindings.kwargs)

    async def main_acall(self, human: str):
        # TODO: work in progress

        # must return an async generator of tokens/pieces that can be appended to create the full response

        if self.main_async_method is None:
            raise RuntimeError("`main_async_method` was not specified so we do not know how to run this app.")

        sig = signature(self.main_async_method)
        bindings = sig.bind(self.app, human) # self.app is app's "self"

        generator = await self.awith_(self.main_async_method, *bindings.args, **bindings.kwargs)

        return generator

__init__(app, methods_to_instrument=None, **kwargs)

Wrap a custom class for recording.

• app: Any -- the custom app object being wrapped.
• More args in App
• More args in AppDefinition
• More args in WithClassInfo

Source code in trulens_eval/trulens_eval/tru_custom_app.py

def __init__(self, app: Any, methods_to_instrument=None, **kwargs):
    """
    Wrap a custom class for recording.

    Arguments:
    - app: Any -- the custom app object being wrapped.
    - More args in App
    - More args in AppDefinition
    - More args in WithClassInfo
    """

    kwargs['app'] = app
    kwargs['root_class'] = Class.of_object(app)

    kwargs['instrument'] = Instrument(
        app=self  # App mixes in WithInstrumentCallbacks
    )

    super().__init__(**kwargs)

    methods_to_instrument = methods_to_instrument or dict()

    # The rest of this code instruments methods explicitly passed to
    # constructor as needing instrumentation and checks that methods
    # decorated with @instrument or passed explicitly belong to some
    # component as per serialized version of this app. If they are not,
    # placeholders are made in `app_extra_json` so that subsequent
    # serialization looks like the components exist.
    json = self.dict()

    for m, path in methods_to_instrument.items():
        method_name = m.__name__

        full_path = JSONPath().app + path

        self.instrument.instrument_method(
            method_name=method_name, obj=m.__self__, query=full_path
        )

        # TODO: DEDUP with next condition

        # Check whether the path/location of the method is in json serialization and
        # if not, add a placeholder to app_extra_json.
        try:
            next(full_path(json))

            print(
                f"{UNICODE_CHECK} Added method {m.__name__} under component at path {full_path}"
            )

        except Exception:
            logger.warning(
                f"App has no component at path {full_path} . "
                f"Specify the component with the `app_extra_json` argument to TruCustomApp constructor. "
                f"Creating a placeholder there for now."
            )

            path.set(
                self.app_extra_json, {
                    PLACEHOLDER:
                        "I was automatically added to `app_extra_json` because there was nothing here to refer to an instrumented method owner.",
                    m.__name__:
                        f"Placeholder for method {m.__name__}."
                }
            )

    # Check that any functions marked with `TruCustomApp.instrument` has been
    # instrumented as a method under some object.
    for f in TruCustomApp.functions_to_instrument:
        obj_ids_methods_and_full_paths = list(self._get_methods_for_func(f))

        if len(obj_ids_methods_and_full_paths) == 0:
            logger.warning(
                f"Function {f} was not found during instrumentation walk. "
                f"Make sure it is accessible by traversing app {app} "
                f"or provide a bound method for it as TruCustomApp constructor argument `methods_to_instrument`."
            )

        else:
            for obj_id, m, full_path in obj_ids_methods_and_full_paths:
                try:
                    next(full_path(json))

                except Exception as e:
                    logger.warning(
                        f"App has no component owner of instrumented method {m} at path {full_path}. "
                        f"Specify the component with the `app_extra_json` argument to TruCustomApp constructor. "
                        f"Creating a placeholder there for now."
                    )

                    path.set(
                        self.app_extra_json, {
                            PLACEHOLDER:
                                "I was automatically added to `app_extra_json` because there was nothing here to refer to an instrumented method owner.",
                            m.__name__:
                                f"Placeholder for method {m.__name__}."
                        }
                    )

    # DB stuff and checks:
    self.post_init()

instrument

Bases: base_instrument

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

Source code in trulens_eval/trulens_eval/tru_custom_app.py

class instrument(base_instrument):
    """
    Decorator for marking methods to be instrumented in custom classes that are
    wrapped by TruCustomApp.
    """

    @classmethod
    def method(self_class, cls: type, name: str) -> None:
        base_instrument.method(cls, name)

        # Also make note of it for verification that it was found by the walk
        # after init.
        TruCustomApp.functions_to_instrument.add(getattr(cls, name))