trulens.apps.nemo¶

trulens.apps.nemo ¶

Additional Dependency Required

To use this module, you must have the trulens-apps-nemo package installed.

pip install trulens-apps-nemo

Classes¶

RailsActionSelect ¶

Bases: Select

Selector shorthands for NeMo Guardrails apps when used for evaluating feedback in actions.

These should not be used for feedback functions given to TruRails but instead for selectors in the FeedbackActions action invoked from with a rails app.

Attributes¶

Tru `class-attribute` `instance-attribute` ¶

Tru: Lens = Lens()

Selector for the tru wrapper (TruLlama, TruChain, etc.).

Record `class-attribute` `instance-attribute` ¶

Record: Lens = __record__

Selector for the record.

App `class-attribute` `instance-attribute` ¶

App: Lens = __app__

Selector for the app.

RecordInput `class-attribute` `instance-attribute` ¶

RecordInput: Lens = main_input

Selector for the main app input.

RecordOutput `class-attribute` `instance-attribute` ¶

RecordOutput: Lens = main_output

Selector for the main app output.

RecordCalls `class-attribute` `instance-attribute` ¶

RecordCalls: Lens = app

Selector for the calls made by the wrapped app.

Laid out by path into components.

RecordCall `class-attribute` `instance-attribute` ¶

RecordCall: Lens = calls[-1]

Selector for the first called method (last to return).

RecordArgs `class-attribute` `instance-attribute` ¶

RecordArgs: Lens = args

Selector for the whole set of inputs/arguments to the first called / last method call.

RecordRets `class-attribute` `instance-attribute` ¶

RecordRets: Lens = rets

Selector for the whole output of the first called / last returned method call.

_PREFIXES `class-attribute` `instance-attribute` ¶

_PREFIXES = [
    ("Select.RecordInput", RecordInput),
    ("Select.RecordOutput", RecordOutput),
    ("Select.RecordArgs", RecordArgs),
    ("Select.RecordRets", RecordRets),
    ("Select.RecordCalls", RecordCalls),
    ("Select.RecordCall", RecordCall),
    ("Select.Record", Record),
    ("Select.App", App),
]

All prefixes/shorthands defined in this class.

Make sure this list is sorted by longest prefix first as some prefixes are prefixes of others.

Action `class-attribute` `instance-attribute` ¶

Action = action

Selector for action call parameters.

Events `class-attribute` `instance-attribute` ¶

Events = events

Selector for events in action call parameters.

Context `class-attribute` `instance-attribute` ¶

Context = context

Selector for context in action call parameters.

Warning

This is not the same "context" as in RAG triad. This is a parameter to rails actions that stores context of the rails app execution.

LLM `class-attribute` `instance-attribute` ¶

LLM = llm

Selector for the language model in action call parameters.

Config `class-attribute` `instance-attribute` ¶

Config = config

Selector for the configuration in action call parameters.

RetrievalContexts `class-attribute` `instance-attribute` ¶

RetrievalContexts = relevant_chunks_sep

Selector for the retrieved contexts chunks returned from a KB search.

Equivalent to $relevant_chunks_sep in colang.

UserMessage `class-attribute` `instance-attribute` ¶

UserMessage = user_message

Selector for the user message.

Equivalent to $user_message in colang.

BotMessage `class-attribute` `instance-attribute` ¶

BotMessage = bot_message

Selector for the bot message.

Equivalent to $bot_message in colang.

LastUserMessage `class-attribute` `instance-attribute` ¶

LastUserMessage = last_user_message

Selector for the last user message.

Equivalent to $last_user_message in colang.

LastBotMessage `class-attribute` `instance-attribute` ¶

LastBotMessage = last_bot_message

Selector for the last bot message.

Equivalent to $last_bot_message in colang.

Functions¶

path_and_method `staticmethod` ¶

path_and_method(select: Lens) -> Tuple[Lens, str]

If select names in method as the last attribute, extract the method name and the selector without the final method name.

dequalify `staticmethod` ¶

dequalify(lens: Lens) -> Lens

If the given selector qualifies record or app, remove that qualification.

context `staticmethod` ¶

context(app: Optional[Any] = None) -> Lens

DEPRECATED: Select the context (retrieval step outputs) of the given app.

for_record `staticmethod` ¶

for_record(lens: Lens) -> Lens

Add the Record prefix to the beginning of the given lens.

for_app `staticmethod` ¶

for_app(lens: Lens) -> Lens

Add the App prefix to the beginning of the given lens.

render_for_dashboard `staticmethod` ¶

render_for_dashboard(lens: Lens) -> str

Render the given lens for use in dashboard to help user specify feedback functions.

RailsInstrument ¶

Bases: Instrument

Instrumentation specification for NeMo Guardrails apps.

Attributes¶

INSTRUMENT `class-attribute` `instance-attribute` ¶

INSTRUMENT = '__tru_instrumented'

Attribute name to be used to flag instrumented objects/methods/others.

APPS `class-attribute` `instance-attribute` ¶

APPS = '__tru_apps'

Attribute name for storing apps that expect to be notified of calls.

Classes¶

Default ¶

Default instrumentation specification.

Attributes¶

MODULES class-attribute instance-attribute ¶

MODULES = union(MODULES)

Modules to instrument by name prefix.

Note that NeMo Guardrails uses LangChain internally for some things.

CLASSES class-attribute instance-attribute ¶

CLASSES = lambda: union(CLASSES())

Instrument only these classes.

METHODS class-attribute instance-attribute ¶

METHODS: List[InstrumentedMethod] = METHODS + [
    InstrumentedMethod("execute_action", ActionDispatcher),
    InstrumentedMethod("generate", LLMRails),
    InstrumentedMethod("generate_async", LLMRails),
    InstrumentedMethod("stream_async", LLMRails),
    InstrumentedMethod("generate_events", LLMRails),
    InstrumentedMethod("generate_events_async", LLMRails),
    InstrumentedMethod(
        "_get_events_for_messages", LLMRails
    ),
    InstrumentedMethod(
        "search_relevant_chunks", KnowledgeBase
    ),
    InstrumentedMethod(
        "generate_user_intent", LLMGenerationActions
    ),
    InstrumentedMethod(
        "generate_next_step", LLMGenerationActions
    ),
    InstrumentedMethod(
        "generate_bot_message", LLMGenerationActions
    ),
    InstrumentedMethod(
        "generate_value", LLMGenerationActions
    ),
    InstrumentedMethod(
        "generate_intent_steps_message",
        LLMGenerationActions,
    ),
    InstrumentedMethod("feedback", FeedbackActions),
]

Instrument only methods with these names and of these classes.

Functions¶

print_instrumentation ¶

print_instrumentation() -> None

Print out description of the modules, classes, methods this class will instrument.

to_instrument_object ¶

to_instrument_object(obj: object) -> bool

Determine whether the given object should be instrumented.

to_instrument_class ¶

to_instrument_class(cls: type) -> bool

Determine whether the given class should be instrumented.

to_instrument_module ¶

to_instrument_module(module_name: str) -> bool

Determine whether a module with the given (full) name should be instrumented.

_have_context `staticmethod` ¶

_have_context() -> bool

Determine whether context vars we need for recording are available.

tracked_method_wrapper ¶

tracked_method_wrapper(
    query: Lens,
    func: Callable,
    method_name: str,
    cls: type,
    obj: object,
    span_type: Optional[SpanType] = None,
    span_attributes: Optional[Attributes] = None,
    full_scoped_span_attributes: Optional[
        Attributes
    ] = None,
    must_be_first_wrapper: bool = False,
)

Wrap a method to capture its inputs/outputs/errors.

instrument_method ¶

instrument_method(method_name: str, obj: Any, query: Lens)

Instrument a method.

instrument_class ¶

instrument_class(cls)

Instrument the given class cls's new method.

This is done so we can be aware when new instances are created and is needed for wrapped methods that dynamically create instances of classes we wish to instrument. As they will not be visible at the time we wrap the app, we need to pay attention to new to make a note of them when they are created and the creator's path. This path will be used to place these new instances in the app json structure.

instrument_object ¶

instrument_object(
    obj, query: Lens, done: Optional[Set[int]] = None
)

Instrument the given object obj and its components.

TruRails ¶

Bases: App

Recorder for apps defined using NeMo Guardrails.

PARAMETER	DESCRIPTION
`app`	A NeMo Guardrails application. TYPE: `LLMRails`

Attributes¶

_context_contexts `class-attribute` `instance-attribute` ¶

_context_contexts = ContextVar(
    "context_contexts", default=set()
)

ContextVars for storing collections of RecordingContext .

_stack_contexts `class-attribute` `instance-attribute` ¶

_stack_contexts = ContextVar('stack_contexts', default={})

ContextVars for storing call stacks.

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.

session `class-attribute` `instance-attribute` ¶

session: TruSession = Field(
    default_factory=TruSession, exclude=True
)

Session for this app.

connector `property` ¶

connector: DBConnector

Database connector.

db `property` ¶

db: DB

Database used by this app.

main_method_name `class-attribute` `instance-attribute` ¶

main_method_name: Optional[str] = Field(None)

Name of the main method of the app to be recorded. For serialization and this is required for OTEL.

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 1if the expected record content cannot be determined before it is produced.

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.

wrap_lazy_values ¶

wrap_lazy_values(
    rets: Any,
    wrap: Callable[[T], T],
    on_done: Callable[[T], T],
    context_vars: Optional[ContextVarsOrValues],
) -> Any

Wrap any lazy values in the return value of a method call to invoke handle_done when the value is ready.

This is used to handle library-specific lazy values that are hidden in containers not visible otherwise. Visible lazy values like iterators, generators, awaitables, and async generators are handled elsewhere.

PARAMETER	DESCRIPTION
`rets`	The return value of the method call. TYPE: `Any`
`wrap`	A callback to be called when the lazy value is ready. Should return the input value or a wrapped version of it. TYPE: `Callable[[T], T]`
`on_done`	Called when the lazy values is done and is no longer lazy. This as opposed to a lazy value that evaluates to another lazy values. Should return the value or wrapper. TYPE: `Callable[[T], T]`
`context_vars`	The contextvars to be captured by the lazy value. If not given, all contexts are captured. TYPE: `Optional[ContextVarsOrValues]`

RETURNS	DESCRIPTION
`Any`	The return value with lazy values wrapped.

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,
    final: bool = False,
) -> 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).

_submit_feedback_functions `staticmethod` ¶

_submit_feedback_functions(
    record: Record,
    feedback_functions: Sequence[Feedback],
    connector: DBConnector,
    app: Optional[AppDefinition] = None,
    on_done: Optional[
        Callable[
            [Union[FeedbackResult, Future[FeedbackResult]]],
            None,
        ]
    ] = None,
) -> List[Tuple[Feedback, Future[FeedbackResult]]]

Schedules to run the given feedback functions.

PARAMETER	DESCRIPTION
`record`	The record on which to evaluate the feedback functions. TYPE: `Record`
`feedback_functions`	A collection of feedback functions to evaluate. TYPE: `Sequence[Feedback]`
`connector`	The database connector to use. TYPE: `DBConnector`
`app`	The app that produced the given record. If not provided, it is looked up from the database of this `TruSession` instance TYPE: `Optional[AppDefinition]` DEFAULT: `None`
`on_done`	A callback to call when each feedback function is done. TYPE: `Optional[Callable[[Union[FeedbackResult, Future[FeedbackResult]]], None]]` DEFAULT: `None`

Returns:

List[Tuple[Feedback, Future[FeedbackResult]]]

Produces a list of tuples where the first item in each tuple is the
feedback function and the second is the future of the feedback result.

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.

del ¶

__del__()

Shut down anything associated with this app that might persist otherwise.

_start_manage_pending_feedback_results ¶

_start_manage_pending_feedback_results() -> None

Start the thread that manages the queue of records with pending feedback results.

This is meant to be run permanently in a separate thread. It will remove records from the set records_with_pending_feedback_results as their feedback results are computed.

_manage_pending_feedback_results `staticmethod` ¶

_manage_pending_feedback_results(
    self_proxy: ProxyType[App],
) -> None

Manage the queue of records with pending feedback results.

This is meant to be run permanently in a separate thread. It will remove records from the set records_with_pending_feedback_results as their feedback results are computed.

wait_for_feedback_results ¶

wait_for_feedback_results(
    feedback_timeout: Optional[float] = None,
) -> Iterable[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
`Iterable[Record]`	An iterable 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.

_tru_post_init ¶

_tru_post_init()

Database-related initialization and additional data checks.

DB

Insert the app into the database.
Insert feedback function definitions into the database.

Checks

In deferred mode, try to serialize and deserialize feedback functions.
Check that feedback function selectors are likely to refer to expected app or record components.

main_call ¶

main_call(human: str) -> str

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

main_acall `async` ¶

main_acall(human: str) -> str

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

_extract_content ¶

_extract_content(value, content_keys=['content'])

Extracts the 'content' from various data types commonly used by libraries like OpenAI, Canopy, LiteLLM, etc. This method navigates nested data structures (pydantic models, dictionaries, lists) to retrieve the 'content' field. If 'content' is not directly available, it attempts to extract from known structures like 'choices' in a ChatResponse. This standardizes extracting relevant text or data from complex API responses or internal data representations.

PARAMETER	DESCRIPTION
`value`	The input data to extract content from. Can be a pydantic model, dictionary, list, or basic data type.

RETURNS	DESCRIPTION
	The extracted content, which may be a single value, a list of values,
	or a nested structure with content extracted from all levels.

json ¶

json(*args, **kwargs)

Create a json string representation of this app.

_check_instrumented ¶

_check_instrumented(func)

Issue a warning and some instructions if a function that has not been instrumented is being used in a with_ call.

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.

_add_future_feedback ¶

_add_future_feedback(
    future_or_result: Union[
        FeedbackResult, Future[FeedbackResult]
    ]
) -> None

Callback used to add feedback results to the database once they are done.

See _handle_record.

_handle_record ¶

_handle_record(
    record: Record,
    feedback_mode: Optional[FeedbackMode] = None,
) -> Optional[
    List[Tuple[Feedback, Future[FeedbackResult]]]
]

Write out record-related info to database if set and schedule feedback functions to be evaluated. If feedback_mode is provided, will use that mode instead of the one provided to constructor.

dummy_record ¶

dummy_record(
    cost: Cost = Cost(),
    perf: Perf = now(),
    ts: 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.

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.

main_input ¶

main_input(
    func: Callable, sig: Signature, bindings: BoundArguments
) -> JSON

Determine the main input string for the given function func with signature sig after it is called with the given bindings and has returned ret.

select_context `classmethod` ¶

select_context(app: Optional[LLMRails] = None) -> Lens

Get the path to the context in the query output.

trulens.apps.nemo¶