文档 / 事件队列/批处理

事件队列/批处理

Langfuse 的客户端 SDK 和集成都被设计为在后台对请求进行队列化和批处理,以优化 API 调用和网络时间。批次由时间和大小(事件数量和批次大小)的组合决定。

配置

所有集成都有合理的默认配置,但你可以自定义批处理行为以满足你的需求。

选项(Python)[SDK 构造函数, 环境变量]	选项(JS)	描述
`flush_at`、`LANGFUSE_FLUSH_AT`	`flushAt`	发送前批量打包的最大事件数。
`flush_interval`、`LANGFUSE_FLUSH_INTERVAL`(秒)	`flushInterval`(秒)	发送批次前等待的最大时间(秒)。

例如,你可以设置 flushAt=1 以立即发送每个事件,或 flushInterval=1 以每秒发送一次。

手动刷新

在短生命周期环境(如 serverless 函数,例如 Vercel Functions、AWS Lambda)中,你应该在进程退出或运行时环境被冻结之前显式刷新 traces。如果你不刷新客户端,可能会丢失事件。

如果你想立即发送一个批次,可以调用客户端的 flush 方法。如果出现网络问题,flush 会记录错误并重试该批次,它永远不会抛出异常。

python

from langfuse import get_client

# access the client directly

langfuse = get_client()

# Flush all pending observations

langfuse.flush()

如果你退出应用,使用 shutdown 方法确保所有请求被刷新,并在进程退出前等待挂起的请求。此函数成功后,将不再向 Langfuse API 发送事件。

python

from langfuse import get_client

langfuse = get_client()

langfuse.shutdown()

LangfuseSpanProcessor 缓冲事件并批量发送,因此最后一次刷新能确保不丢失数据。

你可以从 OTEL SDK 设置文件中导出该 processor。

tsinstrumentation.ts

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

// Export the processor to be able to flush it
export const langfuseSpanProcessor = new LangfuseSpanProcessor();

const sdk = new NodeSDK({
  spanProcessors: [langfuseSpanProcessor],
});

sdk.start();

然后,在你的 serverless 函数 handler 中,在函数退出前调用 forceFlush()。

tshandler.ts

import { langfuseSpanProcessor } from "./instrumentation";

export async function handler(event, context) {
  // ... your application logic ...

  // Flush before exiting
  await langfuseSpanProcessor.forceFlush();
}

python

from langfuse import get_client

# access the client directly
langfuse = get_client()

# Flush all pending observations
langfuse.flush()

python

from langfuse import get_client

langfuse = get_client()

langfuse.flush()

# access the client directly
langfuse_handler.client.flush()

javascript

await langfuseHandler.flushAsync();

如果你退出应用,使用 shutdownAsync 方法确保所有请求被刷新,并在进程退出前等待挂起的请求。

javascript

await langfuseHandler.shutdownAsync();