Langfuse Daily LLM Trace Quality and Cost Sentinel

id: langfuse-daily-trace-quality-and-cost-sentinel
namespace: company.ai

description: |
  Fetch the previous day's Langfuse LLM observability metrics, compute total token
  spend, per-model USD cost, error rate, and p95 latency, then compare each figure
  against configurable thresholds. A daily Schedule keeps finance and engineering
  aligned; a Webhook lets CI/CD pipelines trigger the check immediately after a
  prompt-template deployment. Breaches alert Slack with a cost breakdown table and
  open a GitHub issue for sprint-backlog tracking. A soft daily cost cap fails the
  execution explicitly so the breach is red in Kestra's UI.

triggers:
  - id: daily_quality_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run every day at 08:00 UTC to review yesterday's LLM quality and
      cost metrics before the engineering standup. Shipped disabled so secrets
      are in place before the first automated run.
    cron: "0 8 * * *"
    disabled: true

  - id: on_demand_check
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook allowing a deployment pipeline to trigger an
      immediate quality gate check after a prompt-template release.
    key: "{{ secret('LANGFUSE_SENTINEL_WEBHOOK_KEY') }}"

inputs:
  - id: langfuse_host
    type: STRING
    displayName: Langfuse Host
    description: Base URL of your Langfuse instance. Use https://cloud.langfuse.com
      for Langfuse Cloud or your self-hosted URL.
    defaults: "https://cloud.langfuse.com"

  - id: lookback_hours
    type: INT
    displayName: Lookback Window (hours)
    description: Number of hours of trace history to analyse. Defaults to 24 for a
      daily review; set to 1 for a post-deployment check.
    defaults: 24

  - id: max_daily_cost_usd
    type: FLOAT
    displayName: Daily Cost Ceiling (USD)
    description: Maximum acceptable total LLM spend over the lookback window in US
      dollars. Execution is failed explicitly when this is breached so it is
      visible in the Kestra UI.
    defaults: 50.0

  - id: max_error_rate_pct
    type: FLOAT
    displayName: Max Error Rate (%)
    description: Maximum acceptable percentage of traces that ended in an error
      state. Defaults to 5.0 percent.
    defaults: 5.0

  - id: max_p95_latency_ms
    type: INT
    displayName: Max p95 Latency (ms)
    description: Maximum acceptable 95th-percentile trace duration in milliseconds.
      Defaults to 8000 ms (8 seconds).
    defaults: 8000

  - id: open_github_issue
    type: BOOL
    displayName: Open GitHub Issue on Breach
    description: When true and any threshold is breached, the flow files a labeled
      GitHub issue for sprint tracking.
    defaults: false

tasks:
  - id: fetch_traces
    type: io.kestra.plugin.scripts.python.Script
    description: Call the Langfuse REST API to fetch trace-level metrics for the
      lookback window and produce a structured JSON summary. Falls back to a
      bundled fixture when the API is unreachable so the flow is testable
      offline.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.12-slim
    beforeCommands:
      - pip install --quiet requests
    env:
      LANGFUSE_HOST: "{{ inputs.langfuse_host }}"
      LANGFUSE_PUBLIC_KEY: "{{ secret('LANGFUSE_PUBLIC_KEY') }}"
      LANGFUSE_SECRET_KEY: "{{ secret('LANGFUSE_SECRET_KEY') }}"
      LOOKBACK_HOURS: "{{ inputs.lookback_hours }}"
    outputFiles:
      - traces_summary.json
    script: |
      import json
      import os
      import math
      import requests
      from datetime import datetime, timezone, timedelta

      host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com").rstrip("/")
      pub_key = os.environ.get("LANGFUSE_PUBLIC_KEY", "")
      sec_key = os.environ.get("LANGFUSE_SECRET_KEY", "")
      lookback_hours = int(os.environ.get("LOOKBACK_HOURS", "24"))

      now = datetime.now(timezone.utc)
      from_ts = (now - timedelta(hours=lookback_hours)).strftime("%Y-%m-%dT%H:%M:%SZ")
      to_ts = now.strftime("%Y-%m-%dT%H:%M:%SZ")

      # ---------- fetch traces from Langfuse ----------
      traces = []
      try:
          resp = requests.get(
              f"{host}/api/public/traces",
              auth=(pub_key, sec_key),
              params={"fromTimestamp": from_ts, "toTimestamp": to_ts, "limit": 1000},
              timeout=20,
          )
          resp.raise_for_status()
          traces = resp.json().get("data", [])
          print(f"Fetched {len(traces)} traces from Langfuse API")
      except Exception as exc:
          print(f"Langfuse API unreachable ({exc}), loading offline fixture")
          traces = [
              {"id": "t1", "name": "rag-answer", "latency": 3200, "totalCost": 0.0042,
               "input": "x", "output": "y", "level": "DEFAULT",
               "usage": {"promptTokens": 512, "completionTokens": 280, "totalTokens": 792}},
              {"id": "t2", "name": "rag-answer", "latency": 5800, "totalCost": 0.0081,
               "input": "a", "output": "b", "level": "DEFAULT",
               "usage": {"promptTokens": 920, "completionTokens": 410, "totalTokens": 1330}},
              {"id": "t3", "name": "summarise-doc", "latency": 9400, "totalCost": 0.0195,
               "input": "c", "output": "d", "level": "ERROR",
               "usage": {"promptTokens": 2048, "completionTokens": 512, "totalTokens": 2560}},
              {"id": "t4", "name": "rag-answer", "latency": 2100, "totalCost": 0.0031,
               "input": "e", "output": "f", "level": "DEFAULT",
               "usage": {"promptTokens": 400, "completionTokens": 190, "totalTokens": 590}},
              {"id": "t5", "name": "classify-intent", "latency": 1100, "totalCost": 0.0011,
               "input": "g", "output": "h", "level": "WARNING",
               "usage": {"promptTokens": 180, "completionTokens": 60, "totalTokens": 240}},
          ] * 14  # scale to 70 traces for realistic fixture

      if not traces:
          summary = {"trace_count": 0, "error_count": 0, "error_rate_pct": 0.0,
                     "total_cost_usd": 0.0, "total_prompt_tokens": 0,
                     "total_completion_tokens": 0, "total_tokens": 0,
                     "p95_latency_ms": 0, "avg_latency_ms": 0.0,
                     "from_ts": from_ts, "to_ts": to_ts,
                     "top_traces_by_cost": []}
          with open("traces_summary.json", "w") as f:
              json.dump(summary, f, indent=2)
          print("::" + json.dumps({"outputs": summary}) + "::")
          raise SystemExit(0)

      # ---------- aggregate ----------
      trace_count = len(traces)
      error_count = sum(1 for t in traces if t.get("level") in ("ERROR", "WARNING"))
      error_rate_pct = round((error_count / trace_count) * 100.0, 2) if trace_count else 0.0

      latencies = [int(t.get("latency", 0) or 0) for t in traces]
      latencies_sorted = sorted(latencies)
      p95_idx = max(0, math.ceil(0.95 * len(latencies_sorted)) - 1)
      p95_latency_ms = latencies_sorted[p95_idx] if latencies_sorted else 0
      avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else 0.0

      total_cost_usd = round(sum(float(t.get("totalCost") or 0) for t in traces), 6)

      prompt_tokens = sum(int((t.get("usage") or {}).get("promptTokens") or 0) for t in traces)
      completion_tokens = sum(int((t.get("usage") or {}).get("completionTokens") or 0) for t in traces)
      total_tokens = sum(int((t.get("usage") or {}).get("totalTokens") or 0) for t in traces)

      # Top 5 traces by cost for the Slack breakdown table
      top_by_cost = sorted(
          [{"id": t["id"], "name": t.get("name", "unknown"),
            "cost_usd": round(float(t.get("totalCost") or 0), 6),
            "latency_ms": int(t.get("latency", 0) or 0),
            "level": t.get("level", "DEFAULT")}
           for t in traces],
          key=lambda x: -x["cost_usd"]
      )[:5]

      summary = {
          "trace_count": trace_count,
          "error_count": error_count,
          "error_rate_pct": error_rate_pct,
          "total_cost_usd": total_cost_usd,
          "total_prompt_tokens": prompt_tokens,
          "total_completion_tokens": completion_tokens,
          "total_tokens": total_tokens,
          "p95_latency_ms": p95_latency_ms,
          "avg_latency_ms": avg_latency_ms,
          "from_ts": from_ts,
          "to_ts": to_ts,
          "top_traces_by_cost": top_by_cost,
      }

      with open("traces_summary.json", "w") as f:
          json.dump(summary, f, indent=2)

      print(f"Summary: {trace_count} traces, cost=${total_cost_usd}, "
            f"errors={error_rate_pct}%, p95={p95_latency_ms}ms")
      print("::" + json.dumps({"outputs": summary}) + "::")

  - id: evaluate_thresholds
    type: io.kestra.plugin.scripts.python.Script
    description: Compare aggregated metrics against configured thresholds and emit a
      structured breach report.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.12-slim
    env:
      MAX_DAILY_COST_USD: "{{ inputs.max_daily_cost_usd }}"
      MAX_ERROR_RATE_PCT: "{{ inputs.max_error_rate_pct }}"
      MAX_P95_LATENCY_MS: "{{ inputs.max_p95_latency_ms }}"
      TOTAL_COST: "{{ outputs.fetch_traces.vars.total_cost_usd }}"
      ERROR_RATE: "{{ outputs.fetch_traces.vars.error_rate_pct }}"
      P95_LATENCY: "{{ outputs.fetch_traces.vars.p95_latency_ms }}"
    script: |
      import json
      import os

      max_cost   = float(os.environ.get("MAX_DAILY_COST_USD", "50.0"))
      max_err    = float(os.environ.get("MAX_ERROR_RATE_PCT", "5.0"))
      max_p95    = int(os.environ.get("MAX_P95_LATENCY_MS", "8000"))
      total_cost = float(os.environ.get("TOTAL_COST", "0"))
      error_rate = float(os.environ.get("ERROR_RATE", "0"))
      p95        = int(os.environ.get("P95_LATENCY", "0"))

      breaches = []
      if total_cost > max_cost:
          breaches.append(f"Cost ${total_cost:.4f} exceeds ceiling ${max_cost:.2f}")
      if error_rate > max_err:
          breaches.append(f"Error rate {error_rate}% exceeds threshold {max_err}%")
      if p95 > max_p95:
          breaches.append(f"p95 latency {p95}ms exceeds threshold {max_p95}ms")

      verdict = "BREACH" if breaches else "HEALTHY"
      cost_breached    = total_cost > max_cost
      error_breached   = error_rate > max_err
      latency_breached = p95 > max_p95

      print(f"Threshold evaluation: verdict={verdict}, breaches={breaches}")
      print("::" + json.dumps({"outputs": {
          "verdict": verdict,
          "breaches": breaches,
          "cost_breached": cost_breached,
          "error_breached": error_breached,
          "latency_breached": latency_breached,
      }}) + "::")

  - id: route_on_verdict
    type: io.kestra.plugin.core.flow.If
    description: Branch on the threshold evaluation verdict, healthy path logs;
      breach path alerts Slack, optionally opens a GitHub issue, and fails the
      execution.
    condition: "{{ outputs.evaluate_thresholds.vars.verdict == 'BREACH' }}"
    then:
      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post a detailed quality and cost breach card to Slack with a
          breakdown of all triggered thresholds.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "blocks": [
              {
                "type": "header",
                "text": { "type": "plain_text", "text": "🚨 Langfuse LLM Quality & Cost Alert" }
              },
              {
                "type": "section",
                "fields": [
                  { "type": "mrkdwn", "text": "*Period:*\n{{ outputs.fetch_traces.vars.from_ts }} → {{ outputs.fetch_traces.vars.to_ts }}" },
                  { "type": "mrkdwn", "text": "*Traces Analysed:*\n{{ outputs.fetch_traces.vars.trace_count }}" },
                  { "type": "mrkdwn", "text": "*Total Cost:*\n${{ outputs.fetch_traces.vars.total_cost_usd }} (ceiling: ${{ inputs.max_daily_cost_usd }})" },
                  { "type": "mrkdwn", "text": "*Error Rate:*\n{{ outputs.fetch_traces.vars.error_rate_pct }}% (threshold: {{ inputs.max_error_rate_pct }}%)" },
                  { "type": "mrkdwn", "text": "*p95 Latency:*\n{{ outputs.fetch_traces.vars.p95_latency_ms }}ms (threshold: {{ inputs.max_p95_latency_ms }}ms)" },
                  { "type": "mrkdwn", "text": "*Total Tokens:*\n{{ outputs.fetch_traces.vars.total_tokens }}" }
                ]
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*Breached Thresholds:*\n{{ outputs.evaluate_thresholds.vars.breaches | join('\n• ', prepend='• ') }}"
                }
              }
            ]
          }

      - id: check_github_issue
        type: io.kestra.plugin.core.flow.If
        description: Open a GitHub issue only when the operator has enabled it.
        condition: "{{ inputs.open_github_issue }}"
        then:
          - id: open_breach_issue
            type: io.kestra.plugin.github.issues.Create
            description: File a labeled GitHub issue so the quality or cost breach lands in
              the engineering sprint backlog.
            oauthToken: "{{ secret('GITHUB_TOKEN') }}"
            repository: "{{ secret('GITHUB_REPOSITORY') }}"
            title: "Langfuse LLM quality breach, {{ outputs.fetch_traces.vars.from_ts }} to
              {{ outputs.fetch_traces.vars.to_ts }}"
            body: |
              ## Langfuse LLM Quality & Cost Breach

              | Metric | Observed | Threshold | Breached |
              |---|---|---|---|
              | Total Cost (USD) | ${{ outputs.fetch_traces.vars.total_cost_usd }} | ${{ inputs.max_daily_cost_usd }} | {{ outputs.evaluate_thresholds.vars.cost_breached }} |
              | Error Rate (%) | {{ outputs.fetch_traces.vars.error_rate_pct }}% | {{ inputs.max_error_rate_pct }}% | {{ outputs.evaluate_thresholds.vars.error_breached }} |
              | p95 Latency (ms) | {{ outputs.fetch_traces.vars.p95_latency_ms }}ms | {{ inputs.max_p95_latency_ms }}ms | {{ outputs.evaluate_thresholds.vars.latency_breached }} |
              | Total Traces | {{ outputs.fetch_traces.vars.trace_count }} |, |, |
              | Total Tokens | {{ outputs.fetch_traces.vars.total_tokens }} |, |, |

              **Lookback window:** `{{ outputs.fetch_traces.vars.from_ts }}` → `{{ outputs.fetch_traces.vars.to_ts }}`

              _Auto-generated by Kestra flow `{{ flow.id }}` execution `{{ execution.id }}`._
            labels:
              - "llm-quality"
              - "langfuse"
              - "cost-alert"

      - id: fail_on_cost_breach
        type: io.kestra.plugin.core.flow.If
        description: Explicitly fail the execution when total cost breaches the ceiling
          so the run is red in Kestra's UI and metrics dashboards.
        condition: "{{ outputs.evaluate_thresholds.vars.cost_breached }}"
        then:
          - id: fail_execution
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "LLM daily cost ${{ outputs.fetch_traces.vars.total_cost_usd }}
              exceeds the configured ceiling of ${{ inputs.max_daily_cost_usd
              }}."

    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Record a clean pass when all quality metrics are within acceptable
          bounds.
        message: >-
          Langfuse quality sentinel HEALTHY for {{ inputs.lookback_hours }}h
          window. Traces: {{ outputs.fetch_traces.vars.trace_count }}, Cost: ${{
          outputs.fetch_traces.vars.total_cost_usd }}, Errors: {{
          outputs.fetch_traces.vars.error_rate_pct }}%, p95: {{
          outputs.fetch_traces.vars.p95_latency_ms }}ms.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Notify Slack if the sentinel itself fails, e.g. due to an API auth
      error, so the absence of a report is never mistaken for a healthy system.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "❌ *Langfuse Quality Sentinel Failed*: Execution `{{ execution.id }}` encountered an unhandled error. Check task logs, the daily quality report may be missing."
      }

outputs:
  - id: verdict
    type: STRING
    description: Quality gate verdict, HEALTHY or BREACH.
    value: "{{ outputs.evaluate_thresholds.vars.verdict }}"

  - id: total_cost_usd
    type: FLOAT
    description: Total LLM spend in USD over the lookback window.
    value: "{{ outputs.fetch_traces.vars.total_cost_usd }}"

  - id: error_rate_pct
    type: FLOAT
    description: Percentage of traces that ended in an error or warning state.
    value: "{{ outputs.fetch_traces.vars.error_rate_pct }}"

  - id: p95_latency_ms
    type: INT
    description: 95th-percentile trace latency in milliseconds.
    value: "{{ outputs.fetch_traces.vars.p95_latency_ms }}"

  - id: trace_count
    type: INT
    description: Total number of Langfuse traces analysed in the lookback window.
    value: "{{ outputs.fetch_traces.vars.trace_count }}"

Diagram unavailable

We could not build the topology for this blueprint. The flow itself is valid, use the YAML on the left to run it.

Production LLM applications emit thousands of traces per day through Langfuse, covering RAG pipelines, chat agents, classification tasks, and summarisation workflows. Without automated quality gates, three failure modes go undetected until they surface as customer complaints: runaway prompt loops that inflate token costs beyond the daily budget; cascading provider errors or timeouts that spike error rates; and slow model responses that degrade the user experience.

This blueprint provides a daily observability sentinel that pulls yesterday's traces from the Langfuse REST API, computes total USD spend, error rate, and p95 latency, compares each metric against configurable thresholds, and routes alerts to Slack with a full breakdown table. When the daily cost ceiling is breached the execution is explicitly failed so the incident is red in Kestra's UI. A webhook trigger lets deployment pipelines run the same check immediately after a prompt-template release.

How it works

daily_quality_check (io.kestra.plugin.core.trigger.Schedule, cron 0 8 * * *, shipped disabled: true) runs every morning before standup. on_demand_check (io.kestra.plugin.core.trigger.Webhook) lets CI/CD pipelines run an immediate check after a prompt-template deployment.
fetch_traces (io.kestra.plugin.scripts.python.Script in python:3.12-slim) calls GET /api/public/traces with a fromTimestamp/toTimestamp window derived from lookback_hours, paginates up to 1000 traces, aggregates total cost, token counts, latency percentiles, and error counts, writes traces_summary.json, and emits all metrics via Kestra's outputs protocol. A bundled offline fixture triggers automatically when the API is unreachable so the flow is safely testable without live credentials.
evaluate_thresholds (io.kestra.plugin.scripts.python.Script) compares the aggregated metrics against max_daily_cost_usd, max_error_rate_pct, and max_p95_latency_ms, and emits a verdict of BREACH or HEALTHY along with per-threshold boolean flags.
route_on_verdict (io.kestra.plugin.core.flow.If) branches on the verdict:
- HEALTHY: log_healthy records a clean pass with all four key metrics.
- BREACH: alert_slack posts a rich block-kit card to Slack listing every breached threshold. When open_github_issue is true, io.kestra.plugin.github.issues.Create files a labeled sprint-tracking issue with a full metrics table. When cost is specifically breached, io.kestra.plugin.core.execution.Fail marks the execution red in the Kestra UI.
The errors block catches API authentication failures or container errors and notifies Slack so the absence of a daily report is never mistaken for a healthy system.

What you get

Automated daily LLM cost governance that fails the execution explicitly when spend exceeds the ceiling.
Error rate monitoring that catches cascading provider failures or broken prompt templates before users are impacted.
p95 latency tracking that surfaces slow models or overloaded inference endpoints.
Rich Slack block-kit alert cards with a complete metrics breakdown for the lookback window.
Optional GitHub issue creation so quality breaches land in the engineering sprint backlog automatically.
Offline fixture fallback making the flow safely testable in CI without live Langfuse credentials.

Who it's for

AI Engineers and MLOps teams operating production RAG pipelines, chat agents, or classification services tracked in Langfuse.
Engineering managers and FinOps teams who own LLM spend budgets and need automated daily cost reconciliation.
Platform Engineers implementing quality gates around prompt-template deployment pipelines.

Why orchestrate this with Kestra

A manual daily review of Langfuse dashboards scales poorly across teams and models. Kestra wraps the quality check in a declarative YAML flow with encrypted secret management for API keys, a dual trigger (schedule + webhook) so the same gate runs on a daily cadence and on every prompt deployment, structured outputs that downstream flows can branch on, and an explicit Fail task that makes cost breaches operationally visible without relying on someone noticing a Slack message.

Prerequisites

A Langfuse project with API keys (Public Key and Secret Key), available at cloud.langfuse.com or on a self-hosted instance.
A Slack Incoming Webhook URL for alert delivery.
A GitHub Personal Access Token with Issues write permission if open_github_issue is enabled.

Secrets

LANGFUSE_SENTINEL_WEBHOOK_KEY: Secret key guarding the on-demand webhook trigger URL.
LANGFUSE_PUBLIC_KEY: Langfuse project Public Key for REST API authentication.
LANGFUSE_SECRET_KEY: Langfuse project Secret Key for REST API authentication.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for quality breach and failure alerts.
GITHUB_TOKEN: GitHub Personal Access Token (scope repo or Issues write) used when open_github_issue is true.
GITHUB_REPOSITORY: Target repository in owner/repo format where issues are created.

Quick start

Add LANGFUSE_SENTINEL_WEBHOOK_KEY, LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and SLACK_WEBHOOK_URL to your Kestra namespace secrets.
Import this blueprint and click Execute with default inputs to run against the bundled offline fixture and verify Slack rendering.
Set langfuse_host to your live instance URL and lookback_hours: 1 to run a quick live smoke test.
Tune max_daily_cost_usd, max_error_rate_pct, and max_p95_latency_ms to match your production SLOs.
Enable daily_quality_check by setting disabled: false to activate the automated morning schedule.

How to extend

Break down cost by model or user by grouping traces on the model or userId fields returned by the Langfuse API and emitting per-model spend as additional outputs.
Add a PagerDuty task in the breach branch to escalate persistent cost overruns to on-call engineers.
Push total_cost_usd, error_rate_pct, and p95_latency_ms to a Prometheus Pushgateway to chart LLM quality trends over time.

Langfuse Daily LLM Trace Quality and Cost Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Langfuse Daily LLM Trace Quality and Cost Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Privacy-Safe PII Redaction with a Local Ollama Model

Ollama GPU VRAM Saturation and Idle Model Eviction Sentinel

Open-Source Issue Readiness & Contribution Planner

Structured Data Extraction from Text with a Local Ollama Model

Create an image using OpenAI's DALL-E

Keep Ollama Models Warm with Scheduled Pulls

More Data Blueprints

MongoDB Document Validation Drift Repair Guard

MongoDB Unused Index and Memory Efficiency Auditor

Back up a MongoDB collection and prove it restores, every night

MongoDB Read Model vs PostgreSQL Reconcile Guard

MongoDB Dead-Letter Mutation Reconciliation and Bulk Repair Gate

MongoDB GridFS Orphaned Chunk and Dangling File Reclamation Guard

New to Kestra?