Offline evals score a frozen set; online evals score what production actually served. You need both, and they answer different questions: offline gates the release, online tells you what the release did to users 1. Langfuse is the open-source default for the online half because the four primitives the loop needs (tracing SDK, scores API, annotation queues, datasets) live in one place, and since June 2025 all of them, including the managed LLM-as-a-judge evaluators, are MIT-licensed rather than paywalled 2. Langfuse joined ClickHouse in January 2026 and states there are no planned licensing changes 3. If your observability is OpenTelemetry-first or you are already on Arize, the same loop in Phoenix is the sibling recipe: online eval with Phoenix. Pick one; running both is how teams end up with two dashboards nobody trusts.
This recipe wires the full loop with three scripts: an instrumented app, a judge worker that scores a deterministic 5% sample plus every user-flagged trace, and a promotion job that turns confirmed failures into offline eval cases. The hour breaks down as: fifteen minutes to instrument and see traces, twenty for the worker, fifteen for the dashboard and dataset wiring, ten to decide your sampling and alerting numbers.
What you build
app.py: a support bot instrumented with the v4 SDK; every call is a trace carrying session, user, and a prod tag, and thumbs-down feedback lands as a score plus a local flag queue.judge_worker.py: a cron-shaped worker that pulls recent root observations from the v2 API, judges the sample, and writes BOOLEAN scores back with the critique in the comment.promote.py: a weekly job that moves judged failures into a Langfuse dataset, deduplicated by content hash, linked to their source traces.
Figure: Online eval loop with Langfuse: app.py traces every production call, judge_worker.py scores a deterministic 5% sample plus every thumbs-down (26 of 412 traces judged in the example window: 22 PASS, 4 FAIL), and promote.py dedupes confirmed failures into the support-bot-regressions dataset that the next offline run executes against, closing the loop where production failures gate releases.
Step 1: instrument the app
pip install langfuse openai anthropic
export LANGFUSE_PUBLIC_KEY=pk-lf-...
export LANGFUSE_SECRET_KEY=sk-lf-...
export LANGFUSE_BASE_URL=https://cloud.langfuse.com # or your own host
export OPENAI_API_KEY=... ANTHROPIC_API_KEY=...
The SDK moved to an OpenTelemetry core in v3 and was rewritten in v4 (March 2026); the code below is current against v4 and will not run on v2 snippets you find in older blog posts (langfuse.trace(...) is gone). The three stable entry points are the @observe decorator, the langfuse.openai drop-in client, and get_client() 4.
The platform followed the SDK. Langfuse v4 went live on Cloud on 2026-08-17 with an observations-first data model, and Cloud becomes v4-only on 2026-11-16, when the legacy trace read APIs, trace-level evaluators, and legacy ingestion stop working 5. Two things in this recipe depend on that: the worker reads observations rather than traces, and the managed evaluators run on observations.
"""The instrumented app: every production call becomes a Langfuse trace.
Run:
python app.py
"""
from __future__ import annotations
import json
from langfuse import get_client, observe, propagate_attributes
from langfuse.openai import openai # drop-in client; calls are traced
langfuse = get_client()
MODEL = "gpt-4o-2024-08-06"
SYSTEM = (
"You are the support assistant for a meal-kit company. Answer only from "
"the provided policy context. If the context does not cover the "
"question, say you do not know and offer the human handoff."
)
POLICY = """Cancellations: free until the Wednesday before delivery.
Refunds: spoiled items are refunded within 5 business days of a photo report.
Allergens: boxes are packed in a facility that handles nuts and shellfish."""
@observe()
def answer(*, question: str, context: str, session_id: str, user_id: str) -> dict:
with propagate_attributes(
session_id=session_id, user_id=user_id, tags=["prod", "support-bot"]
):
resp = openai.chat.completions.create(
model=MODEL,
name="support-answer",
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user",
"content": f"CONTEXT:\n{context}\n\nQUESTION: {question}"},
],
)
return {
"answer": resp.choices[0].message.content,
"trace_id": langfuse.get_current_trace_id(),
}
def record_feedback(trace_id: str, thumbs_up: bool) -> None:
"""Wire this to your feedback button. Failures also land in a local
queue so the judge worker scores them at 100%, not the sample rate."""
langfuse.create_score(
trace_id=trace_id, name="user_feedback",
value=1 if thumbs_up else 0, data_type="BOOLEAN",
)
if not thumbs_up:
with open("flagged_traces.jsonl", "a") as f:
f.write(json.dumps({"trace_id": trace_id}) + "\n")
if __name__ == "__main__":
questions = [
"Can I cancel my Friday box on Thursday?",
"My salmon arrived warm, what now?",
"Is the vegan box safe for my peanut allergy?",
"Do you deliver to Alaska?",
]
for i, q in enumerate(questions):
out = answer(question=q, context=POLICY,
session_id=f"sess-{i:03d}", user_id="user-042")
print(f"[{out['trace_id'][:8]}] {q}\n -> {out['answer'][:80]}")
if i == 2: # simulate a thumbs-down on the allergy answer
record_feedback(out["trace_id"], thumbs_up=False)
langfuse.flush()
Run it and open the Langfuse UI: four traces tagged prod, each with the model call nested inside, one carrying a user_feedback score of 0. The trace id returned to the caller is the load-bearing detail; production feedback can only be attached if your API response carries it. Note what is sampled where: LANGFUSE_SAMPLE_RATE drops traces at ingestion (use it only under real volume pressure, since a dropped trace cannot be debugged or judged later), while the judge worker samples at scoring time. Keep ingestion at 100% for as long as you can afford it 4.
Step 2: the judge worker
Two ways to attach judge scores to production traces. The platform way: Langfuse's managed LLM-as-a-judge evaluators run inside Langfuse itself and are the right choice when you want zero infrastructure. You configure an evaluator (judge prompt, model, score type, variable mapping) and a rule (filters and a sampling rate) that picks which incoming observations it runs on. Evaluators now target observations, not traces, and an evaluator sees only the matched observation's own input and output; to judge a whole request, filter the rule with Is Root Observation, which matches the @observe root that app.py creates. Trace-level evaluators are deprecated and stop producing results on Langfuse Cloud on 2026-11-16 6. The worker way, below, is the external evaluation pipeline pattern: you own the judge prompt, the model choice, the sampling logic, and the version pinning, and Langfuse is the system of record for the resulting scores 7. Teams that calibrated a judge offline almost always want the worker, because the exact prompt that cleared calibration is the one that must run online.
"""Scoring worker: judge a 5% sample of production traces plus every
user-flagged trace, write scores back to Langfuse, queue failures.
Run on a schedule (cron every 15 minutes is plenty to start):
python judge_worker.py
"""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from anthropic import Anthropic
from langfuse import get_client
langfuse = get_client()
anthropic = Anthropic()
JUDGE_MODEL = "claude-sonnet-5-5"
SCORE_NAME = "grounded_v3" # bump the suffix when the prompt changes
SAMPLE_PCT = 5
LOOKBACK_HOURS = 24
STATE = Path("scored_ids.json")
FLAGGED = Path("flagged_traces.jsonl")
FAILS = Path("fails.jsonl")
JUDGE_PROMPT = """You are grading one support-bot answer against one rubric.
RUBRIC: The answer must be fully supported by the CONTEXT. If the context
does not cover the question, the only passing answer is one that says so
and offers the handoff. Any unsupported claim is a FAIL.
CONTEXT:
{context}
QUESTION:
{question}
ANSWER UNDER REVIEW:
{answer}
Reply with JSON: {{"label": "PASS" or "FAIL", "critique": "one sentence"}}"""
def in_sample(trace_id: str) -> bool:
"""Deterministic 5% sample: same trace, same verdict, on every rerun."""
return int(hashlib.sha256(trace_id.encode()).hexdigest(), 16) % 100 < SAMPLE_PCT
def as_json(raw):
"""The v2 API returns input and output as raw JSON strings."""
try:
return json.loads(raw) if isinstance(raw, str) else raw
except json.JSONDecodeError:
return raw
def unpack(root) -> tuple[str, str, str]:
"""@observe stores the call's kwargs as the root observation's input and
the return dict as its output. Adjust this one function to your own schema."""
inp, out = as_json(root.input), as_json(root.output)
kwargs = inp.get("kwargs", {}) if isinstance(inp, dict) else {}
answer = out.get("answer", str(out)) if isinstance(out, dict) else str(out)
return kwargs.get("question", ""), kwargs.get("context", ""), answer
def judge(question: str, context: str, answer: str) -> dict:
msg = anthropic.messages.create(
model=JUDGE_MODEL, max_tokens=200,
thinking={"type": "between_tools"}, # Sonnet 5.5: no up-front thinking
messages=[{"role": "user", "content": JUDGE_PROMPT.format(
context=context, question=question, answer=answer)}],
)
text = msg.content[0].text.strip().removeprefix("```json").removesuffix("```")
return json.loads(text)
def fetch_recent_roots() -> list:
"""One root observation per prod trace. v4 has no trace-level input or
output; the @observe root observation carries them."""
now = datetime.now(timezone.utc)
since = now - timedelta(hours=LOOKBACK_HOURS)
iso = "%Y-%m-%dT%H:%M:%SZ"
# `filter` overrides the individual query parameters, so the time window
# is expressed as filter conditions too.
conditions = json.dumps([
{"type": "boolean", "column": "isRootObservation", "operator": "=", "value": True},
{"type": "arrayOptions", "column": "tags", "operator": "all of", "value": ["prod"]},
{"type": "datetime", "column": "startTime", "operator": ">=", "value": since.strftime(iso)},
{"type": "datetime", "column": "startTime", "operator": "<", "value": now.strftime(iso)},
])
roots, cursor = [], None
while True:
page = langfuse.api.observations.get_many(
filter=conditions, fields="core,basic,io,trace_context",
limit=100, cursor=cursor,
)
roots.extend(page.data)
cursor = page.meta.cursor
if not cursor:
return roots
def main() -> None:
scored: set[str] = set(json.loads(STATE.read_text())) if STATE.exists() else set()
flagged: set[str] = set()
if FLAGGED.exists():
flagged = {json.loads(l)["trace_id"]
for l in FLAGGED.read_text().splitlines() if l}
roots = fetch_recent_roots()
todo = [r for r in roots if r.trace_id not in scored
and (r.trace_id in flagged or in_sample(r.trace_id))]
n_pass = 0
for r in todo:
question, context, answer = unpack(r)
verdict = judge(question, context, answer)
langfuse.create_score(
trace_id=r.trace_id, name=SCORE_NAME,
value=1 if verdict["label"] == "PASS" else 0,
data_type="BOOLEAN", comment=verdict["critique"],
)
scored.add(r.trace_id)
n_pass += verdict["label"] == "PASS"
if verdict["label"] == "FAIL":
with FAILS.open("a") as f:
f.write(json.dumps({
"trace_id": r.trace_id, "question": question,
"answer": answer, "critique": verdict["critique"],
}) + "\n")
STATE.write_text(json.dumps(sorted(scored)))
langfuse.flush()
print(f"window: {len(roots)} traces, {len(flagged)} user-flagged")
print(f"judged: {len(todo)} ({SAMPLE_PCT}% sample + flagged), "
f"PASS {n_pass}, FAIL {len(todo) - n_pass}")
Output shape from one worker run against the demo app's traces (yours will differ):
window: 412 traces, 7 user-flagged
judged: 26 (5% sample + flagged), PASS 22, FAIL 4
Two notes on how the worker reads. langfuse.api.trace.list, the call earlier versions of this recipe used, is deprecated: Langfuse Cloud serves it until 2026-11-16, and the documented replacement is api.observations.get_many. It returns observation rows rather than trace objects, so the worker reads the @observe root of each trace and takes its input, output, and tags from there 8. Reads through that API are real-time only on langfuse 4.7.0 or newer; older SDKs can lag by up to 15 minutes 9.
The judge call changed too. It no longer sets temperature=0: the Anthropic Python SDK dropped the parameter in 1.0 and Claude Sonnet 5.5 rejects non-default values, so judge stability now rests on the pinned model id, the versioned score name, and the monthly human audit. The thinking argument is Sonnet 5.5 syntax for turning off up-front thinking, which is on by default there and would otherwise spend the 200-token budget and put thinking blocks ahead of the text 10. claude-sonnet-4-5, the judge in earlier versions of this recipe, was deprecated on 2026-09-30 and retires on 2026-11-30 11; a judge model swap moves the metric, so when you make it, bump the suffix on grounded_v3 and re-run calibration first.
Three decisions in this worker are deliberate. Sampling hashes the trace id instead of calling random(), so reruns and backfills judge the same subset and your daily pass rate is not jittered by the sampler. The score name carries the judge version (grounded_v3), because a judge prompt change moves the metric and an unversioned score series silently splices two different instruments; never reuse a name after the prompt changes. And every FAIL writes the critique into the score comment, so a human reading the trace in Langfuse sees why the judge objected without leaving the page.
Step 3: read the dashboard like an eval, not like a vibe
In Langfuse, chart the daily mean of grounded_v3 next to user_feedback, split by release version. Two reading rules. Alert on the score crossing the offline error bar, not on any movement: if your judge cleared calibration with a dev-set pass rate of 0.91 plus or minus 0.04, a production day at 0.88 is noise and a week at 0.84 is a finding (adding error bars gives you the interval). And treat judge-vs-feedback divergence as its own signal: judge PASS with rising thumbs-downs means the rubric is missing a failure mode users can see; route a week of thumbs-down traces through an annotation-queue review and fix the rubric before trusting the metric again 4.
The judge itself gets audited on a schedule. Once a month, send 50 judge-scored traces to a human, compute agreement, and compare against the TPR/TNR the prompt cleared offline in LLM-judge calibration. An online judge that nobody re-checks decays into a dashboard decoration.
Step 4: close the loop into the offline set
Online eval pays for itself the day a production failure becomes an offline test case. The promotion job below trusts fails.jsonl; in practice you gate it behind the annotation queue, so a human confirms each failure (and writes the expected output) before it becomes a regression case.
"""Promote reviewed failures into the offline eval dataset.
Run weekly, after a human has walked the annotation queue:
python promote.py
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from langfuse import get_client
langfuse = get_client()
DATASET = "support-bot-regressions"
FAILS = Path("fails.jsonl")
SEEN = Path("promoted_hashes.json")
langfuse.create_dataset(
name=DATASET,
description="Production failures promoted from online judge scores",
)
seen: set[str] = set(json.loads(SEEN.read_text())) if SEEN.exists() else set()
added = 0
for line in FAILS.read_text().splitlines() if FAILS.exists() else []:
fail = json.loads(line)
content_hash = hashlib.sha256(fail["question"].encode()).hexdigest()[:16]
if content_hash in seen:
continue # the same failure mode does not get to vote twice
langfuse.create_dataset_item(
dataset_name=DATASET,
input={"question": fail["question"]},
expected_output=None, # filled in during human review
metadata={"critique": fail["critique"], "content_hash": content_hash},
source_trace_id=fail["trace_id"],
)
seen.add(content_hash)
added += 1
SEEN.write_text(json.dumps(sorted(seen)))
langfuse.flush()
print(f"promoted {added} new failures to dataset '{DATASET}'")
source_trace_id is the field that makes this auditable: every dataset item links back to the production trace it came from, so six months later you can still answer "why is this weird question in our eval set" 4. The content hash stops one angry user from contributing the same failure forty times and skewing the offline distribution. Your next offline run executes against this dataset, which is the moment the loop closes: production failures now gate releases.
What this does not do
It does not cover Phoenix or OTel-native instrumentation; that is the sibling recipe. It does not ensemble judges or distill a cheaper one; start with one calibrated judge and add machinery when the audit numbers say you need it. It does not handle multi-turn session scoring, which changes the unit of analysis from trace to session; the multi-turn conversation eval recipe covers that, and the same worker pattern applies with session_id filters. And the 5% number is a starting point, not advice: the right rate is whatever makes judge cost a rounding error against your serving cost. A managed rule asks for a sampling rate and shows the matching volume for the past seven days, so set it from that estimate 6.
TIP
Run the worker from the same repo and CI system as your offline evals, with the judge prompt imported from one shared module. The moment online and offline judges drift apart in wording, their numbers stop being comparable and every "production regressed" conversation turns into an argument about prompts.
What to do next
If the judge has not been through calibration, do LLM-judge calibration first; an uncalibrated online judge automates a guess at production scale. For the conceptual map of where online evaluation sits in a platform, Chip Huyen's essay is the reference 1, and the online vs offline chapter covers the decision in depth.