何时使用
- 需要给 LLM 应用做生产级可观测:记录每次调用的输入/输出、模型、token 成本、延迟,并按 user/session 串联链路。
- 需要对输出做质量评估:人工打分、LLM-as-judge、或基于数据集的回归评估。
- 需要把提示词从代码里解耦,做版本化、按标签(production/staging)灰度部署。
- 已用 OpenAI / LangChain / LlamaIndex / Anthropic / Vercel AI SDK,想低改造接入追踪。
不该用的边界:
- 这不是通用 APM / 日志聚合 / 指标平台,不要拿它替代 Prometheus、ELK、Datadog 去监控非 LLM 的服务。
- 自托管需要自备基础设施;高吞吐场景需额外优化,实时看板有延迟,别用于强实时告警。
- 不替代环境内的真实测试与专家评审;评分逻辑(尤其 LLM-as-judge)本身需要校准。
- 缺少 API Key、账户或成功标准不明确时,先停下来确认,不要盲目埋点。
步骤
- 准备:注册 Langfuse Cloud 或自托管,拿到
public_key/secret_key/host;安装 SDK(Python 或 JS/TS)。 - 选择接入方式:
- 最省事 → OpenAI 直接替换
from langfuse.openai import openai,调用即自动追踪。 - 函数式应用 → 用
@observe()装饰器,嵌套调用自动成为 span。 - LangChain → 注入
CallbackHandler到config={"callbacks": [...]}。 - 需要精细控制 → 手动
trace()→generation()→generation.end()。
- 最省事 → OpenAI 直接替换
- 打分:用
trace.score()记录质量/反馈指标(数值、布尔、或 LLM 评分)。 - 提示词托管:UI/API 创建提示词,代码里
get_prompt(...).compile(...),并把prompt传给 generation 关联版本。 - 评估:建数据集 → 遍历 items 生成响应 → 打分 →
item.link(trace, run_name),对比提示词版本,部署最优。 - 退出前
langfuse.flush()(serverless / 短生命周期进程必做,否则丢数据)。
指令
- 初始化客户端:
Langfuse(public_key, secret_key, host);环境变量已配则可空参Langfuse()。 - 创建链路:
langfuse.trace(name, user_id, session_id, metadata, tags),session_id用于聚合一组相关 trace。 - 记录生成:
trace.generation(name, model, model_parameters, input, prompt=...),结束时generation.end(output, usage={"input": prompt_tokens, "output": completion_tokens})。 - 打分:
trace.score(name, value, comment, data_type="BOOLEAN");数值型用 0–1 标度,二值用 0/1。 - 装饰器内更新与打分:
langfuse_context.update_current_trace(...)、langfuse_context.score_current_trace(...)。 - 提示词写入/读取:
create_prompt(name, prompt, config, labels=["production"]);get_prompt(name, label="production")。
示例
OpenAI 一行接入(自动追踪,支持流式/异步):
from langfuse.openai import openai # OpenAI 客户端的即插即用替换
response = openai.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
name="greeting", session_id="session-123", user_id="user-456",
tags=["test"], metadata={"feature": "chat"},
)
手动埋点 + 成本与打分:
trace = langfuse.trace(name="chat-completion", user_id="user-123", session_id="session-456")
gen = trace.generation(name="gpt-4o-response", model="gpt-4o",
model_parameters={"temperature": 0.7},
input={"messages": [{"role": "user", "content": "Hello"}]})
resp = openai.chat.completions.create(model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}])
gen.end(output=resp.choices[0].message.content,
usage={"input": resp.usage.prompt_tokens, "output": resp.usage.completion_tokens})
trace.score(name="user-feedback", value=1, comment="User clicked helpful")
langfuse.flush() # serverless 退出前必须 flush
装饰器模式(嵌套自动成 span):
from langfuse.decorators import observe, langfuse_context
@observe() # 创建 trace
def chat_handler(user_id, message):
ctx = get_context(message) # 子 span
return generate_response(message, ctx)
@observe(as_type="generation") # LLM 生成 span
def generate_response(message, context):
r = openai.chat.completions.create(model="gpt-4o", messages=[
{"role": "system", "content": f"Context: {context}"},
{"role": "user", "content": message}])
return r.choices[0].message.content
数据集评估(关联 expected_output 并对比版本):
dataset = langfuse.get_dataset("support-qa-v1")
for item in dataset.items:
response = generate_response(item.input["question"])
trace = langfuse.trace(name="eval-run")
trace.generation(name="response", input=item.input, output=response)
trace.score(name="similarity", value=calculate_similarity(response, item.expected_output))
item.link(trace, "eval-run-1") # 把 trace 挂到数据集项
注意事项
- flush 是高频踩坑点:serverless / 脚本 / 短进程结束前不调用
langfuse.flush()会丢追踪数据。 session_id串联多轮对话,user_id关联用户,二者是后续按会话/用户分析的基础,埋点时尽早带上。- 打分标度要统一:同名 score 在数值/布尔间混用会让看板失真;用
data_type明确类型。 - LLM-as-judge 用更便宜的模型(如 gpt-4o-mini)评估以控成本,但要对评分一致性做校准。
- 提示词用
label(production/staging/development)做环境隔离,并把prompt传入 generation 以追溯版本,便于做 A/B 与回归。 - 自托管落地需要基础设施投入;高 QPS 场景关注批量上报与采样,避免追踪本身拖慢主链路。
互见
- 智能/agent 相关:LangGraph / CrewAI(构建被监控的 Agent / 多智能体)。
- 智能/misc:结构化输出 structured-output(抽取流程的可观测与评估)。
- 典型编排:LangGraph Agent + Langfuse 回调全链路追踪;RAG 管线追踪检索与生成并对相关性/准确性打分;带数据集评估的 Agent 系统对比提示词版本后部署最优。
采编自 sickn33/antigravity-awesome-skills(MIT)。