Low Level SDK 使用指南

功能概述

Low Level SDK 提供了更细粒度的控制，让您可以手动管理追踪、观测和事件的生命周期。这种方式特别适合需要精确控制数据采集过程的场景。

使用场景

适用场景

Low Level SDK 特别适用于以下场景：

分布式系统中的手动上下文传递
自定义框架集成
特殊的数据采集需求
复杂的异步处理流程

初始化

初始化注意事项

确保在使用 SDK 之前完成初始化配置。建议在应用启动时进行全局初始化。

from langfuse import Langfuse

langfuse = Langfuse(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="https://xxx.xxx.xxx"
)

Trace 管理

创建 Trace

Trace 创建

Trace 是所有观测数据的顶层容器，建议为每个独立的用户请求或任务创建一个新的 Trace：

# 创建新的 Trace
trace = langfuse.trace(
    name="用户查询处理",
    metadata={"user_id": "123"},
    tags=["production"]
)

# 使用已有 Trace ID
trace = langfuse.trace(
    name="后续处理",
    trace_id="已有的 Trace ID"
)

更新 Trace

Trace 更新

您可以在任何时候更新 Trace 的属性：

trace.update(
    name="更新后的名称",
    metadata={"status": "completed"},
    tags=["processed"]
)

Observation 管理

创建 Span

Span 使用

使用 Span 来记录一段时间内的执行过程：

# 创建并自动开始计时
span = trace.span(
    name="文档处理",
    metadata={"doc_type": "pdf"}
)

try:
    # 执行操作
    process_document()
    span.end()  # 正常结束
except Exception as e:
    span.end(level="ERROR", status_message=str(e))  # 错误结束

创建 Generation

Generation 使用

Generation 是特殊的 Span 类型，用于记录 AI 模型的生成过程：

generation = trace.generation(
    name="内容生成",
    model="gpt-4",
    prompt="生成一个故事",
    metadata={"temperature": 0.7}
)

try:
    response = generate_content()
    generation.end(
        completion=response,
        metadata={"tokens": len(response)}
    )
except Exception as e:
    generation.end(level="ERROR", status_message=str(e))

创建 Event

Event 使用

使用 Event 记录离散的事件点：

trace.event(
    name="配置加载完成",
    metadata={"config_version": "1.0"},
    level="INFO"
)

高级用法

自定义 ID

ID 管理

在分布式系统中，您可能需要手动管理 ID：

trace = langfuse.trace(
    name="分布式处理",
    trace_id="custom-trace-id",
    id="custom-observation-id"
)

批量处理

对于大量数据处理场景，可以使用批量模式：

with langfuse.batch_mode():
    for item in items:
        trace = langfuse.trace(name=f"处理-{item}")
        # 处理逻辑
        trace.end()

分布式追踪

在分布式系统中传递上下文：

# 服务 A
trace = langfuse.trace(name="主请求")
span = trace.span(name="子任务准备")
context = {
    "trace_id": trace.id,
    "parent_observation_id": span.id
}

# 服务 B
child_trace = langfuse.trace(
    name="子任务执行",
    trace_id=context["trace_id"],
    parent_observation_id=context["parent_observation_id"]
)

性能优化

性能注意事项

在高并发场景下，注意以下几点：

# 1. 使用采样控制数据量
langfuse = Langfuse(sample_rate=0.1)  # 只采集 10% 的数据

# 2. 批量模式减少网络请求
with langfuse.batch_mode(max_batch_size=100):
    for item in items:
        process_item(item)

# 3. 异步处理避免阻塞
async def process_async():
    trace = langfuse.trace(name="异步处理")
    await async_operation()
    trace.end()

故障排除

故障排除指南

启用调试模式
```
langfuse = Langfuse(debug=True)
```
验证连接
```
assert langfuse.auth_check()
```
常见问题
- 数据未上传：检查 end() 调用
- ID 冲突：检查自定义 ID 的唯一性
- 内存占用高：调整批处理参数

最佳实践

合理设置超时时间
正确处理异常情况
及时调用 end() 方法
在程序退出前确保数据已发送
适当使用采样控制数据量

使用场景​

初始化​

Trace 管理​

创建 Trace​

更新 Trace​

Observation 管理​

创建 Span​

创建 Generation​

创建 Event​

高级用法​

自定义 ID​

批量处理​

分布式追踪​

性能优化​

故障排除​