阶段 4：Python SDK 快速上手

预计学习时间：1-2 天
SDK 版本：本笔记以 langfuse>=2,<3 为主（v2 有状态 API）

---

目录

1. 环境准备
2. 最小示例
3. 核心 API 详解
4. 嵌套结构实战
5. @observe 装饰器
6. 异步上报与 flush
7. 常见错误排查
8. 学习检查

---

1. 环境准备

1.1 安装

pip install "langfuse>=2,<3"

1.2 获取密钥

在 Langfuse Cloud → Settings → API Keys 创建：

• LANGFUSE_PUBLIC_KEY（形如 pk-lf-...）
• LANGFUSE_SECRET_KEY（形如 sk-lf-...）

1.3 配置环境变量

export LANGFUSE_PUBLIC_KEY="pk-lf-xxx"
export LANGFUSE_SECRET_KEY="sk-lf-xxx"
export LANGFUSE_HOST="https://cloud.langfuse.com"   # 自托管则改成你的域名

生活类比：公钥像「门店编号」，私钥像「店长钥匙」——缺一不可。

---

2. 最小示例

完整可运行代码见 examples/basic_trace.py。

import os
from langfuse import Langfuse

lf = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
)

# 1. 创建 Trace（一张订单）
trace = lf.trace(
    name="demo-chat",
    user_id="user_001",
    input={"question": "北京天气怎么样？"},
    tags=["demo", "learning"],
)

# 2. 创建 Generation（LLM 调用）
generation = trace.generation(
    name="gpt-answer",
    model="gpt-4o",
    input="用户问：北京天气怎么样？",
    model_parameters={"temperature": 0.7},
)

# 模拟 LLM 返回
answer = "北京今天晴，25°C。"
generation.end(
    output=answer,
    usage={
        "input": 20,
        "output": 15,
        "total": 35,
    },
)

# 3. 结束 Trace
trace.update(output={"answer": answer})

# 4. 必须 flush，否则进程退出时数据可能丢失
lf.flush()
print("Done! 去 Langfuse UI 查看 Trace: demo-chat")

运行后，在 Langfuse → Traces 页面搜索 demo-chat。

---

3. 核心 API 详解

3.1 Langfuse 客户端

lf = Langfuse(
    public_key="...",
    secret_key="...",
    host="https://cloud.langfuse.com",
    flush_at=15,        # 累积 15 条事件后自动 flush
    flush_interval=0.5, # 或每 0.5 秒 flush 一次
)

3.2 trace() — 创建一条 Trace

trace = lf.trace(
    id="custom-trace-id",   # 可选，自定义 ID（分布式场景有用）
    name="operation-name",  # UI 里显示的名字
    user_id="user_123",
    session_id="sess_abc",
    input={...},            # 业务输入
    output={...},           # 业务输出（也可最后 update）
    metadata={...},
    tags=["prod"],
)

3.3 generation() — 记录 LLM 调用

gen = trace.generation(
    name="openai-call",
    model="gpt-4o",
    input=messages,               # str 或 list[dict]
    model_parameters={"temperature": 0.5},
    metadata={"retry": 0},
)

# ... 调用 LLM ...

gen.end(
    output=completion_text,
    usage={"input": 100, "output": 50, "total": 150},
    level="DEFAULT",   # 或 "WARNING" / "ERROR"
    status_message="ok",
)

关键：generation 必须用 .end() 结束，才会在 UI 里显示完整耗时和 token。

3.4 span() — 记录普通步骤

span = trace.span(
    name="parse-intent",
    input={"raw": "查天气"},
)

# ... 业务逻辑 ...

span.end(output={"intent": "weather_query"})

3.5 event() — 记录瞬时事件

trace.event(
    name="cache-hit",
    metadata={"key": "weather:beijing"},
)

适合不需要持续时间的点状事件，像「叮咚，门铃响了一下」。

---

4. 嵌套结构实战

4.1 推荐写法：父子嵌套

trace = lf.trace(name="agent-run", input=user_question)

# Agent 思考
plan_span = trace.span(name="planning")
plan_span.end(output={"steps": ["查天气", "生成回复"]})

# Tool 调用
tool_span = trace.span(
    name="tool:weather-api",
    input={"city": "北京"},
    metadata={"type": "tool"},
)
tool_span.end(output={"temp": 25, "weather": "晴"})

# LLM 生成
gen = trace.generation(name="final-answer", model="gpt-4o", input=prompt)
gen.end(output=answer, usage={...})

trace.update(output=answer)
lf.flush()

4.2 树状结构示意

agent-run
├── planning (span)
├── tool:weather-api (span)
└── final-answer (generation)

4.3 生活类比

做一道「宫保鸡丁」：

订单（trace）
├── 备料（span）
├── 腌制（span）
├── 爆炒（generation）← 最关键的「火候」环节
└── 装盘（span）

---

5. @observe 装饰器

SDK 提供装饰器，自动为函数创建 Observation：

from langfuse.decorators import observe, langfuse_context

@observe()
def fetch_weather(city: str) -> dict:
    # 自动创建 span，函数参数作为 input
    return {"city": city, "temp": 25}

@observe()
def generate_answer(question: str) -> str:
    langfuse_context.update_current_observation(
        model="gpt-4o",
    )
    return "北京今天 25°C"

@observe()
def handle_user_query(question: str) -> str:
    weather = fetch_weather("北京")       # 自动嵌套
    return generate_answer(question)      # 自动嵌套

handle_user_query("北京天气？")

优点：代码侵入小
注意：装饰器方案对复杂 Agent 流程可能不够灵活，需要手动 langfuse_context 补充 model/usage。

---

6. 异步上报与 flush

6.1 为什么需要 flush

Langfuse SDK 不会在每次 generation.end() 时立刻发 HTTP 请求。它先在本地队列攒一批，后台线程定时发送。

生活类比：快递员不会每卖出一本书就上门取件，而是攒够一整车再发。

6.2 什么时候必须 flush

场景	是否需要 flush
长期运行的 Web 服务	通常不需要，后台自动 flush
短脚本 / 单元测试	✅ 必须 lf.flush()
CI 流水线	✅ 必须
进程即将退出	✅ 必须

import atexit

lf = Langfuse()
atexit.register(lf.flush)  # 进程退出时自动 flush

6.3 同步等待

lf.flush()  # 阻塞直到队列清空

---

7. 常见错误排查

错误 1：UI 里看不到数据

检查清单：

1. 环境变量 LANGFUSE_PUBLIC_KEY / SECRET_KEY 是否正确
2. 脚本末尾是否调用了 lf.flush()
3. LANGFUSE_HOST 是否指向正确的 Server（Cloud vs 自托管）
4. 项目是否选对（Cloud 左上角 Project 切换）

错误 2：Latency 显示 0.00s

v2 API 需要显式传时间，或确保调用了 .end()：

from datetime import datetime, timezone

gen = trace.generation(name="call", model="gpt-4o", input="hi")
gen.end(
    output="hello",
    end_time=datetime.now(timezone.utc),  # 必要时手动指定
)

错误 3：401 Unauthorized

密钥错误或 host 不对。自托管环境常见问题是 LANGFUSE_HOST 少了 https:// 或端口。

错误 4：数据重复

Hook 和 Watcher 同时启用会导致重复上报（见 6_integration_cases.md）。

---

8. 学习检查

• [ ] 成功配置环境变量并运行 examples/basic_trace.py
• [ ] 能在 UI 里找到 trace 并看到 generation 的 token
• [ ] 能手写 trace → span → generation 三层嵌套
• [ ] 理解为什么短脚本必须 flush()
• [ ] 跑通 examples/rag_agent_trace.py

下一章 → 5_platform_features.md