SDK로 실험하기(Experiments via SDK)

SDK로 실험하기(Experiments via SDK)

SDK로 실험하면 애플리케이션이나 프롬프트를 데이터셋을 통해 프로그래매틱하게 반복하고, 결과에 평가 방법을 선택적으로 적용할 수 있습니다. 이 문서는 Experiment runner SDK의 기본 사용법, 데이터셋 사용, 평가자, 멀티모달, 비동기, 설정 옵션을 다룹니다. Langfuse 호스팅 데이터셋이나 로컬 데이터셋을 실험의 기반으로 사용할 수 있습니다.

출처: 문서

본문

SDK로 실험하면 애플리케이션이나 프롬프트를 데이터셋을 통해 프로그래매틱하게 반복하고, 결과에 평가 방법(Evaluation Methods)을 선택적으로 적용할 수 있습니다. Langfuse에 호스팅된 데이터셋이나 로컬 데이터셋을 실험의 기반으로 사용할 수 있습니다.

SDK로 실험을 실행하는 자세한 내용은 JS/TS SDK reference 및 Python SDK reference 를 참고하세요.

Python이나 JS/TS를 사용하지 않나요? OpenTelemetry 실험 속성 으로 실험 trace를 직접 수집하세요. Experiment runner SDK가 그 속성을 대신 설정해 줍니다.

왜 SDK로 실험하나요?

• 내 애플리케이션 로직을 사용할 수 있는 완전한 유연성
• 단일 아이템과 전체 실행의 출력을 평가하는 커스텀 스코어링 함수 사용
• Langfuse UI에서 결정적 Python/TypeScript 검사를 작성하고 observations와 experiments 전반에서 재사용하려면 code evaluators 사용
• 같은 데이터셋에서 병렬로 여러 실험 실행
• 기존 평가 인프라와 쉽게 통합
• CI/CD에서 실험을 실행 해 출시 전 회귀를 잡기

Experiment runner SDK

Python과 JS/TS SDK 모두 데이터셋에서 실험을 실행하는 고수준 추상화를 제공합니다. 데이터셋은 로컬이거나 Langfuse에 호스팅될 수 있습니다. 데이터셋에서 실험을 실행할 때 Experiment runner를 사용하는 것이 SDK에서 권장하는 방법입니다.

Experiment runner는 자동으로 다음을 처리합니다:

• 동시 실행 — 구성 가능한 제한 사용
• 자동 tracing — 옵저버빌리티를 위한 모든 실행
• 유연한 평가 — 아이템 수준과 run 수준 평가자 모두
• 오류 격리 — 개별 실패가 실험을 멈추지 않음
• 데이터셋 통합 — 쉬운 비교와 추적

Experiment runner SDK는 Langfuse 호스팅 데이터셋과 로컬 데이터셋을 모두 지원합니다. Langfuse v4와 현재 SDK에서 둘 다 Experiments 아래에 나타나고 UI에서 비교할 수 있습니다. 호스팅 데이터셋은 추가로 공유 테스트 케이스와 과거 데이터셋 버전을 제공합니다. Compare experiments 를 참고하세요.

기본 사용법(Basic Usage)

로컬 데이터에서 task 함수를 테스트하는 가장 간단한 실험부터 시작하세요. 이미 Langfuse에 데이터셋이 있으면 아래를 참고하세요.

Python:

from langfuse import get_client
from langfuse.openai import OpenAI

# Initialize client
langfuse = get_client()

# Define your task function
def my_task(*, item, **kwargs):
    question = item["input"]
    response = OpenAI().chat.completions.create(
        model="gpt-4.1", messages=[{"role": "user", "content": question}]
    )

    return response.choices[0].message.content

# Run experiment on local data
local_data = [
    {"input": "What is the capital of France?", "expected_output": "Paris"},
    {"input": "What is the capital of Germany?", "expected_output": "Berlin"},
]

result = langfuse.run_experiment(
    name="Geography Quiz",
    description="Testing basic functionality",
    data=local_data,
    task=my_task,
)

# Use format method to display results
print(result.format())

JS/TS:

import { OpenAI } from "openai";
import { NodeSDK } from "@opentelemetry/sdk-node";

import {
  LangfuseClient,
  ExperimentTask,
  ExperimentItem,
} from "@langfuse/client";
import { observeOpenAI } from "@langfuse/openai";
import { LangfuseSpanProcessor } from "@langfuse/otel";

// Initialize OpenTelemetry
const otelSdk = new NodeSDK({ spanProcessors: [new LangfuseSpanProcessor()] });
otelSdk.start();

// Initialize client
const langfuse = new LangfuseClient();

// Run experiment on local data
const localData: ExperimentItem[] = [
  { input: "What is the capital of France?", expectedOutput: "Paris" },
  { input: "What is the capital of Germany?", expectedOutput: "Berlin" },
];

// Define your task function
const myTask: ExperimentTask = async (item) => {
  const question = item.input;

  const response = await observeOpenAI(new OpenAI()).chat.completions.create({
    model: "gpt-4.1",
    messages: [
      {
        role: "user",
        content: question,
      },
    ],
  });

  return response.choices[0].message.content;
};

// Run the experiment
const result = await langfuse.experiment.run({
  name: "Geography Quiz",
  description: "Testing basic functionality",
  data: localData,
  task: myTask,
});

// Print formatted result
console.log(await result.format());

// Important: shut down OTEL SDK to deliver traces
await otelSdk.shutdown();

JS/TS SDK 참고: trace가 Langfuse로 전달되도록 OpenTelemetry가 올바르게 설정되어야 합니다. 구성 세부 사항은 tracing setup documentation 을 참고하세요. 모든 trace가 전송되도록 실행 끝에 span processor를 항상 플러시하세요.

Langfuse v4와 현재 SDK에서 로컬 데이터 실험은 호스팅 데이터셋 없이 Experiments에 나타납니다. 각 task 실행은 디버깅용 trace도 만듭니다. 로컬 데이터는 입력이 내 프로세스에서 비롯된다는 뜻이며, 실험 데이터는 여전히 내가 구성한 Langfuse 인스턴스로 전송됩니다.

Langfuse 데이터셋 사용

자동 tracing과 비교를 위해 Langfuse에 저장된 데이터셋에서 실험을 직접 실행하세요.

Python:

from langfuse import get_client
from langfuse.openai import OpenAI

# Initialize client
langfuse = get_client()

# Define your task function
def my_task(*, item, **kwargs):
    question = item.input # `run_experiment` passes a `DatasetItem` to the task function. The input of the dataset item is available as `item.input`.
    response = OpenAI().chat.completions.create(
        model="gpt-4.1", messages=[{"role": "user", "content": question}]
    )

    return response.choices[0].message.content

# Get dataset from Langfuse
dataset = langfuse.get_dataset("my-evaluation-dataset")

# Run experiment directly on the dataset
result = dataset.run_experiment(
    name="Production Model Test",
    description="Monthly evaluation of our production model",
    task=my_task # see above for the task definition
)

# Use format method to display results
print(result.format())

JS/TS:

// Get dataset from Langfuse
const dataset = await langfuse.dataset.get("my-evaluation-dataset");

// Run experiment directly on the dataset
const result = await dataset.runExperiment({
  name: "Production Model Test",
  description: "Monthly evaluation of our production model",
  task: myTask, // see above for the task definition
});

// Use format method to display results
console.log(await result.format());

// Important: shut down OpenTelemetry to ensure traces are sent to Langfuse
await otelSdk.shutdown();

Langfuse 데이터셋을 사용하면 데이터셋 run이 Langfuse에 자동으로 생성되고 UI에서 비교할 수 있습니다. 이를 통해 시간 경과에 따른 실험 성능을 추적하고 같은 데이터셋에서 다른 접근법을 비교할 수 있습니다.

기본적으로 데이터셋을 가져오면 최신 버전이 반환됩니다. 가져올 때 버전 타임스탬프를 전달하면 그 과거 상태(이후 업데이트되거나 삭제된 아이템 포함)에서 실행할 수 있습니다. 버전 데이터셋에서 실험 실행 을 참고하세요.

고급 기능(Advanced Features)

평가자와 고급 구성 옵션으로 실험을 강화하세요.

평가자(Evaluators)

평가자는 아이템 수준에서 task 출력의 품질을 평가합니다. 각 아이템의 input, metadata, output, expected output을 받아 Langfuse의 traces에 점수로 보고되는 평가 메트릭을 반환합니다.

이 SDK 평가자 함수는 내 실험 프로세스에서 실행됩니다. Langfuse UI에서 결정적 평가자 코드를 작성하고 Langfuse가 실험 observations에 실행하게 하려면 code evaluators 를 사용하세요.

Python:

from langfuse import Evaluation

# Define evaluation functions
def accuracy_evaluator(*, input, output, expected_output, metadata, **kwargs):
    if expected_output and expected_output.lower() in output.lower():
        return Evaluation(name="accuracy", value=1.0, comment="Correct answer found")

    return Evaluation(name="accuracy", value=0.0, comment="Incorrect answer")

def length_evaluator(*, input, output, **kwargs):
    return Evaluation(name="response_length", value=len(output), comment=f"Response has {len(output)} characters")

# Use multiple evaluators
result = langfuse.run_experiment(
    name="Multi-metric Evaluation",
    data=test_data,
    task=my_task,
    evaluators=[accuracy_evaluator, length_evaluator]
)

print(result.format())

JS/TS:

// Define evaluation functions
const accuracyEvaluator = async ({ input, output, expectedOutput }) => {
  if (
    expectedOutput &&
    output.toLowerCase().includes(expectedOutput.toLowerCase())
  ) {
    return {
      name: "accuracy",
      value: 1.0,
      comment: "Correct answer found",
    };
  }
  return {
    name: "accuracy",
    value: 0.0,
    comment: "Incorrect answer",
  };
};

const lengthEvaluator = async ({ input, output }) => {
  return {
    name: "response_length",
    value: output.length,
    comment: `Response has ${output.length} characters`,
  };
};

// Use multiple evaluators
const result = await langfuse.experiment.run({
  name: "Multi-metric Evaluation",
  data: testData,
  task: myTask,
  evaluators: [accuracyEvaluator, lengthEvaluator],
});

console.log(await result.format());

Run-level 평가자

Run-level 평가자는 전체 실험 결과를 평가하고 집계 메트릭을 계산합니다. Langfuse 데이터셋에서 실행하면 이 점수들이 전체 데이터셋 run에 연결되어 전반적인 실험 성능을 추적합니다.

Python:

from langfuse import Evaluation

def average_accuracy(*, item_results, **kwargs):
    """Calculate average accuracy across all items"""
    accuracies = [
        eval.value for result in item_results
        for eval in result.evaluations
        if eval.name == "accuracy"
    ]

    if not accuracies:
        return Evaluation(name="avg_accuracy", value=None)

    avg = sum(accuracies) / len(accuracies)

    return Evaluation(name="avg_accuracy", value=avg, comment=f"Average accuracy: {avg:.2%}")

result = langfuse.run_experiment(
    name="Comprehensive Analysis",
    data=test_data,
    task=my_task,
    evaluators=[accuracy_evaluator],
    run_evaluators=[average_accuracy]
)

print(result.format())

JS/TS:

const averageAccuracy = async ({ itemResults }) => {
  // Calculate average accuracy across all items
  const accuracies = itemResults
    .flatMap((result) => result.evaluations)
    .filter((evaluation) => evaluation.name === "accuracy")
    .map((evaluation) => evaluation.value as number);

  if (accuracies.length === 0) {
    return { name: "avg_accuracy", value: null };
  }

  const avg = accuracies.reduce((sum, val) => sum + val, 0) / accuracies.length;

  return {
    name: "avg_accuracy",
    value: avg,
    comment: `Average accuracy: ${(avg * 100).toFixed(1)}%`,
  };
};

const result = await langfuse.experiment.run({
  name: "Comprehensive Analysis",
  data: testData,
  task: myTask,
  evaluators: [accuracyEvaluator],
  runEvaluators: [averageAccuracy],
});

console.log(await result.format());

멀티모달 실험

SDK 기반 실험은 input, expectedOutput, metadata에 미디어 첨부를 포함하는 데이터셋에서 실행할 수 있습니다. SDK로 데이터셋을 가져오면 각 미디어 토큰이 기본적으로 서명된 LangfuseMediaReference로 하이드레이션됩니다.

멀티모달 데이터셋은 Python SDK >= 4.10.0과 JS/TS SDK @langfuse/client >= 5.6.0의 SDK 기반 실험에서 지원됩니다. UI 기반 실험은 아직 미디어 첨부가 있는 데이터셋 아이템을 지원하지 않습니다.

Python:

from langfuse import get_client
from langfuse.media import LangfuseMediaReference

langfuse = get_client()

dataset = langfuse.get_dataset("visual-qa")

def my_multi_modal_task(*, item, **kwargs):
    image = item.input["image"]
    assert isinstance(image, LangfuseMediaReference)

    # Use the format expected by your model provider.
    image_data_uri = image.fetch_data_uri()

    # Call your multi-modal application here.
    return run_visual_qa(
        question=item.input["question"],
        image=image_data_uri,
    )

result = dataset.run_experiment(
    name="Visual QA",
    task=my_multi_modal_task,
)

JS/TS:

import {
  LangfuseClient,
  LangfuseMediaReference,
} from "@langfuse/client";

const langfuse = new LangfuseClient();

const dataset = await langfuse.dataset.get("visual-qa");

const result = await dataset.runExperiment({
  name: "Visual QA",
  task: async (item) => {
    const image = item.input.image as LangfuseMediaReference;

    // Use the format expected by your model provider.
    const imageDataUri = await image.fetchDataUri();

    // Call your multi-modal application here.
    return runVisualQa({
      question: item.input.question,
      image: imageDataUri,
    });
  },
});

LangfuseMediaReference는 미디어를 raw bytes, raw base64, data URI로 가져오는 헬퍼를 제공합니다:

SDK	Bytes	Base64	Data URI
Python	fetch_bytes()	fetch_base64()	fetch_data_uri()
JS/TS	fetchBytes()	fetchBase64()	fetchDataUri()

해석된 URL은 서명되며 만료됩니다. 실험이 사용하기 전에 URL이 만료되면 데이터셋을 다시 가져와 새로운 미디어 참조를 받으세요.

비동기 Task와 평가자

task 함수와 평가자 모두 비동기일 수 있습니다.

Python:

import asyncio
from langfuse.openai import AsyncOpenAI

async def async_llm_task(*, item, **kwargs):
    """Async task using OpenAI"""
    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": item["input"]}]
    )

    return response.choices[0].message.content

# Works seamlessly with async functions
result = langfuse.run_experiment(
    name="Async Experiment",
    data=test_data,
    task=async_llm_task,
    max_concurrency=5  # Control concurrent API calls
)

print(result.format())

JS/TS:

import OpenAI from "openai";

const asyncLlmTask = async (item) => {
  // Async task using OpenAI
  const client = new OpenAI();
  const response = await client.chat.completions.create({
    model: "gpt-4",
    messages: [{ role: "user", content: item.input }],
  });

  return response.choices[0].message.content;
};

// Works seamlessly with async functions
const result = await langfuse.experiment.run({
  name: "Async Experiment",
  data: testData,
  task: asyncLlmTask,
  maxConcurrency: 5, // Control concurrent API calls
});

console.log(await result.format());

구성 옵션(Configuration Options)

다양한 구성 옵션으로 실험 동작을 커스터마이즈하세요.

Python:

result = langfuse.run_experiment(
    name="Configurable Experiment",
    run_name="Custom Run Name", # will be dataset run name if dataset is used
    description="Experiment with custom configuration",
    data=test_data,
    task=my_task,
    evaluators=[accuracy_evaluator],
    run_evaluators=[average_accuracy],
    max_concurrency=10,  # Max concurrent executions
    metadata={  # Attached to all traces
        "model": "gpt-4",
        "temperature": 0.7,
        "version": "v1.2.0"
    }
)

print(result.format())

JS/TS:

const result = await langfuse.experiment.run({
  name: "Configurable Experiment",
  runName: "Custom Run Name", // will be dataset run name if dataset is used
  description: "Experiment with custom configuration",
  data: testData,
  task: myTask,
  evaluators: [accuracyEvaluator],
  runEvaluators: [averageAccuracy],
  maxConcurrency: 10, // Max concurrent executions
  metadata: {
    // Attached to all traces
    model: "gpt-4",
    temperature: 0.7,
    version: "v1.2.0",
  },
});

console.log(await result.format());

Autoevals 통합

autoevals 라이브러리 통합을 통해 사전 구축된 평가 함수에 접근하세요.

Python SDK는 직접 통합으로 AutoEvals 평가자를 지원합니다:

from langfuse.experiment import create_evaluator_from_autoevals
from autoevals.llm import Factuality

evaluator = create_evaluator_from_autoevals(Factuality())

result = langfuse.run_experiment(
    name="Autoevals Integration Test",
    data=test_data,
    task=my_task,
    evaluators=[evaluator]
)

print(result.format())

JS SDK는 사전 구축된 평가 함수를 위해 AutoEvals 라이브러리와 원활하게 통합됩니다:

import { Factuality, Levenshtein } from "autoevals";
import { createEvaluatorFromAutoevals } from "@langfuse/client";

// Convert AutoEvals evaluators to Langfuse-compatible format
const factualityEvaluator = createEvaluatorFromAutoevals(Factuality());
const levenshteinEvaluator = createEvaluatorFromAutoevals(Levenshtein());

// Use with additional parameters
const customFactualityEvaluator = createEvaluatorFromAutoevals(
  Factuality,
  { model: "gpt-4o" } // Additional AutoEvals parameters
);

const result = await langfuse.experiment.run({
  name: "AutoEvals Integration Test",
  data: testDataset,
  task: myTask,
  evaluators: [
    factualityEvaluator,
    levenshteinEvaluator,
    customFactualityEvaluator,
  ],
});

console.log(await result.format());

선택: UI에서 SDK 실험 트리거

SDK로 Experiments를 설정할 때 실험 실행을 Langfuse UI에서 트리거할 수 있게 하는 것이 유용할 수 있습니다.

Langfuse의 트리거 요청을 받을 HTTP 엔드포인트를 설정해야 합니다.

데이터셋으로 이동

• 내 프로젝트 > Datasets로 이동
• 원격 실험 트리거를 설정할 데이터셋을 클릭

설정 페이지 열기

Start Experiment를 클릭해 설정 페이지를 엽니다. Custom Experiment 아래의 **⚡**를 클릭합니다.

웹훅 구성

실험이 트리거될 때 웹훅을 받을 외부 평가 서비스의 URL을 입력합니다. 웹훅으로 보낼 기본 구성을 지정합니다. 사용자는 실험 트리거 시 이를 수정할 수 있습니다.

원격 실험 트리거 요청은 선택적으로 프롬프트 웹훅 이 사용하는 것과 같은 HMAC 서명 x-langfuse-signature 헤더를 포함할 수 있습니다. 설정 폼에서 요청 서명을 활성화하고 Advanced Options에서 추가 커스텀 헤더를 구성하세요. 서명이나 커스텀 헤더가 없는 기존 트리거는 계속 작동합니다. 내 엔드포인트는 빠르게 2xx 응답을 반환하고 실험을 비동기로 실행해야 합니다.

실험 트리거

구성 후 팀원들은 Custom Experiment 옵션 아래의 Run 버튼으로 원격 실험을 트리거할 수 있습니다. Langfuse는 데이터셋 메타데이터(ID와 이름)와 커스텀 구성을 내 웹훅으로 보냅니다.

일반적인 워크플로우: 내 웹훅이 요청을 받고, Langfuse에서 데이터셋을 가져오고, 데이터셋 아이템에 대해 내 애플리케이션을 실행하고, 결과를 평가하고, 점수를 새 Experiment run으로 Langfuse에 다시 수집합니다.

더 알아보기 (Learn more)

• 출처 문서: SDK로 실험하기(Experiments via SDK)