고급 기능

고급 기능 (Advanced features)

이 메서드들을 사용하면 Langfuse 계측을 더 단단하게 만들고, 민감한 데이터를 보호하며, SDK를 당신의 특정 환경에 맞게 조정할 수 있어요.

출처: 문서

본문

계측 범위로 필터링하기

Langfuse는 이제 두 SDK 모두에 기본 span 필터를 적용해서 추가 설정 없이도 내보내기(export)를 LLM 중심으로 유지해요.

기본적으로 다음 중 하나라도 참이면 span이 내보내져요:

• Langfuse SDK로 생성됨 (instrumentation_scope.name == "langfuse-sdk")
• gen_ai.* 속성을 하나 이상 가짐
• 알려진 LLM 계측 범위에서 옴 (예: openinference.*, langsmith, haystack, litellm, agent_framework, strands-agents, vllm, opentelemetry.instrumentation.anthropic)

기본 계측 범위 허용 목록에 다른 통합을 추가하고 싶다면 langfuse/langfuse에 범위 이름과 샘플 span을 포함한 이슈를 열어 주세요.

span의 계측 범위는 Langfuse의 metadata.scope.name 아래에서 확인할 수 있어요. 필터링된 span은 UI에 나타나지 않아요.

필터링된 범위를 식별하려면:

• 디버그 로깅을 켜세요 (Python: Langfuse(debug=True) 또는 LANGFUSE_DEBUG="True", JS/TS: LANGFUSE_DEBUG="true" 또는 LANGFUSE_LOG_LEVEL="DEBUG").
• 애플리케이션을 실행하고 로그에서 드롭된 span 메시지와 계측 범위 이름을 확인하세요.
• is_default_export_span / isDefaultExportSpan으로 조합해 그 범위들을 허용 목록 로직에 추가하세요.
• 선택 사항: 모든 span을 검사하려면 임시로 should_export_span=lambda span: True 또는 shouldExportSpan: () => true를 사용한 뒤 필터링을 복원하세요.

동작 변경

이전 SDK 버전은 기본적으로 차단되지 않은 모든 span을 내보냈어요. 그 동작을 복원하려면 항상 true를 반환하는 커스텀 필터 콜백을 제공하세요.

span 필터링이 트레이스 트리를 깨뜨릴 수 있음

span을 필터링하면 트레이스의 부모-자식 관계가 깨질 수 있어요. 예를 들어 부모 span을 필터링하고 자식은 유지하면 Langfuse UI에 "고아(orphaned)" observation이 보일 수 있어요. 위의 디버깅 흐름을 사용해 필터링된 span을 다시 추가하세요.

Python SDKJS/TS SDK

기본 동작(권장):

from langfuse import Langfuse

# Smart default filter (Langfuse + GenAI/LLM spans)
langfuse = Langfuse()

모두 내보내기:

from langfuse import Langfuse

langfuse = Langfuse(should_export_span=lambda span: True)

should_export_span을 전달하면 기본 필터를 대체해요. 기본 동작을 유지하면서 확장하려면 is_default_export_span과 조합하세요.

내장 술어(predicate)와 커스텀 로직 조합:

from langfuse import Langfuse
from langfuse.span_filter import is_default_export_span

langfuse = Langfuse(
    should_export_span=lambda span: (
        is_default_export_span(span)
        or (
            span.instrumentation_scope is not None
            and span.instrumentation_scope.name.startswith("my_framework")
        )
    )
)

Langfuse SDK로 생성된 span만 내보내기:

from langfuse import Langfuse
from langfuse.span_filter import is_langfuse_span

langfuse = Langfuse(should_export_span=is_langfuse_span)

Python 헬퍼: is_default_export_span, is_langfuse_span, is_genai_span, is_known_llm_instrumentor, KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES.

blocked_instrumentation_scopes는 하위 호환성으로 여전히 동작하지만 deprecated이며 향후 버전에서 제거될 예정이에요. 거부 규칙은 should_export_span에 표현하는 것을 권장해요.

deprecated 호환 예시:

from langfuse import Langfuse

langfuse = Langfuse(
    should_export_span=lambda span: True,
    blocked_instrumentation_scopes=["sqlalchemy", "psycopg"],
)

기본 동작(권장):

instrumentation.ts

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

const sdk = new NodeSDK({
  // Smart default filter (Langfuse + GenAI/LLM spans)
  spanProcessors: [new LangfuseSpanProcessor()],
});

sdk.start();

커스텀 필터링:

instrumentation.ts

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

const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  otelSpan.instrumentationScope.name !== "express";

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

sdk.start();

shouldExportSpan을 전달하면 기본 필터를 대체해요. 기본 동작을 확장(대체가 아닌)하려면 콜백에 기본 조건을 포함하세요.

기본 필터와 조합:

instrumentation.ts

import { isDefaultExportSpan, type ShouldExportSpan } from "@langfuse/otel";

const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  isDefaultExportSpan(otelSpan) ||
  otelSpan.instrumentationScope.name.startsWith("my-framework");

@langfuse/otel의 JS/TS 헬퍼: isDefaultExportSpan, isLangfuseSpan, isGenAISpan, isKnownLLMInstrumentor, KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES.

모두 내보내기:

instrumentation.ts

new LangfuseSpanProcessor({ shouldExportSpan: () => true });

기존 OpenTelemetry 설정과 Langfuse 사용하기에 대해 더 읽어보세요.

민감한 데이터 마스킹

트레이스 데이터에 PII나 비밀 같은 민감한 정보가 포함될 수 있다면, span을 Langfuse로 보내기 전에 마스킹 훅을 구성하세요. Python SDK 애플리케이션에서는 mask_otel_spans를 권장해요. 이 함수는 내보내기 단계에서 raw OpenTelemetry span 속성에 대해 실행되며, 제3자 계측으로 생성된 span도 포함해요.

Python SDKJS/TS SDK

내보내기 전에 변경해야 할 OpenTelemetry span에 대해 sparse 패치를 반환하도록 mask_otel_spans를 사용하세요.

import re
from typing import Optional

from langfuse import Langfuse
from langfuse.types import (
    MaskOtelSpansParams,
    MaskOtelSpansResult,
    OtelSpanPatch,
)

email_pattern = re.compile(r"\b[\w.-]+?@[\w.-]+?\.\w+?\b")

def mask_otel_spans(
    *, params: MaskOtelSpansParams
) -> Optional[MaskOtelSpansResult]:
    patches = {}

    for identifier, span in params.spans.items():
        replacements = {}

        for key, value in span.attributes.items():
            if isinstance(value, str):
                masked_value = email_pattern.sub("[EMAIL_REDACTED]", value)

                if masked_value != value:
                    replacements[key] = masked_value

        if replacements:
            patches[identifier] = OtelSpanPatch(set_attributes=replacements)

    return MaskOtelSpansResult(span_patches=patches)

langfuse = Langfuse(mask_otel_spans=mask_otel_spans)

mask_otel_spans는 동기적이며 보통 OpenTelemetry 배치 span processor 워커 스레드에서 실행돼요. flush()와 shutdown() 중에는 호출자 스레드에서 실행될 수 있어요. 내보내기 큐가 밀리지 않도록 함수를 빠르게 유지하세요. 전체 동작, 오류 처리, 레거시 mask와의 비교는 masking을 참고하세요.

LangfuseSpanProcessor에 mask 함수를 제공할 수 있어요. 이 함수는 모든 observation의 input, output, metadata에 적용돼요.

함수는 { data } 객체를 받는데, 여기서 data는 속성 값의 stringify된 JSON이에요. 마스킹된 데이터를 반환하면 돼요.

instrumentation.ts

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

const spanProcessor = new LangfuseSpanProcessor({
  mask: ({ data }) =>
    data.replace(/\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b/g, "***MASKED_CREDIT_CARD***"),
});

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

sdk.start();

로깅 & 디버깅

Langfuse SDK는 애플리케이션 문제를 해결하는 데 도움이 되는 상세한 로깅 및 디버깅 정보를 노출할 수 있어요.

Python SDKJS/TS SDK

환경 변수로:

LANGFUSE_DEBUG 환경 변수로 로그 레벨을 설정해 디버그 모드를 켤 수 있어요.

export LANGFUSE_DEBUG="True"

코드에서:

Langfuse SDK는 Python의 표준 logging 모듈을 사용해요. 메인 로거 이름은 "langfuse"예요. 상세한 디버그 로깅을 켜려면 다음 중 하나를 할 수 있어요:

• Langfuse 클라이언트 초기화 시 debug=True 파라미터 설정
• "langfuse" 로거를 수동으로 구성:

import logging

langfuse_logger = logging.getLogger("langfuse")
langfuse_logger.setLevel(logging.DEBUG)

langfuse 로거의 기본 로그 레벨은 logging.WARNING이에요.

전역 SDK 로거를 구성해 로그 출력의 상세도를 제어할 수 있어요. 디버깅에 유용해요.

환경 변수로:

LANGFUSE_LOG_LEVEL 환경 변수로 로그 레벨을 설정해 디버그 모드를 켤 수 있어요.

export LANGFUSE_LOG_LEVEL="DEBUG"

코드에서:

import { configureGlobalLogger, LogLevel } from "@langfuse/core";

// Set the log level to DEBUG to see all log messages
configureGlobalLogger({ level: LogLevel.DEBUG });

사용 가능한 로그 레벨은 DEBUG, INFO, WARN, ERROR예요.

샘플링

샘플링을 사용하면 트레이스의 일부만 Langfuse로 보낼 수 있어요. 고볼륨 애플리케이션에서 비용과 노이즈를 줄이는 데 유용해요.

Python SDKJS/TS SDK

코드에서:

클라이언트 초기화 시 sample_rate 파라미터를 설정해 SDK가 트레이스를 샘플링하도록 구성할 수 있어요. 이 값은 0.0(트레이스 0% 샘플링)과 1.0(트레이스 100% 샘플링) 사이의 float이어야 해요.

샘플링되지 않은 트레이스는 그 observations이나 관련 score가 Langfuse로 전송되지 않아요.

from langfuse import Langfuse

# Sample approximately 20% of traces
langfuse_sampled = Langfuse(sample_rate=0.2)

환경 변수로:

LANGFUSE_SAMPLE_RATE 환경 변수로 샘플 비율을 설정할 수도 있어요.

export LANGFUSE_SAMPLE_RATE="0.2"

코드에서:

Langfuse는 OpenTelemetry의 샘플링 결정을 존중해요. 어떤 트레이스가 Langfuse에 도달할지 제어하고 고볼륨 워크로드의 노이즈/비용을 줄이려면 OTEL NodeSDK에 sampler를 구성하세요.

instrumentation.ts

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

const sdk = new NodeSDK({
  sampler: new TraceIdRatioBasedSampler(0.2),
  spanProcessors: [new LangfuseSpanProcessor()],
});

sdk.start();

환경 변수로:

LANGFUSE_SAMPLE_RATE 환경 변수로 샘플 비율을 설정할 수도 있어요.

export LANGFUSE_SAMPLE_RATE="0.2"

분리된 TracerProvider

Langfuse와 함께 사용할 별도의 OpenTelemetry TracerProvider를 구성할 수 있어요. 이렇게 하면 Langfuse 트레이싱과 다른 관찰성 시스템 사이에 격리가 생겨요.

격리의 장점:

• Langfuse span이 다른 관찰성 백엔드(예: Datadog, Jaeger, Zipkin)로 전송되지 않음
• 제3자 라이브러리 span이 Langfuse로 전송되지 않음
• 독립적인 구성과 샘플링 비율

TracerProvider는 분리되어 있지만 활성 span 추적을 위해 동일한 OpenTelemetry 컨텍스트를 공유해요. 이로 인해 span 관계 문제가 발생할 수 있어요:

• 한 TracerProvider의 부모 span이 다른 TracerProvider의 자식을 가질 수 있음
• 부모 span이 다른 TracerProvider에 속하면 일부 span이 "고아"로 보일 수 있음
• 트레이스 계층이 불완전하거나 혼란스러울 수 있음

혼란스러운 트레이스 구조를 피하려면 계측을 신중히 계획하세요.

Python SDKJS/TS SDK

from opentelemetry.sdk.trace import TracerProvider
from langfuse import Langfuse

langfuse_tracer_provider = TracerProvider() # do not set to global tracer provider to keep isolation
langfuse = Langfuse(tracer_provider=langfuse_tracer_provider)
langfuse.start_observation(name="myspan").end() # Span will be isolated from remaining OTEL instrumentation

커스텀 provider로 Langfuse span을 격리하고 다른 exporter로 보내지 않게 하세요.

import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { setLangfuseTracerProvider } from "@langfuse/tracing";

// Create a new TracerProvider and register the LangfuseSpanProcessor
// do not set this TracerProvider as the global TracerProvider
const langfuseTracerProvider = new NodeTracerProvider({
  spanProcessors: [new LangfuseSpanProcessor()],
})

// Register the isolated TracerProvider
setLangfuseTracerProvider(langfuseTracerProvider)

기존 OpenTelemetry 설정과 Langfuse 사용하기에 대해 더 읽어보세요.

다중 프로젝트 설정

Python SDKJS/TS SDK

다중 프로젝트 설정은 Python SDK에서 실험적이며 제3자 OpenTelemetry 통합에 중요한 제약이 있어요.

Langfuse Python SDK는 여러 공개 키를 사용해 같은 애플리케이션 내에서 트레이스를 다른 프로젝트로 라우팅하는 것을 지원해요. Langfuse SDK가 생성하는 모든 span에 공개 키를 담은 특정 span 속성을 추가하기 때문에 동작해요.

작동 방식:

• Span 속성: Langfuse SDK가 생성하는 span에 공개 키를 담은 특정 span 속성을 추가함
• 여러 Processor: 여러 span processor가 글로벌 tracer provider에 등록되고, 각각 특정 공개 키에 바인딩된 exporter를 가짐
• 필터링: 각 span processor 내에서 공개 키 속성의 존재와 값을 기준으로 span을 필터링함

제3자 라이브러리의 중요한 제약:

OpenTelemetry span을 자동으로 내보내는 제3자 라이브러리(예: HTTP 클라이언트, 데이터베이스, 다른 계측 라이브러리)는 Langfuse 공개 키 span 속성이 없어요. 결과적으로:

• 내보내기 필터를 통과하고 공개 키가 없는 제3자 span은 특정 프로젝트로 라우팅될 수 없음
• 이 span들은 모든 span processor에 의해 처리되어 모든 프로젝트로 전송될 수 있음
• 기본 필터에서 이는 주로 제3자 계측기의 GenAI/LLM span에 영향을 줌 (인프라 span은 보통 필터링됨)

왜 실험적인가? 이 접근 방식은 모든 통합에서 모든 Langfuse SDK 실행에 public_key 파라미터를 전달해 올바른 라우팅을 보장해야 하며, 필터링을 통과한 제3자 span이 모든 프로젝트에 나타날 수 있어요.

초기화

여러 프로젝트를 설정하려면 각 프로젝트에 대해 별도의 Langfuse 클라이언트를 초기화하세요:

from langfuse import Langfuse

# Initialize clients for different projects
project_a_client = Langfuse(
    public_key="«redacted:pk-lf-…»...",
    secret_key="«redacted:sk-…»...",
    base_url="https://cloud.langfuse.com"
)

project_b_client = Langfuse(
    public_key="«redacted:pk-lf-…»...",
    secret_key="«redacted:sk-…»...",
    base_url="https://cloud.langfuse.com"
)

통합 사용법

다중 프로젝트 설정의 모든 통합에서 트레이스가 올바른 프로젝트로 라우팅되도록 public_key 파라미터를 지정해야 해요.

Observe 데코레이터:

최상위 observed 함수(데코레이터가 아니라)에 langfuse_public_key를 키워드 인자로 전달하세요. Python SDK >= 3.2.2부터 중첩된 데코레이트 함수는 현재 실행 중인 실행 컨텍스트에서 공개 키를 자동으로 가져와요. 또한 get_client 호출도 데코레이트된 함수 실행 컨텍스트의 현재 langfuse_public_key를 인지하므로 여기서 다시 전달할 필요가 없어요.

from langfuse import observe

@observe
def nested():
    # get_client call is context aware
    # if it runs inside another decorated function that has
    # langfuse_public_key passed, it does not need passing here again

@observe
def process_data_for_project_a(data):
    # passing `langfuse_public_key` here again is not necessarily
    # as it is stored in execution context
    nested()

    return {"processed": data}

@observe
def process_data_for_project_b(data):
    # passing `langfuse_public_key` here again is not necessarily
    # as it is stored in execution context
    nested()

    return {"enhanced": data}

# Route to Project A
# Top-most decorated function needs `langfuse_public_key` kwarg
result_a = process_data_for_project_a(
    data="input data",
    langfuse_public_key="«redacted:pk-lf-…»..."
)

# Route to Project B
# Top-most decorated function needs `langfuse_public_key` kwarg
result_b = process_data_for_project_b(
    data="input data",
    langfuse_public_key="«redacted:pk-lf-…»..."
)

OpenAI 통합:

OpenAI 실행에 langfuse_public_key를 키워드 인자로 추가하세요:

from langfuse.openai import openai

client = openai.OpenAI()

# Route to Project A
response_a = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello from Project A"}],
    langfuse_public_key="«redacted:pk-lf-…»..."
)

# Route to Project B
response_b = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello from Project B"}],
    langfuse_public_key="«redacted:pk-lf-…»..."
)

Langchain 통합:

CallbackHandler 생성자에 public_key를 추가하세요:

from langfuse.langchain import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Create handlers for different projects
handler_a = CallbackHandler(public_key="«redacted:pk-lf-…»...")
handler_b = CallbackHandler(public_key="«redacted:pk-lf-…»...")

llm = ChatOpenAI(model_name="gpt-4o")
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm

# Route to Project A
response_a = chain.invoke(
    {"topic": "machine learning"},
    config={"callbacks": [handler_a]}
)

# Route to Project B
response_b = chain.invoke(
    {"topic": "data science"},
    config={"callbacks": [handler_b]}
)

중요한 고려사항:

• 모든 통합의 모든 Langfuse SDK 실행에 적절한 공개 키 파라미터를 포함해야 함
• 공개 키 파라미터가 누락되면 트레이스가 기본 프로젝트로 라우팅되거나 손실될 수 있음
• 필터링을 통과한 제3자 OpenTelemetry span은 Langfuse 공개 키 속성이 없으므로 모든 프로젝트에 나타날 수 있음

SDK가 트레이스를 여러 Langfuse 프로젝트로 보내도록 구성할 수 있어요. 이는 멀티테넌트 애플리케이션이나 다른 환경으로 트레이스를 보내는 데 유용해요. 각각 자체 자격 증명을 가진 여러 LangfuseSpanProcessor 인스턴스를 등록하기만 하면 돼요.

instrumentation.ts

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

const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: "«redacted:pk-lf-…»",
      secretKey: "«redacted:sk-…»",
    }),
    new LangfuseSpanProcessor({
      publicKey: "«redacted:pk-lf-…»",
      secretKey: "«redacted:sk-…»",
    }),
  ],
});

sdk.start();

이 구성은 각 processor의 필터가 수락한 모든 span을 두 프로젝트 모두로 보내요. 각 processor에 커스텀 shouldExportSpan 필터를 구성해 어떤 트레이스가 어떤 프로젝트로 가는지 제어할 수 있어요.

첫 토큰까지의 시간 (Time to first token, TTFT)

LLM 호출의 시간-투-퍼스트-토큰(TTFT)을 수동으로 설정할 수 있어요. LLM 호출의 지연을 측정하고 느린 LLM 호출을 식별하는 데 유용해요.

Python SDKJS/TS SDK

completion_start_time 속성을 사용해 LLM 호출의 TTFT를 수동으로 설정할 수 있어요. LLM 호출의 지연을 측정하고 느린 LLM 호출을 식별하는 데 유용해요.

from langfuse import get_client
import datetime, time

langfuse = get_client()

with langfuse.start_as_current_observation(as_type="generation", name="TTFT-Generation") as generation:
    time.sleep(3)
    generation.update(
        completion_start_time=datetime.datetime.now(),
        output="some response",
    )

langfuse.flush()

completionStartTime 속성을 사용해 LLM 호출의 TTFT를 수동으로 설정할 수 있어요. LLM 호출의 지연을 측정하고 느린 LLM 호출을 식별하는 데 유용해요.

import { startActiveObservation } from "@langfuse/tracing";

startActiveObservation("llm-call", async (span) => {
  span.update({
    completionStartTime: new Date().toISOString(),
  });
});

자체 서명 SSL 인증서 (셀프 호스트 Langfuse)

셀프 호스트 Langfuse를 사용하면서 자체 서명 SSL 인증서를 쓰려면 SDK가 자체 서명 인증서를 신뢰하도록 구성해야 해요:

SSL 설정 변경은 환경에 따라 주요 보안 영향을 미쳐요. 진행 전에 반드시 그 영향을 이해하세요.

1. OpenTelemetry span exporter가 자체 서명 인증서를 신뢰하도록 설정

.env

OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE="/path/to/my-selfsigned-cert.crt"

2. Langfuse 인스턴스에 대한 다른 모든 API 요청에 HTTPX가 인증서를 신뢰하도록 설정

main.py

import os

import httpx

from langfuse import Langfuse

httpx_client = httpx.Client(verify=os.environ["OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE"])

langfuse = Langfuse(httpx_client=httpx_client)

Sentry와 함께 설정하기

애플리케이션에서 Sentry와 Langfuse를 둘 다 사용한다면, 두 도구 모두 트레이싱에 OpenTelemetry를 사용하므로 커스텀 OpenTelemetry 설정을 구성해야 해요. 이 가이드는 Sentry에 오류 모니터링 데이터를 보내면서 동시에 Langfuse에서 LLM 관찰성 트레이스를 캡처하는 방법을 보여줘요.

스레드 풀과 멀티프로세싱

Python SDK

컨텍스트가 워커 스레드 간에 흐르도록 OpenTelemetry threading instrumentor를 사용하세요.

from opentelemetry.instrumentation.threading import ThreadingInstrumentor

ThreadingInstrumentor().instrument()

멀티프로세싱은 OpenTelemetry 가이드를 따르세요. Pydantic Logfire를 사용한다면 distributed_tracing=True를 켜세요. 별도 서비스나 프로세스 간 트레이싱은 트레이스 ID & 분산 트레이싱을 참고하세요.

더 알아보기 (Learn more)

• 출처 문서: 고급 기능 (Advanced features)