fix: langchainjs works with langfuse at 2510

Author: uptonking

langchain/chain-groq1-chat-local-langfuse.ts

@@ -1,13 +1,18 @@
-// import '@dotenvx/dotenvx/config';
+import '@dotenvx/dotenvx/config';
 import './langfuse-instrumentation';
 
 import { HumanMessage } from '@langchain/core/messages';
 import { ChatOpenAI } from '@langchain/openai';
+import { CallbackHandler } from '@langfuse/langchain';
 
-import { startActiveObservation, startObservation } from '@langfuse/tracing';
+import { langfuseSpanProcessor } from './langfuse-instrumentation';
 
-// import { NodeSDK } from "@opentelemetry/sdk-node";
-// import { LangfuseSpanProcessor } from "@langfuse/otel";
+// Initialize the Langfuse callback handler with tracing configuration
+const langfuseHandler = new CallbackHandler({
+  sessionId: 'user-session-123', // Track user session
+  userId: 'user-abc', // Track user identity
+  tags: ['langchain-test'], // Add searchable tags
+});
 
 const model = new ChatOpenAI({
   model: 'qwen/qwen3-vl-4b',
@@ -18,60 +23,17 @@ const model = new ChatOpenAI({
   // temperature: 0.5,
 });
 
-// import { CallbackHandler } from "@langfuse/langchain";
-// const langfuseHandler = new CallbackHandler();
-
 console.log(';; langfuse ', process.env['LANGFUSE_PUBLIC_KEY']);
 
-// const sdk = new NodeSDK({
-//   spanProcessors: [new LangfuseSpanProcessor()],
-// });
-
-// sdk.start();
-
-// await startActiveObservation("user-request", async (span) => {
-//   span.update({
-//     input: { query: "What is the capital of France?" },
-//   });
-
-//   // This generation will automatically be a child of "user-request"
-//   const generation = startObservation(
-//     "llm-call",
-//     {
-//       model: "gpt-4",
-//       input: [{ role: "user", content: "What is the capital of France?" }],
-//     },
-//     { asType: "generation" },
-//   );
-
-//   // ... LLM call logic ...
-//   // refers to https://github.com/calcajack3/auto-tool-eval/blob/main/src/utils/utils.ts
+const messages = [
+  new HumanMessage('give an brief intro to codemirror in less than 80 words '),
+];
 
-//   const messages = [
-//     new HumanMessage('give an brief intro to codemirror in less than 80 words '),
-//   ];
-
-//   const res = await model.invoke(messages, {
-//     // callbacks: [langfuseHandler]
-//   });
-//   console.log('res');
-
-//   generation
-//     .update({
-//       output: { content: "The capital of France is Paris." },
-//     })
-//     .end();
-
-//   span.update({ output: "Successfully answered." });
-// });
-
-async function main() {
-  await startActiveObservation('my-first-trace', async (span) => {
-    span.update({
-      input: 'Hello, Langfuse!',
-      output: 'This is my first trace!',
-    });
-  });
-}
+const res = await model.invoke(messages, {
+  callbacks: [langfuseHandler],
+  // Name for the trace (if no active span)
+  runName: 'joke-generator',
+});
+console.log(';; res ', res.content);
 
-main();
+await langfuseSpanProcessor.forceFlush();

langchain/langfuse-instrumentation.ts

@@ -1,8 +1,18 @@
 import { NodeSDK } from '@opentelemetry/sdk-node';
 import { LangfuseSpanProcessor } from '@langfuse/otel';
 
+// Export the processor to be able to flush it later
+// This is important for ensuring all spans are sent to Langfuse
+export const langfuseSpanProcessor = new LangfuseSpanProcessor({
+  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
+  secretKey: process.env.LANGFUSE_SECRET_KEY!,
+  baseUrl: process.env.LANGFUSE_HOST ?? 'https://us.cloud.langfuse.com', // Default to cloud if not specified
+  environment: process.env.NODE_ENV ?? 'development', // Default to development if not specified
+});
+
+// Initialize the OpenTelemetry SDK with our Langfuse processor
 const sdk = new NodeSDK({
-  spanProcessors: [new LangfuseSpanProcessor()],
+  spanProcessors: [langfuseSpanProcessor],
 });
 
 sdk.start();

langchain/langfuse-tracing-test1.ts

@@ -0,0 +1,70 @@
+import './langfuse-instrumentation';
+
+// Import necessary functions from the tracing package
+import {
+  startActiveObservation,
+  startObservation,
+  updateActiveTrace,
+  updateActiveObservation,
+} from '@langfuse/tracing';
+import { langfuseSpanProcessor } from './langfuse-instrumentation';
+
+// Start a new span with automatic context management
+await startActiveObservation('context-manager', async (span) => {
+  // Log the initial user query
+  span.update({
+    input: { query: 'What is the capital of France?' },
+  });
+
+  // Create a new generation span that will automatically be a child of "context-manager"
+  const generation = startObservation(
+    'llm-call',
+    {
+      model: 'gpt-4',
+      input: [{ role: 'user', content: 'What is the capital of France?' }],
+    },
+    { asType: 'generation' },
+  );
+
+  // ... LLM call logic would go here ...
+
+  // Update the generation with token usage statistics
+  generation.update({
+    usageDetails: {
+      input: 10, // Number of input tokens
+      output: 5, // Number of output tokens
+      cache_read_input_tokens: 2, // Tokens read from cache
+      some_other_token_count: 10, // Custom token metric
+      total: 17, // Optional: automatically calculated if not provided
+    },
+  });
+
+  // End the generation with the LLM response
+  generation
+    .update({
+      output: { content: 'The capital of France is Paris.' },
+    })
+    .end();
+
+  // Example user information
+  const user = { id: 'user-5678', name: 'Jane Doe', sessionId: '123' };
+
+  // Add an optional log level of type warning to the active span
+  updateActiveObservation({
+    level: 'WARNING',
+    statusMessage: 'This is a warning',
+  });
+
+  // Update the trace with user context
+  updateActiveTrace({
+    userId: user.id,
+    sessionId: user.sessionId,
+    metadata: { userName: user.name },
+  });
+
+  // Mark the span as complete with final output
+  span.update({ output: 'Successfully answered.' });
+});
+
+// Ensure all spans are sent to Langfuse
+await langfuseSpanProcessor.forceFlush();