Migrate from Arize AX to Langfuse

# Before
import os
from arize.otel import register
from openinference.instrumentation.anthropic import AnthropicInstrumentor  # or OpenAIInstrumentor, ...

tracer_provider = register(
    space_id=os.environ["ARIZE_SPACE_ID"],
    api_key=os.environ["ARIZE_API_KEY"],
    project_name=os.environ["ARIZE_PROJECT_NAME"],
    endpoint=os.environ["ARIZE_OTLP_ENDPOINT"],  # copy from the AX Get started page
)
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)

# After
from langfuse import get_client
from openinference.instrumentation.anthropic import AnthropicInstrumentor

get_client()
AnthropicInstrumentor().instrument()

// Before
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
import { resourceFromAttributes } from "@opentelemetry/resources";
import { SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { SEMRESATTRS_PROJECT_NAME } from "@arizeai/openinference-semantic-conventions";
import { AnthropicInstrumentation } from "@arizeai/openinference-instrumentation-anthropic";
import Anthropic from "@anthropic-ai/sdk";

const instrumentation = new AnthropicInstrumentation();
instrumentation.manuallyInstrument(Anthropic);

const provider = new NodeTracerProvider({
  resource: resourceFromAttributes({
    [SEMRESATTRS_PROJECT_NAME]: process.env.ARIZE_PROJECT_NAME!,
  }),
  spanProcessors: [
    new SimpleSpanProcessor(
      new OTLPTraceExporter({
        // Get started value is .../v1; the JS exporter needs .../v1/traces
        url: `${process.env.ARIZE_OTLP_ENDPOINT}/traces`,
        headers: {
          "arize-space-id": process.env.ARIZE_SPACE_ID!,
          "arize-api-key": process.env.ARIZE_API_KEY!,
        },
      }),
    ),
  ],
});
provider.register();

// After
import { NodeSDK } from "@opentelemetry/sdk-node";
import {
  LangfuseSpanProcessor,
  isDefaultExportSpan,
  type ShouldExportSpan,
} from "@langfuse/otel";
import { AnthropicInstrumentation } from "@arizeai/openinference-instrumentation-anthropic";
import Anthropic from "@anthropic-ai/sdk";

const afterInstrumentation = new AnthropicInstrumentation();
afterInstrumentation.manuallyInstrument(Anthropic);

const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  isDefaultExportSpan(otelSpan) ||
  otelSpan.instrumentationScope.name.startsWith("@arizeai/openinference");

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

import os
import requests
from urllib.parse import quote
from langfuse import get_client

AX_BASE = os.environ["ARIZE_API_BASE"]  # e.g. https://api.eu-west-1a.arize.com
AX_KEY = os.environ["ARIZE_API_KEY"]
DATASET_ID = os.environ["ARIZE_DATASET_ID"]
DATASET_NAME = "support-golden"

examples = []
cursor = None
while True:
    params = {"limit": 500}
    if cursor:
        params["cursor"] = cursor
    payload = requests.get(
        f"{AX_BASE}/v2/datasets/{quote(DATASET_ID, safe='')}/examples",
        headers={"Authorization": f"Bearer {AX_KEY}", "Accept": "application/json"},
        params=params,
        timeout=30,
    ).json()
    examples.extend(payload.get("examples") or [])
    pagination = payload.get("pagination") or {}
    if not pagination.get("has_more"):
        break
    cursor = pagination.get("next_cursor")
    if not cursor:
        break

langfuse = get_client()
langfuse.create_dataset(name=DATASET_NAME)

for row in examples:
    langfuse.create_dataset_item(
        id=row["id"],  # stable retry key in Langfuse; this is the AX example id
        dataset_name=DATASET_NAME,
        input=row.get("input") or row.get("query") or row.get("question") or row,
        expected_output=row.get("expected_output")
        or row.get("output")
        or row.get("answer"),
        metadata={
            k: v
            for k, v in row.items()
            if k
            not in {
                "id",
                "created_at",
                "updated_at",
                "input",
                "expected_output",
                "query",
                "question",
                "output",
                "answer",
                "annotations",
            }
        },
    )

import { listDatasetExamples } from "@arizeai/ax-client";
import { LangfuseClient } from "@langfuse/client";

// Reads ARIZE_API_KEY and ARIZE_BASE_URL (EU: https://api.eu-west-1a.arize.com)
const DATASET = process.env.ARIZE_DATASET_ID!; // dataset name or ID
const SPACE = process.env.ARIZE_SPACE_ID; // required when DATASET is a name
const DATASET_NAME = "support-golden";
const SKIP = new Set([
  "id",
  "createdAt",
  "updatedAt",
  "annotations",
  "input",
  "query",
  "question",
  "expected_output",
  "expectedOutput",
  "output",
  "answer",
]);

const examples: Record<string, unknown>[] = [];
let cursor: string | undefined;
while (true) {
  const { data, pagination } = await listDatasetExamples({
    dataset: DATASET,
    space: SPACE,
    limit: 500,
    cursor,
  });
  examples.push(...data);
  if (!pagination.hasMore) break;
  cursor = pagination.nextCursor;
  if (!cursor) break;
}

const langfuse = new LangfuseClient();
await langfuse.api.datasets.create({ name: DATASET_NAME });

for (const row of examples) {
  await langfuse.dataset.createItem({
    id: String(row.id), // stable retry key in Langfuse; this is the AX example id
    datasetName: DATASET_NAME,
    input: row.input ?? row.query ?? row.question ?? row,
    expectedOutput:
      row.expected_output ?? row.expectedOutput ?? row.output ?? row.answer,
    metadata: Object.fromEntries(
      Object.entries(row).filter(([key]) => !SKIP.has(key)),
    ),
  });
}

from langfuse import get_client

langfuse = get_client()
dataset = langfuse.get_dataset("support-golden")


def task(*, item, **kwargs):
    return run_your_app(item.input)  # same path as production


dataset.run_experiment(name="after-migration", task=task)

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

const langfuse = new LangfuseClient();
const dataset = await langfuse.dataset.get("support-golden");

await dataset.runExperiment({
  name: "after-migration",
  task: async (item) => runYourApp(item.input), // same path as production
});