Skip to content

Langfuse 是 LLM 应用的可观测性平台,因 OpenRouter 兼容 OpenAI API schema,可直接使用 Langfuse 的 OpenAI SDK 包装器对 OpenRouter 调用进行自动追踪。核心配置:用 from langfuse.openai import openai 替换原有 openai 导入,将 base_url 指向 OpenRouter,即可在 Langfuse Dashboard 中查看每次请求的输入、输出、token 消耗和延迟。嵌套调用链可用 @observe() 装饰器捕获。

Langfuse 是面向 LLM 应用的可观测性与分析平台,帮助团队追踪、监控和调试 AI 模型调用。由于 OpenRouter 使用 OpenAI API schema,可以直接复用 Langfuse 与 OpenAI SDK 的原生集成,无需额外适配。

不想修改客户端代码?可以通过 OpenRouter Broadcast 在服务端自动将追踪数据发送到 Langfuse。

安装

bash
pip install langfuse openai

配置环境变量

python
import os

# Langfuse API key(在 Langfuse 控制台创建)
LANGFUSE_SECRET_KEY = "sk-lf-..."
LANGFUSE_PUBLIC_KEY = "pk-lf-..."
# EU 区域
LANGFUSE_HOST = "https://cloud.langfuse.com"
# US 区域:LANGFUSE_HOST = "https://us.cloud.langfuse.com"

# OpenRouter API key(作为 OPENAI_API_KEY 传入)
os.environ["OPENAI_API_KEY"] = "<OPENROUTER_API_KEY>"

基本用法:单次调用追踪

from langfuse.openai import openai 替换原有的 import openai,Langfuse 自动拦截并记录每次调用:

python
from langfuse.openai import openai

client = openai.OpenAI(
    base_url="https://openrouter.ai/api/v1",
    default_headers={
        "HTTP-Referer": "<YOUR_SITE_URL>",   # 可选:你的站点 URL
        "X-OpenRouter-Title": "<YOUR_SITE_NAME>",  # 可选:你的站点名称
    }
)

response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4.6",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Tell me a fun fact about space."}
    ],
    name="fun-fact-request"  # 可选:在 Langfuse 中显示的 generation 名称
)

print(response.choices[0].message.content)

name 参数会在 Langfuse Dashboard 中作为这条 generation 的标识名显示,便于后续筛选。

进阶追踪:嵌套调用链

当一个请求包含多个串联的 LLM 调用时,使用 @observe() 装饰器可以将整个调用链作为一个 trace 呈现,每个子调用作为 span:

python
from langfuse import observe
from langfuse.openai import openai

client = openai.OpenAI(
    base_url="https://openrouter.ai/api/v1",
)

@observe()
def analyze_text(text: str):
    summary = summarize_text(text).choices[0].message.content
    sentiment = analyze_sentiment(summary).choices[0].message.content
    return {"summary": summary, "sentiment": sentiment}

@observe()
def summarize_text(text: str):
    return client.chat.completions.create(
        model="openai/gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You summarize texts concisely."},
            {"role": "user", "content": f"Summarize:\n{text}"}
        ],
        name="summarize-text"
    )

@observe()
def analyze_sentiment(summary: str):
    return client.chat.completions.create(
        model="openai/gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You analyze text sentiment."},
            {"role": "user", "content": f"Analyze sentiment:\n{summary}"}
        ],
        name="analyze-sentiment"
    )

result = analyze_text("OpenRouter's unified API has advanced AI development.")
print(result)

@observe() 自动将嵌套函数调用树映射为 Langfuse 的 trace → span 层级结构,无需手动传递 trace context。

更多文档

常见问题

Q: Langfuse 包装器会影响 OpenRouter 的请求性能吗?

A: 几乎没有影响。Langfuse SDK 以异步方式上报追踪数据,不会阻塞主请求路径,通常增加不超过 1ms 的本地开销。

Q: 追踪数据存储在哪里,是否会将 API 响应内容发送给第三方?

A: 追踪数据发送到 Langfuse Cloud(EU 或 US 区域)。如需完全自托管,可以部署 Langfuse 开源版本,追踪数据不离开你的环境。

Q: 如何只追踪部分调用,而不是所有调用?

A: 只需在需要追踪的位置使用 from langfuse.openai import openai 创建的 client,其他位置继续用普通 openai client 即可,两者可以在同一个应用中共存。