App(Definition)¶
Apps in trulens derive from two classes, AppDefinition and App. The first contains only serialized or serializable components in a JSON-like format while the latter contains the executable apps that may or may not be serializable.
Bases: WithClassInfo, SerialModel
Serialized fields of an app here whereas App contains non-serialized fields.
Attributes¶
class-attribute
instance-attribute
¶
feedback_definitions: Sequence[FeedbackDefinition] = []
Feedback functions to evaluate on each record.
class-attribute
instance-attribute
¶
feedback_mode: FeedbackMode = WITH_APP_THREAD
How to evaluate feedback functions upon producing a record.
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.
class-attribute
¶
root_callable: FunctionOrMethod
App's main method.
This is to be filled in by subclass.
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.
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.
Functions¶
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 |
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).
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.
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¶
class-attribute
instance-attribute
¶
Feedback functions to evaluate on each record.
class-attribute
instance-attribute
¶
Workspace manager.
If this is not povided, a singleton Tru will be made (if not already) and used.
class-attribute
instance-attribute
¶
Database interface.
If this is not provided, a singleton SQLAlchemyDB will be made (if not already) and used.
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.
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.
class-attribute
instance-attribute
¶
Mapping of instrumented methods (by id(.) of owner object and the function) to their path in this app.
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.
class-attribute
instance-attribute
¶
Thread for manager of pending feedback results queue.
See _manage_pending_feedback_results.
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.
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¶
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.
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.
If available, a single text to a single text invocation of this app.
async
¶
If available, a single text to a single text invocation of this app.
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(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.
Called by instrumentation system for every function requested to be instrumented by this app.
Get the methods (rather the inner functions) matching the given func
and the path of each.
Get the path of the instrumented function method relative to this app.
on_new_record(func) -> Iterable[RecordingContext]
Called at the start of record creation.
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.
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.
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.
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_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.
Iteration over instrumented components and their categories.
format_instrumented_methods() -> str
Build a string containing a listing of instrumented methods.
print_instrumented_components() -> None
Print instrumented components and their categories.
Manager of the creation of records from record calls.
An instance of this class is produced when using an App as a context mananger, 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¶
instance-attribute
¶
calls: List[RecordAppCall] = []
A record (in terms of its RecordAppCall) in process of being created.
instance-attribute
¶
Lock blocking access to calls and records when adding calls or finishing a record.
instance-attribute
¶
record_metadata = record_metadata
Metadata to attach to all records produced in this context.