Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions examples/plugins/langfuse/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
# Langfuse Plugin

Emits [Langfuse](https://langfuse.com/) generation traces for every LLM completion routed through OmniRoute.

Records **prompt, response, model, provider, token usage, latency, and error details** to Langfuse cloud (`cloud.langfuse.com` or `us.cloud.langfuse.com`) or a self-hosted Langfuse instance.

## Install

Copy the `examples/plugins/langfuse/` directory to your OmniRoute plugins path, or install directly from the marketplace UI.

## Configuration

Fill these fields in the plugin config panel:

| Key | Required | Default | Notes |
|---|---|---|---|
| `publicKey` | Yes | `""` | Langfuse public key (`pk-lf-...`) |
| `secretKey` | Yes | `""` | Langfuse secret key (`sk-lf-...`) |
| `host` | No | `https://cloud.langfuse.com` | Also `https://us.cloud.langfuse.com` or self-hosted URL |
| `enabled` | No | `true` | Set to `false` to make the plugin a no-op without uninstalling |
| `sampleRate` | No | `1.0` | `0.1` = trace 10% of requests |
| `flushAt` | No | `15` | Events to buffer before flushing |
| `flushInterval` | No | `10000` | Max ms between flushes |
| `redactBody` | No | `false` | Set `true` to strip prompt + completion from traces (metadata still recorded) |

Get keys at [cloud.langfuse.com](https://cloud.langfuse.com) → Settings → API keys.

## What gets traced

Each LLM completion emits one Langfuse `generation` observation inside a per-request trace:

- **Trace:** `omniroute:<model>` with `userId`, `provider`, `requestId` metadata
- **Generation:** `chat.completion` with:
- `model` — full model ID
- `modelParameters` — `temperature`, `max_tokens`, `top_p`
- `input` — messages array (redacted if `redactBody: true`)
- `output` — assistant message (redacted if `redactBody: true`)
- `usage.promptTokens`, `usage.completionTokens`, `usage.totalTokens`
- `startTime`, `endTime` — for latency

Errors emit a generation with `level: "ERROR"` and `statusMessage`.

## Runtime dependency

The plugin lazy-loads the [`langfuse`](https://www.npmjs.com/package/langfuse) npm SDK on first request. Install it in the plugin's own directory so a broken SDK cannot crash the gateway:

```bash
cd examples/plugins/langfuse
npm install langfuse
```

## Privacy

- Prompts and completions are sent to Langfuse cloud (or your self-host) in cleartext unless `redactBody: true`
- API keys are stored in the plugin config, encrypted at rest by OmniRoute
- Set `sampleRate < 1.0` to reduce data volume
- Set `enabled: false` to disable without removing configuration

## Related

- [Langfuse docs](https://langfuse.com/docs)
- [Langfuse OpenTelemetry endpoint](https://langfuse.com/docs/opentelemetry/get-started) (alternative path — this plugin uses the native SDK)
- OmniRoute plugin SDK: `docs/frameworks/PLUGIN_SDK.md`
158 changes: 158 additions & 0 deletions examples/plugins/langfuse/index.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
/**
* Langfuse Plugin — emits generation traces for every LLM completion.
*
* Records model, provider, prompt, completion, token usage, latency,
* and error details to Langfuse cloud or self-hosted.
*
* @module langfuse
*/

let langfuseClient = null;

/**
* Lazy-init the Langfuse SDK. First trace request creates the client; the SDK
* batches events and flushes per config.flushAt / config.flushInterval.
*/
async function getClient(config) {
if (langfuseClient) return langfuseClient;
if (!config.publicKey || !config.secretKey) return null;
try {
const { Langfuse } = await import("langfuse");
langfuseClient = new Langfuse({
publicKey: config.publicKey,
secretKey: config.secretKey,
baseUrl: config.host || "https://cloud.langfuse.com",
flushAt: config.flushAt ?? 15,
flushInterval: config.flushInterval ?? 10000,
});
return langfuseClient;
} catch (err) {
console.error("[langfuse] SDK import failed:", err.message);
return null;
}
}

function shouldSample(rate) {
if (rate >= 1) return true;
if (rate <= 0) return false;
return Math.random() < rate;
}

function truncateBody(body, redact) {
if (redact) return "[REDACTED]";
if (body === undefined || body === null) return undefined;
return body;
}

/**
* onRequest — mark trace start, capture request metadata.
*/
export async function onRequest(ctx) {
const config = ctx?.config || {};
if (config.enabled === false) return;
if (!shouldSample(config.sampleRate ?? 1)) return;
if (ctx?.metadata) {
ctx.metadata.__langfuseStart = Date.now();
ctx.metadata.__langfuseSampled = true;
}
}

/**
* onResponse — emit a Langfuse generation event with prompt, completion, usage.
*/
export async function onResponse(ctx) {
const config = ctx?.config || {};
if (config.enabled === false) return;
if (!ctx?.metadata?.__langfuseSampled) return;

const client = await getClient(config);
if (!client) return;

const start = ctx.metadata.__langfuseStart || Date.now();
const end = Date.now();
const body = ctx?.body || {};
const response = ctx?.response || {};
const usage = response.usage || {};

try {
const trace = client.trace({
name: `omniroute:${body.model || "unknown"}`,
userId: ctx?.userId,
metadata: {
provider: ctx?.provider,
requestId: ctx?.requestId,
omnirouteVersion: ctx?.omnirouteVersion,
},
});
trace.generation({
name: "chat.completion",
model: body.model || "unknown",
modelParameters: {
temperature: body.temperature,
max_tokens: body.max_tokens,
top_p: body.top_p,
},
input: truncateBody(body.messages, config.redactBody),
output: truncateBody(response.choices?.[0]?.message, config.redactBody),
usage: {
promptTokens: usage.prompt_tokens,
completionTokens: usage.completion_tokens,
totalTokens: usage.total_tokens,
},
startTime: new Date(start),
endTime: new Date(end),
});
} catch (err) {
console.error("[langfuse] trace emit failed:", err.message);
}
}

/**
* onError — emit a failed generation with error details.
*/
export async function onError(ctx) {
const config = ctx?.config || {};
if (config.enabled === false) return;
if (!ctx?.metadata?.__langfuseSampled) return;

const client = await getClient(config);
if (!client) return;

const start = ctx.metadata.__langfuseStart || Date.now();
const body = ctx?.body || {};
const error = ctx?.error || {};

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Defaulting error to {} means that if ctx?.error is undefined, error becomes an empty object. Later, String(error) will evaluate to "[object Object]". It is better to keep error as nullable and handle the fallback gracefully.

Suggested change
const error = ctx?.error || {};
const error = ctx?.error;


try {
const trace = client.trace({
name: `omniroute:${body.model || "unknown"}`,
userId: ctx?.userId,
metadata: {
provider: ctx?.provider,
requestId: ctx?.requestId,
},
});
trace.generation({
name: "chat.completion",
model: body.model || "unknown",
input: truncateBody(body.messages, config.redactBody),
level: "ERROR",
statusMessage: error.message || String(error),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Improve the error message fallback to avoid logging "[object Object]" when error is undefined or not an Error instance.

Suggested change
statusMessage: error.message || String(error),
statusMessage: error?.message || (error ? String(error) : "Unknown error"),

startTime: new Date(start),
endTime: new Date(),
});
} catch (err) {
console.error("[langfuse] error trace emit failed:", err.message);
}
}

/**
* onShutdown — flush pending events before OmniRoute exits.
*/
export async function onShutdown() {
if (!langfuseClient) return;
try {
await langfuseClient.shutdownAsync();
} catch (err) {
console.error("[langfuse] shutdown failed:", err.message);
}
}
Comment on lines +148 to +158

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

OmniRoute does not support an onShutdown hook. Instead, use the onDeactivate lifecycle hook to flush pending events and clean up the client. Additionally, resetting langfuseClient to null is crucial to ensure that if the plugin is deactivated and reactivated with a new configuration (e.g., updated API keys or host), a new client is instantiated with the updated settings.

Suggested change
/**
* onShutdown — flush pending events before OmniRoute exits.
*/
export async function onShutdown() {
if (!langfuseClient) return;
try {
await langfuseClient.shutdownAsync();
} catch (err) {
console.error("[langfuse] shutdown failed:", err.message);
}
}
/**
* onDeactivate — flush pending events and clean up the client when the plugin is deactivated.
*/
export async function onDeactivate() {
if (!langfuseClient) return;
try {
await langfuseClient.shutdownAsync();
} catch (err) {
console.error("[langfuse] shutdown failed:", err.message);
} finally {
langfuseClient = null;
}
}

67 changes: 67 additions & 0 deletions examples/plugins/langfuse/plugin.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
{
"name": "langfuse",
"version": "1.0.0",
"description": "Emits Langfuse generation traces for every LLM completion. Records prompt, response, model, usage, latency, and error details to Langfuse cloud or self-hosted.",
"author": "chirag127",
"license": "MIT",
"main": "index.mjs",
"source": "local",
"tags": ["observability", "tracing", "langfuse", "monitoring"],
"hooks": {
"onRequest": true,
"onResponse": true,
"onError": true
},
Comment on lines +10 to +14

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The onShutdown hook used in index.mjs is not a recognized lifecycle event in OmniRoute. To properly flush pending events and clean up resources when the plugin is deactivated or updated, you should register and implement the onDeactivate hook instead.

  "hooks": {
    "onRequest": true,
    "onResponse": true,
    "onError": true,
    "onDeactivate": true
  },

"requires": {
"permissions": []
},
"enabledByDefault": false,
"configSchema": {
"publicKey": {
"type": "string",
"default": "",
"description": "Langfuse public key (pk-lf-...)"
},
"secretKey": {
"type": "string",
"default": "",
"description": "Langfuse secret key (sk-lf-...)"
},
"host": {
"type": "string",
"default": "https://cloud.langfuse.com",
"description": "Langfuse host (cloud.langfuse.com, us.cloud.langfuse.com, or self-hosted URL)"
},
"enabled": {
"type": "boolean",
"default": true,
"description": "Emit traces when true; act as a no-op when false"
},
"sampleRate": {
"type": "number",
"default": 1.0,
"min": 0.0,
"max": 1.0,
"description": "Fraction of requests to trace (1.0 = all, 0.1 = 10%)"
},
"flushAt": {
"type": "number",
"default": 15,
"min": 1,
"max": 100,
"description": "Number of events to buffer before flushing to Langfuse"
},
"flushInterval": {
"type": "number",
"default": 10000,
"min": 1000,
"max": 60000,
"description": "Max ms between flushes (in-flight events flushed on timer)"
},
"redactBody": {
"type": "boolean",
"default": false,
"description": "Redact prompt + completion bodies from traces (metadata still recorded)"
}
}
}
Loading