高级使用指南

概述

本文档介绍 Langfuse JS/TS SDK 的高级特性，帮助您：

构建分层的观测数据结构
实现分布式追踪
添加质量评分
优化性能
处理敏感数据

观测值嵌套

嵌套说明

观测值（Spans、Events、Generations）的嵌套可以帮助您：

构建层次化的追踪结构
分析调用关系
计算各层级耗时
定位性能瓶颈

默认嵌套

const trace = langfuse.trace({ name: "chat-app-session" });

const span = trace.span({ name: "chat-interaction" });

// 在 span 下创建子观测值
span.event({ name: "get-user-profile" });
span.generation({
  name: "chat-completion",
  model: "gpt-3.5-turbo",
});

手动嵌套（分布式场景）

分布式注意事项

在分布式应用中，需要手动管理观测值的嵌套关系。主要场景包括：

跨服务调用
异步任务处理
消息队列
定时任务

// 服务 A
const trace = langfuse.trace({
  name: "chat-app-session",
});

const span = langfuse.span({
  name: "chat-interaction",
  traceId: trace.id,
});

// 传递上下文到服务 B
const context = {
  traceId: trace.id,
  parentObservationId: span.id,
};

// 服务 B
const event = langfuse.event({
  traceId: context.traceId,
  parentObservationId: context.parentObservationId,
  name: "get-user-profile",
});

const generation = langfuse.generation({
  traceId: context.traceId,
  parentObservationId: context.parentObservationId,
  name: "chat-completion",
});

性能优化

批量处理

批量处理场景

批量处理适用于：

大规模数据处理
批量文档分析
并行任务处理
数据迁移任务

async function processBatch(items: string[]) {
  const trace = langfuse.trace({ name: "batch-processing" });
  const results = [];

  for (const item of items) {
    const span = trace.span({
      name: "process-item",
      metadata: { item },
    });

    try {
      const result = await processItem(item);
      span.end({
        output: result,
        level: "SUCCESS",
      });
      results.push(result);
    } catch (error) {
      span.end({
        level: "ERROR",
        statusMessage: error.message,
      });
    }
  }

  return results;
}

采样控制

采样策略

采样控制可以帮助您：

降低数据量
控制存储成本
保持关键数据
平衡性能影响

const langfuse = new Langfuse({
  // 只采集 10% 的数据
  sampleRate: 0.1,

  // 自定义采样逻辑
  shouldSample: (trace) => {
    // 错误情况总是采集
    if (trace.level === "ERROR") return true;
    // 特定用户总是采集
    if (trace.userId === "vip_user") return true;
    // 其他情况随机采样
    return Math.random() < 0.1;
  },
});

异步处理

异步优化

异步处理可以：

提高响应速度
优化资源使用
处理并发请求
避免阻塞操作

async function processWithAsync() {
  const trace = langfuse.trace({ name: "async-processing" });

  // 并行处理多个任务
  const [result1, result2] = await Promise.all([
    processTask1(trace),
    processTask2(trace),
  ]);

  // 确保数据发送完成
  await langfuse.flushAsync();

  return { result1, result2 };
}

数据处理

敏感数据处理

数据安全

处理敏感数据时的注意事项：

过滤个人信息
脱敏关键数据
遵守隐私政策
符合合规要求

function maskSensitiveData(data: any): any {
  if (typeof data === "string") {
    // 过滤邮箱
    data = data.replace(
      /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/g,
      "[EMAIL]"
    );
    // 过滤手机号
    data = data.replace(/1[3-9]\d{9}/g, "[PHONE]");
  }
  return data;
}

const generation = trace.generation({
  name: "user-profile-generation",
  input: maskSensitiveData(userInput),
  model: "gpt-4",
});

generation.end({
  output: maskSensitiveData(response),
});

自定义元数据

元数据使用场景

自定义元数据可用于记录：

环境信息
版本数据
业务参数
调试信息

const span = trace.span({
  name: "api-request",
  metadata: {
    endpoint: "/api/users",
    method: "POST",
    requestId: "req_123",
    environment: process.env.NODE_ENV,
    version: "1.0.0",
  },
});

错误处理

错误处理策略

完善的错误处理应该：

记录错误详情
分类错误类型
提供重试机制
保留上下文信息

async function processWithErrorHandling() {
  const trace = langfuse.trace({ name: "error-handling-demo" });

  try {
    const result = await riskyOperation();
    trace.event({
      name: "operation-success",
      level: "SUCCESS",
      output: result,
    });
    return result;
  } catch (error) {
    trace.event({
      name: "operation-failed",
      level: "ERROR",
      statusMessage: error.message,
      metadata: {
        errorType: error.name,
        stackTrace: error.stack,
      },
    });

    // 重试逻辑
    if (error.isRetryable) {
      return await retryOperation(trace);
    }

    throw error;
  }
}

最佳实践

最佳实践建议

合理使用观测类型
- 使用 Event 记录关键节点
- 使用 Span 记录耗时操作
- 使用 Generation 记录 AI 模型调用
优化数据量
- 合理设置采样率
- 避免记录过大的数据
- 定期清理不必要的数据
安全性考虑
- 过滤敏感信息
- 使用环境变量管理密钥
- 定期更新 SDK 版本
性能优化
- 使用异步处理
- 批量处理数据
- 合理设置超时时间

注意事项

在生产环境中禁用调试模式
定期检查数据质量
监控错误率和延迟
及时处理积压的数据
保持代码的可维护性

观测值嵌套​

默认嵌套​

手动嵌套（分布式场景）​

性能优化​

批量处理​

采样控制​

异步处理​

数据处理​

敏感数据处理​

自定义元数据​

错误处理​

最佳实践​