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

2026-02-04 04:51:31作者：侯霆垣

你是否还在为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支持每秒数千事件的高吞吐量处理，采用三级架构确保可靠性：

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

以下是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"
  }'