langfuse

golangLeetcode

发布于 2026-03-18 18:39:30

1210

langfuse是大模型开发流程的可观测工具，Langfuse 的可观测性能力包含三个层级的追踪体系，实现了从单轮执行到多轮对话再到用户全局行为的全维度覆盖：

Trace（单轮执行追踪）：聚焦单条 LLM 请求的完整执行链路，记录从输入、中间步骤到输出的全量数据，包括层级化执行树、耗时分布、Token 与成本消耗等细节，为单轮问题的根因分析提供颗粒度最细的依据。
Session（多轮对话追踪）：将同一上下文的关联请求（如用户的多轮连续问答）聚合为会话，还原完整对话逻辑，解决 “单轮 Trace 孤立无关联” 的问题，便于复盘用户多轮交互中的行为模式与问题。
User（用户维度追踪）：基于用户唯一 ID 聚合其所有 Session 和 Trace，实现 “用户 - 对话 - 单轮执行” 的三层关联，支持从用户视角全局分析行为特征、问题分布，是面向真实业务场景（如用户留存、个性化优化）的顶层观测维度。

官方文档路径：https://github.com/langfuse/langfuse

https://langfuse.com/self-hosting/deployment/docker-compose

读过dify源码的都知道，dify内部也是使用了langfuse来做监控观测的。下面我们体验下如何使用，简单分析下它的原理。

首先下载源码，然后启动服务器

cd langfuse
docker compose up
[+] Running 0/5
 ⠏ langfuse-worker Pulling        34.0s
 ⠋ postgres Pulling                 34.0s
 ⠋ clickhouse Pulling                  34.0s
 ⠋ langfuse-web Pulling                34.0s
 ⠏ minio Pulling

启动成功后可以看到下面的提示

 |▲ Next.js 15.5.9
langfuse-web-1     |    - Local:        http://ff7bf206810a:3000
langfuse-web-1     |    - Network:      http://ff7bf206810a:3000

打开http://127.0.0.1:3000后可以看到下面的界面

然后申请api key

申请完api key后可以开始代码层面接入了。首先定义.env文件，然后填入上面申请的密钥

LANGFUSE_SECRET_KEY = "sk-lf-00820f1a-e42b-4794-90f5-7967e5ec92ca"
LANGFUSE_PUBLIC_KEY = "pk-lf-1763ff0c-45f9-481c-b3eb-83efb5e5854f"
LANGFUSE_BASE_URL = "http://127.0.0.1:3000"

然后代码层面接入，可以看到，只需要在函数上添加装饰器

@observe()

就可以接入

from langfuse import observe
from langfuse.openai import openai # OpenAI integration
from dotenv import load_dotenv
from openai import OpenAI
# 加载.env文件
load_dotenv()
@observe()
def story():
    api_key = "xx"  # 替换为实际API密钥
    base_url = ""  # 替换为实际API路径
    client = OpenAI(base_url=base_url, api_key=api_key)
    return client.chat.completions.create(
        model="xxx" ,                          # 指定模型版本
        messages=[{"role": "user", "content": "What is Langfuse?"}],
    ).choices[0].message.content
@observe()
def main():
    return story()
main()

尝试运行

uv init     
uv add -i https://mirrors.aliyun.com/pypi/simple/  langfuse openai 
uv add -i https://mirrors.aliyun.com/pypi/simple/ dotenv
uv run ./main.py

然后打开链接http://127.0.0.1:3000就可以看到下面的监控结果。

具体看下这个装饰器的定义langfuse/__init__.py

from ._client.observe import observe

__all__ = [
    "Langfuse",
    "get_client",
    "observe",

它引用了client包langfuse/_client/observe.py

  @overload
    def observe(self, func: F) -> F: ...
    @overload
    def observe(
        self,
        func: None = None,
        *,
        name: Optional[str] = None,
        as_type: Optional[ObservationTypeLiteralNoEvent] = None,
        capture_input: Optional[bool] = None,
        capture_output: Optional[bool] = None,
        transform_to_string: Optional[Callable[[Iterable], str]] = None,
    ) -> Callable[[F], F]: ...
    def observe(
        self,
        func: Optional[F] = None,
        *,
        name: Optional[str] = None,
        as_type: Optional[ObservationTypeLiteralNoEvent] = None,
        capture_input: Optional[bool] = None,
        capture_output: Optional[bool] = None,
        transform_to_string: Optional[Callable[[Iterable], str]] = None,
    ) -> Union[F, Callable[[F], F]]:
    
    
     with _set_current_public_key(public_key):
                langfuse_client = get_client(public_key=public_key)
                context_manager: Optional[
                    Union[
                        _AgnosticContextManager[LangfuseGeneration],
                        _AgnosticContextManager[LangfuseSpan],
                        _AgnosticContextManager[LangfuseAgent],
                        _AgnosticContextManager[LangfuseTool],
                        _AgnosticContextManager[LangfuseChain],
                        _AgnosticContextManager[LangfuseRetriever],
                        _AgnosticContextManager[LangfuseEvaluator],
                        _AgnosticContextManager[LangfuseEmbedding],
                        _AgnosticContextManager[LangfuseGuardrail],
                    ]
                ] = (
                    langfuse_client.start_as_current_observation(
                        name=final_name,
                        as_type=as_type or "span",
                        trace_context=trace_context,
                        input=input,
                        end_on_exit=False,  # when returning a generator, closing on exit would be to early
                    )
                    if langfuse_client
                    else None
                )

可以看到，函数内部主要实现了日志和上报的功能，和普通的日志装饰器差别不大。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-12-21，如有侵权请联系 cloudcommunity@tencent.com 删除

模型