From 5e5fb61fccecb19aed772965be7d7da28ad26d28 Mon Sep 17 00:00:00 2001 From: Chirag Singhal <76880977+chirag127@users.noreply.github.com> Date: Wed, 8 Jul 2026 04:02:28 +0530 Subject: [PATCH] feat(plugins): add langfuse plugin (#6577) feat(plugins): add langfuse example plugin Integrated into release/v3.8.47. (thanks @chirag127) --- examples/plugins/langfuse/README.md | 63 ++++++++++ examples/plugins/langfuse/index.mjs | 158 ++++++++++++++++++++++++++ examples/plugins/langfuse/plugin.json | 67 +++++++++++ 3 files changed, 288 insertions(+) create mode 100644 examples/plugins/langfuse/README.md create mode 100644 examples/plugins/langfuse/index.mjs create mode 100644 examples/plugins/langfuse/plugin.json diff --git a/examples/plugins/langfuse/README.md b/examples/plugins/langfuse/README.md new file mode 100644 index 000000000..f30f4af40 --- /dev/null +++ b/examples/plugins/langfuse/README.md @@ -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:` 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` diff --git a/examples/plugins/langfuse/index.mjs b/examples/plugins/langfuse/index.mjs new file mode 100644 index 000000000..e647b5b0c --- /dev/null +++ b/examples/plugins/langfuse/index.mjs @@ -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 || {}; + + 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), + 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); + } +} diff --git a/examples/plugins/langfuse/plugin.json b/examples/plugins/langfuse/plugin.json new file mode 100644 index 000000000..52c1fd8c4 --- /dev/null +++ b/examples/plugins/langfuse/plugin.json @@ -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 + }, + "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)" + } + } +}