Observability and Tracing for the Vercel AI SDK

// instrumentation.ts
 
import { LangfuseSpanProcessor, ShouldExportSpan } from "@langfuse/otel";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
 
// Optional: filter our NextJS infra spans
const shouldExportSpan: ShouldExportSpan = (span) => {
  return span.otelSpan.instrumentationScope.name !== "next.js";
};
 
export const langfuseSpanProcessor = new LangfuseSpanProcessor({
  shouldExportSpan,
});
 
const tracerProvider = new NodeTracerProvider({
  spanProcessors: [langfuseSpanProcessor],
});
 
tracerProvider.register();

// route.ts
 
import { streamText } from "ai";
import { after } from "next/server";
 
import { openai } from "@ai-sdk/openai";
import {
  observe,
  updateActiveObservation,
  updateActiveTrace,
} from "@langfuse/tracing";
import { trace } from "@opentelemetry/api";
 
import { langfuseSpanProcessor } from "@/src/instrumentation";
 
const handler = async (req: Request) => {
  const {
    messages,
    chatId,
    userId,
  }: { messages: UIMessage[]; chatId: string; userId: string } =
    await req.json();
 
  // Set session id and user id on active trace
  const inputText = messages[messages.length - 1].parts.find(
    (part) => part.type === "text"
  )?.text;
 
  updateActiveObservation({
    input: inputText,
  });
 
  updateActiveTrace({
    name: "my-ai-sdk-trace",
    sessionId: chatId,
    userId,
    input: inputText,
  });
 
  const result = streamText({
    // ... other streamText options ...
    experimental_telemetry: {
      isEnabled: true,
    },
    onFinish: async (result) => {
      updateActiveObservation({
        output: result.content,
      });
      updateActiveTrace({
        output: result.content,
      });
 
      // End span manually after stream has finished
      trace.getActiveSpan().end();
    },
    onError: async (error) => {
      updateActiveObservation({
        output: error,
        level: "ERROR"
      });
      updateActiveTrace({
        output: error,
      });
 
      // End span manually after stream has finished
      trace.getActiveSpan()?.end();
    },
  });
 
  // Important in serverless environments: schedule flush after request is finished
  after(async () => await langfuseSpanProcessor.forceFlush());
 
  return result.toUIMessageStreamResponse();
};
 
export const POST = observe(handler, {
  name: "handle-chat-message",
  endOnExit: false, // end observation _after_ stream has finished
});

import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";
import { LangfuseClient } from "@langfuse/client"; // Add this import       
 
const langfuse = new LangfuseClient();
 
// Get current `production` version
const prompt = await langfuse.prompt.get("movie-critic");
 
// Insert variables into prompt template
const compiledPrompt = prompt.compile({
  someVariable: "example-variable",
});
 
const { text } = await generateText({
  model: openai("gpt-5"),
  prompt: compiledPrompt,
  experimental_telemetry: { 
    isEnabled: true,
    metadata: {
      langfusePrompt: prompt.toJSON() // This links the Generation to your prompt in Langfuse
    },
  },
});