Langfuse v3 Python SDK 构建在 OpenTelemetry 之上，心智模型只有三条：①
start_as_current_span/start_as_current_generation是上下文管理器，with块进入即开始、退出即结束并自动记录耗时；②嵌套靠with的词法嵌套自动建立父子关系——内层with自动成为外层的 child，你不用手动传 parent；③span是通用工作单元，generation是它的特化版（专用于 LLM 调用，多了model / model_parameters / usage_details / cost_details字段）。记住：model 和 usage 只属于 generation，普通 span 没有这些字段。
一、两个核心上下文管理器：span 与 generation
先装包并初始化。v3 SDK 用全局 get_client() 拿单例客户端，凭证从环境变量读取，无需每次显式传 key。start_as_current_span 用于检索、解析、工具调用等非 LLM 的工作单元；start_as_current_generation 用于每一次 LLM 请求。两者签名几乎一致，区别在 generation 多了 LLM 专属字段。
# Langfuse v3 Python SDK（基于 OpenTelemetry）
pip install -U "langfuse>=3.0.0"
# 凭证写进环境变量（云版在 cloud.langfuse.com 项目设置里拿；自托管填你的 host）
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com" # 自托管改成 http://localhost:3000from langfuse import get_client
# get_client() 返回进程内单例，凭证从 LANGFUSE_* 环境变量自动读取
langfuse = get_client()
# 建议启动时验证连通性（凭证错 / host 不通会在这里暴露）
assert langfuse.auth_check(), "Langfuse 认证失败，检查 PUBLIC/SECRET KEY 与 HOST"
# --- 一个普通 span：包裹一段非 LLM 的工作（这里模拟检索）---
with langfuse.start_as_current_span(
name="retrieve-docs", # 在 UI 里显示的 observation 名字
input={"query": "什么是 Langfuse"}, # 入参，任意可 JSON 序列化对象
) as span:
# ... 这里执行真实检索逻辑 ...
docs = ["Langfuse 是开源 LLM 工程平台", "提供可观测/评估/Prompt 管理"]
# 工作完成后回填 output（也可在创建时就传，但检索结果通常事后才有）
span.update(
output={"docs": docs, "hit_count": len(docs)},
metadata={"retriever": "bm25", "top_k": 2},
)
# 退出 with 块时 span 自动结束并记录耗时。
# 短命脚本结尾务必 flush，确保数据在进程退出前上报：
langfuse.flush()from langfuse import get_client
langfuse = get_client()
# --- 一个 generation：专门记录一次 LLM 调用，带 model / usage / cost ---
with langfuse.start_as_current_generation(
name="answer-llm",
model="gpt-4o-mini", # 模型名（Langfuse 据此算成本）
model_parameters={"temperature": 0.2, "max_tokens": 512},
input=[ # 通常传 chat messages 列表
{"role": "system", "content": "你是简洁的助手"},
{"role": "user", "content": "用一句话解释 Langfuse"},
],
) as gen:
# ... 调用你的 LLM，拿到回复与 token 用量 ...
completion_text = "Langfuse 是开源的 LLM 可观测与评估平台。"
prompt_tokens, completion_tokens = 28, 17
# 手动回填输出与用量。usage_details 的 key 是自定义的，
# input/output/total 是 Langfuse 识别成本的约定 key。
gen.update(
output=completion_text,
usage_details={
"input": prompt_tokens,
"output": completion_tokens,
"total": prompt_tokens + completion_tokens,
},
# 若你知道单价也可直接传 cost_details，否则 Langfuse 按 model 自动估算
# cost_details={"input": 0.0000012, "output": 0.0000048},
)
langfuse.flush()二、嵌套：检索 span 内嵌 LLM generation
嵌套不需要任何手动连线——内层
with在外层with的作用域里打开，OpenTelemetry 的上下文会自动把内层 observation 挂到外层之下。这正是手动 tracing 的威力：你想要几层就嵌几层，结构与代码的缩进一一对应。下面这段是经典 RAG 形状：外层rag-pipelinespan 包住整个流程，里面先是retrievespan，再是generategeneration。
from langfuse import get_client
langfuse = get_client()
def rag_answer(question: str) -> str:
# 外层 span：代表整个 RAG 流程，是本次所有子步骤的父节点
with langfuse.start_as_current_span(
name="rag-pipeline",
input={"question": question},
) as root:
# —— 子步骤 1：检索（普通 span，自动成为 rag-pipeline 的 child）——
with langfuse.start_as_current_span(name="retrieve") as retrieve_span:
docs = [
"Langfuse 提供 tracing、评估与 prompt 管理。",
"v3 SDK 基于 OpenTelemetry。",
]
retrieve_span.update(
input={"query": question},
output={"docs": docs},
metadata={"top_k": 2, "store": "qdrant"},
)
context = "\n".join(docs)
# —— 子步骤 2：LLM 生成（generation，同样自动挂在 rag-pipeline 下）——
with langfuse.start_as_current_generation(
name="generate",
model="gpt-4o-mini",
input=[
{"role": "system", "content": f"仅依据以下资料回答：\n{context}"},
{"role": "user", "content": question},
],
) as gen:
answer = "Langfuse 是基于 OpenTelemetry 的开源 LLM 工程平台。"
gen.update(
output=answer,
usage_details={"input": 60, "output": 22, "total": 82},
)
# 把最终答案回填到外层 span 作为整条流程的 output
root.update(output={"answer": answer})
return answer
print(rag_answer("什么是 Langfuse？"))
langfuse.flush()
# UI 里你会看到三层结构：
# rag-pipeline (span)
# ├─ retrieve (span)
# └─ generate (generation, 带 model/token/cost)三、Session 与 User：把零散调用聚合成会话与用户视图
session_id、user_id、tags、metadata 是 Trace 级属性，不是 span 级——它们描述「这一整条 Trace 属于哪个会话、哪个用户」。设置方式是在 Trace 的某个 span 上下文里调用 langfuse.update_current_trace(...)。同一个 session_id 的多条 Trace 会在 Langfuse 的 Sessions 页聚合成一条时间线；同一个 user_id 则在 Users 页聚合，并自动汇总该用户的成本与调用量。
from langfuse import get_client
langfuse = get_client()
def handle_turn(session_id: str, user_id: str, user_msg: str) -> str:
# 任意一个 span 上下文里调 update_current_trace，作用于「整条 Trace」
with langfuse.start_as_current_span(name="chat-turn", input=user_msg) as span:
# 关键：注入会话与用户维度 + 业务标签 + 元数据
langfuse.update_current_trace(
session_id=session_id, # 同一会话的多轮对话用同一个 id -> Sessions 聚合
user_id=user_id, # 同一用户 -> Users 页聚合成本/调用量
tags=["prod", "chatbot"], # 标签用于 UI 过滤
metadata={"channel": "web", "locale": "zh-CN"}, # 任意自定义元数据
input=user_msg, # 也能在这里覆盖 Trace 级 input/output
)
with langfuse.start_as_current_generation(
name="reply", model="gpt-4o-mini",
input=[{"role": "user", "content": user_msg}],
) as gen:
reply = f"收到：{user_msg}"
gen.update(output=reply, usage_details={"input": 12, "output": 6, "total": 18})
span.update(output=reply)
return reply
# 模拟同一会话、同一用户的多轮对话：三条 Trace 会聚合到一个 Session
sid, uid = "session-2026-0042", "user-alice"
handle_turn(sid, uid, "你好")
handle_turn(sid, uid, "Langfuse 能做什么？")
handle_turn(sid, uid, "怎么看会话维度？")
langfuse.flush()| 你想做的事 | 用哪个 API | 落在哪个层级 |
|---|---|---|
| 记录一段非 LLM 工作（检索/解析/工具） | start_as_current_span | Observation (span) |
| 记录一次 LLM 调用并统计 token/成本 | start_as_current_generation | Observation (generation) |
| 设置会话/用户/标签/Trace 级元数据 | update_current_trace | Trace |
| 给整条 Trace 打分（质量/反馈） | score_current_trace | Trace |
| 给当前这个 span/generation 单独打分 | score_current_span | Observation |
口诀干活→span/generation；描述整条链→update_current_trace；评价→score_current_*
四、给 Trace 挂分数：score_current_trace
Score 是 Langfuse 里质量与反馈的统一落点：后面评估章节的 LLM-as-judge、数据集实验、用户点赞点踩，最终都化为挂在 Trace 或 observation 上的 score。在手动 tracing 里，你用 score_current_trace() 给当前 Trace 打分（不用持有 trace 对象引用，它从 OTel 上下文找当前 Trace），用 score_current_span() 给当前 observation 打分。score 有三种 data_type：NUMERIC（数值）、BOOLEAN（是/否）、CATEGORICAL（枚举）。
from langfuse import get_client
langfuse = get_client()
with langfuse.start_as_current_span(name="qa-pipeline", input="问题X") as span:
answer = "这是模型的回答"
span.update(output=answer)
# 1) 数值分：比如内置启发式打分（答案长度合规度 0~1）
langfuse.score_current_trace(
name="length-ok",
value=0.9,
data_type="NUMERIC",
comment="长度在合理区间",
)
# 2) 布尔分：是否包含拒答
langfuse.score_current_trace(
name="refused",
value=False,
data_type="BOOLEAN",
)
# 3) 给「当前这个 span」单独打分（区别于整条 Trace）
langfuse.score_current_span(
name="relevance",
value="high",
data_type="CATEGORICAL",
)
langfuse.flush()
# 进阶：拿到 trace_id 后，可在请求结束、异步收集到用户反馈时"事后补分"：
# trace_id = span.trace_id # 在 with 块内可取
# langfuse.create_score(trace_id=trace_id, name="user-thumb",
# value=1, data_type="NUMERIC") # 比如用户点了赞五、完整示例：四步 Agent 的端到端嵌套埋点
把前面所有要素拼成一个可运行的多步 Agent：①检索（span）→②推理是否要用工具（generation）→③调用工具（span）→④基于工具结果总结（generation）。外层 Agent span 注入 session/user/tags，结束时打一个整体质量分。代码用占位的检索与工具函数，把真实 LLM/工具换上即可直接跑。
from langfuse import get_client
langfuse = get_client()
# —— 占位的外部依赖（换成你的真实实现）——
def search_kb(q: str) -> list[str]:
return ["北京今天多云", "气温 18-26 摄氏度"]
def call_weather_tool(city: str) -> dict:
return {"city": city, "temp": 23, "cond": "多云"}
def run_agent(question: str, session_id: str, user_id: str) -> str:
# 外层：整个 Agent 一条 Trace 的根 span
with langfuse.start_as_current_span(
name="weather-agent", input={"question": question}
) as root:
# 在根 span 上注入 Trace 级维度
langfuse.update_current_trace(
session_id=session_id,
user_id=user_id,
tags=["agent", "weather"],
metadata={"agent_version": "v2"},
)
# 步骤 1：检索（span）
with langfuse.start_as_current_span(name="step1-retrieve") as s1:
kb = search_kb(question)
s1.update(input={"q": question}, output={"kb": kb})
# 步骤 2：让 LLM 决定是否调用工具（generation）
with langfuse.start_as_current_generation(
name="step2-plan", model="gpt-4o-mini",
input=[{"role": "user",
"content": f"资料:{kb}\n问题:{question}\n要查天气工具吗?"}],
) as s2:
decision = {"use_tool": True, "tool": "weather", "city": "北京"}
s2.update(output=decision,
usage_details={"input": 80, "output": 15, "total": 95})
# 步骤 3：执行工具调用（span）——条件嵌套，结构随逻辑而变
tool_result = None
if decision["use_tool"]:
with langfuse.start_as_current_span(name="step3-tool") as s3:
tool_result = call_weather_tool(decision["city"])
s3.update(input=decision, output=tool_result,
level="DEFAULT") # level: DEFAULT/DEBUG/WARNING/ERROR
# 步骤 4：基于工具结果总结成最终回答（generation）
with langfuse.start_as_current_generation(
name="step4-summarize", model="gpt-4o-mini",
input=[{"role": "user",
"content": f"工具结果:{tool_result}\n请回答:{question}"}],
) as s4:
answer = f"{decision['city']}今天{tool_result['cond']}，约 {tool_result['temp']} 度。"
s4.update(output=answer,
usage_details={"input": 50, "output": 18, "total": 68})
# 回填整条 Trace 的最终输出 + 打一个整体质量分
root.update(output={"answer": answer})
langfuse.score_current_trace(name="answered", value=True, data_type="BOOLEAN")
return answer
print(run_agent("北京今天天气怎么样？", session_id="sess-77", user_id="user-bob"))
langfuse.flush()
# UI 中的结构：
# weather-agent (span) [session-77 / user-bob / tags / score=answered:true]
# ├─ step1-retrieve (span)
# ├─ step2-plan (generation, gpt-4o-mini, 95 tok)
# ├─ step3-tool (span)
# └─ step4-summarize (generation, gpt-4o-mini, 68 tok)✓推荐做法
- 一律用 start_as_current_* + with，让父子关系、计时、结束全自动
- LLM 调用用 generation 并填 model + usage_details，才有 token/成本分析
- 在根 span 里用 update_current_trace 注入 session_id / user_id / tags
- 脚本/请求结束前调用 langfuse.flush()，确保数据上报
- 把 level=ERROR 用在异常/降级分支，UI 里能一眼筛出问题 observation
✗不推荐
- 用普通 span 包 LLM 调用——会丢失 model、token、成本维度
- 在 span 上塞 session_id/user_id——它们是 Trace 级，要用 update_current_trace
- 在长驻服务里每条 Trace 都 flush——flush 是阻塞的，按批/按生命周期 flush
- 用裸 start_span 却忘了 .end()，导致 span 永不闭合、Trace 显示异常
- 把超大对象整个塞进 input/output——会拖慢上报且占用存储
⚠常见误区
- usage_details 的成本识别 key 约定是 input/output/total，拼错则成本算不出
- 异步/多线程下 with 词法嵌套不一定传递上下文，需用 parent 显式连或在同协程内
- update_current_trace 必须在某个 span 上下文内调用，否则找不到当前 Trace
- v2 SDK 的 trace()/span() 链式 API 与 v3 不兼容，照着旧文档写会报错
UI 中能看到与代码缩进一致的多层嵌套结构，每个 generation 显示正确的 model/token/cost，Trace 带上 session_id 与 user_id 并能在 Sessions / Users 页聚合，整条 Trace 与关键 observation 上都挂着可检索的 score。
看不见的链路无法被改进。手动 tracing 的全部价值，就是让你按真实业务结构而非框架默认结构去度量——结构对了，会话、用户、成本、分数才落得进同一张图。
— 本章小结