이벤트 큐잉/배칭

이벤트 큐잉/배칭 (Event Queuing/Batching)

Langfuse의 클라이언트 SDK와 통합은 모두 API 호출과 네트워크 시간을 최적화하기 위해 요청을 백그라운드에서 큐에 쌓고 배치로 보내도록 설계됐어요. 배치는 시간과 크기(이벤트 수와 배치 크기)의 조합으로 결정돼요.

출처: 문서

본문

Langfuse의 클라이언트 SDK와 통합은 API 호출과 네트워크 시간을 최적화하기 위해 요청을 백그라운드에서 큐에 쌓고 배치하는 방식으로 설계됐어요. 배치는 시간과 크기(이벤트 수와 배치 크기)의 조합으로 결정돼요.

구성 (Configuration)

모든 통합은 합리적인 기본 구성을 갖지만, 자신의 필요에 맞게 배칭 동작을 커스터마이즈할 수 있어요.

Option (Python) [SDK constructor, Environment]	Option (JS)	Description
flush_at, LANGFUSE_FLUSH_AT	flushAt	보내기 전에 배치로 모을 최대 이벤트 수.
flush_interval, LANGFUSE_FLUSH_INTERVAL (s)	flushInterval (seconds)	배치를 보내기 전에 기다리는 최대 시간(초).

예를 들어 flushAt=1로 설정하면 모든 이벤트를 즉시 보내고, flushInterval=1로 설정하면 1초마다 보낼 수 있어요.

수동 플러시 (Manual flushing)

서버리스 함수(Vercel Functions, AWS Lambda 등) 같은 단기 환경에서는 프로세스가 종료되거나 런타임 환경이 동결되기 전에 trace를 명시적으로 플러시해야 해요. 클라이언트를 플러시하지 않으면 이벤트를 잃을 수 있어요.

Python SDK

from langfuse import get_client

# access the client directly

langfuse = get_client()

# Flush all pending observations

langfuse.flush()

배치를 즉시 보내고 싶다면 클라이언트의 flush 메서드를 호출할 수 있어요. 네트워크 문제가 있으면 flush는 에러를 로그하고 배치를 재시도하며, 예외를 던지지 않아요.

from langfuse import get_client

langfuse = get_client()

langfuse.shutdown()

애플리케이션을 종료할 때는 shutdown 메서드를 사용해 모든 요청이 플러시되고 프로세스가 종료되기 전에 보류 중인 요청이 완료되도록 해요. 이 함수가 성공하면 더 이상 이벤트가 Langfuse API로 보내지지 않아요.

JS/TS SDK

LangfuseSpanProcessor는 이벤트를 버퍼링하고 배치로 보내므로, 최종 flush가 데이터 손실을 방지해요.

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();

OTEL SDK 설정 파일에서 processor를 내보낼 수 있어요.

import { langfuseSpanProcessor } from "./instrumentation";

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

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

서버리스 함수 핸들러에서 함수가 종료되기 전에 forceFlush()를 호출하세요.

from langfuse import get_client

# access the client directly
langfuse = get_client()

# Flush all pending observations
langfuse.flush()

Langchain (Python)

from langfuse import get_client

langfuse = get_client()

langfuse.flush()

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

Langchain (JS)

await langfuseHandler.flushAsync();

await langfuseHandler.shutdownAsync();

애플리케이션을 종료할 때는 shutdownAsync 메서드를 사용해 모든 요청이 플러시되고 프로세스가 종료되기 전에 보류 중인 요청이 완료되도록 해요.

더 알아보기 (Learn more)

출처 문서: 이벤트 큐잉/배칭 (Event Queuing/Batching)