简介

        Langfuse 是一个开源 LLMOps 工程平台，专注于帮助团队协作完成 AI 应用的全生命周期管理，包括开发、监控、评估及调试。其核心优势在于轻量化部署（可在几分钟内完成自托管）和实战可靠性，已在大量实际场景中得到验证。

• 平台涵盖的主要领域：
  • 可观测性（Observability)：查看 LLM 运行时发生的情况。追踪成本、发现错误并了解用户如何与您的应用交互。无需再猜测故障原因或 OpenAI 上个月向您收取了多少费用。
  • 提示管理(Prompt Management)：将所有提示集中存储并进行版本控制，而不是分散在代码库各处。您的团队无需修改生产代码即可测试不同版本。
  • 评估（Evaluations）：衡量您的学习领导力模型 (LLM) 的性能。设置自动化测试，检查回复是否合理、是否切题或是否符合质量标准。

Observability 简介

      Langfuse 的可观测性与可追踪性，核心是为 LLM 应用提供全链路、可视化的运行状态监控方案，解决传统日志无法覆盖的 LLM 专属需求，适配复杂场景的调试与优化。

• 弥补传统日志短板：传统日志仅能记录函数是否运行，无法捕捉 LLM 应用的关键细节（如文档检索是否准确、提示词格式是否合规、哪个中间步骤耗时）。
• 适配复杂架构：针对生产级 RAG 等多步骤场景（文档加载、分块、多检索器、重排、生成），实现全链路无遗漏追踪。

可追踪性的核心价值

• 高效调试：快速定位错误根源（如检索错误、提示词格式问题），替代 “猜测式排查”。
• 性能优化：精准识别 latency 瓶颈（嵌入查询、向量搜索、LLM 调用等环节）。
• 成本管控：实时追踪单个模型调用的具体成本，便于预算管理。
• 评估支撑：为 LLM 应用效果评估提供数据基础。
• 客服提效：通过会话快速还原用户问题场景，节省排查时间。

三大维度

        Langfuse 的可观测性能力包含三个层级的追踪体系，实现了从单轮执行到多轮对话再到用户全局行为的全维度覆盖：

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

数据类型

        Langfuse 的 Observation Types（观测类型） 是对追踪数据的分类标签，核心作用是为 Span（追踪跨度）补充上下文、明确操作属性，同时支持高效筛选，覆盖 AI 应用全链路关键环节。

备注：

• event 是最外层，每一个trace 数据都是一个 event

• span 是基础类型

• 其他的是特殊场景下的类型

核心参数

Trace上报

Trace更新

跨服务传播(基于标准协议)

核心原理：OpenTelemetry- Baggage

     Baggage是一种跨信号、跨服务传递上下文信息的机制，本质是与 Context 绑定的键值存储，核心作用是让关键数据能在分布式系统中流转并赋能可观测性分析。

• 核心工作原理：依赖于传播器（Propagator）。在微服务的一次调用中，发起方服务会将Baggage从上下文（Context）中注入（Inject） 到HTTP请求头等载体中；接收方服务则从传入的请求头中提取（Extract） 这些信息到自己的上下文里，从而获取到这些数据-。

• 本质属性：是与 OpenTelemetry Context 绑定的键值存储，和 Context 一起在服务、进程间传递，不属于追踪（Trace）、指标（Metric）、日志（Log）任一单独信号，但能为所有信号提供数据支持。

• 核心价值：解决 “分布式系统中关键信息跨服务传递” 的问题，无需在代码中反复复制数据（如用户 ID、账户标识），就能让这些信息在下游服务的追踪、指标、日志中可用。

• 参考文献：https://opentelemetry.io/docs/concepts/signals/baggage/

使用 Baggage 时需重点关注两个风险：

1. 敏感信息泄露风险：Baggage 会随网络请求（如 HTTP 头）自动传播，网络流量中可见，可能被第三方 API 或下游服务扩散到外部网络，因此禁止存储密码、令牌等敏感数据。

2. 数据完整性无保障：Baggage 没有内置的完整性校验机制，无法确认数据是否被篡改或是否来自可信来源，读取时需谨慎判断有效性。

实现过程

OpenTelemetry 协议标准 Http Header 头注入

1. 通过OpenTelemetry创建Span后，TraceID，SpanId 会缓存到SpanContext 上下文中；

2. 如果需要额外的透传参数如Session_id,User_id ，OpenTelemetry 建议通过组装Baggage 的方式，先放入上下文对象中。
3. 通过OpenTelemetry 传播对象，将上下文对象中的SpanContent 和 Baggage 传入Header 头，此时Header 头中包含两个对象traceparent 和 baggage ；
   1. traceparent  主要包含TraceId，SpanId，
   2. baggage 是用户自行组装的内容。

OpenTelemetry 协议标准Http Header 头提取

1. 从header 头中，使用 OpenTelemetry 传播对象的extract 方法，提取current_span的span_context,获取TraceId,SpanId
2. 使用baggage.getall,获取header 头中的baggage 参数。

跨服务传播实现

1. 在创建current_span的时候传递

   trace_context={"trace_id": trace_id, "parent_span_id": parent_span_id}

2. 通过相同的 trace_id 和确认当前span 的父类Id，parent_span_id，此时即可实现跨服务传播的过程。

Langfuse实现方式(Java)

        langfuse 官网提供了 langfuse-java-sdk ,是基于langfuse的openAPI 自动生成的Java

端封装。但是官网有明确说明，Java 端建议使用OpenTelemetry-java-sdk 的方式去接入Langfuse。

基于（langfuse-java-sdk)

• 创建 Langfuse 客户端

// 创建 Langfuse 客户端
        LangfuseClient client = LangfuseClient.builder()
                .url("https://localhost:3000") // 本地部署
                .credentials("public-key", "private-key") // 替换为您的 API 密钥
                .build();

• 创建最外层的Trace-event

 // 1. 创建 Trace
TraceEvent traceEvent = TraceEvent.builder()
                .id(UUID.randomUUID().toString())
                .timestamp(timestamp)
                .body(TraceBody.builder()
                        .id(traceId)
                        .name("LLM Call Trace")
                        .userId("user-789")
                        .sessionId("session-101")
                        .input("User question about capitals")
                        .environment("production")
                        .build())
                .build();

• 创建span

// 2. 创建 Span（数据预处理）
CreateSpanEvent preprocessingSpan = CreateSpanEvent.builder()
                .id(UUID.randomUUID().toString())
                .timestamp(timestamp)
                .body(CreateSpanBody.builder()
                        .id(UUID.randomUUID().toString())
                        .traceId(traceId)
                        .name("Preprocessing")
                        .input("User question: What is the capital of France?")
                        .output("Formatted prompt: What is the capital of France?")
                        .level(ObservationLevel.DEFAULT)
                        .startTime(OffsetDateTime.now(ZoneOffset.UTC))
                        .endTime(OffsetDateTime.now(ZoneOffset.UTC).plusSeconds(1))
                        .build())
                .build();

• 创建Observation
  • 目前Java-SDK-Observation只支持 event，span，Generation（LLM调用） 三种类型

// 3. 创建 Generation（LLM 调用）
String generationId = UUID.randomUUID().toString();
CreateObservationEvent generationEvent = CreateObservationEvent.builder()
                .id(UUID.randomUUID().toString())
                .timestamp(timestamp)
                .body(ObservationBody.builder()
                        .type(ObservationType.GENERATION)  // 必需：设置观察类型
                        .id(generationId)
                        .traceId(traceId)
                        .name("OpenAI GPT-4 Call")
                        .input("What is the capital of France?")
                        .output("The capital of France is Paris.")
                        .model("gpt-4")
                        .modelParameters(createModelParameters())
                        .usage(createUsageForObservation())
                        .completionStartTime(OffsetDateTime.now(ZoneOffset.UTC))
                        .startTime(OffsetDateTime.now(ZoneOffset.UTC))
                        .endTime(OffsetDateTime.now(ZoneOffset.UTC).plusSeconds(2))
                        .level(ObservationLevel.DEFAULT)
                        .metadata(Map.of("provider", "openai", "temperature", 0.7))
                        .build())
                .build();

• 批量上报

// 批量上报所有事件
IngestionRequest request = IngestionRequest.builder()
                .addBatch(IngestionEvent.traceCreate(traceEvent))
                .addBatch(IngestionEvent.spanCreate(preprocessingSpan))
                .addBatch(IngestionEvent.observationCreate(generationEvent))
                .addBatch(IngestionEvent.spanCreate(postprocessingSpan))
                .build();

// 执行批量上报
IngestionResponse response = client.ingestion().batch(request);
System.out.println("成功上报: " + response.getSuccesses().size());
System.out.println("失败数量: " + response.getErrors().size());

• 更新 Trace

 //更新 Trace（使用相同的 traceId 但不同的事件 ID）
TraceEvent updateTraceEvent = TraceEvent.builder()
                .id(UUID.randomUUID().toString()) // 新的事件 ID
                .timestamp(OffsetDateTime.now(ZoneOffset.UTC).toString())
                .body(TraceBody.builder()
                        .id(traceId) // 相同的 traceId
                        .output("Updated output")
                        .metadata(Map.of("status", "completed", "updated_at", System.currentTimeMillis()))
                        .build())
                .build();

基于OpenTelemetry-java-sdk

初始化

OTEL 上报方法

• 初始化 OpenTelemetry SDK

 //初始化 OpenTelemetry SDK
        OpenTelemetry openTelemetry = OpenTelemetrySdk.builder()
                .setTracerProvider(tracerProvider)
                .setPropagators(contextPropagators)  // 设置 propagator
                .buildAndRegisterGlobal();

• 初始化traceProvider

 // 配置 Tracer Provider（批量上报 Span，提升性能）
        tracerProvider = SdkTracerProvider.builder()
                .setResource(resource)
                .addSpanProcessor(
                        BatchSpanProcessor.builder(otlpExporter)
                                .setMaxExportBatchSize(512)
                                .setScheduleDelay(5, TimeUnit.SECONDS)
                                .build()
                )
                .build();

• BatchSpanProcessor：批量上报Span,可以配置每次上报的最大batchSize、上报时间
• Resource：配置资源（服务名称、版本、环境等）

// 配置资源（服务名称、版本、环境等）
        Resource resource = Resource.getDefault()
                .merge(Resource.create(Attributes.of(
                        AttributeKey.stringKey("service.name"), SERVICE_NAME,
                        AttributeKey.stringKey("service.version"), "1.0.0",
                        AttributeKey.stringKey("deployment.environment"), "dev"
                )));

• 初始化OtlpHttpSpanExporter

 OtlpHttpSpanExporter otlpExporter = OtlpHttpSpanExporter.builder()
                .setEndpoint(fullEndpoint)
                .addHeader("Authorization", "Basic " + authString)
                .setTimeout(30, TimeUnit.SECONDS)
                .build();

• fullEndpoint：追踪上报地址，即Langfuse 上报地址：https://localhost:3000/api/public/otel/v1/traces
• Header：请求Langfuse 的鉴权，是固定Base64编码的 pubilckey:secretKey

/**
     * 生成 Basic Auth 认证头
     */
    private static String generateAuthHeader(String publicKey, String secretKey) {
        String credentials = publicKey + ":" + secretKey;
        return Base64.getEncoder().encodeToString(credentials.getBytes());
    }

• 创建Span

/**
     * 示例：用户查询接口
     * 演示如何创建 Span 并设置 Langfuse 相关属性
     */
    public String getUserInfo(String userId) {
        // 1. 创建 Span（描述当前操作）
        Span span = tracer.spanBuilder("user-service.get-user-info") // Span 名称（建议：服务名.接口名）
                .startSpan();

        try (var scope = span.makeCurrent()) { // 绑定 Span 到当前线程
            // 设置 Langfuse Trace 级别属性（可选）
            span.setAttribute("langfuse.trace.name", "Get User Info");
            span.setAttribute("langfuse.user.id", userId);
            span.setAttribute("langfuse.trace.input", "userId: " + userId);

            // 2. 业务逻辑（模拟接口调用）
            String userInfo = fetchUserFromDB(userId);

            // 3. 记录 Span 结果（可选：添加输出、耗时等元数据）
            span.setAttribute("user.info", userInfo);
            span.setAttribute("langfuse.trace.output", userInfo);
            span.setStatus(StatusCode.OK); // 标记成功
            return userInfo;
        } catch (Exception e) {
            // 4. 异常处理（标记失败，记录异常信息）
            span.setStatus(StatusCode.ERROR, e.getMessage());
            span.recordException(e);
            span.setAttribute("langfuse.trace.output", "Error: " + e.getMessage());
            throw e;
        } finally {
            // 5. 结束 Span（必须调用，否则数据不会上报）
            span.end();
        }
    }

• 创建子节点Span

/**
     * 模拟数据库查询
     * 演示如何创建子 Span（嵌套追踪）
     */
    private String fetchUserFromDB(String userId) {
        // 子 Span（可选：追踪内部调用，如 DB 查询、第三方 API）
        Span childSpan = tracer.spanBuilder("user-service.fetch-from-db")
                .startSpan();
        try (var scope = childSpan.makeCurrent()) {
            // 设置 Langfuse Observation 级别属性（可选）
            childSpan.setAttribute("langfuse.observation.type", "span");
            childSpan.setAttribute("langfuse.observation.input", "userId: " + userId);

            Thread.sleep(50); // 模拟耗时
            String result = "User{id=" + userId + ", name='test'}";
            
            childSpan.setAttribute("langfuse.observation.output", result);
            return result;
        } catch (InterruptedException e) {
            childSpan.setStatus(StatusCode.ERROR);
            childSpan.recordException(e);
            childSpan.setAttribute("langfuse.observation.status_message", e.getMessage());
            throw new RuntimeException(e);
        } finally {
            childSpan.end();
        }
    }

• 实时推送

tracerProvider.forceFlush().join(10, TimeUnit.SECONDS);

• 优雅关闭

tracerProvider.shutdown().join(10, TimeUnit.SECONDS);

Langfuse实现方式（python)

        我们可以直接使用 langfuse-python-sdk 实现 langfuse 的接入。

初始化

一、必填参数（无默认值，必须配置）

二、核心可选参数（常用配置，适配不同部署 / 网络场景）

三、进阶配置参数（控制数据上报 / 调试 / 敏感数据处理）

Python SDK Method

tracing相关

start_span()：

• 创建新的 span，但不自动设置为当前 span

• 需要手动调用 .end() 结束

start_as_current_span()：

• 创建 span 并自动设置为当前 span（上下文管理器）

• 退出时自动结束，适合 with 语句

start_observation()：

• 创建指定类型的 observation（span, generation, agent, tool 等）

• 通过 as_type 参数指定类型

start_as_current_observation()：

• 创建并设置为当前 observation（上下文管理器）

• 支持多种类型：span, generation, agent, tool, chain 等

update_current_span()：更新当前活动的 span 信息

update_current_trace():更新当前 trace 的元数据（用户 ID、会话 ID、标签等）

create_trace_id():生成唯一的 trace ID（32 位十六进制字符串）

create_observation_id()：生成唯一的 observation ID（16 位十六进制字符串）

get_current_trace_id()：获取当前活动 span 的 trace ID

get_current_observation_id()：获取当前活动 span 的 observation ID

create_event():创建 EVENT 类型的 observation（瞬时事件，立即结束）

工具方法

flush():强制刷新所有待发送的数据到 Langfuse API

shutdown():优雅关闭客户端，确保数据已发送

get_trace_url():获取在 Langfuse UI 中查看 trace 的 URL

resolve_media_references()：将媒体引用字符串替换为 base64 数据 URI

auth_check()：检查提供的凭证是否有效

基于LangChain 框架

• 使用方式：初始化CallbackHandler并将其添加到您的 Langchain 调用中，可以全局添加，也可以按调用添加。
• 

• 原理：回调处理函数会将各种 Langchain 事件映射到 Langfuse 观测值，Langchain事件包括：
• Chains (on_chain_start, on_chain_end, on_chain_error): Traced as spans.
• LLMs (on_llm_start, on_llm_end, on_llm_error, on_chat_model_start): Traced as generations, capturing model name, prompts, responses, and usage if available from the LLM provider.
• Tools (on_tool_start, on_tool_end, on_tool_error): Traced as spans, capturing tool input and output.
• Retrievers (on_retriever_start, on_retriever_end, on_retriever_error): Traced as spans, capturing the query and retrieved documents.
• Agents (on_agent_action, on_agent_finish): Agent actions and final finishes are captured within their parent chain/agent span.

拓展方法

propagate_attributes()：将属性拓展到子节点

• 值必须是长度不超过 200 个字符的字符串。
• 元数据键：仅限字母数字字符（不含空格或特殊字符）
• 在跟踪初期就进行调用，以确保涵盖所有观测结果。这样可以确保 Langfuse 中的所有指标都准确无误。

跨业务传播

        对于跨多个服务的分布式追踪，请使用as_baggage参数（更多详细信息请参阅OpenTelemetry 文档）通过 HTTP 标头传播属性。

将新生成的观测数据（如 Span）关联到外部已存在的追踪（Trace）

• 适用场景：当你拥有外部来源的 trace_id（可选 parent_span_id）时，需要将 Langfuse 中新建的 Span 关联到这个已有追踪上。例如：外部服务、批处理任务已生成了追踪 ID，现在要在 Langfuse 中记录该任务下游操作的 Span，并让它归属到同一个追踪链路中。
• 前提说明：OpenTelemetry 本身支持 “已接入 OTEL 的服务间” 自动传递追踪上下文，无需手动处理；该方法仅用于对接未自动传递上下文的外部系统，或需要手动关联追踪的特殊场景。

参考文献

• 开源代码地址：https://github.com/langfuse/langfuse
• 文档地址：https://langfuse.com/docs
• API接口文档： https://api.reference.langfuse.com/#tag/organizations
• 优秀教程：https://www.datacamp.com/tutorial/langfuse
• 性能基准测试参考：https://blog.csdn.net/gitblog_00342/article/details/150961332
• OpenTelemetry-otel-trace：https://opentelemetry.io/docs/specs/otel/trace/
• OpenTelemetry-java:https://opentelemetry.io/docs/languages/java/intro/
• Langfuse-OpenTelemetry支持：https://langfuse.com/integrations/native/opentelemetry