๐ Prompt Management and Version Evaluationยถ
Prompts change more often than the application around them. This notebook stores a prompt in the TruLens database, creates three immutable versions of it, and evaluates each one over the same inputs so a change can be accepted or rejected on evidence rather than on a reading of the diff.
Everything here runs locally against SQLite. There is no API key, no Snowflake account, and no network call. A deterministic local callable stands in for a model, and the last section shows where a real provider goes instead.
# !pip install trulens trulens-core plotly
from trulens.core import TruSession
session = TruSession(database_url="sqlite:///prompt_management.sqlite")
session.reset_database()
The evaluation setยถ
The checked-in fixture is eight synthetic support questions, each with the policy id it belongs to and the answer a good response would carry. It is small on purpose: the point is the workflow, not the corpus.
To use your own data, replace the two lines below with any DataFrame that has an input
column, or read from a table your warehouse already holds. RunConfig also accepts
source_type="TABLE" with a table name in dataset_name.
import pandas as pd
dataset = pd.read_json("data/prompt_management_fixture.jsonl", lines=True)
dataset
One prompt, three versionsยถ
A Prompt is the stable identity. Its type is fixed at creation. Each
create_prompt_version call writes an immutable, content-addressed version and moves the
latest label onto it. Creating identical content twice returns the same version rather
than a duplicate.
The three versions below are a realistic sequence: a baseline, a change that asks for a citation, and a change that also forces brevity.
prompt = session.create_prompt(
slug="support-assistant",
name="Support assistant",
prompt_type="chat",
description="Answers support questions from the internal policy set.",
tags=["support"],
)
POLICY_INSTRUCTION = "Answer the question using the support policy."
v1 = session.create_prompt_version(
prompt=prompt,
messages=[
{"role": "system", "content": POLICY_INSTRUCTION},
{"role": "user", "content": "{{question}}"},
],
variables=["question"],
model_defaults={"temperature": 0.0},
change_note="Baseline",
)
v2 = session.create_prompt_version(
prompt=prompt,
messages=[
{
"role": "system",
"content": POLICY_INSTRUCTION + " Cite the policy id.",
},
{"role": "user", "content": "{{question}}"},
],
variables=["question"],
model_defaults={"temperature": 0.0},
change_note="Ask for the policy id",
)
v3 = session.create_prompt_version(
prompt=prompt,
messages=[
{
"role": "system",
"content": POLICY_INSTRUCTION
+ " Cite the policy id. Answer in one short sentence.",
},
{"role": "user", "content": "{{question}}"},
],
variables=["question"],
model_defaults={"temperature": 0.0},
change_note="Ask for the policy id and force brevity",
)
session.get_prompt_versions(prompt)[
["version_id", "change_note", "parent_version_id"]
]
Resolve exact versions, not labelsยถ
Labels such as production and latest move. An evaluation that resolves a label is not
reproducible, because the same notebook run tomorrow can score different content. Every
comparison below therefore asks for a version id.
session.get_prompt(prompt, label="production") is the right call in application code,
where following the label is the point.
candidates = {
"v1-baseline": session.get_prompt(prompt, version_id=v1.version_id),
"v2-cite": session.get_prompt(prompt, version_id=v2.version_id),
"v3-cite-brief": session.get_prompt(prompt, version_id=v3.version_id),
}
{name: resolved.version_id for name, resolved in candidates.items()}
Render over identical inputsยถ
render interpolates the declared variables and returns provider-neutral message
dictionaries, the model defaults merged with any caller overrides, and the response
format separately. It calls no model and needs no credentials. Missing or unexpected
values are an error, so a renamed variable fails here rather than silently reaching a
model.
first_question = dataset["question"].iloc[0]
for name, resolved in candidates.items():
request = resolved.render(question=first_question)
print(name)
print(" system:", request.messages[0]["content"])
print(" user: ", request.messages[1]["content"])
print(" model_args:", request.model_args)
try:
candidates["v1-baseline"].render(quesiton=first_question)
except Exception as e:
print(type(e).__name__, e)
A deterministic local modelยถ
This stands in for a hosted model so the notebook runs with no credentials and produces the same numbers every time. It reads the rendered system message the way a model would be asked to, so editing the prompt changes the output.
To use a real provider, replace the body with your own client call and forward
request.messages, request.model_args, and request.response_format. Nothing else in
the notebook changes.
POLICY_BOOK = {
row["policy_id"]: row["expected_answer"] for _, row in dataset.iterrows()
}
QUESTION_TO_POLICY = {
row["question"]: row["policy_id"] for _, row in dataset.iterrows()
}
def local_support_model(messages, model_args):
system = next((m["content"] for m in messages if m["role"] == "system"), "")
question = next((m["content"] for m in messages if m["role"] == "user"), "")
policy_id = QUESTION_TO_POLICY.get(question)
if policy_id is None:
return "I do not have a policy covering that."
answer = POLICY_BOOK[policy_id]
if "one short sentence" in system:
answer = answer.split(",")[0].split(".")[0] + "."
if "Cite the policy id" in system:
answer = f"{answer} (policy {policy_id})"
return answer
The applicationยถ
The app holds one resolved version and renders it per input. The prompt_lineage context
manager writes the prompt id, slug, exact version id, requested label, and a hash of the
rendered content onto the generation span, so nothing in the app has to carry those
identifiers around or pass them to the run. Six months from now, when the production
label has moved twice and the version that produced a given trace is no longer the one it
points at, the span still names the version it used.
What lands on the span is identifiers and a hash, not the prompt body, so lineage survives with GenAI content capture off, which is the default.
from trulens.core.otel.instrument import instrument
from trulens.core.otel.instrument import prompt_lineage
from trulens.otel.semconv.trace import SpanAttributes
class SupportAssistant:
def __init__(self, resolved):
self.resolved = resolved
@instrument(span_type=SpanAttributes.SpanType.GENERATION)
def generate(self, question: str) -> str:
request = self.resolved.render(question=question)
with prompt_lineage(request):
return local_support_model(request.messages, request.model_args)
def answer(self, question: str) -> str:
return self.generate(question)
Metricsยถ
These are ordinary Metric objects. The implementations are local and deterministic so
the notebook stays offline; swap either implementation for a provider method such as
provider.groundedness_measure_with_cot_reasons and the rest of the workflow is
unchanged.
Answer Coverage reads the ground truth carried on the record root, which the
dataset_spec below maps from the fixture.
import re
import numpy as np
from trulens.core import Metric
from trulens.core import Selector
POLICY_ID_PATTERN = re.compile(r"\b[A-Z]{3}-\d{3}\b")
def answer_coverage(response: str, ground_truth: str) -> float:
expected = {word for word in re.findall(r"[a-z]{4,}", ground_truth.lower())}
if not expected:
return 0.0
got = set(re.findall(r"[a-z]{4,}", response.lower()))
return len(expected & got) / len(expected)
def policy_citation(response: str) -> float:
return 1.0 if POLICY_ID_PATTERN.search(response) else 0.0
m_coverage = Metric(
implementation=answer_coverage,
name="Answer Coverage",
selectors={
"response": Selector.select_record_output(),
"ground_truth": Selector(
span_type=SpanAttributes.SpanType.RECORD_ROOT,
span_attribute=SpanAttributes.RECORD_ROOT.GROUND_TRUTH_OUTPUT,
),
},
agg=np.mean,
)
m_citation = Metric(
implementation=policy_citation,
name="Policy Citation",
selectors={"response": Selector.select_record_output()},
agg=np.mean,
)
metrics = [m_coverage, m_citation]
metric_names = [m.name for m in metrics]
One run per exact versionยถ
Each version gets a normal Run over the identical input frame. Run metadata currently
takes a description and a label, so the prompt identity, version id, and content hash go
into the description as JSON. That is what lets a run be traced back to the exact prompt
that produced it after the labels have moved on.
import json
from trulens.apps.app import TruApp
from trulens.core.run import RunConfig
input_df = dataset.rename(
columns={"question": "input", "expected_answer": "ground_truth_output"}
)[["input", "ground_truth_output"]]
tru_apps = {}
runs = {}
for name, resolved in candidates.items():
app = SupportAssistant(resolved)
tru_app = TruApp(
app,
app_name="Support assistant",
app_version=name,
main_method=app.answer,
feedbacks=metrics,
connector=session.connector,
)
run = tru_app.add_run(
run_config=RunConfig(
run_name=f"prompt_{name.replace('-', '_')}",
dataset_name="support_questions",
source_type="DATAFRAME",
dataset_spec={
"input": "input",
"ground_truth_output": "ground_truth_output",
},
invocation_max_workers=1,
label=resolved.prompt.slug,
description=json.dumps({
"prompt_id": resolved.prompt.prompt_id,
"prompt_slug": resolved.prompt.slug,
"prompt_version_id": resolved.version_id,
"prompt_content_hash": resolved.version.content_hash,
}),
)
)
run.start(input_df=input_df)
run.compute_metrics(metrics)
tru_apps[name] = tru_app
runs[name] = run
Resultsยถ
Metric computation is asynchronous, so wait for the scored columns to land before reading them.
records = {}
for name, run in runs.items():
run_records = run.get_records()
tru_apps[name].retrieve_feedback_results(
record_ids=run_records["record_id"].tolist()
)
records[name] = run.get_records()
records["v3-cite-brief"][["input", "output", "latency"] + metric_names]
Quality, latency, and cost side by side. Cost is zero because the generation and both metrics are deterministic Python calls with no model behind them, so no provider tokens are bought. Against a real provider the endpoint records usage and the column carries it.
summary = (
session.get_leaderboard()
.loc["Support assistant"]
.loc[list(candidates)]
.join(
pd.Series(
{name: resolved.version_id for name, resolved in candidates.items()},
name="version_id",
)
)
)
summary
Compare versionsยถ
Run.compare matches records by input and reports, per metric, the mean delta with a
95% confidence interval, a permutation p-value, and how many items regressed.
runs["v1-baseline"].compare(runs["v2-cite"]).summary()
Asking for the policy id is a clean win: every answer now cites one and nothing else moved.
The next comparison is the interesting one. Forcing brevity kept the citation and cost real coverage.
diff = runs["v2-cite"].compare(runs["v3-cite-brief"])
diff.summary()
diff.items_df("Answer Coverage").sort_values("delta")
Prompt lineage on the spansยถ
Content capture is off, and the generation spans still say which exact version produced them.
events = session.connector.db.get_events(
app_name="Support assistant",
app_version=None,
record_ids=None,
start_time=None,
)
lineage = pd.DataFrame([
{
"slug": attrs[SpanAttributes.PROMPT.SLUG],
"version_id": attrs[SpanAttributes.PROMPT.VERSION_ID],
"rendered_content_hash": attrs[
SpanAttributes.PROMPT.RENDERED_CONTENT_HASH
],
}
for attrs in events["record_attributes"]
if SpanAttributes.PROMPT.VERSION_ID in attrs
])
lineage.groupby(["slug", "version_id"]).size().rename("spans").reset_index()
Chartยถ
import plotly.graph_objects as go
fig = go.Figure()
for metric in metric_names:
fig.add_bar(x=summary.index, y=summary[metric], name=metric)
fig.update_layout(
barmode="group",
title="Mean metric score by prompt version",
yaxis_title="score",
yaxis_range=[0, 1.05],
height=380,
)
fig
Promote the version you pickedยถ
v2-cite won, so point production at it. Rolling back later is the same call with an
older version id; the versions themselves are never edited and application code that
looks the label up needs no change.
session.set_prompt_label(prompt, "production", v2, moved_by="notebook")
session.get_prompt_label_history(prompt, label="production")
session.get_prompt(prompt, label="production").version_id == v2.version_id
Dashboardยถ
The Prompts page lists prompts, their labels, and their version history, and it can diff versions, move labels, roll back, and render a local preview. It never invokes a model and never stores credentials.
from trulens.dashboard import run_dashboard
run_dashboard(session)