文档 / 通过 SDK 做实验

通过 SDK 做实验

通过 SDK 做实验用于以编程方式将你的应用或提示词在数据集上循环,并可选地对结果应用评估方法。你可以使用托管在 Langfuse 上的数据集或本地数据集作为实验的基础。

关于通过 SDK 运行实验的更多细节,另见 JS/TS SDK 参考和 Python SDK 参考。

为什么使用通过 SDK 做实验?

完全灵活地使用你自己的应用逻辑
使用自定义打分函数评估单个条目和完整运行的输出
当你想在 Langfuse UI 中编写确定性的 Python 或 TypeScript 检查并在 observations 和实验间复用时,使用代码评估器
在同一数据集上并行运行多个实验
易于与你现有的评估基础设施集成
在 CI/CD 中运行你的实验,在发布前捕获回归

实验运行器 SDK

Python 和 JS/TS SDK 都提供了在数据集上运行实验的高级抽象。数据集可以是本地的或托管在 Langfuse 上的。使用实验运行器是用我们的 SDK 在数据集上运行实验的推荐方式。

实验运行器自动处理:

带可配置限制的并发执行任务
对所有执行的自动追踪以实现可观测性
带条目级和运行级评估器的灵活评估
错误隔离,使单个失败不会停止实验
数据集集成,便于对比和追踪

实验运行器 SDK 支持托管在 Langfuse 上的数据集和本地托管的数据集。如果你的实验使用托管在 Langfuse 上的数据集,SDK 会自动为你创建一个数据集运行,你可以在 Langfuse UI 中检查和对比。对于不在 Langfuse 上的本地托管数据集,只有 traces 和 scores(如果使用评估)会在 Langfuse 中追踪。

基本用法

从最简单的实验开始,在本地数据上测试你的任务函数。如果你已经在 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())

确保 OpenTelemetry 已正确设置,以便 traces 能送达 Langfuse。配置细节参见追踪设置文档。始终在执行结束时刷新 span processor,以确保所有 traces 被发送。

typescript

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;
};

// 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 注意:OpenTelemetry 必须正确设置,以便 traces 能送达 Langfuse。配置细节参见追踪设置文档。始终在执行结束时刷新 span processor,以确保所有 traces 被发送。

在本地数据上运行实验时,Langfuse 中只创建 traces——不生成数据集运行。每次任务执行都会创建一个单独的 trace,用于可观测性和调试。

使用 Langfuse 数据集

直接在存储于 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())

typescript

// 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 数据集时,数据集运行会在 Langfuse 中自动创建,并可在 UI 中对比。这使你能够随时间追踪实验性能,并在同一数据集上对比不同方法。

实验始终在实验时的最新数据集版本上运行。对特定数据集版本运行实验的支持即将添加到 SDK。

多模态实验

基于 SDK 的实验可以在 input、expectedOutput 或 metadata 中包含媒体附件的数据集上运行。当你通过 SDK 获取数据集时,每个媒体 token 默认会被 hydrate 为签名的 LangfuseMediaReference。

多模态数据集支持基于 SDK 的实验,需要 Python SDK >= 4.10.0 和 JS/TS SDK @langfuse/client >= 5.6.0。基于 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,
)

typescript

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 暴露了将媒体获取为原始字节、原始 base64 或 data URI 的辅助方法:

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

解析后的 URL 是签名的且会过期。如果 URL 在你的实验使用它之前过期,请重新获取数据集以接收新的媒体引用。

高级功能

用评估器和高级配置选项增强你的实验。

评估器

评估器在条目级评估任务输出的质量。它们接收每个条目的输入、元数据、输出和预期输出,并返回作为 Langfuse 中 traces 上的分数报告的评估指标。

这些 SDK 评估器函数在你的实验进程中运行。如果你想在 Langfuse UI 中编写确定性的评估器代码并让 Langfuse 在实验 observations 上执行它,请使用代码评估器。

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

typescript

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

运行级评估器

运行级评估器评估完整的实验结果并计算聚合指标。在 Langfuse 数据集上运行时,这些分数附加到完整的数据集运行,用于追踪整体实验性能。

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

typescript

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

异步任务和评估器

任务函数和评估器都可以是异步的。

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

typescript

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

配置选项

用各种配置选项自定义实验行为。

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

typescript

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 评估器:

python

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 库的无缝集成,用于预构建的评估函数:

typescript

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 做实验时,允许从 Langfuse UI 触发实验运行可能很有用。

你需要设置一个 webhook 来接收来自 Langfuse 的触发请求。

典型工作流:你的 webhook 接收请求,从 Langfuse 获取数据集,针对数据集条目运行你的应用,评估结果,并将分数作为新的实验运行接入回 Langfuse。