Langfuse Plugin

The Langfuse Plugin integrates @iqai/adk agents with Langfuse for robust tracing, monitoring, and observability of LLM and tool executions. It automatically records user messages, agent actions, model generations, and tool calls, including errors and token usage.

This plugin is especially useful for:

• Observability of agent workflows
• Debugging tool or model errors
• Tracking model usage and token consumption
• Correlating events across complex agent hierarchies

Key Features

• Automatic tracing: Tracks every invocation, agent, tool, and model generation
• Error monitoring: Captures LLM or tool errors with full context
• Token and model usage tracking: Aggregates input/output tokens and models used per invocation
• Hierarchical spans: Supports nested agents and tool calls with parent-child relationships
• Flexible serialization: Converts complex Content and Event objects into structured logs for Langfuse
• Flush and shutdown control: Manually flush or shutdown Langfuse client if needed

Installation

pnpm add @iqai/adk langfuse

Usage

import {
  AgentBuilder,
  InMemorySessionService,
  LangfusePlugin,
} from "@iqai/adk";
import { openrouter } from "@openrouter/ai-sdk-provider";
import { randomUUID } from "crypto";

const langfusePlugin = new LangfusePlugin({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  flushAt: 5,
});

const sessionService = new InMemorySessionService();

const { runner } = await AgentBuilder.withModel(
  openrouter("openai/gpt-4.1-mini"),
)
  .withDescription("Calendar scheduling assistant")
  .withInstruction("Help users manage calendar events.")
  .withTools(/* calendar tools */)
  .withSessionService(sessionService)
  .withQuickSession({
    sessionId: randomUUID(),
    appName: "calendar-agent",
  })
  .withPlugins(langfusePlugin)
  .build();

import { LlmAgent, LangfusePlugin } from "@iqai/adk";

const agent = new LlmAgent({
  name: "calendar_specialist",
  description: "Agent for calendar scheduling",
  plugins: [
    new LangfusePlugin({
      publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
      secretKey: process.env.LANGFUSE_SECRET_KEY!,
    }),
  ],
});

import { LlmAgent, LangfusePlugin } from "@iqai/adk";

const langfusePlugin = new LangfusePlugin({
  name: "my_langfuse_plugin",
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: "https://eu.cloud.langfuse.com", // EU region
  release: "v1.2.3", // Release version for tracking
  flushAt: 10, // Flush after 10 events
  flushInterval: 5000, // Flush every 5 seconds
});

const agent = new LlmAgent({
  name: "production_agent",
  description: "Production agent with comprehensive monitoring",
  plugins: [langfusePlugin],
});

Configuration Options

Callback Hooks

The plugin intercepts key agent events to send structured logs to Langfuse:

User messages

• onUserMessageCallback: Records user input and triggers a user_message event

Agent execution

• beforeAgentCallback / afterAgentCallback: Starts and ends agent spans. Tracks input, output, models used, and sub-agent hierarchy

Model calls

• beforeModelCallback / afterModelCallback: Captures model request and response, including token usage and generation metadata
• onModelErrorCallback: Logs errors for LLM calls with full context

Tool calls

• beforeToolCallback / afterToolCallback: Captures tool invocation and results
• onToolErrorCallback: Records tool failures and metadata

Run lifecycle

• beforeRunCallback / afterRunCallback: Tracks overall invocation start and completion
• onEventCallback: Captures all events in the invocation, including function calls and final outputs

Advanced Usage

Customizing Serialization

LangfusePlugin converts complex Content or Event objects to structured logs. You can extend:

class CustomLangfusePlugin extends LangfusePlugin {
  protected toPlainText(data: any): string {
    // Custom logic to extract plain text
    return super.toPlainText(data);
  }
}

Manual Flush and Shutdown

By default, the Langfuse client batches events and flushes them automatically based on the flushAt and flushInterval configuration options. In some scenarios, however, you may want explicit control over when data is sent or when the client is shut down.

flush()

await langfusePlugin.flush();

Forces all buffered traces, spans, and events to be sent to Langfuse immediately.

Use this when:

• You are running in short-lived environments (e.g. serverless functions, scripts, or cron jobs)
• You want to ensure all telemetry is delivered before continuing execution
• You need deterministic delivery for testing or debugging

Internally, this calls flushAsync() on the Langfuse client and does not shut it down, so the plugin can continue recording events afterward.

close()

await langfusePlugin.close();

Gracefully shuts down the Langfuse client after flushing any remaining buffered data.

Use this when:

• Your application or worker is terminating
• You no longer plan to record any additional traces or events
• You want to release resources and prevent further writes

Once close() is called, the plugin should be considered inactive.

Recommended Usage

In long-running applications (e.g. servers), you typically do not need to call either method manually. In short-lived processes, a common pattern is:

await langfusePlugin.flush();
await langfusePlugin.close();

This ensures all telemetry is delivered before the process exits.

Integration Examples

With Multiple Plugins

import { InMemoryRunner, LlmAgent } from "@iqai/adk";
import {
  LangfusePlugin,
  ToolOutputFilterPlugin,
  ReflectAndRetryToolPlugin,
} from "@iqai/adk";

const plugins = [
  // Monitor everything
  new LangfusePlugin({
    publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
    secretKey: process.env.LANGFUSE_SECRET_KEY!,
  }),

  // Filter large outputs
  new ToolOutputFilterPlugin({
    filterModel: "gemini-1.5-flash",
  }),

  // Retry failed operations
  new ReflectAndRetryToolPlugin({
    maxRetries: 3,
  }),
];

const runner = new InMemoryRunner(agent, {
  appName: "production-app",
  plugins,
});

Environment-specific Configuration

import { LangfusePlugin } from "@iqai/adk";

// Different configurations for different environments
const getLangfuseConfig = () => {
  if (process.env.NODE_ENV === "production") {
    return {
      publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
      secretKey: process.env.LANGFUSE_SECRET_KEY!,
      baseUrl: process.env.LANGFUSE_BASE_URL || "https://us.cloud.langfuse.com",
      flushAt: 10,
      flushInterval: 5000,
    };
  } else {
    return {
      publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
      secretKey: process.env.LANGFUSE_SECRET_KEY!,
      baseUrl: process.env.LANGFUSE_BASE_URL || "https://us.cloud.langfuse.com",
      flushAt: 1, // Immediate flushing for development
      flushInterval: 1000,
    };
  }
};

const langfusePlugin = new LangfusePlugin(getLangfuseConfig());

Benefits

• Centralized observability for multi-agent workflows
• Debugging complex tool and LLM interactions
• Token and model usage analytics
• Error tracking with detailed context
• Supports nested agents and tool spans

Best Practices

Configuration

• Use appropriate flush settings: Higher values for production, lower for development
• Set meaningful release tags: Helps track performance across versions
• Configure proper base URL: Use EU or US region based on your requirements

Security

• Secure API keys: Store Langfuse keys in environment variables
• Use different projects: Separate development and production data
• Monitor usage: Track API key usage and rotate regularly

Performance

• Batch appropriately: Balance between data freshness and API calls
• Monitor flush intervals: Adjust based on your application's needs
• Handle network issues: The plugin includes retry logic for network failures

Troubleshooting

Common Issues

No data appearing in Langfuse:

• Verify API keys are correct
• Check network connectivity to Langfuse
• Ensure plugin is properly registered
• Check flush settings aren't too high

Missing traces or spans:

• Verify all plugin callbacks are implemented
• Check for errors in plugin execution
• Ensure proper shutdown/flush on application exit

Performance issues:

• Increase flush interval for batching
• Reduce flushAt threshold
• Check for network latency to Langfuse API

Authentication errors:

• Verify publicKey and secretKey are correct
• Check API key permissions in Langfuse dashboard
• Ensure keys are for the correct project