、为什么要用 LangFuse当你把 LLM 应用从 Demo 推向生产一定会遇到这几个问题一次用户请求LLM 被调了几次每次的 prompt 和输出是什么哪次调用耗时最长哪次 token 消耗最多同一个用户的多轮对话能不能串起来看出了 bad case怎么快速定位是哪一步出了问题LangFuse 就是解决这些问题的。它是一个开源的 LLM 可观测性平台由德国团队开发你可以用它的 SaaS 服务cloud.langfuse.com也可以自部署。它的核心价值给你的 LLM 应用加上全链路追踪能力让你在控制台里看到每一次调用的完整细节。回到顶部二、环境准备2.1 安装依赖pip install langfuse langgraph langchain langchain-community2.2 配置环境变量# LangFuse 密钥在 LangFuse 控制台 → Settings → API Keys 中获取 $env:LANGFUSE_PUBLIC_KEYpk-lf-xxx $env:LANGFUSE_SECRET_KEYsk-lf-xxx $env:LANGFUSE_BASE_URLhttps://cloud.langfuse.com # LLM API 密钥本文使用通义千问 $env:DASHSCOPE_API_KEYsk-xxx提示LangFuse 云服务部署在海外国内访问可能有延迟。如果遇到 OpenTelemetry 超时报刷屏可以加一行代码抑制logging.getLogger(opentelemetry).setLevel(logging.CRITICAL)回到顶部三、方式1observe 装饰器推荐最简洁这是 LangFuse 最推荐的集成方式。核心思想给函数加个装饰器追踪就自动完成了。3.1 最小可运行代码from langfuse import observe, get_client from langchain_community.llms import Tongyi from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm Tongyi(model_nameqwen-turbo-latest, dashscope_api_keyos.getenv(DASHSCOPE_API_KEY)) observe(as_typegeneration) def call_llm(prompt_text: str) - str: prompt ChatPromptTemplate.from_template({text}) chain prompt | llm | StrOutputParser() result chain.invoke({text: prompt_text}) return result observe() def qa(question: str) - str: return call_llm(f请用一句话回答{question}) observe(namemy_agent) def agent_entry(user_input: str): return qa(user_input) # 直接调用无需任何其他配置 result agent_entry(什么是大语言模型)就这么多。不需要创建 handler不需要传 config不需要手动 flush。程序退出时数据自动上报。3.2 observe 的三种用法装饰器写法记录类型适用场景observe(namexxx)Trace顶层入口标记整个请求的入口name 是你在控制台看到的 trace 名称observe()Span中间步骤标记业务流程中的子步骤如翻译、摘要、分类observe(as_typegeneration)GenerationLLM 调用标记实际的 LLM 调用会自动记录 input/output/耗时3.3 嵌套调用 → 自动形成追踪树当被observe装饰的函数互相调用时LangFuse 会自动构建父子关系。在我们的示例中observe(namemulti_tool_agent) def agent_entry(user_input: str): answer qa(user_input) # 问答 summary summarize(answer) # 摘要 translated translate(summary) # 翻译 return {answer: answer, summary: summary, translated: translated}在 LangFuse 控制台中你会看到这样的追踪结构└─ multi_tool_agent (trace) ├─ qa (span) │ └─ call_llm (generation) ← 自动记录 input/output/耗时 ├─ summarize (span) │ └─ call_llm (generation) └─ translate (span) └─ call_llm (generation)一次调用三层层级关系全自动生成。你不需要写任何追踪逻辑。3.4 给 Generation 补充详细信息如果你想让 LangFuse 控制台显示更完整的 LLM 调用信息模型名称、完整 prompt 等可以用update_current_generation()observe(as_typegeneration) def call_llm(prompt_text: str, model_name: str qwen-turbo-latest) - str: prompt ChatPromptTemplate.from_template({text}) chain prompt | llm | StrOutputParser() result chain.invoke({text: prompt_text}) # 显式上报 model 和 prompt控制台中能看到完整的 LLM 请求上下文 get_client().update_current_generation( modelmodel_name, inputprompt_text, outputresult, ) return result回到顶部四、方式2observe 动态上下文适合多用户/多会话在实际生产环境中你通常需要按用户user_id追踪这个用户的所有请求按会话session_id归组同一次对话的多轮问答按标签tags筛选比如只看生产环境或v2版本的请求这就需要动态注入上下文信息。4.1 核心代码from opentelemetry import trace as otel_trace observe(as_typechain) def traced_graph_invoke(question, user_id, session_id, tagsNone, metadataNone): # 通过 OpenTelemetry span 属性注入用户上下文 span otel_trace.get_current_span() if span.is_recording(): span.set_attribute(langfuse.user.id, user_id) span.set_attribute(langfuse.session.id, session_id) if tags: span.set_attribute(langfuse.tags, tags) # 附加自定义元数据 if metadata: get_client().update_current_span(metadatametadata) # 执行业务逻辑 app build_graph() return app.invoke({question: question})关键点LangFuse 基于 OpenTelemetry 架构所以它能自动识别 OTEL span 中的特定属性Span 属性作用langfuse.user.id关联到 LangFuse 的 Users 视图langfuse.session.id同一 session_id 的 trace 会被归为一组langfuse.tags用于筛选和过滤4.2 多轮对话示例session_id fsession-{uuid.uuid4().hex[:8]} questions [ (user_001, 什么是 Python), (user_001, 它和 Java 有什么区别), # 同一用户同一会话 (user_002, 推荐一个入门编程语言), # 不同用户 ] for user_id, question in questions: result traced_graph_invoke( questionquestion, user_iduser_id, session_idsession_id, tags[langgraph, multi-turn], metadata{app_version: 1.0.0}, )在 LangFuse 控制台中可以按user_001/user_002筛选不同用户的追踪三条 trace 因为共享session_id而归为同一会话可以用langgraph标签快速过滤出这批请求回到顶部五、与 LangGraph 结合上面的方式2已经展示了 LangGraph 的集成。核心模式很简单observe 装饰外层函数 └─ 内部调用 graph.invoke() └─ graph 的节点调用 observe 标记的函数 └─ 自动形成完整的追踪链在我们的示例中LangGraph 图的结构是class State(TypedDict): question: str answer: Optional[str] def chat_node(state: State) - dict: answer call_llm(f请用一句话回答{state[question]}) return {answer: answer} def build_graph(): graph StateGraph(State) graph.add_node(chat, chat_node) graph.set_entry_point(chat) graph.add_edge(chat, END) return graph.compile()chat_node内部调用了call_llm而call_llm被observe(as_typegeneration)装饰。因此 LangGraph 执行图的时候LLM 调用会自动被追踪到不需要对 LangGraph 本身做任何修改。回到顶部六、关于 flushLangFuse 的数据上报是异步批量的。关于何时需要手动 flush场景是否需要手动 flush脚本执行完自然退出不需要程序退出时自动 flush多轮对话想实时看到每轮数据每轮结束后调用get_client().flush()长时间运行的服务如 Web 服务建议在请求结束时 flush# 手动 flush get_client().flush()回到顶部七、两种方式对比总结对比维度方式1observe 装饰器方式2observe 动态上下文代码侵入性极低加装饰器即可低需在入口函数注入属性user_id / session_id不支持动态传入支持通过 OTEL span 属性tags / metadata不支持动态传入支持灵活设置适用场景简单应用、快速验证、脚本生产环境、多用户多会话与 LangGraph 配合节点函数加 observe外层包装 图内 observe日常开发推荐方式1够简单够快。上生产需要按用户/会话追踪时切换到方式2。回到顶部八、完整代码以下是可直接运行的完整代码复制到本地.py文件即可执行。运行前确保设置好环境变量然后python 文件名.py#!/usr/bin/env python # -*- coding: utf-8 -*- LangGraph LangFuse 集成演示 LangFuse 基于 OpenTelemetry 架构提供两种追踪方式 方式1推荐observe 装饰器 - 最简洁直接修饰函数 - 自动追踪函数内的所有 LLM 调用 - 支持嵌套被装饰的函数互相调用时自动形成父子 span 方式2observe OTEL span 属性 - 适用于需要动态传入 user_id / session_id / tags 的场景 - 通过 OpenTelemetry span 属性注入用户上下文 环境变量配置Windows PowerShell $env:DASHSCOPE_API_KEYsk-xxx $env:LANGFUSE_PUBLIC_KEYpk-lf-xxx $env:LANGFUSE_SECRET_KEYsk-lf-xxx $env:LANGFUSE_BASE_URLhttps://cloud.langfuse.com # 可选 import os import uuid import logging from typing import TypedDict, Optional import warnings warnings.filterwarnings(ignore) # 修复 langchain 版本兼容性 import langchain for attr in (verbose, debug, llm_cache): if not hasattr(langchain, attr): setattr(langchain, attr, False if attr ! llm_cache else None) from langchain_community.llms import Tongyi from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.graph import StateGraph, END # LangFuse 导入 from langfuse import observe, get_client # 抑制 OpenTelemetry 网络超时日志 logging.getLogger(opentelemetry).setLevel(logging.CRITICAL) # 全局配置 LANGFUSE_ENABLED bool( os.getenv(LANGFUSE_PUBLIC_KEY) and os.getenv(LANGFUSE_SECRET_KEY) ) llm Tongyi( model_nameqwen-turbo-latest, dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 方式1observe 装饰器推荐 observe(as_typegeneration) def call_llm(prompt_text: str, model_name: str qwen-turbo-latest) - str: prompt ChatPromptTemplate.from_template({text}) chain prompt | llm | StrOutputParser() result chain.invoke({text: prompt_text}) get_client().update_current_generation( modelmodel_name, inputprompt_text, outputresult, ) return result observe() def translate(text: str) - str: return call_llm(f请将以下内容翻译成英文只返回译文\n{text}) observe() def summarize(text: str) - str: return call_llm(f请用一句话总结以下内容\n{text}) observe() def qa(question: str) - str: return call_llm(f请用一句话回答{question}) observe(namemulti_tool_agent) def agent_entry(user_input: str): answer qa(user_input) summary summarize(answer) translated translate(summary) return {answer: answer, summary: summary, translated: translated} # 方式2observe 动态设置 trace 属性 class State(TypedDict): question: str answer: Optional[str] def chat_node(state: State) - dict: answer call_llm(f请用一句话回答{state[question]}) return {answer: answer} def build_graph(): graph StateGraph(State) graph.add_node(chat, chat_node) graph.set_entry_point(chat) graph.add_edge(chat, END) return graph.compile() observe(as_typechain) def traced_graph_invoke(question, user_id, session_id, tagsNone, metadataNone): from opentelemetry import trace as otel_trace span otel_trace.get_current_span() if span.is_recording(): span.set_attribute(langfuse.user.id, user_id) span.set_attribute(langfuse.session.id, session_id) if tags: span.set_attribute(langfuse.tags, tags) if metadata: get_client().update_current_span(metadatametadata) app build_graph() return app.invoke({question: question}) # 运行演示 def demo_observe_decorator(): print(\n * 60) print(方式1observe 装饰器) print( * 60) result agent_entry(什么是大语言模型) print(f 回答: {result[answer]}) print(f 摘要: {result[summary]}) print(f 译文: {result[translated]}) def demo_observe_with_context(): print(\n * 60) print(方式2observe 动态上下文) print( * 60) session_id fsession-{uuid.uuid4().hex[:8]} questions [ (user_001, 什么是 Python), (user_001, 它和 Java 有什么区别), (user_002, 推荐一个入门编程语言), ] for user_id, question in questions: print(f\n [{user_id}] {question}) result traced_graph_invoke( questionquestion, user_iduser_id, session_idsession_id, tags[langgraph, multi-turn], metadata{app_version: 1.0.0}, ) print(f 助手: {result[answer]}) if LANGFUSE_ENABLED: get_client().flush() if __name__ __main__: print(fLangFuse: {已启用 if LANGFUSE_ENABLED else 未配置跳过追踪}) demo_observe_decorator() demo_observe_with_context() if LANGFUSE_ENABLED: get_client().flush() print(\n运行完毕。在 LangFuse 控制台可查看追踪数据。)