trulens.core.session¶

trulens.core.session ¶

Classes¶

TruSession ¶

Bases: _WithExperimentalSettings, PydanticSingleton

TruSession is the main class that provides an entry points to trulens.

TruSession lets you:

Log app prompts and outputs
Log app Metadata
Run and log feedback functions
Run streamlit dashboard to view experiment results

By default, all data is logged to the current working directory to "default.sqlite". Data can be logged to a SQLAlchemy-compatible url referred to by database_url.

Supported App Types

TruChain: Langchain apps.

TruLlama: Llama Index apps.

TruRails: NeMo Guardrails apps.

TruBasicApp: Basic apps defined solely using a function from str to str.

TruApp: Custom apps containing custom structures and methods. Requires annotation of methods to instrument.

TruVirtual: Virtual apps that do not have a real app to instrument but have a virtual structure and can log existing captured data as if they were trulens records.

PARAMETER	DESCRIPTION
`connector`	Database Connector to use. If not provided, a default DefaultDBConnector is created. TYPE: `Optional[DBConnector]` DEFAULT: `None`
`experimental_feature_flags`	Experimental feature flags. TYPE: `Optional[Union[Mapping[Feature, bool], List[Feature]]]` DEFAULT: `None`
`otel_exporter`	OpenTelemetry exporter to use. Set to `"otlp"` to export TruLens traces and GenAI metrics using OTLP. If omitted, the existing TruLens/Snowflake exporter remains the default. TYPE: `Optional[str]` DEFAULT: `None`
`otlp_endpoint`	Optional OTLP endpoint. This is only valid with `otel_exporter="otlp"`. If omitted, the standard OpenTelemetry environment variables are used. TYPE: `Optional[str]` DEFAULT: `None`
`otlp_protocol`	Optional OTLP transport protocol, `"grpc"` or `"http/protobuf"`. This is only valid with `otel_exporter="otlp"`. If omitted, the standard `OTEL_EXPORTER_OTLP_PROTOCOL` environment variable is honored; when that is also unset, `"grpc"` is used for backward compatibility. TYPE: `Optional[str]` DEFAULT: `None`
`span_exporter`	An explicit `SpanExporter` instance for sending traces. Use this when you need full control over the exporter (for example, exporting to a Snowflake AI Gateway). Cannot be combined with `otel_exporter="otlp"`. TYPE: `Optional[SpanExporter]` DEFAULT: `None`
`**kwargs`	All other arguments are used to initialize DefaultDBConnector. Mutually exclusive with `connector`. TYPE: `Any` DEFAULT: `{}`

Attributes¶

RETRY_RUNNING_SECONDS `class-attribute` `instance-attribute` ¶

RETRY_RUNNING_SECONDS: float = 60.0

How long to wait (in seconds) before restarting a feedback function that has already started

A feedback function execution that has started may have stalled or failed in a bad way that did not record the failure.

RETRY_FAILED_SECONDS `class-attribute` `instance-attribute` ¶

RETRY_FAILED_SECONDS: float = 5 * 60.0

How long to wait (in seconds) to retry a failed feedback function run.

DEFERRED_NUM_RUNS `class-attribute` `instance-attribute` ¶

DEFERRED_NUM_RUNS: int = 32

Number of futures to wait for when evaluating deferred feedback functions.

RECORDS_BATCH_TIMEOUT_IN_SEC `class-attribute` `instance-attribute` ¶

RECORDS_BATCH_TIMEOUT_IN_SEC: int = 10

Time to wait before inserting a batch of records into the database.

GROUND_TRUTHS_BATCH_SIZE `class-attribute` `instance-attribute` ¶

GROUND_TRUTHS_BATCH_SIZE: int = 100

Time to wait before inserting a batch of ground truths into the database.

connector `class-attribute` `instance-attribute` ¶

connector: Optional[DBConnector] = Field(None, exclude=True)

Database Connector to use. If not provided, a default is created and used.

experimental_otel_exporter `property` ¶

experimental_otel_exporter: Optional[SpanExporter]

EXPERIMENTAL(otel_tracing): OpenTelemetry SpanExporter to send spans to.

Only works if the trulens.core.experimental.Feature.OTEL_TRACING flag is set. The setter will set and lock the flag as enabled.

sampling_controller `property` `writable` ¶

sampling_controller

The active :class:SamplingController, or None.

prompt_label_cache `property` ¶

prompt_label_cache: LabelCache

Process-local time-to-live cache of prompt label to exact version.

Disabled by default. Set session.prompt_label_cache.ttl to a positive number of seconds to reduce label reads. Correctness never depends on it: exact-version lookups bypass it entirely and every write invalidates the label it touched.

Functions¶

force_flush ¶

force_flush(timeout_millis: int = 300000) -> bool

Force flush the OpenTelemetry exporters.

PARAMETER	DESCRIPTION
`timeout_millis`	The maximum amount of time to wait for spans and metrics to be processed. TYPE: `int` DEFAULT: `300000`

RETURNS	DESCRIPTION
`bool`	False if the timeout is exceeded, feature is not enabled, or the
`bool`	provider doesn't exist, True otherwise.

configure_online_eval ¶

configure_online_eval(
    sample_rate: Union[float, Dict[str, float]] = 1.0,
    throttle: Optional[int] = None,
    cost_budget: Optional[float] = None,
    feedbacks: Optional[Sequence[Metric]] = None,
) -> None

Configure sampling for automatic post-ingest evaluation.

A second call replaces the previous configuration entirely.

This does not affect explicit compute_metrics() / compute_now() calls -- those always evaluate everything.

PARAMETER	DESCRIPTION
`sample_rate`	Probability (0--1) that a record is evaluated, or a `{app_name: rate}` dict for per-app rates. TYPE: `Union[float, Dict[str, float]]` DEFAULT: `1.0`
`throttle`	Max evaluations per minute (`None` = unlimited). TYPE: `Optional[int]` DEFAULT: `None`
`cost_budget`	Daily USD cap (`None` = unlimited). Only enforceable for providers whose `reports_costs` property is `True`. TYPE: `Optional[float]` DEFAULT: `None`
`feedbacks`	Optional list of metrics/feedbacks whose providers should be checked for cost-tracking support when `cost_budget` is set. If omitted, the warning is generic. TYPE: `Optional[Sequence[Metric]]` DEFAULT: `None`

App ¶

App(*args, app: Optional[Any] = None, **kwargs) -> App

Create an App from the given App constructor arguments by guessing which app type they refer to.

This method intentionally prints out the type of app being created to let user know in case the guess is wrong.

Basic ¶

Basic(*args, **kwargs) -> App

Deprecated

Use trulens.core.session.TruSession.App instead.

Virtual ¶

Virtual(*args, **kwargs) -> App

Deprecated

Use trulens.core.session.TruSession.App instead.

Chain ¶

Chain(*args, **kwargs) -> App

Deprecated

Use trulens.core.session.TruSession.App instead.

Llama ¶

Llama(*args, **kwargs) -> App

Deprecated

Use trulens.core.session.TruSession.App instead.

Rails ¶

Rails(*args, **kwargs) -> App

Deprecated

Use trulens.core.session.TruSession.App instead.

find_unused_port ¶

find_unused_port(*args, **kwargs)

Deprecated

Use trulens.dashboard.run.find_unused_port instead.

run_dashboard ¶

run_dashboard(*args, **kwargs)

Deprecated

Use trulens.dashboard.run.run_dashboard instead.

start_dashboard ¶

start_dashboard(*args, **kwargs)

Deprecated

Use trulens.dashboard.run.run_dashboard instead.

stop_dashboard ¶

stop_dashboard(*args, **kwargs)

Deprecated

Use trulens.dashboard.run.stop_dashboard instead.

update_record ¶

update_record(*args, **kwargs)

Deprecated

Use trulens.core.session.TruSession.connector .db.insert_record instead.

reset_database ¶

reset_database()

Reset the database. Clears all tables.

See DB.reset_database.

migrate_database ¶

migrate_database(**kwargs: Dict[str, Any])

Migrates the database.

This should be run whenever there are breaking changes in a database created with an older version of trulens.

PARAMETER	DESCRIPTION
`**kwargs`	Keyword arguments to pass to migrate_database of the current database. TYPE: `Dict[str, Any]` DEFAULT: `{}`

See DB.migrate_database.

add_record ¶

add_record(
    record: Optional[Record] = None, **kwargs: dict
) -> RecordID

Add a record to the database.

PARAMETER	DESCRIPTION
`record`	The record to add. TYPE: `Optional[Record]` DEFAULT: `None`
`**kwargs`	Record fields to add to the given record or a new record if no `record` provided. TYPE: `dict` DEFAULT: `{}`

RETURNS	DESCRIPTION
`RecordID`	Unique record identifier `str`.

add_record_nowait ¶

add_record_nowait(record: Record) -> None

Add a record to the queue to be inserted in the next batch.

run_feedback_functions ¶

run_feedback_functions(
    record: Record,
    feedback_functions: Sequence[Metric],
    app: Optional[AppDefinition] = None,
    wait: bool = True,
) -> Union[
    Iterable[FeedbackResult],
    Iterable[Future[FeedbackResult]],
]

Run a collection of feedback functions and report their result.

PARAMETER	DESCRIPTION
`record`	The record on which to evaluate the feedback functions. TYPE: `Record`
`app`	The app that produced the given record. If not provided, it is looked up from the given database `db`. TYPE: `Optional[AppDefinition]` DEFAULT: `None`
`feedback_functions`	A collection of feedback functions to evaluate. TYPE: `Sequence[Metric]`
`wait`	If set (default), will wait for results before returning. TYPE: `bool` DEFAULT: `True`

YIELDS	DESCRIPTION
`Union[Iterable[FeedbackResult], Iterable[Future[FeedbackResult]]]`	One result for each element of `feedback_functions` of FeedbackResult if `wait` is enabled (default) or `concurrent.futures.Future` of FeedbackResult if `wait` is disabled.

add_app ¶

add_app(app: AppDefinition) -> AppID

Add an app to the database and return its unique id.

PARAMETER	DESCRIPTION
`app`	The app to add to the database. TYPE: `AppDefinition`

RETURNS	DESCRIPTION
`AppID`	A unique app identifier `str`.

delete_app ¶

delete_app(app_id: AppID) -> None

Deletes an app from the database based on its app_id.

PARAMETER	DESCRIPTION
`app_id`	The unique identifier of the app to be deleted. TYPE: `AppID`

add_feedback ¶

add_feedback(
    feedback_result_or_future: Optional[
        Union[FeedbackResult, Future[FeedbackResult]]
    ] = None,
    **kwargs: dict
) -> FeedbackResultID

Add a single feedback result or future to the database and return its unique id.

PARAMETER	DESCRIPTION
`feedback_result_or_future`	If a `concurrent.futures.Future` is given, call will wait for the result before adding it to the database. If `kwargs` are given and a FeedbackResult is also given, the `kwargs` will be used to update the FeedbackResult otherwise a new one will be created with `kwargs` as arguments to its constructor. TYPE: `Optional[Union[FeedbackResult, Future[FeedbackResult]]]` DEFAULT: `None`
`**kwargs`	Fields to add to the given feedback result or to create a new FeedbackResult with. TYPE: `dict` DEFAULT: `{}`

RETURNS	DESCRIPTION
`FeedbackResultID`	A unique result identifier `str`.

add_feedbacks ¶

add_feedbacks(
    feedback_results: Iterable[
        Union[FeedbackResult, Future[FeedbackResult]]
    ],
) -> List[FeedbackResultID]

Add multiple feedback results to the database and return their unique ids.

PARAMETER	DESCRIPTION
`feedback_results`	An iterable with each iteration being a FeedbackResult or `concurrent.futures.Future` of the same. Each given future will be waited. TYPE: `Iterable[Union[FeedbackResult, Future[FeedbackResult]]]`

RETURNS	DESCRIPTION
`List[FeedbackResultID]`	List of unique result identifiers `str` in the same order as input `feedback_results`.

get_app ¶

get_app(app_id: AppID) -> Optional[JSONized[AppDefinition]]

Look up an app from the database.

This method produces the JSON-ized version of the app. It can be deserialized back into an AppDefinition with model_validate:

Example

from trulens.core.schema import app
app_json = session.get_app(app_id="app_hash_85ebbf172d02e733c8183ac035d0cbb2")
app = app.AppDefinition.model_validate(app_json)

Warning

Do not rely on deserializing into App as its implementations feature attributes not meant to be deserialized.

PARAMETER	DESCRIPTION
`app_id`	The unique identifier `str` of the app to look up. TYPE: `AppID`

RETURNS	DESCRIPTION
`Optional[JSONized[AppDefinition]]`	JSON-ized version of the app.

get_apps ¶

get_apps() -> List[JSONized[AppDefinition]]

Look up all apps from the database.

RETURNS	DESCRIPTION
`List[JSONized[AppDefinition]]`	A list of JSON-ized version of all apps in the database.

Warning

Same Deserialization caveats as get_app.

get_conversations ¶

get_conversations(
    app_id: Optional[AppID] = None,
) -> Dict[ConversationID, Conversation]

Get conversations reconstructed from recorded spans.

get_records_by_conversation ¶

get_records_by_conversation(
    conversation_id: ConversationID,
    app_id: Optional[AppID] = None,
) -> List[Record]

Get a conversation's records ordered chronologically.

get_records_and_feedback ¶

get_records_and_feedback(
    app_ids: Optional[List[AppID]] = None,
    app_name: Optional[AppName] = None,
    app_version: Optional[AppVersion] = None,
    app_versions: Optional[List[AppVersion]] = None,
    run_name: Optional[RunName] = None,
    record_ids: Optional[List[RecordID]] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
) -> Tuple[DataFrame, List[str]]

Get records, their feedback results, and feedback names.

PARAMETER	DESCRIPTION
`app_ids`	A list of app ids to filter records by. If empty or not given, all apps' records will be returned. TYPE: `Optional[List[AppID]]` DEFAULT: `None`
`app_name`	A name of the app to filter records by. If given, only records for this app will be returned. TYPE: `Optional[AppName]` DEFAULT: `None`
`app_version`	A version of the app to filter records by. If given, only records for this app version will be returned. TYPE: `Optional[AppVersion]` DEFAULT: `None`
`app_versions`	A list of app versions to filter records by. If given, only records for these app versions will be returned. TYPE: `Optional[List[AppVersion]]` DEFAULT: `None`
`run_name`	A run name to filter records by. If given, only records for this run will be returned. TYPE: `Optional[RunName]` DEFAULT: `None`
`record_ids`	An optional list of record ids to filter records by. TYPE: `Optional[List[RecordID]]` DEFAULT: `None`
`offset`	Record row offset. TYPE: `Optional[int]` DEFAULT: `None`
`limit`	Limit on the number of records to return. TYPE: `Optional[int]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	Tuple of:
`List[str]`	DataFrame of records with their feedback results.
`Tuple[DataFrame, List[str]]`	List of feedback names that are columns in the DataFrame.

get_leaderboard ¶

get_leaderboard(
    app_ids: Optional[List[AppID]] = None,
    group_by_metadata_key: Optional[str] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
    format: str = "dataframe",
) -> Union[DataFrame, str, List[Dict[str, Any]]]

Get a leaderboard for the given apps.

PARAMETER	DESCRIPTION
`app_ids`	A list of app ids to filter records by. If empty or not given, all apps will be included in leaderboard. TYPE: `Optional[List[AppID]]` DEFAULT: `None`
`group_by_metadata_key`	A key included in record metadata that you want to group results by. TYPE: `Optional[str]` DEFAULT: `None`
`limit`	Limit on the number of records to aggregate to produce the leaderboard. TYPE: `Optional[int]` DEFAULT: `None`
`offset`	Record row offset to select which records to use to aggregate the leaderboard. TYPE: `Optional[int]` DEFAULT: `None`
`format`	Output format for the leaderboard. Supported values are `"dataframe"` for the existing pandas DataFrame behavior, `"dict"` for `DataFrame.to_dict(orient="records")`, and `"json"` for `DataFrame.to_json(orient="records")`. TYPE: `str` DEFAULT: `'dataframe'`

RETURNS	DESCRIPTION
`Union[DataFrame, str, List[Dict[str, Any]]]`	Leaderboard data in the requested format. If `format` is
`Union[DataFrame, str, List[Dict[str, Any]]]`	`"dataframe"` and `group_by_metadata_key` is provided, the
`Union[DataFrame, str, List[Dict[str, Any]]]`	DataFrame will be grouped by the specified key.

add_ground_truth_to_dataset ¶

add_ground_truth_to_dataset(
    dataset_name: str,
    ground_truth_df: DataFrame,
    dataset_metadata: Optional[Dict[str, Any]] = None,
)

Create a new dataset, if not existing, and add ground truth data to it. If the dataset with the same name already exists, the ground truth data will be added to it.

PARAMETER	DESCRIPTION
`dataset_name`	Name of the dataset. TYPE: `str`
`ground_truth_df`	DataFrame containing the ground truth data. TYPE: `DataFrame`
`dataset_metadata`	Additional metadata to add to the dataset. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

curate_records_to_dataset ¶

curate_records_to_dataset(
    dataset_name: str,
    records: DataFrame,
    mapping: TraceDatasetMapping | None = None,
    expected_response_fn: (
        Callable[[Series], str | None] | None
    ) = None,
    dataset_metadata: dict[str, Any] | None = None,
    on_error: str = ON_ERROR_RAISE,
    batch_size: int = DEFAULT_BATCH_SIZE,
    include_provenance: bool = True,
) -> CurationResult

Curate recorded traces into a persisted evaluation dataset.

Turns a records dataframe into GroundTruth rows: it validates the mapping against the dataframe, resolves recorded inputs and outputs into stable text, normalizes expected contexts, preserves provenance in metadata, and writes in bounded batches. Ground truth ids are content-addressed, so curating the same rows twice does not create duplicate rows.

Accepted inputs:

a dataframe from get_records_and_feedback;
a dataframe carrying a record_id column but not the record content, which is resolved through the session before mapping;
a review export (CSV, JSON, ...) that the caller has loaded into a dataframe.

Example

result = session.curate_records_to_dataset(
    dataset_name="production-failures",
    records=records_df,
    mapping=TraceDatasetMapping(
        query="input",
        expected_response="corrected_output",
        metadata={"groundedness": "Groundedness"},
    ),
    on_error="collect",
)

PARAMETER	DESCRIPTION
`dataset_name`	Name of the dataset to write to. It is created if it does not exist. TYPE: `str`
`records`	The rows to curate. TYPE: `DataFrame`
`mapping`	Which dataframe columns supply which ground truth fields. Values are column names, never expressions. Defaults to `TraceDatasetMapping()`, which reads `input` and `record_id`. TYPE: `TraceDatasetMapping \| None` DEFAULT: `None`
`expected_response_fn`	Synchronous callback returning the expected response for a row. Only consulted when the mapped `expected_response` column is absent or empty for that row — a mapped correction always wins. TruLens never calls an LLM on your behalf here. TYPE: `Callable[[Series], str \| None] \| None` DEFAULT: `None`
`dataset_metadata`	Metadata for the dataset itself. TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`on_error`	`"raise"` to fail on the first unusable row, or `"collect"` to skip it and report it in the result. Malformed rows are never written in either mode. TYPE: `str` DEFAULT: `ON_ERROR_RAISE`
`batch_size`	Ground truths written per database round trip. Peak memory is one batch beyond the input dataframe. TYPE: `int` DEFAULT: `DEFAULT_BATCH_SIZE`
`include_provenance`	Whether to copy the source record id and app name/version into each ground truth's metadata. A ground truth id covers its metadata, so leaving this on means two records with identical content but different provenance stay distinct; turn it off to deduplicate purely on example content. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`CurationResult`	A CurationResult with the
`CurationResult`	accepted, duplicate and rejected counts, the ids written, and one
`CurationResult`	error row per rejection.

RAISES	DESCRIPTION
`ValueError`	If `on_error` or `batch_size` is invalid, or a mapped column is missing from `records`. Column validation happens before anything is written.
`CurationRowError`	In `"raise"` mode, on the first unusable row. Batches written before that point stay written; because ids are content-addressed, re-running after fixing the data is safe.

get_ground_truth ¶

get_ground_truth(
    dataset_name: Optional[str] = None,
    user_table_name: Optional[str] = None,
    user_schema_mapping: Optional[Dict[str, str]] = None,
    user_schema_name: Optional[str] = None,
) -> DataFrame

Get ground truth data from the dataset. If user_table_name and user_schema_mapping are provided, load a virtual dataset from the user's table using the schema mapping. If dataset_name is provided, load ground truth data from the dataset by name. dataset_name: Name of the dataset. user_table_name: Name of the user's table to load ground truth data from. user_schema_mapping: Mapping of user table columns to internal GroundTruth schema fields. user_schema_name: Name of the user's schema to load ground truth data from.

create_prompt ¶

create_prompt(
    slug: str,
    name: Optional[str] = None,
    prompt_type: Union[PromptType, str] = TEXT,
    description: Optional[str] = None,
    tags: Optional[Sequence[str]] = None,
) -> Prompt

Create a prompt, or update the metadata of an existing slug.

PARAMETER	DESCRIPTION
`slug`	Stable key such as `support-assistant`. TYPE: `str`
`name`	Display name. Defaults to the slug. TYPE: `Optional[str]` DEFAULT: `None`
`prompt_type`	`text` or `chat`. Fixed after creation. TYPE: `Union[PromptType, str]` DEFAULT: `TEXT`
`description`	Free-text description. TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Tags for grouping prompts. TYPE: `Optional[Sequence[str]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Prompt`	The stored Prompt.

create_prompt_version ¶

create_prompt_version(
    prompt: Union[Prompt, PromptID, str],
    text: Optional[str] = None,
    messages: Optional[Sequence[Dict[str, Any]]] = None,
    variables: Optional[Sequence[str]] = None,
    model_defaults: Optional[Dict[str, Any]] = None,
    response_format: Optional[Dict[str, Any]] = None,
    change_note: Optional[str] = None,
    parent_version_id: Optional[PromptVersionID] = None,
    created_by: Optional[str] = None,
) -> PromptVersion

Create an immutable prompt version and move latest onto it.

Creating the same content twice returns the same version.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`
`text`	Template string for a text prompt. TYPE: `Optional[str]` DEFAULT: `None`
`messages`	Ordered messages for a chat prompt. TYPE: `Optional[Sequence[Dict[str, Any]]]` DEFAULT: `None`
`variables`	Declared variable names. Inferred from the content when omitted. TYPE: `Optional[Sequence[str]]` DEFAULT: `None`
`model_defaults`	Provider-neutral model settings. Never credentials. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`response_format`	Response-format metadata for provider adapters. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`change_note`	Why the version was created. TYPE: `Optional[str]` DEFAULT: `None`
`parent_version_id`	The version this one derives from. Defaults to whatever `latest` currently points at. TYPE: `Optional[PromptVersionID]` DEFAULT: `None`
`created_by`	Who created the version. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`PromptVersion`	The stored
`PromptVersion`	PromptVersion.

set_prompt_label ¶

set_prompt_label(
    prompt: Union[Prompt, PromptID, str],
    label: str,
    version: Union[PromptVersion, PromptVersionID],
    moved_by: Optional[str] = None,
) -> PromptLabel

Point a label at one exact version and record the movement.

Rolling back is the same call with an older version. Application code that looks the label up needs no change.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`
`label`	The label name, for example `production`. TYPE: `str`
`version`	The version or version id to point at. TYPE: `Union[PromptVersion, PromptVersionID]`
`moved_by`	Caller label written to the history entry. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`PromptLabel`	The resulting
`PromptLabel`	PromptLabel.

get_prompt ¶

get_prompt(
    prompt: Union[Prompt, PromptID, str],
    label: Optional[str] = None,
    version_id: Optional[PromptVersionID] = None,
    use_cache: bool = True,
) -> ResolvedPrompt

Resolve a prompt to one exact version.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`
`label`	Label to resolve. Defaults to `latest` when no `version_id` is given. TYPE: `Optional[str]` DEFAULT: `None`
`version_id`	Exact version to load. Wins over `label`. TYPE: `Optional[PromptVersionID]` DEFAULT: `None`
`use_cache`	Set false to bypass the label cache for this call. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`ResolvedPrompt`	A ResolvedPrompt,
`ResolvedPrompt`	which renders locally without calling a model.

get_prompts ¶

get_prompts() -> DataFrame

Get every prompt in the database.

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with one row per prompt.

get_prompt_versions ¶

get_prompt_versions(
    prompt: Union[Prompt, PromptID, str],
) -> DataFrame

Get every version of a prompt, oldest first.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with one row per version.

get_prompt_labels ¶

get_prompt_labels(
    prompt: Union[Prompt, PromptID, str],
) -> DataFrame

Get every label of a prompt and the version it points at.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with one row per label.

get_prompt_label_history ¶

get_prompt_label_history(
    prompt: Union[Prompt, PromptID, str],
    label: Optional[str] = None,
) -> DataFrame

Get the append-only history of label movements, newest first.

PARAMETER	DESCRIPTION
`prompt`	A prompt, prompt id, or slug. TYPE: `Union[Prompt, PromptID, str]`
`label`	Restrict to one label when given. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with one row per movement.

start_evaluator ¶

start_evaluator(
    restart: bool = False,
    fork: bool = False,
    disable_tqdm: bool = False,
    run_location: Optional[FeedbackRunLocation] = None,
    return_when_done: bool = False,
) -> Optional[Union[Process, Thread]]

Start a deferred feedback function evaluation thread or process.

PARAMETER	DESCRIPTION
`restart`	If set, will stop the existing evaluator before starting a new one. TYPE: `bool` DEFAULT: `False`
`fork`	If set, will start the evaluator in a new process instead of a thread. NOT CURRENTLY SUPPORTED. TYPE: `bool` DEFAULT: `False`
`disable_tqdm`	If set, will disable progress bar logging from the evaluator. TYPE: `bool` DEFAULT: `False`
`run_location`	Run only the evaluations corresponding to run_location. TYPE: `Optional[FeedbackRunLocation]` DEFAULT: `None`
`return_when_done`	Instead of running asynchronously, will block until no feedbacks remain. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Optional[Union[Process, Thread]]`	If return_when_done is True, then returns None. Otherwise, the started process or thread that is executing the deferred feedback evaluator.

Relevant constants

RETRY_RUNNING_SECONDS

RETRY_FAILED_SECONDS

DEFERRED_NUM_RUNS

MAX_THREADS

stop_evaluator ¶

stop_evaluator()

Stop the deferred feedback evaluation thread.

wait_for_records ¶

wait_for_records(
    record_ids: List[str],
    timeout: float = 10,
    poll_interval: float = 0.5,
) -> None

Wait for specific record_ids to appear in the TruLens session.

PARAMETER	DESCRIPTION
`record_ids`	The record ids to wait for. TYPE: `List[str]`
`timeout`	Maximum time to wait in seconds. TYPE: `float` DEFAULT: `10`
`poll_interval`	How often to poll in seconds. TYPE: `float` DEFAULT: `0.5`

wait_for_feedback_results ¶

wait_for_feedback_results(
    record_ids: List[str],
    feedback_names: List[str],
    timeout: float = 60,
    poll_interval: float = 0.5,
) -> None

Wait for feedback evaluation results to appear in the database.

This method waits for EVAL_ROOT spans to appear for all combinations of the given record_ids and feedback_names. It should be called after feedback computation has been triggered (e.g., after compute_now).

PARAMETER	DESCRIPTION
`record_ids`	The record IDs to wait for feedback results for. TYPE: `List[str]`
`feedback_names`	The names of the feedback functions to wait for. TYPE: `List[str]`
`timeout`	Maximum time to wait in seconds. TYPE: `float` DEFAULT: `60`
`poll_interval`	How often to poll in seconds. TYPE: `float` DEFAULT: `0.5`

RAISES	DESCRIPTION
`RuntimeError`	If not all feedback results are found within the timeout.

add_feedback_result ¶

add_feedback_result(
    record: Record,
    feedback_name: str,
    feedback_result: Union[float, int],
    higher_is_better: bool,
) -> None

Add a feedback result for a given record.

PARAMETER	DESCRIPTION
`record`	The Record object to add feedback for. TYPE: `Record`
`feedback_name`	The name of the feedback function. TYPE: `str`
`feedback_result`	The feedback score/result (float or int). TYPE: `Union[float, int]`
`higher_is_better`	Whether higher values are better. TYPE: `bool`

compute_feedbacks_on_events ¶

compute_feedbacks_on_events(
    events: DataFrame,
    feedbacks: List[Metric],
    raise_error_on_no_feedbacks_computed: bool = False,
) -> None

Compute feedbacks/metrics on events.

PARAMETER	DESCRIPTION
`events`	Events to compute feedbacks on. This can be from multiple records. TYPE: `DataFrame`
`feedbacks`	Feedback functions to compute. TYPE: `List[Metric]`
`raise_error_on_no_feedbacks_computed`	Raise an error if no feedbacks were computed. Default is False. TYPE: `bool` DEFAULT: `False`

get_events ¶

get_events(
    app_name: Optional[str],
    app_version: Optional[str],
    record_ids: Optional[List[str]] = None,
    start_time: Optional[datetime] = None,
) -> DataFrame

Get events/spans from the database in OTel mode.

PARAMETER	DESCRIPTION
`app_name`	The app name to filter events by. TYPE: `Optional[str]`
`app_version`	The app version to filter events by. TYPE: `Optional[str]`
`record_ids`	The record ids to filter events by. TYPE: `Optional[List[str]]` DEFAULT: `None`
`start_time`	The minimum time to consider events from. TYPE: `Optional[datetime]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A pandas DataFrame of all relevant events/spans.

experimental_enable_feature ¶

experimental_enable_feature(
    flag: Union[str, Feature],
) -> bool

Enable the given feature flag.

RAISES	DESCRIPTION
`ValueError`	If the flag is already frozen to disabled.

experimental_disable_feature ¶

experimental_disable_feature(
    flag: Union[str, Feature],
) -> bool

Disable the given feature flag.

RAISES	DESCRIPTION
`ValueError`	If the flag is already frozen to enabled.

experimental_feature ¶

experimental_feature(
    flag: Union[str, Feature], *, freeze: bool = False
) -> bool

Determine the value of the given feature flag.

If freeze is set, the flag will be frozen to the value returned.

experimental_set_features ¶

experimental_set_features(
    flags: Optional[
        Union[
            Iterable[Union[str, Feature]],
            Mapping[Union[str, Feature], bool],
        ]
    ],
    freeze: bool = False,
)

Set multiple feature flags.

If freeze is set, the flags will be frozen to the values given.

RAISES	DESCRIPTION
`ValueError`	If any flag is already frozen to a different value than provided.

trulens.core.session¶