快速上手：SDK 接入与第一条 Trace

三种接入不是三选一，而是分层叠加：装饰器负责「业务结构」（哪个函数是哪一步），openai 集成负责「LLM 细节」（哪次调用花了多少 token）。最佳实践是外层用 @observe 划分 trace 结构，内层用 langfuse.openai 自动记录每次模型调用，二者在同一 trace 里自动父子关联，无需手动串联。

前置：安装与环境变量

本页所有示例基于 Python SDK v3（langfuse>=3.0，底层已切换到 OpenTelemetry）与 JS SDK v3。先装包、再配三个环境变量（云版从 Project Settings 拿 key，自托管把 host 换成你的地址）。

# Python：核心 SDK
pip install "langfuse>=3.0.0"
# 若用 OpenAI 集成，额外装 openai
pip install openai

# 三个环境变量（云版 EU 区示例；US 区用 us.cloud.langfuse.com；自托管填你的地址）
export LANGFUSE_PUBLIC_KEY="pk-lf-xxxxxxxx"
export LANGFUSE_SECRET_KEY="sk-lf-xxxxxxxx"
export LANGFUSE_HOST="https://cloud.langfuse.com"

# 调用真实 OpenAI 还需要
export OPENAI_API_KEY="sk-xxxxxxxx"

方式一：@observe 装饰器（自动捕获函数 I/O）

@observe() 是最快的零侵入方式：给函数加一行注解，Langfuse 自动把函数入参记成 input、返回值记成 output，并按调用栈生成嵌套结构——最外层被装饰的函数成为 trace 的根 span，里面再调用的被装饰函数自动成为子 span。

from langfuse import observe, get_client

# 内层函数：自动成为子 span，入参/返回值被捕获
@observe()
def retrieve_context(question: str) -> str:
    # 模拟一次检索
    return f"[doc] 关于 '{question}' 的背景资料"

# 内层函数：另一个子 span
@observe()
def build_answer(question: str, context: str) -> str:
    return f"针对「{question}」，根据 {context} 给出回答。"

# 最外层函数：成为 trace 的根 span
@observe()
def rag_pipeline(question: str) -> str:
    context = retrieve_context(question)   # 自动成为子 span
    answer = build_answer(question, context)  # 自动成为子 span
    return answer

if __name__ == "__main__":
    result = rag_pipeline("Langfuse 是什么？")
    print(result)

    # 关键：短生命周期脚本必须 flush，把异步队列里的数据强制上报
    get_client().flush()

把函数标成 generation 并补字段

默认装饰器产出的是普通 span。如果这个函数本身就是一次 LLM 调用，用 @observe(as_type='generation') 把它标成 generation，再用 update_current_observation 补上 model / usage_details / input / output 等字段——这样 UI 才会按 generation 的样式展示 token 与成本。

from langfuse import observe, get_client

# as_type='generation' 让这个 observation 在 UI 里按「模型调用」展示
@observe(as_type="generation")
def call_my_llm(prompt: str) -> str:
    # 假设这是你自己封装的、非 OpenAI 的模型调用
    completion = "这是模型返回的文本"
    prompt_tokens, completion_tokens = 42, 18

    client = get_client()
    # 在当前 observation 上补充 generation 专有字段
    client.update_current_generation(
        model="my-custom-model-v1",
        input=prompt,
        output=completion,
        usage_details={
            "input": prompt_tokens,
            "output": completion_tokens,
            "total": prompt_tokens + completion_tokens,
        },
        metadata={"temperature": 0.7},
    )
    return completion

if __name__ == "__main__":
    call_my_llm("用一句话解释 Langfuse")
    get_client().flush()

方式二：langfuse.openai 一行替换（自动追 token 与成本）

如果你已经在用 OpenAI 官方 SDK，这是侵入性最低的方式：把 from openai import OpenAI 换成 from langfuse.openai import OpenAI，其余代码一字不改。Langfuse 会拦截每次调用，自动记录 model、prompt/completion、token 用量并按价目表算成本——全部出现在 UI 的 generation 里。

# 唯一改动：import 来源从 openai 换成 langfuse.openai
from langfuse.openai import OpenAI
from langfuse import observe, get_client

client = OpenAI()  # 用法与官方 SDK 完全相同

# 叠加用法：外层 @observe 划 trace 结构，内层自动追踪每次模型调用
@observe()
def ask(question: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是简洁的助手"},
            {"role": "user", "content": question},
        ],
        # langfuse 专有参数：直接挂在调用上，给这条 generation 命名/打标
        name="ask-llm",
        metadata={"feature": "qa"},
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    print(ask("什么是可观测性？"))
    get_client().flush()

方式三：JS/TS 用 observeOpenAI 包裹客户端

JS/TS 端语义与 Python 的 langfuse.openai 完全对齐：不替换 import，而是用 observeOpenAI() 包裹一个已有的 OpenAI 客户端实例，返回的代理对象用法不变，但每次调用自动上报。

# JS/TS：装 langfuse 集成包与 openai
npm install langfuse openai
# 或 pnpm add langfuse openai / yarn add langfuse openai

import OpenAI from "openai";
import { observeOpenAI } from "langfuse";

// 环境变量：LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_BASEURL
// 用 observeOpenAI 包裹原始客户端，得到一个自动追踪的代理
const openai = observeOpenAI(new OpenAI(), {
  // 可选：给本次追踪命名、挂 session / user
  generationName: "ask-llm",
  sessionId: "session-123",
  userId: "user-abc",
  metadata: { feature: "qa" },
});

async function main() {
  const res = await openai.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [
      { role: "system", content: "你是简洁的助手" },
      { role: "user", content: "什么是可观测性？" },
    ],
  });
  console.log(res.choices[0].message.content);

  // 与 Python 同理：进程退出前 flush，确保异步上报完成
  await openai.flushAsync();
}

main();

在 UI 查看第一条 Trace

1. 运行上面任一脚本，确认控制台打印出模型/函数返回，且没有抛错。
2. 打开 Langfuse（云版 https://cloud.langfuse.com，自托管为你的 host），进入对应 Project。
3. 左侧导航点 Tracing → Traces，列表顶部应出现刚才那条（按时间倒序，几秒内到）。
4. 点开 trace：左边是 span/generation 的树状结构，右边显示每个节点的 Input / Output；generation 节点会额外展示 model、token 用量和换算后的成本。
5. 看不到？先确认脚本最后调了 flush；再核对 host 与 key 指向同一个 Project（key 区分项目，填错会上报到别处）。

现在你已经能让一条 Trace 出现在 UI 里了。下一章深入 Tracing：如何手动控制嵌套 span 的边界、用 update_current_trace 补 trace 级元数据，并把多次调用按 Session 串成会话、按 User 维度聚合。