LangFuse 实战指南：用 @observe 三行代码给 LLM 应用加上全链路追踪

作者：Work Hard Work Smart

4 次浏览

1 次

2026-8-5

编辑推荐:

本文以一个完整的代码示例，演示如何在 LangChain / LangGraph 项目中集成 LangFuse，实现对 LLM 调用的可观测性。全文不讲空泛理论，只看代码和效果, 希望对你的学习有帮助。
本文来自于博客园，由火龙果软件Alice编辑推荐。

一、为什么要用 LangFuse？

当你把 LLM 应用从 Demo 推向生产，一定会遇到这几个问题：

一次用户请求，LLM 被调了几次？每次的 prompt 和输出是什么？
哪次调用耗时最长？哪次 token 消耗最多？
同一个用户的多轮对话，能不能串起来看？
出了 bad case，怎么快速定位是哪一步出了问题？

LangFuse 就是解决这些问题的。它是一个开源的 LLM 可观测性平台（由德国团队开发），你可以用它的 SaaS 服务（cloud.langfuse.com），也可以自部署。

它的核心价值：给你的 LLM 应用加上"全链路追踪"能力，让你在控制台里看到每一次调用的完整细节。

二、环境准备

2.1 安装依赖

pip install langfuse langgraph langchain langchain-community

2.2 配置环境变量

# LangFuse 密钥（在 LangFuse 控制台 → Settings → API Keys 中获取）
$env:LANGFUSE_PUBLIC_KEY="pk-lf-xxx"
$env:LANGFUSE_SECRET_KEY="sk-lf-xxx"
$env:LANGFUSE_BASE_URL="https://cloud.langfuse.com"

# LLM API 密钥（本文使用通义千问）
$env:DASHSCOPE_API_KEY="sk-xxx"

提示： LangFuse 云服务部署在海外，国内访问可能有延迟。如果遇到 OpenTelemetry 超时报刷屏，可以加一行代码抑制：

logging.getLogger("opentelemetry").setLevel(logging.CRITICAL)

三、方式1：@observe 装饰器（推荐，最简洁）

这是 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_name="qwen-turbo-latest", dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"))

@observe(as_type="generation")
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(name="my_agent")
def agent_entry(user_input: str):
    return qa(user_input)

# 直接调用，无需任何其他配置
result = agent_entry("什么是大语言模型？")

就这么多。不需要创建 handler，不需要传 config，不需要手动 flush。程序退出时数据自动上报。

3.2 @observe 的三种用法

装饰器写法	记录类型	适用场景
@observe(name="xxx")	Trace （顶层入口）	标记整个请求的入口，name 是你在控制台看到的 trace 名称
@observe()	Span （中间步骤）	标记业务流程中的子步骤，如翻译、摘要、分类
@observe(as_type="generation")	Generation （LLM 调用）	标记实际的 LLM 调用，会自动记录 input/output/耗时

3.3 嵌套调用 → 自动形成追踪树

当被 @observe 装饰的函数互相调用时，LangFuse 会自动构建父子关系。在我们的示例中：

@observe(name="multi_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_type="generation")
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(
        model=model_name,
        input=prompt_text,
        output=result,
    )
    return result

四、方式2：@observe + 动态上下文（适合多用户/多会话）

在实际生产环境中，你通常需要：

按用户（user_id）追踪：这个用户的所有请求
按会话（session_id）归组：同一次对话的多轮问答
按标签（tags）筛选：比如只看"生产环境"或"v2版本"的请求

这就需要动态注入上下文信息。

4.1 核心代码

from opentelemetry import trace as otel_trace

@observe(as_type="chain")
def traced_graph_invoke(question, user_id, session_id, tags=None, metadata=None):
    # 通过 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(metadata=metadata)

    # 执行业务逻辑
    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 = f"session-{uuid.uuid4().hex[:8]}"

questions = [
    ("user_001", "什么是 Python？"),
    ("user_001", "它和 Java 有什么区别？"),      # 同一用户，同一会话
    ("user_002", "推荐一个入门编程语言"),          # 不同用户
]

for user_id, question in questions:
    result = traced_graph_invoke(
        question=question,
        user_id=user_id,
        session_id=session_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_type="generation") 装饰。因此 LangGraph 执行图的时候，LLM 调用会自动被追踪到，不需要对 LangGraph 本身做任何修改。

六、关于 flush

LangFuse 的数据上报是异步批量的。关于何时需要手动 flush：

场景	是否需要手动 flush
脚本执行完自然退出	不需要，程序退出时自动 flush
多轮对话，想实时看到每轮数据	每轮结束后调用 get_client().flush()
长时间运行的服务（如 Web 服务）	建议在请求结束时 flush

# 手动 flush
get_client().flush()

七、两种方式对比总结

对比维度	方式1：@observe 装饰器	方式2：@observe + 动态上下文
代码侵入性	极低，加装饰器即可	低，需在入口函数注入属性
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

方式2：@observe + OTEL span 属性
  - 适用于需要动态传入 user_id / session_id / tags 的场景
  - 通过 OpenTelemetry span 属性注入用户上下文

环境变量配置（Windows PowerShell）：
  $env:DASHSCOPE_API_KEY="sk-xxx"
  $env:LANGFUSE_PUBLIC_KEY="pk-lf-xxx"
  $env:LANGFUSE_SECRET_KEY="sk-lf-xxx"
  $env:LANGFUSE_BASE_URL="https://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_name="qwen-turbo-latest",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
)


# ====== 方式1：@observe 装饰器（推荐） ======

@observe(as_type="generation")
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(
        model=model_name, input=prompt_text, output=result,
    )
    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(name="multi_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}


# ====== 方式2：@observe + 动态设置 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_type="chain")
def traced_graph_invoke(question, user_id, session_id, tags=None, metadata=None):
    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(metadata=metadata)
    app = build_graph()
    return app.invoke({"question": question})


# ====== 运行演示 ======

def demo_observe_decorator():
    print("\n" + "=" * 60)
    print("方式1：@observe 装饰器")
    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("方式2：@observe + 动态上下文")
    print("=" * 60)
    session_id = f"session-{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(
            question=question, user_id=user_id,
            session_id=session_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(f"LangFuse: {'已启用' if LANGFUSE_ENABLED else '未配置（跳过追踪）'}")
    demo_observe_decorator()
    demo_observe_with_context()
    if LANGFUSE_ENABLED:
        get_client().flush()
    print("\n运行完毕。在 LangFuse 控制台可查看追踪数据。")

在 LangFuse 控制台（Traces 页面）即可看到完整的追踪数据，包括每次 LLM 调用的 prompt、输出、耗时、token 用量等信息。

4 次浏览

1 次