upgrade langfuse

src/app/index.ts

@@ -74,6 +74,12 @@ async function main(): Promise<void> {
 
   scheduleCron(config, pipeline, logger);
 
+  const shutdown = () => {
+    tracer.shutdown().catch((err) => logger.error({ err }, 'Error shutting down Langfuse tracer'));
+  };
+  process.on('SIGTERM', shutdown);
+  process.on('SIGINT', shutdown);
+
   if (config.API_ENABLED) {
     const app = buildServer({
       secretKey: config.SECRET_KEY,

src/core/langfuse/LangfusePromptGateway.ts

@@ -1,11 +1,11 @@
-import { Langfuse } from 'langfuse';
+import { LangfuseClient } from '@langfuse/client';
 import type { PromptFetcher, RawPrompt } from './PromptGateway';
 
 export class LangfusePromptGateway implements PromptFetcher {
-  private readonly client: Langfuse;
+  private readonly client: LangfuseClient;
 
   constructor(opts: { publicKey: string; secretKey: string; baseUrl: string }) {
-    this.client = new Langfuse({
+    this.client = new LangfuseClient({
       publicKey: opts.publicKey,
       secretKey: opts.secretKey,
       baseUrl: opts.baseUrl,
@@ -13,7 +13,7 @@ export class LangfusePromptGateway implements PromptFetcher {
   }
 
   async fetch(promptName: string): Promise<RawPrompt> {
-    const prompt = await this.client.getPrompt(promptName, undefined, { type: 'chat', cacheTtlSeconds: 0 });
+    const prompt = await this.client.prompt.get(promptName, { type: 'chat' });
     return {
       name: prompt.name,
       version: prompt.version,

src/core/langfuse/LangfuseTracer.ts

@@ -1,33 +1,43 @@
-import { Langfuse } from 'langfuse';
+import { NodeSDK } from '@opentelemetry/sdk-node';
+import { LangfuseSpanProcessor } from '@langfuse/otel';
+import { startObservation } from '@langfuse/tracing';
+import type { LangfuseGeneration, LangfuseSpan } from '@langfuse/tracing';
 import { Tracer } from '../llm/Tracer';
 import type { GenerationHandle, GenerationInput, GenerationOutput, TraceHandle } from '../llm/Tracer';
 
 export class LangfuseTracer extends Tracer {
-  private readonly client: Langfuse;
+  private readonly processor: LangfuseSpanProcessor;
+  private readonly sdk: NodeSDK;
 
   constructor(opts: { publicKey: string; secretKey: string; baseUrl: string }) {
     super();
-    this.client = new Langfuse({
+    this.processor = new LangfuseSpanProcessor({
       publicKey: opts.publicKey,
       secretKey: opts.secretKey,
       baseUrl: opts.baseUrl,
     });
+    this.sdk = new NodeSDK({ spanProcessors: [this.processor] });
+    this.sdk.start();
   }
 
   startTrace(name: string, metadata?: Record<string, unknown>): TraceHandle {
-    const trace = this.client.trace({ name, metadata });
+    const span: LangfuseSpan = startObservation(name, { metadata });
     return {
-      id: trace.id,
+      id: span.traceId,
       startGeneration(input: GenerationInput): GenerationHandle {
-        const generation = trace.generation({
-          name: input.name,
-          model: input.model,
-          input: input.input,
-          metadata: input.metadata,
-        });
+        const generation: LangfuseGeneration = span.startObservation(
+          input.name,
+          { model: input.model, input: input.input, metadata: input.metadata },
+          { asType: 'generation' }
+        );
         return {
           end(output: GenerationOutput) {
-            generation.end({ output: output.output, usage: output.usage, metadata: output.metadata });
+            generation.update({
+              output: output.output,
+              usageDetails: output.usage as Record<string, number> | undefined,
+              metadata: output.metadata,
+            });
+            generation.end();
           },
         };
       },
@@ -35,6 +45,10 @@ export class LangfuseTracer extends Tracer {
   }
 
   async flush(): Promise<void> {
-    await this.client.flushAsync();
+    await this.processor.forceFlush();
+  }
+
+  async shutdown(): Promise<void> {
+    await this.sdk.shutdown();
   }
 }