核心概念：Trace / Span / Generation 数据模型

==一次请求 = 一棵树==。根是 Trace，枝干是 Observation。决定建哪种 Observation 的唯一标准：这一步是不是在调大模型？是 → Generation；不是、但有耗时（检索 / 工具 / 子流程）→ Span；不是、且是个瞬时事件（缓存命中、规则触发）→ Event。

四种节点：各自管什么

树是怎么连起来的：trace_id 与 parent_observation_id

Observation 之间的父子关系不靠代码缩进，而靠两个 id 字段显式声明。每个 Observation 都带一个 trace_id（属于哪棵树）和一个 parent_observation_id（父节点是谁，为空则直接挂在 Trace 根上）。Langfuse 服务端收到这些扁平记录后，按 id 关系重建出树形结构。理解这点很重要：只要 id 对得上，跨函数、跨线程、甚至跨服务上报的节点都能拼进同一棵树。

# 安装（v3 SDK）：pip install "langfuse>=3.0.0"
# 环境变量：LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_HOST

from langfuse import get_client

langfuse = get_client()  # 自动读取上面三个环境变量

# 1) 手动建一棵树，观察 id / parent 是怎么串起来的
# start_as_current_span 会把这个 span 设为「当前上下文」，
# 其内部再创建的 observation 自动以它为 parent。
with langfuse.start_as_current_span(
    name="rag-request",          # 这个 span 是整棵树的第一个工作单元
    input={"question": "Langfuse 怎么记成本？"},
) as root_span:

    # 2) 在 root_span 上下文里建一个检索 Span —— parent 自动 = root_span
    with root_span.start_as_current_span(
        name="retrieve-docs",
        input={"query": "langfuse cost tracking"},
    ) as retrieval:
        docs = ["doc-1", "doc-2"]            # 假装做了向量检索
        retrieval.update(output={"docs": docs})

    # 3) 建一个 Generation（LLM 调用），parent 也自动 = root_span
    with root_span.start_as_current_generation(
        name="answer-llm",
        model="gpt-4o",                      # ← Generation 专属，Span 没有
        input=[{"role": "user", "content": "根据文档回答"}],
        model_parameters={"temperature": 0.2},
    ) as gen:
        answer = "Langfuse 靠 Generation 上报 usage 来统计 token。"
        gen.update(
            output=answer,
            usage_details={"input": 320, "output": 48},  # ← token 用量
        )

    root_span.update(output={"answer": answer})

# 4) 在 Trace 根上挂横切维度（user / session / tags）
langfuse.update_current_trace(
    user_id="user-42",
    session_id="chat-session-001",
    tags=["rag", "prod"],
    metadata={"region": "cn"},
)

langfuse.flush()  # 进程退出前 flush，确保异步队列里的数据都上报
# 现在 Langfuse UI 里会看到：rag-request 下挂着 retrieve-docs(Span) + answer-llm(Generation)

横切维度：Session / User / Tags / Metadata / Score 挂在哪

# Score 的两种落点：整条 Trace vs 某个具体 Observation
from langfuse import get_client

langfuse = get_client()

with langfuse.start_as_current_span(name="qa-request") as span:
    with langfuse.start_as_current_generation(
        name="llm-call", model="gpt-4o",
        input=[{"role": "user", "content": "2+2=?"}],
    ) as gen:
        gen.update(output="4", usage_details={"input": 8, "output": 1})

        # (A) 把分数打在「这次 LLM 生成」上 —— 评的是单个 observation 的质量
        gen.score(name="correctness", value=1.0, comment="答案正确")

    # (B) 把分数打在「整条 Trace」上 —— 评的是这次请求的整体表现
    # score_trace 评当前上下文所属的 trace
    langfuse.score_current_trace(
        name="user-feedback",
        value="good",            # 分类型分数（categorical）
        data_type="CATEGORICAL",
    )

langfuse.flush()
# UI 里：correctness 显示在 llm-call 这个 generation 行；
#        user-feedback 显示在整条 trace 的汇总区。

把它放进一棵真实的 RAG 树看落点

下面是一次「用户提问 → 检索 → 重排 → 生成 → 校验」的 RAG 请求，展开成 Observation 树后每个节点的类型。注意每一行的类型选择都对应前面那张决策表：调模型的两步是 Generation，有耗时的检索 / 重排 / 校验是 Span，缓存命中是瞬时 Event。

Trace: "rag-chat-request"                      ← 根容器（一次完整请求）
│   user_id=user-42  session_id=chat-001       ← 横切维度挂在 Trace 上
│   tags=[rag, prod]  metadata={region: cn}
│
├─ Event:      "cache-miss"                     ← 瞬时点，无时长
│
├─ Span:       "retrieve"                       ← 工作单元（向量检索，有耗时）
│   ├─ Generation: "embed-query"               ← LLM 调用（嵌入模型）→ 记 usage
│   │     model=text-embedding-3-small
│   └─ Span:        "vector-search"            ← 子工作单元（查向量库）
│
├─ Span:       "rerank"                         ← 工作单元（重排，有耗时）
│
├─ Generation: "answer"                         ← 核心 LLM 调用 → 记 token + cost
│     model=gpt-4o  usage={input:1840, output:96}
│     └─ score: "hallucination" = 0.02          ← Score 挂在这个 generation 上
│
└─ Span:       "guardrail-check"                ← 工作单元（输出校验）

     trace 级 score: "user-feedback" = good     ← Score 也可挂在整条 Trace 上

搞清楚每个节点该落在树的哪一层，比急着写埋点代码更省时间——埋错层，整个成本与质量看板都是错的。

— 本章小结