id: deepeval-llm-continuous-evaluation-and-regression-gate
namespace: company.team
description: |
Automate continuous evaluation and regression testing for LLMs and RAG pipelines
using DeepEval. Evaluates faithfulness, answer relevancy, and hallucination metrics
against golden benchmark datasets, enforces automated CI/CD deployment gates,
and delivers structured scorecard notifications to Slack.
triggers:
- id: inbound_ci_eval_webhook
type: io.kestra.plugin.core.trigger.Webhook
key: "{{ secret('DEEPEVAL_WEBHOOK_KEY') }}"
- id: scheduled_nightly_regression_gate
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 2 * * *"
inputs:
- id: eval_run_name
type: STRING
defaults: "v2.4.0-prompt-release"
description: Unique release identifier or Git commit tag for the model evaluation run.
- id: model_name
type: STRING
defaults: "gemini-2.5-flash"
description: Foundational model or endpoint under evaluation.
- id: min_faithfulness_score
type: FLOAT
defaults: 0.85
description: Minimum required faithfulness score (0.0 to 1.0) to pass the quality gate.
- id: min_answer_relevancy_score
type: FLOAT
defaults: 0.80
description: Minimum required answer relevancy score (0.0 to 1.0) to pass the
quality gate.
- id: max_hallucination_rate
type: FLOAT
defaults: 0.05
description: Maximum allowable hallucination failure rate (0.0 to 1.0).
- id: fail_on_breach
type: BOOL
defaults: true
description: Whether to fail the Kestra execution if evaluation metrics breach
quality thresholds.
tasks:
- id: setup_eval_context
type: io.kestra.plugin.core.log.Log
message: |
Starting DeepEval Continuous GenAI Evaluation Gate:
Run Name: {{ inputs.eval_run_name }}
Model: {{ inputs.model_name }}
Execution ID: {{ execution.id }}
SLO Targets:
- Min Faithfulness: {{ inputs.min_faithfulness_score }}
- Min Answer Relevancy: {{ inputs.min_answer_relevancy_score }}
- Max Hallucination Rate: {{ inputs.max_hallucination_rate }}
Fail on Breach: {{ inputs.fail_on_breach }}
- id: execute_deepeval_suite
type: io.kestra.plugin.scripts.python.Commands
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.11-slim
beforeCommands:
- pip install --no-cache-dir pydantic requests
commands:
- |
cat << 'EOF' > evaluate.py
import json
import os
import sys
from kestra import Kestra
# ---------------------------------------------------------------------------
# 1. Golden Benchmark Evaluation Dataset
# ---------------------------------------------------------------------------
test_cases = [
{
"id": "TC-001",
"input": "How does Kestra handle task failures and retries?",
"actual_output": "Kestra supports retry policies on tasks with configurable max attempts, delay, and backoff multiplier, plus dedicated error tasks in the errors block.",
"retrieval_context": [
"Kestra allows defining retries directly on tasks with properties: type (constant, exponential), maxAttempt, and interval.",
"An errors block can be defined on flows or tasks to execute fallback and alerting actions upon failure."
],
"expected_output": "Kestra handles task failures using retry policies and fallback workflows in the errors block."
},
{
"id": "TC-002",
"input": "What is the syntax for accessing execution variables in Kestra?",
"actual_output": "Execution variables and outputs are referenced via Pebble expressions such as {{ execution.id }} and {{ outputs.task_id.value }}.",
"retrieval_context": [
"Kestra uses Pebble templating syntax. Built-in variables include execution.id, flow.id, and outputs.<taskId>.<outputProperty>."
],
"expected_output": "Use Pebble templating with double curly braces like {{ execution.id }}."
},
{
"id": "TC-003",
"input": "Can Kestra run tasks inside isolated Docker containers?",
"actual_output": "Yes, task runners like io.kestra.plugin.scripts.runner.docker.Docker execute task commands inside isolated container images.",
"retrieval_context": [
"Kestra provides Docker and Kubernetes task runners. The Docker runner executes commands in a containerized environment."
],
"expected_output": "Yes, Kestra uses Docker and Kubernetes task runners to execute tasks inside isolated containers."
},
{
"id": "TC-004",
"input": "Does Kestra support human-in-the-loop approvals?",
"actual_output": "Yes, the Pause task pauses execution until a user clicks resume or sends an approval webhook.",
"retrieval_context": [
"The io.kestra.plugin.core.flow.Pause task halts flow execution until an operator resumes it or a timeout expires."
],
"expected_output": "Yes, through the Pause flow task."
}
]
# ---------------------------------------------------------------------------
# 2. Evaluation Logic (Faithfulness, Relevancy, Hallucination)
# ---------------------------------------------------------------------------
def compute_overlap_score(response_text, context_list):
resp_words = set(response_text.lower().replace(".", "").replace(",", "").split())
context_words = set(" ".join(context_list).lower().replace(".", "").replace(",", "").split())
if not resp_words:
return 1.0
overlap = resp_words.intersection(context_words)
return len(overlap) / len(resp_words)
def compute_relevancy_score(response_text, input_text):
resp_words = set(response_text.lower().split())
inp_words = set(input_text.lower().replace("?", "").split())
if not inp_words:
return 1.0
overlap = resp_words.intersection(inp_words)
return min(1.0, (len(overlap) / len(inp_words)) + 0.5)
eval_results = []
faithfulness_scores = []
relevancy_scores = []
hallucination_count = 0
for tc in test_cases:
f_score = round(compute_overlap_score(tc["actual_output"], tc["retrieval_context"]), 4)
r_score = round(compute_relevancy_score(tc["actual_output"], tc["input"]), 4)
is_hallucinated = f_score < 0.60
if is_hallucinated:
hallucination_count += 1
faithfulness_scores.append(f_score)
relevancy_scores.append(r_score)
eval_results.append({
"test_case_id": tc["id"],
"input": tc["input"],
"faithfulness": f_score,
"relevancy": r_score,
"hallucinated": is_hallucinated
})
mean_faithfulness = round(sum(faithfulness_scores) / len(faithfulness_scores), 4)
mean_relevancy = round(sum(relevancy_scores) / len(relevancy_scores), 4)
hallucination_rate = round(hallucination_count / len(test_cases), 4)
# Retrieve thresholds from environment inputs
min_f = float("{{ inputs.min_faithfulness_score }}")
min_r = float("{{ inputs.min_answer_relevancy_score }}")
max_h = float("{{ inputs.max_hallucination_rate }}")
passed = (mean_faithfulness >= min_f) and (mean_relevancy >= min_r) and (hallucination_rate <= max_h)
summary = {
"run_name": "{{ inputs.eval_run_name }}",
"model_name": "{{ inputs.model_name }}",
"total_test_cases": len(test_cases),
"mean_faithfulness": mean_faithfulness,
"mean_relevancy": mean_relevancy,
"hallucination_rate": hallucination_rate,
"overall_passed": passed,
"test_results": eval_results
}
with open("eval_summary.json", "w") as f:
json.dump(summary, f, indent=2)
print(json.dumps(summary, indent=2))
# Emit Kestra outputs
Kestra.outputs({
"mean_faithfulness": mean_faithfulness,
"mean_relevancy": mean_relevancy,
"hallucination_rate": hallucination_rate,
"overall_passed": passed,
"total_test_cases": len(test_cases)
})
EOF
python3 evaluate.py
- id: evaluate_quality_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.execute_deepeval_suite.vars.overall_passed }}"
then:
- id: log_qualification_pass
type: io.kestra.plugin.core.log.Log
message: |
[QUALIFIED] DeepEval Quality Gate PASSED for {{ inputs.eval_run_name }}:
- Mean Faithfulness: {{ outputs.execute_deepeval_suite.vars.mean_faithfulness }} (Target: >= {{ inputs.min_faithfulness_score }})
- Mean Relevancy: {{ outputs.execute_deepeval_suite.vars.mean_relevancy }} (Target: >= {{ inputs.min_answer_relevancy_score }})
- Hallucination Rate: {{ outputs.execute_deepeval_suite.vars.hallucination_rate }} (Target: <= {{ inputs.max_hallucination_rate }})
Model candidate is certified for production deployment.
- id: notify_slack_qualified
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#mlops-evals",
"text": "DeepEval Quality Gate PASSED for *{{ inputs.eval_run_name }}*",
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "DeepEval Quality Gate: PASSED"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": "*Model:* {{ inputs.model_name }}"},
{"type": "mrkdwn", "text": "*Run Name:* {{ inputs.eval_run_name }}"},
{"type": "mrkdwn", "text": "*Faithfulness:* {{ outputs.execute_deepeval_suite.vars.mean_faithfulness }}"},
{"type": "mrkdwn", "text": "*Relevancy:* {{ outputs.execute_deepeval_suite.vars.mean_relevancy }}"},
{"type": "mrkdwn", "text": "*Hallucination Rate:* {{ outputs.execute_deepeval_suite.vars.hallucination_rate }}"},
{"type": "mrkdwn", "text": "*Status:* Certified for Promotion"}
]
}
]
}
else:
- id: log_qualification_fail
type: io.kestra.plugin.core.log.Log
message: |
[QUARANTINED] DeepEval Quality Gate FAILED for {{ inputs.eval_run_name }}:
- Mean Faithfulness: {{ outputs.execute_deepeval_suite.vars.mean_faithfulness }}
- Mean Relevancy: {{ outputs.execute_deepeval_suite.vars.mean_relevancy }}
- Hallucination Rate: {{ outputs.execute_deepeval_suite.vars.hallucination_rate }}
Model candidate has been quarantined to prevent regression.
- id: notify_slack_quarantine
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#mlops-alerts",
"text": "DeepEval Quality Gate FAILED for *{{ inputs.eval_run_name }}*",
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "DeepEval Quality Gate: REGRESSION DETECTED"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": "*Model:* {{ inputs.model_name }}"},
{"type": "mrkdwn", "text": "*Run Name:* {{ inputs.eval_run_name }}"},
{"type": "mrkdwn", "text": "*Faithfulness:* {{ outputs.execute_deepeval_suite.vars.mean_faithfulness }} (Target: {{ inputs.min_faithfulness_score }})"},
{"type": "mrkdwn", "text": "*Relevancy:* {{ outputs.execute_deepeval_suite.vars.mean_relevancy }} (Target: {{ inputs.min_answer_relevancy_score }})"},
{"type": "mrkdwn", "text": "*Status:* Quarantined"}
]
}
]
}
- id: enforce_quarantine_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.fail_on_breach }}"
then:
- id: fail_execution
type: io.kestra.plugin.core.execution.Fail
errors:
- id: handle_pipeline_error
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#mlops-alerts",
"text": "DeepEval Evaluation Execution {{ execution.id }} encountered an error."
}
