Langfuse: LLM tracing & evals

pip install langfuse
export LANGFUSE_PUBLIC_KEY=pk-lf-...
export LANGFUSE_SECRET_KEY=sk-lf-...
export LANGFUSE_BASE_URL=https://cloud.langfuse.com       # or your self-hosted URL

from langfuse import get_client

langfuse = get_client()             # reads the LANGFUSE_* environment variables
assert langfuse.auth_check()        # fail fast on bad keys (development only; it makes a network call)

from langfuse import get_client, observe, propagate_attributes
from openai import OpenAI

langfuse = get_client()
openai_client = OpenAI()

@observe(as_type="retriever")                                # also: span (default), tool, agent, …
def retrieve(question: str) -> list[dict]:
    return [h.doc for h in index.search(question, k=4)]

@observe(as_type="generation", name="chat")
def generate(question: str, docs: list[dict]) -> str:
    r = openai_client.responses.create(model="gpt-5.6-luna", instructions=INSTRUCTIONS,
                                       input=build_input(question, docs))
    langfuse.update_current_generation(
        model="gpt-5.6-luna",
        usage_details={"input_tokens": r.usage.input_tokens, "output_tokens": r.usage.output_tokens})
    return r.output_text

@observe()
def answer(question: str, user_id: str, session_id: str) -> str:
    with propagate_attributes(user_id=user_id, session_id=session_id,
                              tags=["docs-copilot", "prod"], metadata={"prompt_version": "answer-v3"}):
        docs = retrieve(question)
        text = generate(question, docs)
        langfuse.score_current_trace(name="grounded", value=1.0 if has_valid_citations(text, docs) else 0.0)
        return text

# when answering: return the trace id to the frontend
trace_id = langfuse.get_current_trace_id()

# later, when the user clicks 👍 or 👎:
langfuse.create_score(trace_id=trace_id, name="user_feedback", value=1 if thumbs_up else 0,
                      data_type="BOOLEAN", comment=user_comment)

langfuse.create_prompt(
    name="answer", type="text", labels=["production"],
    prompt="Answer using only the sources; cite [n]; say I don't know if unsure.\n"
           "{{sources}}\n<question>{{question}}</question>",
    config={"model": "gpt-5.6-luna"})

prompt = langfuse.get_prompt("answer")                 # the "production" version by default; cached client-side
text = prompt.compile(sources=sources_block, question=question)

with langfuse.start_as_current_observation(as_type="generation", name="chat",
                                           model=prompt.config["model"], prompt=prompt) as gen:
    r = openai_client.responses.create(model=prompt.config["model"], input=text)
    gen.update(output=r.output_text)

langfuse.create_dataset(name="docs-golden")
for question, lessons in GOLDEN:
    langfuse.create_dataset_item(dataset_name="docs-golden", input=question, expected_output=sorted(lessons))

dataset = langfuse.get_dataset("docs-golden")
result = dataset.run_experiment(name="hybrid-k5", task=task, evaluators=[hit_at_5, reciprocal_rank],
                                run_evaluators=[average("hit@5")], max_concurrency=4)
print(result.format())

from dataclasses import dataclass
from statistics import mean

from mini_rag import GOLDEN, HybridIndex, lessons_in
from site_docs import load_sections

@dataclass
class Evaluation:                                  # same fields as langfuse.Evaluation
    name: str
    value: float
    comment: str | None = None

index = HybridIndex(load_sections())

def task(*, item, **kwargs):                        # Langfuse passes the dataset item
    return lessons_in(index.rank(item["input"]), index.docs, k=5)

def hit_at_5(*, input, output, expected_output, **kwargs):
    hit = bool(set(expected_output) & set(output))
    return Evaluation("hit@5", float(hit), None if hit else f"got {output[0]}")

def reciprocal_rank(*, input, output, expected_output, **kwargs):
    rank = next((i for i, url in enumerate(output, 1) if url in expected_output), None)
    return Evaluation("reciprocal_rank", 1 / rank if rank else 0.0)

def average(name):                                  # a run-level evaluator
    def run_evaluator(*, item_results, **kwargs):
        return Evaluation(f"avg_{name}", mean(e.value for r in item_results for e in r["evaluations"]
                                              if e.name == name))
    return run_evaluator

def run_experiment_locally(name, data, task, evaluators, run_evaluators=()):
    item_results = []
    for item in data:
        output = task(item=item)
        evals = [ev(input=item["input"], output=output, expected_output=item["expected_output"])
                 for ev in evaluators]
        item_results.append({"item": item, "output": output, "evaluations": evals})
    print(f"experiment {name!r}: {len(item_results)} items")
    for rev in run_evaluators:
        e = rev(item_results=item_results)
        print(f"  {e.name} = {e.value:.2f}")
    misses = [r for r in item_results if r["evaluations"][0].value == 0]
    for r in misses:
        print(f"  miss: {r['item']['input'][:52]!r:56} {r['evaluations'][0].comment}")

data = [{"input": q, "expected_output": sorted(lessons)} for q, lessons in GOLDEN]
run_experiment_locally("hybrid-k5", data, task, [hit_at_5, reciprocal_rank],
                       run_evaluators=[average("hit@5"), average("reciprocal_rank")])