目录
- 一、整体目的概述
- 二、代码逐段解说
- 1️⃣ 获取 Langfuse 客户端
- 2️⃣ 创建一个 Span（业务处理阶段）
- 3️⃣ 更新 Span 的输出信息
- 4️⃣ 在 Span 内创建一个 Generation（LLM 调用）
- 5️⃣ 更新 Generation 的输出
- 6️⃣ 自动关闭所有 Observation
- 7️⃣ 显式 Flush（短生命周期程序必需）
- 三、典型使用场景总结
下面对这段 Langfuse 代码进行逐行、分层次的技术解说，重点说明其观测模型（Observation Model）、Span / Generation 的语义差异以及典型使用场景。
一、整体目的概述
这段代码演示了如何手动使用 Langfuse SDK 对一次业务处理流程进行可观测性埋点，包括：
- 创建一个顶层 Span，表示一次业务处理过程
- 在该 Span 内部嵌套一个 Generation，表示一次 LLM 调用
- 自动记录执行时间、层级关系、输入输出
- 在短生命周期程序中显式
flush数据
适用于：
- LangGraph / Agent / Workflow 中的关键节点埋点
- 希望比
@observe装饰器更精细地控制 trace 结构的场景
二、代码逐段解说
1️⃣ 获取 Langfuse 客户端
from langfuse import get_clientlangfuse = get_client()
作用说明：
get_client()会基于当前环境变量（如LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY、LANGFUSE_HOST）初始化 Langfuse 客户端- 该客户端是创建 Trace / Span / Generation 的入口
在 Web 服务中通常是全局单例；在脚本或任务中则按需创建。
2️⃣ 创建一个 Span（业务处理阶段）
with langfuse.start_as_current_observation(as_type="span",name="process-request"
) as span:
关键点解释：
| 参数 | 含义 |
|---|---|
as_type="span" |
创建一个 Span 类型的 Observation |
name="process-request" |
Span 的逻辑名称，表示业务阶段 |
Span 的语义：
-
表示一个非 LLM 的逻辑处理单元
-
典型用途：
- 请求处理
- 规则判断
- 意图识别
- 多步骤 pipeline 中的某一环
上下文管理器的意义：
- 进入
with块时：Span 自动 start - 退出
with块时：Span 自动 end - 无需手动调用
end()，避免遗漏
3️⃣ 更新 Span 的输出信息
span.update(output="Processing complete")
作用说明：
-
向当前 Span 追加或更新元数据
-
output通常用于记录：- 处理结果摘要
- 状态描述
- 关键中间结论
在真实项目中，这里往往是结构化 JSON，例如规则命中情况、决策结果等。
4️⃣ 在 Span 内创建一个 Generation（LLM 调用）
with langfuse.start_as_current_observation(as_type="generation",name="llm-response",model="gpt-3.5-turbo"
) as generation:
Generation 与 Span 的核心区别：
| 维度 | Span | Generation |
|---|---|---|
| 语义 | 业务逻辑阶段 | LLM 调用 |
| 关注点 | 流程、规则、耗时 | Prompt / Output / Token |
| Langfuse UI | 普通节点 | 可查看 Prompt / Completion |
参数说明：
as_type="generation"
→ 明确告知 Langfuse：这是一次 模型生成行为model="gpt-3.5-turbo"
→ 用于统计不同模型的调用情况与成本
层级关系：
- 该 Generation 会自动成为当前 Span 的子节点
- 在 Langfuse Trace UI 中呈现为树状结构
5️⃣ 更新 Generation 的输出
generation.update(output="Generated response")
实际生产中通常会包含：
-
LLM 返回的文本
-
或结构化 JSON（如分类结果、评分）
-
有时还会同时更新：
input（prompt）metadata（temperature、top_p 等）
这样可以在 Langfuse 中完整回放一次 LLM 调用。
6️⃣ 自动关闭所有 Observation
# All spans are automatically closed when exiting their context blocks
机制说明：
-
所有
start_as_current_observation都基于上下文栈 -
with结束即自动：- 结束当前 Span / Generation
- 恢复父级 Observation
-
避免手动管理层级与生命周期错误
7️⃣ 显式 Flush（短生命周期程序必需）
langfuse.flush()
为什么需要？
-
Langfuse 默认是异步批量上报
-
在以下场景必须调用
flush()：- CLI 脚本
- Cron / Batch Job
- Serverless / FaaS
- Notebook 执行完即退出
否则可能导致：
- Trace 尚未发送，进程已结束
- Langfuse 控制台看不到数据
三、典型使用场景总结
这类写法非常适合你当前关注的场景，例如：
-
LangGraph 中：
- 每个 Node = 一个 Span
- LLM Node = Generation
-
销售 Agent 中：
- 决策判断（Span）
- 话术生成（Generation）
-
对 Objection / Intent 分类效果做可观测与回溯
相比 @observe：
- 更底层
- 更可控
- 适合复杂流程与多层嵌套
如果你希望，我可以进一步：
- 将这段代码改写成 LangGraph Node 标准模板
- 给出 Span / Generation 命名规范建议
- 或结合你当前的 销售智能体架构，给出一套推荐的 Langfuse 埋点结构设计