突破LLM应用集成瓶颈：Langfuse公共API实战指南

你是否还在为LLM应用的可观测性与第三方系统集成而头疼？本文将带你零门槛掌握Langfuse公共API的使用方法，通过10分钟快速上手教程和3个实战案例，解决从数据采集到高级分析的全流程集成难题。读完本文你将获得：API认证最佳实践、事件批量处理方案、跨平台数据同步技巧，以及完整的错误处理策略。

API架构概览

Langfuse公共API采用RESTful设计风格，提供了覆盖LLM应用全生命周期的接口服务。核心API端点位于web/src/pages/api/public目录下，主要包含五大功能模块：

• 数据采集：ingestion.ts处理事件批量导入
• 追踪系统：traces与sessions接口记录用户交互
• 评估工具：scores与score-configs支持模型性能评估
• 提示管理：prompts接口实现提示模板版本控制
• 数据集管理：datasets与dataset-items支持测试集管理

快速入门：10分钟接入指南

环境准备

首先通过Git获取最新代码库：

git clone https://gitcode.com/GitHub_Trending/la/langfuse.git
cd langfuse

API认证配置

Langfuse采用双密钥认证机制，在项目设置中创建API凭证后，配置环境变量：

LANGFUSE_SECRET_KEY="sk-lf-..."
LANGFUSE_PUBLIC_KEY="pk-lf-..."
LANGFUSE_HOST="http://localhost:3000"  # 本地部署地址

认证逻辑在ApiAuthService中实现，支持项目级和组织级权限控制。

首次事件发送

使用curl发送测试事件到 ingestion 端点：

curl -X POST http://localhost:3000/api/public/ingestion \
  -H "Authorization: PublicKey pk-lf-..." \
  -H "Content-Type: application/json" \
  -d '{"batch":[{"type":"generation","name":"test","input":"Hello Langfuse"}]}'

成功响应将返回事件处理结果：

{
  "success": true,
  "batchId": "batch-123456",
  "processed": 1,
  "failed": 0
}

核心功能实战

批量事件处理

ingestion.ts支持每秒数千事件的高吞吐量处理，采用三级架构确保可靠性：

1. 请求验证：Zod模式验证确保数据格式正确
2. 异步队列：Redis队列实现削峰填谷
3. 错误重试：内置指数退避机制处理临时故障

以下是Python批量发送事件的示例代码：

import requests
import json

url = "http://localhost:3000/api/public/ingestion"
headers = {
    "Authorization": "PublicKey pk-lf-...",
    "Content-Type": "application/json"
}

events = [
    {"type": "trace", "id": "trace-1", "name": "user-query"},
    {"type": "generation", "traceId": "trace-1", "input": "Hello"}
]

response = requests.post(url, headers=headers, data=json.dumps({"batch": events}))
print(response.json())

追踪数据查询

通过traceId查询完整调用链：

curl -X GET "http://localhost:3000/api/public/traces/trace-1" \
  -H "Authorization: PublicKey pk-lf-..."

响应包含追踪详情和关联事件，可用于构建用户交互时间线：

{
  "id": "trace-1",
  "name": "user-query",
  "timestamp": "2025-10-12T08:30:00Z",
  "events": [...],
  "metadata": {"userId": "user-123"}
}

评估结果提交

使用scores接口记录模型评估结果：

curl -X POST http://localhost:3000/api/public/scores \
  -H "Authorization: PublicKey pk-lf-..." \
  -H "Content-Type: application/json" \
  -d '{
    "name": "accuracy",
    "value": 0.85,
    "observationId": "obs-123",
    "comment": "Manual evaluation"
  }'

高级应用案例

案例1：LangChain集成

通过Langfuse的LangChain回调处理器，自动记录链执行过程：

from langchain.chains import LLMChain
from langfuse.langchain import LangfuseCallbackHandler

handler = LangfuseCallbackHandler(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="http://localhost:3000"
)

chain = LLMChain.from_string(llm=llm, prompt=prompt)
chain.run(input="Hello", callbacks=[handler])

相关实现代码位于LangChain集成模块。

案例2：前端性能监控

在React应用中集成事件追踪：

import { LangfuseBrowser } from 'langfuse-browser';

const langfuse = new LangfuseBrowser({
  publicKey: 'pk-lf-...',
  host: 'http://localhost:3000'
});

// 记录用户交互
langfuse.trace({ name: 'user-search' }).submit();

案例3：数据导出自动化

使用批量导出API定期备份数据：

curl -X POST http://localhost:3000/api/public/batch-exports \
  -H "Authorization: SecretKey sk-lf-..." \
  -H "Content-Type: application/json" \
  -d '{
    "datasetId": "dataset-123",
    "format": "jsonl",
    "destination": "s3://my-bucket/exports"
  }'

常见问题与解决方案

认证失败排查

若收到401错误，请检查：

1. 密钥是否与项目匹配（区分大小写）
2. 请求头格式是否正确（Authorization: PublicKey <key>）
3. 服务端时间是否同步（密钥有时间戳验证）

相关错误处理逻辑见ingestion.ts第74-88行。

性能优化建议

1. 批量发送事件（建议每批100-500条）
2. 使用压缩（设置Content-Encoding: gzip）
3. 实现客户端重试机制（指数退避策略）

完整文档资源

• 官方API文档：README.md
• OpenAPI规范：web/public/generated/api
• 错误码参考：BaseError

总结与展望

Langfuse公共API为LLM应用开发提供了标准化的数据采集与分析接口，通过本文介绍的认证机制、核心端点和实战案例，你可以快速实现与现有系统的集成。即将发布的v2版本将新增 GraphQL 接口和实时数据流功能，进一步降低集成复杂度。

建议收藏本文作为参考手册，并关注CHANGELOG获取最新更新。如有疑问，可在GitHub Discussions获取社区支持。

本文配套代码示例已上传至examples/api-integration目录，包含完整的Python、JavaScript和curl示例。