Tru Custom App¶
trulens_eval.tru_custom_app.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
.
Using the @instrument
decorator
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()
Using instrument.method
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 ..."
)
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>
Defining feedback functions with instrumented methods
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.
Using the TruCustomApp
recorder
from trulens_eval import TruCustomApp
tru_recorder = TruCustomApp(custom_app,
app_id="Custom Application v1",
feedbacks=[f_context_relevance])
with tru_recorder as recording:
custom_app.respond_to_query("What is the capital of Indonesia?")
See Feedback Functions for instantiating feedback functions.
PARAMETER | DESCRIPTION |
---|---|
app |
Any class.
TYPE:
|
**kwargs |
Additional arguments to pass to App and AppDefinition
TYPE:
|
Attributes¶
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.
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
¶
Feedback functions to evaluate on each record.
tru
class-attribute
instance-attribute
¶
Workspace manager.
If this is not povided, a singleton Tru will be made (if not already) and used.
db
class-attribute
instance-attribute
¶
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
¶
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
¶
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 of the custom app.
main_method
class-attribute
instance-attribute
¶
Serialized version of the main method.
Functions¶
on_method_instrumented
¶
Called by instrumentation system for every function requested to be instrumented by this app.
get_method_path
¶
Get the path of the instrumented function method
relative to this app.
get_methods_for_func
¶
Get the methods (rather the inner functions) matching the given func
and the path of each.
on_new_record
¶
on_new_record(func) -> Iterable[RecordingContext]
Called at the start of record creation.
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.
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:
|
app |
The app to continue the session with.
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
AppDefinition
|
A new |
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.
wait_for_feedback_results
¶
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
¶
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
¶
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
.
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
.
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
¶
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
¶
Iteration over instrumented components and their categories.
format_instrumented_methods
¶
format_instrumented_methods() -> str
Build a string containing a listing of instrumented methods.
print_instrumented_components
¶
print_instrumented_components() -> None
Print instrumented components and their categories.