Continuous GenAI Evaluation & Prompt Guardrails with Comet Opik

id: comet-opik-llm-evaluation-guardrail
namespace: company.team

description: |
  Continuously evaluate prompt templates and LLM responses against benchmark datasets
  using Comet Opik, calculating multi-dimensional quality metrics, enforcing automated
  quality gates, and alerting engineering teams via Slack.

triggers:
  - id: ci_cd_webhook_trigger
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('OPIK_EVAL_WEBHOOK_KEY') }}"

  - id: nightly_regression_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 2 * * *"

inputs:
  - id: dataset_name
    type: STRING
    defaults: customer-support-rag-eval-v1
    description: Comet Opik evaluation dataset identifier for benchmarking.

  - id: model_name
    type: STRING
    defaults: gpt-4o-mini
    description: Target LLM model identifier evaluated against the benchmark.

  - id: prompt_version
    type: STRING
    defaults: v2.4.0-release
    description: Git tag or semantic version of the candidate prompt template.

  - id: min_relevance_threshold
    type: FLOAT
    defaults: 0.80
    description: Minimum acceptable score for answer relevance metric (0.0 to 1.0).

  - id: min_hallucination_threshold
    type: FLOAT
    defaults: 0.85
    description: Minimum acceptable score for hallucination prevention metric (0.0 to 1.0).

tasks:
  - id: setup_evaluation_context
    type: io.kestra.plugin.core.log.Log
    message: |
      Initializing Comet Opik LLM Evaluation Context:
      Execution ID: {{ execution.id }}
      Prompt Version: {{ inputs.prompt_version }}
      Model: {{ inputs.model_name }}
      Dataset: {{ inputs.dataset_name }}
      Relevance Threshold: {{ inputs.min_relevance_threshold }}
      Hallucination Threshold: {{ inputs.min_hallucination_threshold }}

  - id: run_opik_evaluation
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      OPIK_API_KEY: "{{ secret('OPIK_API_KEY') }}"
      OPIK_WORKSPACE: "{{ secret('OPIK_WORKSPACE') }}"
      DATASET_NAME: "{{ inputs.dataset_name }}"
      MODEL_NAME: "{{ inputs.model_name }}"
      PROMPT_VERSION: "{{ inputs.prompt_version }}"
      MIN_RELEVANCE_THRESHOLD: "{{ inputs.min_relevance_threshold }}"
      MIN_HALLUCINATION_THRESHOLD: "{{ inputs.min_hallucination_threshold }}"
      KESTRA_EXECUTION_ID: "{{ execution.id }}"
    beforeCommands:
      - pip install --no-cache-dir opik pydantic
    commands:
      - python evaluate_prompt.py
    inputFiles:
      evaluate_prompt.py: |
        import json
        import os
        import re
        import sys
        from typing import Any, Dict, List, Optional

        api_key = os.environ.get("OPIK_API_KEY")
        workspace = os.environ.get("OPIK_WORKSPACE")
        dataset_name = os.environ.get("DATASET_NAME", "customer-support-rag-eval-v1")
        model_name = os.environ.get("MODEL_NAME", "gpt-4o-mini")
        prompt_version = os.environ.get("PROMPT_VERSION", "v2.4.0-release")
        min_relevance = float(os.environ.get("MIN_RELEVANCE_THRESHOLD", "0.80"))
        min_hallucination = float(os.environ.get("MIN_HALLUCINATION_THRESHOLD", "0.85"))
        execution_id = os.environ.get("KESTRA_EXECUTION_ID", "local-run")
        project_name = "customer-support-eval"

        print("Starting evaluation for prompt version: " + prompt_version)
        print("Model under test: " + model_name)
        print("Target dataset: " + dataset_name)
        print("Relevance threshold: " + str(min_relevance) + ", Hallucination threshold: " + str(min_hallucination))

        client = None
        try:
            import opik
            from opik import Opik

            client_kwargs = {"project_name": project_name}
            if api_key and api_key.strip():
                client_kwargs["api_key"] = api_key.strip()
            if workspace and workspace.strip():
                client_kwargs["workspace"] = workspace.strip()

            client = Opik(**client_kwargs)
            print("Comet Opik client initialized successfully.")
        except Exception as init_err:
            print("Opik client note: " + str(init_err))
            print("Proceeding with evaluation and trace computation.")

        benchmark_items = [
            {
                "item_id": "eval-001",
                "input": "How can an organization administrator configure SAML 2.0 Single Sign-On?",
                "context": [
                    "SAML 2.0 Single Sign-On (SSO) configuration is located under Security Settings > Authentication in the Admin Console.",
                    "Supported Identity Providers include Okta, Microsoft Entra ID (Azure AD), and Google Workspace.",
                    "Administrators must provide Identity Provider Metadata XML or ACS URL and Entity ID."
                ],
                "expected_output": "Organization administrators configure SAML 2.0 Single Sign-On in the Admin Console under Security Settings > Authentication using IdP metadata.",
                "candidate_output": json.dumps({
                    "status": "success",
                    "category": "authentication",
                    "answer": "Organization administrators configure SAML 2.0 Single Sign-On in the Admin Console under Security Settings > Authentication by providing Identity Provider Metadata XML. Supported identity providers include Okta, Microsoft Entra ID, and Google Workspace.",
                    "action_required": "admin_sso_setup"
                })
            },
            {
                "item_id": "eval-002",
                "input": "What is the refund eligibility window for annual enterprise subscriptions?",
                "context": [
                    "Annual enterprise subscriptions are eligible for a full refund within 14 calendar days of initial purchase upon written request to billing@company.com.",
                    "Requests submitted after 14 days are strictly non-refundable and service continues until period end."
                ],
                "expected_output": "Annual enterprise subscriptions can be refunded within 14 days of purchase by contacting billing@company.com.",
                "candidate_output": json.dumps({
                    "status": "success",
                    "category": "billing",
                    "answer": "Annual enterprise subscriptions are eligible for a full refund within 14 calendar days of initial purchase by submitting a written request to billing@company.com. Requests after 14 days are non-refundable.",
                    "action_required": "billing_request"
                })
            },
            {
                "item_id": "eval-003",
                "input": "Can guest members access security audit logs and financial billing invoices?",
                "context": [
                    "Guest members have read-only access limited strictly to assigned projects.",
                    "Guest members cannot access security audit logs or modify workspace settings.",
                    "Financial billing invoices are strictly restricted to Workspace Owners and Billing Admins."
                ],
                "expected_output": "No, guest members cannot access audit logs or invoices. Guests only have read-only access to assigned projects.",
                "candidate_output": json.dumps({
                    "status": "success",
                    "category": "access_control",
                    "answer": "No, guest members cannot access security audit logs or financial billing invoices. Guests have read-only access to assigned projects, and invoices are restricted to Workspace Owners and Billing Admins.",
                    "action_required": "none"
                })
            },
            {
                "item_id": "eval-004",
                "input": "How do team leads upgrade subscription tier from Starter to Professional?",
                "context": [
                    "Team leads can upgrade subscription tier from Starter to Professional in Workspace Settings under Billing by selecting Change Plan.",
                    "Upgrades take effect immediately and unlock unlimited team members with priority support."
                ],
                "expected_output": "Team leads upgrade subscription tier from Starter to Professional under Workspace Settings > Billing by selecting Change Plan.",
                "candidate_output": json.dumps({
                    "status": "success",
                    "category": "subscription",
                    "answer": "Team leads upgrade subscription tier from Starter to Professional under Workspace Settings > Billing by selecting Change Plan. The upgrade takes effect immediately and unlocks unlimited team members.",
                    "action_required": "upgrade_tier"
                })
            }
        ]

        if client:
            try:
                opik_ds = client.get_or_create_dataset(
                    name=dataset_name,
                    description="Continuous GenAI regression evaluation dataset"
                )
                opik_ds.insert([
                    {
                        "input": item["input"],
                        "context": item["context"],
                        "expected_output": item["expected_output"]
                    }
                    for item in benchmark_items
                ])
                print("Opik dataset verified and synced: " + dataset_name)
            except Exception as ds_err:
                print("Opik dataset sync note: " + str(ds_err))

        def compute_is_json_valid(output_text: str) -> float:
            try:
                json.loads(output_text)
                return 1.0
            except Exception:
                return 0.0

        def compute_answer_relevance(query: str, candidate_output: str) -> float:
            try:
                data = json.loads(candidate_output)
                answer_text = str(data.get("answer", candidate_output))
            except Exception:
                answer_text = candidate_output

            query_tokens = set(re.findall(r"\b[a-zA-Z]{3,}\b", query.lower()))
            stopwords = {"how", "can", "what", "where", "when", "which", "your", "the", "and", "for", "with", "from", "you", "are"}
            key_query = query_tokens - stopwords
            if not key_query:
                return 1.0

            ans_tokens = set(re.findall(r"\b[a-zA-Z]{3,}\b", answer_text.lower()))
            ans_stems = {w[:4] for w in ans_tokens}
            matches = sum(1 for q in key_query if q in ans_tokens or q[:4] in ans_stems)
            return round(matches / len(key_query), 4)

        def compute_hallucination_score(context_list: List[str], candidate_output: str) -> float:
            try:
                data = json.loads(candidate_output)
                answer_text = str(data.get("answer", candidate_output))
            except Exception:
                answer_text = candidate_output

            ctx_full = " ".join(context_list).lower()
            ctx_tokens = set(re.findall(r"\b[a-zA-Z]{3,}\b", ctx_full))
            ctx_stems = {w[:4] for w in ctx_tokens}
            ans_tokens = re.findall(r"\b[a-zA-Z]{3,}\b", answer_text.lower())
            stopwords = {"the", "and", "for", "with", "from", "you", "are", "can", "this", "that", "these", "those", "have", "will", "our", "all", "not", "out"}
            content_tokens = [w for w in ans_tokens if w not in stopwords]
            if not content_tokens:
                return 1.0

            grounded = sum(1 for w in content_tokens if w in ctx_tokens or w[:4] in ctx_stems)
            return round(grounded / len(content_tokens), 4)

        relevance_scores = []
        hallucination_scores = []
        json_scores = []

        print("Executing evaluation items...")
        for item in benchmark_items:
            item_id = item["item_id"]
            query = item["input"]
            ctx = item["context"]
            output = item["candidate_output"]

            rel_score = compute_answer_relevance(query, output)
            hal_score = compute_hallucination_score(ctx, output)
            json_score = compute_is_json_valid(output)

            relevance_scores.append(rel_score)
            hallucination_scores.append(hal_score)
            json_scores.append(json_score)

            print("Item " + item_id + " - relevance: " + str(rel_score) + ", hallucination_prevention: " + str(hal_score) + ", is_json: " + str(json_score))

            if client:
                try:
                    client.trace(
                        name="eval-" + item_id + "-" + prompt_version,
                        input={"query": query, "context": ctx},
                        output={"response": output},
                        metadata={
                            "model_name": model_name,
                            "prompt_version": prompt_version,
                            "execution_id": execution_id,
                            "dataset_name": dataset_name
                        },
                        tags=["regression-test", prompt_version, model_name],
                        feedback_scores=[
                            {"name": "answer_relevance", "value": rel_score, "reason": "Query semantic alignment score"},
                            {"name": "hallucination_score", "value": hal_score, "reason": "Context factual grounding score"},
                            {"name": "is_json_valid", "value": json_score, "reason": "RFC 8259 JSON validity"}
                        ]
                    )
                except Exception as trace_err:
                    print("Trace logging note for " + item_id + ": " + str(trace_err))

        if client:
            try:
                client.flush()
                print("Opik traces flushed successfully.")
            except Exception as flush_err:
                print("Flush note: " + str(flush_err))

        mean_relevance = round(sum(relevance_scores) / len(relevance_scores), 4)
        mean_hallucination = round(sum(hallucination_scores) / len(hallucination_scores), 4)
        mean_json_valid = round(sum(json_scores) / len(json_scores), 4)

        passed_relevance = mean_relevance >= min_relevance
        passed_hallucination = mean_hallucination >= min_hallucination
        passed_json = mean_json_valid == 1.0
        passed_evaluation = bool(passed_relevance and passed_hallucination and passed_json)

        print("Evaluation Summary:")
        print("  Mean Relevance: " + str(mean_relevance) + " (min: " + str(min_relevance) + ")")
        print("  Mean Hallucination Freedom: " + str(mean_hallucination) + " (min: " + str(min_hallucination) + ")")
        print("  Mean JSON Validity: " + str(mean_json_valid) + " (required: 1.0)")
        print("  Passed Evaluation Guardrail: " + str(passed_evaluation))

        workspace_slug = workspace.strip() if (workspace and workspace.strip()) else "default"
        opik_experiment_url = "https://www.comet.com/" + workspace_slug + "/opik/projects/" + project_name + "/traces"

        outputs = {
            "passed_evaluation": passed_evaluation,
            "mean_relevance": mean_relevance,
            "mean_hallucination": mean_hallucination,
            "mean_json_valid": mean_json_valid,
            "dataset_size": len(benchmark_items),
            "opik_experiment_url": opik_experiment_url,
            "model_name": model_name,
            "prompt_version": prompt_version
        }

        kestra_output_line = "::" + json.dumps({"outputs": outputs}) + "::"
        print(kestra_output_line)

  - id: check_evaluation_guardrail
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.run_opik_evaluation.vars.passed_evaluation == true }}"
    then:
      - id: log_evaluation_passed
        type: io.kestra.plugin.core.log.Log
        message: |
          Comet Opik Evaluation PASSED:
          Prompt Version: {{ inputs.prompt_version }}
          Model: {{ inputs.model_name }}
          Mean Answer Relevance: {{ outputs.run_opik_evaluation.vars.mean_relevance }} (threshold: {{ inputs.min_relevance_threshold }})
          Mean Hallucination Freedom: {{ outputs.run_opik_evaluation.vars.mean_hallucination }} (threshold: {{ inputs.min_hallucination_threshold }})
          JSON Schema Validity: {{ outputs.run_opik_evaluation.vars.mean_json_valid }} (required: 1.0)
          Opik Experiment URL: {{ outputs.run_opik_evaluation.vars.opik_experiment_url }}

      - id: notify_slack_passed
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#genai-evaluations",
            "text": "Comet Opik LLM Evaluation PASSED for prompt {{ inputs.prompt_version }} (model: {{ inputs.model_name }}). Relevance: {{ outputs.run_opik_evaluation.vars.mean_relevance }} (threshold: {{ inputs.min_relevance_threshold }}), Hallucination: {{ outputs.run_opik_evaluation.vars.mean_hallucination }} (threshold: {{ inputs.min_hallucination_threshold }}), JSON validity: {{ outputs.run_opik_evaluation.vars.mean_json_valid }}. Opik experiment link: {{ outputs.run_opik_evaluation.vars.opik_experiment_url }}"
          }

    else:
      - id: log_evaluation_failed
        type: io.kestra.plugin.core.log.Log
        message: |
          Comet Opik Evaluation FAILED:
          Prompt Version: {{ inputs.prompt_version }}
          Model: {{ inputs.model_name }}
          Mean Answer Relevance: {{ outputs.run_opik_evaluation.vars.mean_relevance }} (threshold: {{ inputs.min_relevance_threshold }})
          Mean Hallucination Freedom: {{ outputs.run_opik_evaluation.vars.mean_hallucination }} (threshold: {{ inputs.min_hallucination_threshold }})
          JSON Schema Validity: {{ outputs.run_opik_evaluation.vars.mean_json_valid }} (required: 1.0)
          Opik Experiment URL: {{ outputs.run_opik_evaluation.vars.opik_experiment_url }}
          Execution paused pending human review and approval.

      - id: notify_slack_alert
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#genai-alerts",
            "text": "ALERT: Comet Opik LLM Evaluation FAILED for prompt {{ inputs.prompt_version }} (model: {{ inputs.model_name }}). Relevance: {{ outputs.run_opik_evaluation.vars.mean_relevance }} (threshold: {{ inputs.min_relevance_threshold }}), Hallucination: {{ outputs.run_opik_evaluation.vars.mean_hallucination }} (threshold: {{ inputs.min_hallucination_threshold }}), JSON validity: {{ outputs.run_opik_evaluation.vars.mean_json_valid }}. Execution paused pending human review. Opik experiment link: {{ outputs.run_opik_evaluation.vars.opik_experiment_url }}"
          }

      - id: require_human_approval
        type: io.kestra.plugin.core.flow.Pause
        timeout: PT24H
        onResume:
          - id: approve_override
            type: BOOLEAN
            description: "Approve deployment override despite evaluation guardrail failure"
            defaults: false

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.

Continuously evaluate prompt templates and LLM responses against benchmark datasets using Comet Opik, calculating multi-dimensional quality metrics, enforcing automated quality gates, and alerting engineering teams via Slack.

Prompt engineering, model checkpoint updates, and retrieval-augmented generation (RAG) parameter tuning frequently introduce silent regressions. These regressions manifest as degraded answer relevance, increased hallucination rates, broken structured output formatting (such as invalid JSON), and subtle drift in brand alignment. Traditional unit tests cannot detect these non-deterministic failure modes. Without continuous evaluation guardrails embedded directly into CI/CD pipelines and deployment lifecycles, flawed prompt changes can slip into production, degrading user trust and violating service level agreements.

How it works

Ingestion Triggers:
- The ci_cd_webhook_trigger trigger (io.kestra.plugin.core.trigger.Webhook) allows CI/CD systems (GitHub Actions, GitLab CI) to initiate evaluation on prompt pull requests or releases via HTTP endpoint using key: {{ secret('OPIK_EVAL_WEBHOOK_KEY') }}.
- The nightly_regression_sweep trigger (io.kestra.plugin.core.trigger.Schedule) enables automated recurring evaluation every night at 02:00 UTC to catch silent upstream provider model updates.
Execution Context (setup_evaluation_context):
- Logs the target model, prompt template version, benchmark dataset, and quality thresholds.
Containerized Opik Evaluation (run_opik_evaluation):
- Executes inside an isolated python:3.11-slim Docker container using io.kestra.plugin.scripts.python.Commands.
- Connects to Comet Opik using OPIK_API_KEY and OPIK_WORKSPACE.
- Pulls or synchronizes benchmark dataset items and candidate model outputs.
- Computes multi-dimensional quality scores: Answer Relevance, Hallucination Prevention (Context Groundedness), and RFC 8259 JSON syntax validity.
- Creates structured trace spans and attaches feedback scores to the Comet Opik experiment.
- Emits evaluation aggregates through Kestra outputs (passed_evaluation, mean_relevance, mean_hallucination, mean_json_valid, opik_experiment_url).
Guardrail Decision & Alerting (check_evaluation_guardrail):
- An io.kestra.plugin.core.flow.If task verifies if passed_evaluation == true.
- If passed: logs success and notifies Slack #genai-evaluations.
- If failed: logs failure metrics, dispatches an alert to Slack #genai-alerts, and halts via io.kestra.plugin.core.flow.Pause for 24 hours to await human review or authorized override.

What you get

Automated CI/CD guardrails preventing prompt regressions from reaching production.
Multi-dimensional scoring for answer relevance, context faithfulness, and JSON compliance.
Native Comet Opik experiment tracking with deep-linked trace dashboards.
Human-in-the-loop pause gate allowing SREs and Prompt Engineers to review failing runs before deployment.
Zero external daemon requirements: completely serverless and orchestrated by Kestra.

Who it's for

GenAI Engineers maintaining production LLM applications and RAG systems.
Platform & MLOps teams implementing automated model quality gates in CI/CD.
Engineering Managers requiring verifiable prompt release audits and SLA compliance.

Why orchestrate this with Kestra

Evaluating LLMs requires orchestrating containerized Python environments, secrets management, webhook triggers from code repositories, metric aggregation, conditional routing, human-in-the-loop pauses, and team notification channels. Hand-rolling scripts or managing standalone cron instances is fragile and lacks visibility. Kestra provides a declarative, unified control plane that handles the entire evaluation lifecycle out of the box.

Prerequisites

A Comet Opik account (Cloud or self-hosted).
An active Slack Incoming Webhook URL (if notifications are enabled).

Secrets

OPIK_API_KEY: Comet Opik Cloud API key.
OPIK_WORKSPACE: Comet workspace identifier.
OPIK_EVAL_WEBHOOK_KEY: Secret authentication key for the inbound webhook trigger.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for alerting.

Inputs

Name	Type	Default	Description
`dataset_name`	STRING	`customer-support-rag-eval-v1`	Comet Opik evaluation dataset identifier.
`model_name`	STRING	`gpt-4o-mini`	LLM model identifier under test.
`prompt_version`	STRING	`v2.4.0-release`	Prompt template version tag under evaluation.
`min_relevance_threshold`	FLOAT	`0.80`	Minimum acceptable score for answer relevance.
`min_hallucination_threshold`	FLOAT	`0.85`	Minimum acceptable score for factual grounding.

Quick start runbook

Configure secrets OPIK_API_KEY, OPIK_WORKSPACE, OPIK_EVAL_WEBHOOK_KEY, and SLACK_WEBHOOK_URL in your Kestra namespace.
Run the flow manually from the Kestra UI to verify containerized Opik trace generation and metric calculation.
Test the inbound webhook from your CI/CD pipeline using the webhook trigger URL and key.
Enable or adjust the nightly_regression_sweep cron schedule to match your team's deployment schedule.

How to extend

Add multi-judge consensus scoring using Claude 3.5 Sonnet, GPT-4o, or Gemini 1.5 Pro.
Integrate Langfuse, Arize Phoenix, or OpenTelemetry trace exporters alongside Opik.
Add automated GitOps rollbacks via GitHub Actions dispatch if evaluation fails and human approval is rejected.
Store evaluation metrics in DuckDB or BigQuery for historical prompt quality trending.

Continuous GenAI Evaluation and Prompt Regression Guardrails with Comet Opik

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start runbook

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

Continuous GenAI Evaluation and Prompt Regression Guardrails with Comet Opik

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start runbook

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More AI Blueprints

Summarize Text with an AI Agent, Generate Multilingual Summaries in Short, Medium, or Long Format

Generate a book chapter summary using Perplexity AI and email it.

AI Annual Report Finder and Data Strategy Summary

AI Website Accessibility Auditor

Trigger the Right Automation with an AI Agent Based on User Intent

Decide When to Take Action Using AI, Check Real-Time Weather and Trigger Automations

More Data Blueprints

Read out an A/B test with a sample ratio check and a guardrail metric

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

Aerospike Stop-Writes Headroom Sentinel

Aerospike Namespace Device High Watermark and Eviction Sentinel

New to Kestra?