Advanced features

from langfuse import Langfuse
import re
 
def pii_masker(data: any, **kwargs) -> any:
    if isinstance(data, str):
        return re.sub(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", "[EMAIL_REDACTED]", data)
    elif isinstance(data, dict):
        return {k: pii_masker(data=v) for k, v in data.items()}
    elif isinstance(data, list):
        return [pii_masker(data=item) for item in data]
    return data
 
langfuse = Langfuse(mask=pii_masker)

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

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

import { configureGlobalLogger, LogLevel } from "@langfuse/core";
 
configureGlobalLogger({ level: LogLevel.DEBUG });

export LANGFUSE_LOG_LEVEL="DEBUG"

from langfuse import Langfuse
 
langfuse = Langfuse(sample_rate=0.2)

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

from langfuse import Langfuse
 
langfuse = Langfuse(blocked_instrumentation_scopes=["sqlalchemy", "psycopg"])

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

import { ShouldExportSpan } from "@langfuse/otel";
 
const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  ["langfuse-sdk", "ai"].includes(otelSpan.instrumentationScope.name);

from opentelemetry.sdk.trace import TracerProvider
from langfuse import Langfuse
 
langfuse_tracer_provider = TracerProvider()
langfuse = Langfuse(tracer_provider=langfuse_tracer_provider)
langfuse.start_span(name="isolated").end()

import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { setLangfuseTracerProvider } from "@langfuse/tracing";
import { LangfuseSpanProcessor } from "@langfuse/otel";
 
const langfuseTracerProvider = new NodeTracerProvider({
  spanProcessors: [new LangfuseSpanProcessor()],
});
 
setLangfuseTracerProvider(langfuseTracerProvider);

from langfuse import Langfuse, observe
 
project_a = Langfuse(public_key="pk-lf-project-a-...", secret_key="sk-lf-project-a-...")
project_b = Langfuse(public_key="pk-lf-project-b-...", secret_key="sk-lf-project-b-...")
 
@observe
def process_data_for_project_a(data, langfuse_public_key="pk-lf-project-a-..."):
    return {"processed": data}
 
@observe
def process_data_for_project_b(data, langfuse_public_key="pk-lf-project-b-..."):
    return {"processed": data}

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
 
const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: "pk-lf-public-key-project-1",
      secretKey: "sk-lf-secret-key-project-1",
    }),
    new LangfuseSpanProcessor({
      publicKey: "pk-lf-public-key-project-2",
      secretKey: "sk-lf-secret-key-project-2",
    }),
  ],
});
 
sdk.start();

from opentelemetry.instrumentation.threading import ThreadingInstrumentor
 
ThreadingInstrumentor().instrument()

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

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

import os, httpx
from langfuse import Langfuse
 
httpx_client = httpx.Client(verify=os.environ["OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE"])
langfuse = Langfuse(httpx_client=httpx_client)

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
 
export const langfuseSpanProcessor = new LangfuseSpanProcessor({
  exportMode: "immediate",
});
 
const sdk = new NodeSDK({ spanProcessors: [langfuseSpanProcessor] });
 
sdk.start();

import { langfuseSpanProcessor } from "./instrumentation";
 
export async function handler(event, context) {
  // ... logic ...
  await langfuseSpanProcessor.forceFlush();
}

import { LangfuseWeb } from "langfuse";
 
export function UserFeedbackComponent(props: { traceId: string }) {
  const langfuseWeb = new LangfuseWeb({
    publicKey: env.NEXT_PUBLIC_LANGFUSE_PUBLIC_KEY,
    baseUrl: "https://cloud.langfuse.com",
  });
 
  const handleUserFeedback = async (value: number) =>
    await langfuseWeb.score({ traceId: props.traceId, name: "user_feedback", value });
 
  return (
    <div>
      <button onClick={() => handleUserFeedback(1)}>👍</button>
      <button onClick={() => handleUserFeedback(0)}>👎</button>
    </div>
  );
}

import * as Sentry from "@sentry/node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import {
  SentryPropagator,
  SentrySampler,
  SentrySpanProcessor,
} from "@sentry/opentelemetry";
 
const sentryClient = Sentry.init({
  dsn: "<your dsn>",
  skipOpenTelemetrySetup: true,
  tracesSampleRate: 1,
});
 
const provider = new NodeTracerProvider({
  sampler: sentryClient ? new SentrySampler(sentryClient) : undefined,
  spanProcessors: [new LangfuseSpanProcessor(), new SentrySpanProcessor()],
});
 
provider.register({
  propagator: new SentryPropagator(),
  contextManager: new Sentry.SentryContextManager(),
});