๐ Curate Production Traces into an Evaluation Datasetยถ
Your production traces already contain the examples a regression test needs: the questions users actually asked, and the answers your app actually gave. Turning those into a persisted ground truth dataset normally means extracting record content by hand, joining in review corrections, normalizing fields, deduplicating, and deciding what to do with malformed rows.
TruSession.curate_records_to_dataset() does that preparation for you. This
notebook walks the full loop on synthetic data, so it runs with no API keys:
- record a small synthetic app and score it with a metric,
- select the low-scoring records in pandas,
- write corrected expected responses,
- curate them into a persisted dataset,
- load the dataset back and use it as a ground truth metric.
# !pip install trulens-core trulens-feedback
Set up a sessionยถ
Everything below writes to a local SQLite database. No provider credentials are used anywhere in this notebook.
from trulens.core import TruSession
session = TruSession()
session.reset_database()
1. Record a synthetic appยถ
A deliberately imperfect support bot: it answers some questions and hedges on others. The hedged answers are the ones worth turning into regression cases.
from trulens.apps.app import TruApp
from trulens.apps.app import instrument
ANSWERS = {
"what is trulens?": "TruLens is a library for evaluating LLM apps.",
"how do i install trulens?": "I'm not sure, please check the docs.",
"what is a feedback function?": "I'm not sure, please check the docs.",
"how do i log to snowflake?": "You log to Snowflake with a connector.",
}
class SupportBot:
@instrument
def answer(self, question: str) -> str:
return ANSWERS.get(
question.strip().lower(), "I'm not sure, please check the docs."
)
bot = SupportBot()
A metric that needs no LLM: it simply flags answers that hedge. Any metric works here โ this one is deterministic so the notebook always produces the same result.
from trulens.core import Metric
from trulens.core import Selector
HEDGES = ("i'm not sure", "i am not sure", "check the docs")
def answered(response: str) -> float:
"""1.0 when the bot actually answered, 0.0 when it hedged."""
lowered = (response or "").lower()
return 0.0 if any(hedge in lowered for hedge in HEDGES) else 1.0
f_answered = Metric(
implementation=answered,
name="Answered",
selectors={"response": Selector.select_record_output()},
)
app = TruApp(
bot,
app_name="support-bot",
app_version="v1",
feedbacks=[f_answered],
)
with app as recording:
for question in ANSWERS:
bot.answer(question)
session.force_flush()
app.compute_feedbacks()
session.force_flush()
records, feedback_columns = session.get_records_and_feedback()
records[["record_id", "input", "output"] + list(feedback_columns)]
2. Select the records worth curatingยถ
This is plain pandas โ filter however you like. Low scores, errors, a particular app version, a time window, or a hand-picked list of record ids.
failures = records[records["Answered"] < 1.0].copy()
failures[["record_id", "input", "output", "Answered"]]
3. Write the corrected answersยถ
TruLens never generates expected outputs for you. Type them in here, paste them
from a review spreadsheet, or pass expected_response_fn= to
curate_records_to_dataset to fill them in with your own function.
CORRECTIONS = {
"how do i install trulens?": "Run `pip install trulens`.",
"what is a feedback function?": "A function that scores an app's behaviour.",
}
failures["corrected_output"] = (
failures["input"].str.strip().str.lower().map(CORRECTIONS)
)
failures[["input", "corrected_output"]]
4. Curate them into a persisted datasetยถ
The mapping says which dataframe column supplies which ground truth field. Values are column names โ never expressions โ so the mapping is validated against the dataframe before anything is written.
on_error="collect" keeps going past rows that cannot be curated and reports
them instead of raising. Malformed rows are never written in either mode.
from trulens.core.dataset import TraceDatasetMapping
result = session.curate_records_to_dataset(
dataset_name="support-bot-regressions",
records=failures,
mapping=TraceDatasetMapping(
query="input",
query_id="record_id",
expected_response="corrected_output",
metadata={"answered": "Answered"},
),
on_error="collect",
)
print(f"accepted: {result.accepted}")
print(f"duplicates: {result.duplicates}")
print(f"rejected: {result.rejected}")
Rejected rows come back as a dataframe you can inspect, fix and re-curate.
result.errors_df()
Idempotencyยถ
Ground truth ids are content-addressed, so curating the same rows again writes nothing new โ no duplicate rows, and no need to track what you already sent.
again = session.curate_records_to_dataset(
dataset_name="support-bot-regressions",
records=failures,
mapping=TraceDatasetMapping(
query="input",
query_id="record_id",
expected_response="corrected_output",
metadata={"answered": "Answered"},
),
)
assert again.ground_truth_ids == result.ground_truth_ids
5. Load the dataset backยถ
This is the same get_ground_truth() you would use for any hand-built dataset.
Note the metadata: every example remembers the record, app and metric score it
came from, so you can always trace a regression case back to the trace that
produced it.
ground_truth = session.get_ground_truth(dataset_name="support-bot-regressions")
ground_truth[["query", "expected_response", "meta"]]
ground_truth["meta"].iloc[0]
6. Use it as a ground truth metricยถ
A ground truth metric looks up the expected response for a query and scores the
app's answer against it. The one below is an exact match so that the notebook
stays credential-free; with a provider configured you would use
GroundTruthAgreement(ground_truth, provider=...).agreement_measure instead for
semantic similarity.
EXPECTED = dict(zip(ground_truth["query"], ground_truth["expected_response"]))
def matches_expected(prompt: str, response: str) -> float:
"""1.0 when the answer matches the curated expected response."""
expected = EXPECTED.get((prompt or "").strip())
if expected is None:
return float("nan") # not part of the regression set
return float((response or "").strip().lower() == expected.strip().lower())
f_ground_truth = Metric(
implementation=matches_expected,
name="Matches Expected",
selectors={
"prompt": Selector.select_record_input(),
"response": Selector.select_record_output(),
},
)
Now fix the bot and re-run it against the curated regression set.
FIXED_ANSWERS = {**ANSWERS, **CORRECTIONS}
class FixedSupportBot:
@instrument
def answer(self, question: str) -> str:
return FIXED_ANSWERS.get(
question.strip().lower(), "I'm not sure, please check the docs."
)
fixed_bot = FixedSupportBot()
app_v2 = TruApp(
fixed_bot,
app_name="support-bot",
app_version="v2",
feedbacks=[f_ground_truth],
)
with app_v2 as recording:
for question in ground_truth["query"]:
fixed_bot.answer(question)
session.force_flush()
app_v2.compute_feedbacks()
session.force_flush()
after, after_columns = session.get_records_and_feedback(
app_name="support-bot", app_version="v2"
)
after[["input", "output"] + list(after_columns)]
Both regression cases now pass.
Where to go from hereยถ
- Only have record ids? Pass a dataframe with a
record_idcolumn and your corrections; the recorded content is resolved through the session before mapping, and any column you supply yourself wins over the stored one. - Reviewing in a spreadsheet? Export to CSV, load it with
pd.read_csv, and pass it straight in. - Expected contexts? Map
expected_chunksto a column of chunk lists; strings and dicts are both normalized into the ground truth chunk shape. - Deduplicating across records? Provenance metadata is part of a ground
truth's content-addressed id, so two records with identical content stay
distinct. Pass
include_provenance=Falseto deduplicate on example content alone.