Migrate from LangSmith to Langfuse

pip install "langsmith[otel]" opentelemetry-sdk opentelemetry-exporter-otlp-proto-http

# Before: traces go to LangSmith only
import anthropic
from langsmith import traceable
from langsmith.wrappers import wrap_anthropic


@traceable(run_type="chain", name="support-agent")
def answer(question: str) -> str:
    client = wrap_anthropic(anthropic.Anthropic())
    ...

# After: same instrumentation, spans fan out to LangSmith and Langfuse
import base64
import os

os.environ["LANGSMITH_OTEL_ENABLED"] = "true"  # set before the first traced call

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

auth = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()
provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(
            endpoint=os.environ.get("LANGFUSE_BASE_URL", "https://cloud.langfuse.com").rstrip("/")
            + "/api/public/otel/v1/traces",
            headers={
                "Authorization": f"Basic {auth}",
                "x-langfuse-ingestion-version": "4",  # real-time ingestion
            },
        )
    )
)
trace.set_tracer_provider(provider)  # LangSmith detects and reuses this provider

# Application code stays unchanged:
import anthropic
from langsmith import traceable
from langsmith.wrappers import wrap_anthropic

npm install langsmith @langfuse/otel @opentelemetry/sdk-trace-node \
  @opentelemetry/api @opentelemetry/sdk-trace-base \
  @opentelemetry/exporter-trace-otlp-proto @opentelemetry/context-async-hooks

// OTel bootstrap: must run before the first traced call
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { initializeOTEL } from "langsmith/experimental/otel/setup";

const langfuseProcessor = new LangfuseSpanProcessor(); // reads LANGFUSE_* env vars
const provider = new NodeTracerProvider({
  spanProcessors: [langfuseProcessor],
});
provider.register();
initializeOTEL({
  globalTracerProvider: provider,
  skipGlobalContextManagerSetup: true,
});

// Application code, unchanged from the source state:
import Anthropic from "@anthropic-ai/sdk";
import { Client } from "langsmith";
import { traceable } from "langsmith/traceable";
import { wrapAnthropic } from "langsmith/wrappers/anthropic";

const anthropic = wrapAnthropic(new Anthropic());
const lsClient = new Client();

export const answer = traceable(
  async (question: string) => {
    // ... your agent logic and anthropic.messages.create() call ...
  },
  { name: "support-agent", run_type: "chain", client: lsClient },
);

// Required in short-lived scripts (batched spans are lost at exit otherwise):
await lsClient.awaitPendingTraceBatches();
await langfuseProcessor.forceFlush();

pip install langfuse langchain

# Before: LANGSMITH_TRACING=true in the environment; no tracing code
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-haiku-4-5-20251001", max_tokens=1024)  # your existing model
agent = create_agent(model, tools, system_prompt=SYSTEM_PROMPT)

agent.invoke({"messages": [{"role": "user", "content": question}]})

# After: add the Langfuse callback; LangSmith env tracing keeps working in parallel
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langfuse import get_client
from langfuse.langchain import CallbackHandler

langfuse_handler = CallbackHandler()

model = ChatAnthropic(model="claude-haiku-4-5-20251001", max_tokens=1024)
agent = create_agent(model, tools, system_prompt=SYSTEM_PROMPT)

agent.invoke(
    {"messages": [{"role": "user", "content": question}]},
    config={"callbacks": [langfuse_handler]},
)

get_client().flush()  # required in short-lived scripts

npm install @langfuse/langchain @langfuse/otel @opentelemetry/sdk-node

// Before: LANGSMITH_TRACING=true in the environment; no tracing code
import { ChatAnthropic } from "@langchain/anthropic";
import { createReactAgent } from "@langchain/langgraph/prebuilt";

const model = new ChatAnthropic({ model: "claude-haiku-4-5-20251001" }); // your existing model
const agent = createReactAgent({ llm: model, tools, prompt: SYSTEM_PROMPT });

await agent.invoke({ messages: [{ role: "user", content: question }] });

// After: add the Langfuse processor + callback; LangSmith env tracing keeps working in parallel
import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { CallbackHandler } from "@langfuse/langchain";
import { ChatAnthropic } from "@langchain/anthropic";
import { createReactAgent } from "@langchain/langgraph/prebuilt";

const langfuseProcessor = new LangfuseSpanProcessor(); // reads LANGFUSE_* env vars
const sdk = new NodeSDK({ spanProcessors: [langfuseProcessor] });
sdk.start();

const langfuseHandler = new CallbackHandler();

const model = new ChatAnthropic({ model: "claude-haiku-4-5-20251001" });
const agent = createReactAgent({ llm: model, tools, prompt: SYSTEM_PROMPT });

await agent.invoke(
  { messages: [{ role: "user", content: question }] },
  { callbacks: [langfuseHandler] },
);

await langfuseProcessor.forceFlush(); // short-lived scripts only

from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult


class LangfuseAttributeMapper(SpanExporter):
    """Rewrite LangSmith OTel attributes to the names Langfuse maps natively."""

    def __init__(self, inner: SpanExporter):
        self._inner = inner

    def export(self, spans) -> SpanExportResult:
        for span in spans:
            attrs = dict(span._attributes or {})
            mapped = dict(attrs)
            if "langsmith.metadata.session_id" in attrs:
                mapped["session.id"] = attrs["langsmith.metadata.session_id"]
            if "langsmith.metadata.user_id" in attrs:
                mapped["user.id"] = attrs["langsmith.metadata.user_id"]
            if "langsmith.span.tags" in attrs:  # arrives comma-joined
                mapped["langfuse.trace.tags"] = [
                    t.strip() for t in str(attrs["langsmith.span.tags"]).split(",")
                ]
            # langsmith.span.kind is not mapped either; without this, chain
            # roots land in Langfuse as generic spans.
            kind = str(attrs.get("langsmith.span.kind", "")).lower()
            if kind in ("chain", "tool", "llm"):
                mapped["langfuse.observation.type"] = (
                    "generation" if kind == "llm" else kind
                )
            if mapped != attrs:
                # relies on SDK internals; verified with opentelemetry-sdk 1.44
                span._attributes = mapped
        return self._inner.export(spans)

    def shutdown(self):
        self._inner.shutdown()

    def force_flush(self, timeout_millis: int = 30000):
        return self._inner.force_flush(timeout_millis)


# In the Step 1 setup, wrap the exporter:
# BatchSpanProcessor(LangfuseAttributeMapper(OTLPSpanExporter(endpoint=..., headers=...)))

import { trace as otelApi } from "@opentelemetry/api";

// inside the root traceable function:
const activeSpan = otelApi.getActiveSpan();
activeSpan?.setAttribute("langfuse.session.id", threadId); // → trace.sessionId
activeSpan?.setAttribute("langfuse.user.id", userId); // → trace.userId
activeSpan?.setAttribute("langfuse.trace.tags", ["support", "js-migration"]); // string[] → trace.tags

from langfuse import Langfuse
from langsmith import Client

ls = Client()  # reads LANGSMITH_API_KEY / LANGSMITH_ENDPOINT
langfuse = Langfuse()  # reads LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_BASE_URL

DATASET = "support-golden"

ls_dataset = ls.read_dataset(dataset_name=DATASET)
examples = list(ls.list_examples(dataset_id=ls_dataset.id))

langfuse.create_dataset(
    name=DATASET,
    description=ls_dataset.description or "",
    metadata={"migrated_from": "langsmith", "langsmith_dataset_id": str(ls_dataset.id)},
)
for ex in examples:
    langfuse.create_dataset_item(
        dataset_name=DATASET,
        id=str(ex.id),  # stable retry key: the LangSmith example id
        input=ex.inputs,
        expected_output=ex.outputs,
        metadata={**(ex.metadata or {}), "langsmith_example_id": str(ex.id)},
    )
print(f"migrated {len(examples)} items")

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

const ls = new Client(); // reads LANGSMITH_API_KEY / LANGSMITH_ENDPOINT
const langfuse = new LangfuseClient(); // reads LANGFUSE_* env vars

const DATASET = "support-golden";

const lsDataset = await ls.readDataset({ datasetName: DATASET });

await langfuse.api.datasets.create({
  name: DATASET,
  description: lsDataset.description ?? "",
  metadata: {
    migrated_from: "langsmith",
    langsmith_dataset_id: String(lsDataset.id),
  },
});

let count = 0;
for await (const ex of ls.listExamples({ datasetId: lsDataset.id })) {
  await langfuse.dataset.createItem({
    id: String(ex.id), // stable retry key: the LangSmith example id
    datasetName: DATASET,
    input: ex.inputs,
    expectedOutput: ex.outputs,
    metadata: { ...(ex.metadata ?? {}), langsmith_example_id: String(ex.id) },
  });
  count += 1;
}
console.log(`migrated ${count} dataset items`);

import re

from langchain_core.prompts import MessagesPlaceholder
from langfuse import Langfuse
from langsmith import Client

ls = Client()
langfuse = Langfuse()

pulled = ls.pull_prompt("support-agent")  # ChatPromptTemplate; requires langchain-core

role_map = {"System": "system", "Human": "user", "AI": "assistant"}
messages = []
for msg in pulled.messages:
    if isinstance(msg, MessagesPlaceholder):  # e.g. chat history
        messages.append({"type": "placeholder", "name": msg.variable_name})
        continue
    role = role_map[msg.__class__.__name__.replace("MessagePromptTemplate", "")]
    template = msg.prompt.template
    if msg.prompt.template_format == "f-string":  # mustache is already {{var}}
        template = re.sub(r"\{(\w+)\}", r"{{\1}}", template)  # {var} -> {{var}}
    messages.append({"role": role, "content": template})

langfuse.create_prompt(
    name="support-agent",
    type="chat",
    prompt=messages,
    labels=["production"],
    commit_message="Migrated from LangSmith Prompt Hub",
)

import { LangfuseClient } from "@langfuse/client";
import type { ChatPromptTemplate } from "@langchain/core/prompts";
import { pull } from "langchain/hub";

const langfuse = new LangfuseClient();

const pulled = await pull<ChatPromptTemplate>("support-agent"); // uses LANGSMITH_API_KEY

const roleMap: Record<string, "system" | "user" | "assistant"> = {
  System: "system",
  Human: "user",
  AI: "assistant",
};
const messages = pulled.promptMessages.map((msg) => {
  if (msg.constructor.name === "MessagesPlaceholder") // e.g. chat history
    return { type: "placeholder", name: (msg as any).variableName };
  const key = msg.constructor.name.replace("MessagePromptTemplate", "");
  const role = roleMap[key];
  if (!role) throw new Error(`Unmapped message type: ${key}`);
  let template: string = (msg as any).prompt.template;
  if ((msg as any).prompt.templateFormat !== "mustache") // mustache is already {{var}}
    template = template.replace(/\{(\w+)\}/g, "{{$1}}"); // f-string {var} -> {{var}}
  return { role, content: template };
});

await langfuse.prompt.create({
  name: "support-agent",
  type: "chat",
  prompt: messages,
  labels: ["production"],
  commitMessage: "Migrated from LangSmith Prompt Hub",
});

# Before: LangSmith
from langsmith import Client

ls = Client()


def target(inputs: dict) -> dict:
    return {"answer": run_your_app(inputs["question"])}


def keyword_match(outputs: dict, reference_outputs: dict) -> dict:
    got = (outputs.get("answer") or "").lower()
    want = (reference_outputs.get("answer") or "").lower()
    keywords = [w for w in want.replace(".", " ").split() if len(w) > 4][:3]
    hit = all(k in got for k in keywords) if keywords else False
    return {"key": "keyword_match", "score": 1.0 if hit else 0.0}


ls.evaluate(target, data="support-golden", evaluators=[keyword_match])

# After: Langfuse
from langfuse import Evaluation, Langfuse

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


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


def keyword_match(*, input, output, expected_output, **kwargs):
    got = (output or "").lower()
    want = ((expected_output or {}).get("answer") or "").lower()
    keywords = [w for w in want.replace(".", " ").split() if len(w) > 4][:3]
    hit = all(k in got for k in keywords) if keywords else False
    return Evaluation(name="keyword_match", value=1.0 if hit else 0.0)


result = dataset.run_experiment(
    name="after-migration",
    task=task,
    evaluators=[keyword_match],
)
print(result.format())

import { LangfuseClient, type Evaluation } from "@langfuse/client";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { NodeSDK } from "@opentelemetry/sdk-node";

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

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

const keywordMatch = async ({
  output,
  expectedOutput,
}: {
  output: any;
  expectedOutput?: any;
}): Promise<Evaluation> => {
  const got = String(output ?? "").toLowerCase();
  const want = String(expectedOutput?.answer ?? "").toLowerCase();
  const keywords = want
    .replaceAll(".", " ")
    .split(/\s+/)
    .filter((w) => w.length > 4)
    .slice(0, 3);
  const hit = keywords.length > 0 && keywords.every((k) => got.includes(k));
  return { name: "keyword_match", value: hit ? 1.0 : 0.0 };
};

const result = await dataset.runExperiment({
  name: "after-migration",
  task: async (item) => runYourApp(item.input.question), // same path as production
  evaluators: [keywordMatch],
});
console.log(await result.format());
await otelSdk.shutdown(); // flush pending spans