评估、Scores 与 Datasets 实验

想清楚「分挂在哪、是什么类型」，评估就不会乱。Score 是 Langfuse 唯一的评估数据单位。挂载点决定它回答什么问题：挂 trace 回答「这次完整请求好不好」，挂 observation（span/generation）回答「这一步好不好」，挂 dataset run item 回答「这个测试样例在某个实验里得了几分」。类型由 data_type 决定：NUMERIC（连续分，如 0.0–1.0 相关性）、CATEGORICAL（枚举标签，如 good/neutral/bad）、BOOLEAN（true/false，存储为 1/0）。

① Scores：手动打分与用户反馈

Python SDK v3（基于 OpenTelemetry）写分的核心是 langfuse.create_score(...)。最常见的模式是在 @observe 装饰的函数里，先拿到当前 trace 的 id，再把分写回去。下面是一个覆盖三种 data_type 的最小可运行示例。

# Langfuse Python SDK v3（OTel 架构）
pip install "langfuse>=3.0.0" openai

# 三个环境变量：自托管把 HOST 换成你的地址（如 http://localhost:3000）
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com"   # EU 区；US 区用 us.cloud.langfuse.com

from langfuse import get_client, observe

langfuse = get_client()  # 从环境变量读取 keys / host

@observe()  # 自动开一条 trace + 一个根 observation
def answer_question(question: str) -> str:
    # ... 你的真实 LLM 调用，这里用占位返回 ...
    answer = f"回答：{question} 的答案是 42"

    # 拿到当前 trace_id（v3 从 active span 上下文获取）
    trace_id = langfuse.get_current_trace_id()

    # 1) NUMERIC 数值分：0.0 ~ 1.0，常用于相关性/准确度
    langfuse.create_score(
        trace_id=trace_id,
        name="relevance",
        value=0.92,                 # float
        data_type="NUMERIC",
        comment="答案与问题高度相关",
    )

    # 2) CATEGORICAL 类别分：value 必须是字符串
    langfuse.create_score(
        trace_id=trace_id,
        name="tone",
        value="professional",       # str
        data_type="CATEGORICAL",
    )

    # 3) BOOLEAN 布尔分：value 传 0/1，UI 显示 True/False
    langfuse.create_score(
        trace_id=trace_id,
        name="contains_pii",
        value=0,                     # 0=False, 1=True
        data_type="BOOLEAN",
    )
    return answer

answer_question("地球到月球多远？")
langfuse.flush()  # 进程退出前 flush，确保异步队列发出

用户反馈分：把前端的 thumbs up/down 写回

生产里最高价值的信号往往是真实用户反馈。模式是：后端接口返回时把 trace_id 一并给前端，用户点赞/点踩后再带着这个 id 回调一个打分接口。下面用 FastAPI 演示这个回写端点。

from fastapi import FastAPI
from pydantic import BaseModel
from langfuse import get_client

app = FastAPI()
langfuse = get_client()

class Feedback(BaseModel):
    trace_id: str          # 生成回答时返回给前端的 trace id
    thumbs_up: bool        # True=赞, False=踩
    comment: str | None = None

@app.post("/feedback")
def submit_feedback(fb: Feedback):
    # 用户反馈用 BOOLEAN：赞=1 踩=0
    langfuse.create_score(
        trace_id=fb.trace_id,
        name="user_feedback",
        value=1 if fb.thumbs_up else 0,
        data_type="BOOLEAN",
        comment=fb.comment,
    )
    langfuse.flush()
    return {"status": "recorded"}

② LLM-as-judge：自动给 trace 打分

人工打分不可规模化，LLM-as-judge 让一个模型去评估另一个模型的输出。Langfuse 提供两条路：UI 配置的 Managed Evaluator（无代码、自动对新 trace 跑、内置 hallucination / relevance / toxicity 等模板），和 SDK 自定义 judge（你自己写裁判 prompt 与解析逻辑，灵活但要自己跑）。

1. UI 路径：Evaluators → New evaluator，选模板（或自定义 prompt），用变量 {{input}} {{output}} 映射 trace 字段，设采样率与目标 filter，Langfuse 会异步对匹配的 trace 自动打分。
2. SDK 路径：自己调用一个评估模型，解析出分值，再 create_score 写回——适合需要私有评分逻辑、或想在离线批处理里集中跑的场景。
3. 两者写出的都是同一个 Score 对象，可以在同一个 Dashboard 里和人工分、用户反馈分一起聚合对比。

# SDK 自定义 LLM-as-judge：用一个模型评估另一个模型的回答
# pip install "langfuse>=3.0.0" openai
import json
from openai import OpenAI
from langfuse import get_client

langfuse = get_client()
oai = OpenAI()

JUDGE_PROMPT = """你是严格的评估员。判断【回答】是否忠实于【问题】，
只输出 JSON：{{"score": <0到1的小数>, "reason": "<简短理由>"}}
问题：{question}
回答：{answer}"""

def llm_judge(trace_id: str, question: str, answer: str):
    resp = oai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": JUDGE_PROMPT.format(question=question, answer=answer)}],
        response_format={"type": "json_object"},
        temperature=0,
    )
    verdict = json.loads(resp.choices[0].message.content)

    # 把裁判结果写回成 trace 上的 NUMERIC score
    langfuse.create_score(
        trace_id=trace_id,
        name="faithfulness_judge",
        value=float(verdict["score"]),
        data_type="NUMERIC",
        comment=verdict["reason"],   # 把裁判理由存进 comment，方便人工复核
    )
    return verdict

# 用法：对某条已存在的 trace 补打裁判分
llm_judge(
    trace_id="abc123...",
    question="地球到月球多远？",
    answer="约 38.4 万公里",
)
langfuse.flush()

③ Datasets 与实验：离线对比 prompt / 模型版本

Score 解决「线上单条好不好」，Dataset 解决「换 prompt / 换模型后整体是变好还是变差」。流程三步：建数据集与样例 → 遍历样例跑实验并通过 item.run() 链接 trace → 写回 score。每次跑用一个 run name 区分，UI 会自动把同名 run 聚合成一行，多个 run 横向对比就是离线 A/B。

# 第一步：创建 dataset 和 dataset items（只需跑一次）
from langfuse import get_client

langfuse = get_client()

# 创建数据集（幂等：同名重复调用不会重复创建）
langfuse.create_dataset(
    name="qa-eval-set",
    description="问答质量回归测试集",
    metadata={"owner": "ak", "version": 1},
)

# 灌入测试样例：input 是输入，expected_output 是参照答案（gold）
samples = [
    {"q": "地球到月球多远？", "gold": "约 38.4 万公里"},
    {"q": "水的沸点是多少？",   "gold": "标准大气压下 100°C"},
    {"q": "光速是多少？",       "gold": "约 30 万公里每秒"},
]
for s in samples:
    langfuse.create_dataset_item(
        dataset_name="qa-eval-set",
        input={"question": s["q"]},        # dict，对应你的应用入参
        expected_output={"answer": s["gold"]},
    )
print("dataset 灌入完成")

# 第二步：遍历 dataset 跑实验，用 item.run() 链接 trace 并写回 score
from langfuse import get_client
from langfuse.openai import openai   # 用 drop-in 包装，generation 自动进 trace

langfuse = get_client()

def my_app(question: str, model: str) -> str:
    """被测应用：换 model 参数就是换实验变量"""
    resp = openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": question}],
        temperature=0,
    )
    return resp.choices[0].message.content

def exact_match(output: str, gold: str) -> float:
    """简单评估函数：gold 子串命中算对。真实场景可换成 LLM-as-judge"""
    return 1.0 if gold in output else 0.0

def run_experiment(run_name: str, model: str):
    dataset = langfuse.get_dataset("qa-eval-set")
    for item in dataset.items:
        # item.run() 上下文管理器：自动创建一条 trace 并 link 到该 run
        with item.run(
            run_name=run_name,
            run_metadata={"model": model},
        ) as root_span:
            question = item.input["question"]
            output = my_app(question, model)

            # 把应用输出更新到 trace 上，方便 UI 查看
            root_span.update_trace(input=item.input, output={"answer": output})

            # 写回 score：用 root_span.score_trace 自动绑定本次 run 的 trace
            gold = item.expected_output["answer"]
            root_span.score_trace(
                name="exact_match",
                value=exact_match(output, gold),
                data_type="NUMERIC",
            )
    langfuse.flush()
    print(f"实验 {run_name} 完成")

# 跑两个 run：对比两个模型版本
run_experiment(run_name="gpt-4o-mini-baseline", model="gpt-4o-mini")
run_experiment(run_name="gpt-4o-candidate",     model="gpt-4o")

无度量、无对照、无回滚 = 无进步。Score 是度量，Dataset run 是对照，prompt 版本是回滚点——评估体系本质是给 LLM 应用装上可验证的反馈回路。

— Karpathy Lens