使用 LangFuse 追踪 LLM 调用链路与成本
使用 LangFuse 追踪 LLM 调用链路与成本
一、问题背景：你的 LLM 应用正在「裸奔」
2026 年，几乎所有应用都在接入大模型。Chat 机器人、代码助手、RAG 问答、Agent 工作流——LLM 已经从实验项目变成了生产系统的核心组件。然而一个尴尬的现实是：绝大多数团队对自己线上 LLM 调用的情况一无所知。
笔者最近接手了一个基于 LangChain 搭建的企业知识库问答系统，上线三个月后 PM 来问：「咱们每个月到底花了多少 Token？用户问的问题都靠谱吗？有没有被 Prompt Injection 攻击过？」面对三连问，团队只能面面相觑——日志散落在各个微服务的 stdout 里，token 用量全凭 OpenAI 后台账单倒推，至于每条链路的实际表现，完全是个黑盒。
这不是个例。LLM 应用的可观测性面临着与微服务时代完全不同的挑战：
- 不确定性：同一个 Prompt 两次调用可能得到不同结果，如何判断是模型波动还是系统退化？
- 嵌套调用：Agent 内部可能触发多次 LLM 调用（思考→工具选择→执行→总结），故障定位极难
- 成本黑洞：GPT-4 一次复杂 Agent 调用可能消耗 $0.5，没有细粒度追踪根本不知道钱花在哪
- Prompt 版本管理混乱：改了某个 system prompt 之后性能是变好了还是变差了？没有基线对比
经过调研，我选择了LangFuse作为团队的 LLM 可观测性平台。这篇文章将分享从选型、接入到生产落地的完整经验。
二、为什么是 LangFuse？
LLM 可观测性赛道目前有几个主要选手：
| 方案 | 优势 | 劣势 |
|---|---|---|
| LangSmith | LangChain 原生集成，功能全面 | 闭源收费，免费版用量限制严格 |
| Weights & Biases | 实验追踪能力强 | LLM 场景不够专精，配置复杂 |
| Phoenix (Arize) | 开源，OpenTelemetry 原生 | 部署运维成本高，文档相对薄弱 |
| LangFuse | 开源自部署，LangChain/LlamaIndex 深度集成，UI 直观 | 社区相对较新（2023 年成立） |
LangFuse 打动我们的几个点：
- 真正的开源：MIT 协议，可以 Docker 一键自部署，数据不出企业内网
- 框架无关：原生支持 LangChain、LlamaIndex、OpenAI SDK，也可以通过装饰器或 context manager 自定义接入
- 三大核心能力合一：Tracing（调用链追踪）+ Prompt Management（Prompt 版本管理）+ Evaluation（评估），不需要同时维护三套系统
- 成本追踪：细粒度到单次调用的 token 消耗和金额统计，支持 OpenAI / Anthropic / Google 等多厂商
三、快速接入：5 分钟跑通第一条 Trace
3.1 部署 LangFuse
如果只是验证，直接用官方 Cloud 版（cloud.langfuse.com）注册即可，有免费额度。生产环境建议自部署：
# 克隆仓库gitclone https://github.com/langfuse/langfuse.gitcdlangfuse# 启动（依赖 Docker Compose）cp.env.dev.example .envdockercompose up-d启动后访问http://localhost:3000，注册账号，在 Settings → API Keys 创建密钥。
3.2 Python SDK 接入
pipinstalllangfuse设置环境变量：
exportLANGFUSE_SECRET_KEY="sk-lf-..."exportLANGFUSE_PUBLIC_KEY="pk-lf-..."exportLANGFUSE_HOST="http://localhost:3000"# 自部署时指定最简接入——直接装饰你的函数：
fromlangfuseimportLangfusefromlangfuse.decoratorsimportobserve langfuse=Langfuse()@observe()defask_llm(question:str)->str:# 你的 LLM 调用逻辑response=openai.chat.completions.create(model="gpt-4",messages=[{"role":"user","content":question}])returnresponse.choices[0].message.content# 调用answer=ask_llm("什么是 Transformer？")langfuse.flush()# 确保数据发送完毕运行后刷新 LangFuse UI，第一条 Trace 已经出现在面板上了。这个@observe()装饰器会自动捕获函数名、入参、返回值、耗时，形成一个 Span。
3.3 与 LangChain 深度集成
如果你的项目用 LangChain，接入更简单——LangFuse 提供了 Callback Handler，一行代码搞定：
fromlangfuse.callbackimportCallbackHandler# 创建 callbacklangfuse_handler=CallbackHandler()# 传给 LangChainfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate llm=ChatOpenAI(model="gpt-4")prompt=ChatPromptTemplate.from_template("将以下内容翻译成英文：{text}")chain=prompt|llm result=chain.invoke({"text":"今天天气真好"},config={"callbacks":[langfuse_handler]})LangFuse Callback 会自动拆解 LangChain 的执行流程：RunnableSequence → ChatPromptTemplate → ChatOpenAI，每一步的耗时、token 用量都会被精准记录。这在排查「是 Prompt 构造慢了还是模型推理慢了」时价值巨大。
四、核心能力深入
4.1 Tracing：不止于「看调用」
Tracing 是 LangFuse 的基石，但它做得比简单的「记录输入输出」深得多。
自定义 Span 层级：对于复杂的 Agent 工作流，建议手动定义 Span 结构，而不是依赖自动追踪：
fromlangfuse.decoratorsimportlangfuse_context,observe@observe()defagent_workflow(user_query: