AI Prompt Drift and Quality Sentinel

id: ai-prompt-drift-and-quality-sentinel
namespace: company.team
description: |
  Automated LLM quality regression and prompt drift sentinel. Dispatches standardized benchmark prompts
  to OpenAI models on schedule, validates JSON schema adherence and semantic classification, and alerts Slack.

concurrency:
  limit: 1

inputs:
  - id: model_name
    type: STRING
    displayName: Model Name
    defaults: "gpt-4o-mini"
    description: "OpenAI model version to evaluate against the golden benchmark test suite."

  - id: golden_prompt
    type: STRING
    displayName: Benchmark Golden Prompt
    defaults: "Extract the core customer intent, sentiment, and urgency from this
      ticket: 'I need to cancel my enterprise subscription immediately, your
      service has had three outages this week.' Return valid JSON with keys:
      intent, sentiment, urgency."
    description: "Standard benchmark golden prompt used to evaluate model output
      consistency."

  - id: expected_intent_keyword
    type: STRING
    displayName: Expected Intent Keyword
    defaults: "cancel"
    description: "Mandatory keyword expected in the structured intent field to pass
      accuracy assertion."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#ai-platform"
    description: "Slack channel destination for model regression and output drift alerts."

triggers:
  - id: daily_model_evaluation
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Runs golden model benchmark daily at 08:00 UTC."
    cron: "0 8 * * *"

tasks:
  - id: execute_model_inference
    type: io.kestra.plugin.openai.ChatCompletion
    description: "Sends standardized golden prompt to OpenAI ChatCompletion API."
    apiKey: "{{ secret('OPENAI_API_KEY') }}"
    model: "{{ inputs.model_name }}"
    messages:
      - role: system
        content: "You are an automated support classifier. Always reply strictly with
          valid JSON."
      - role: user
        content: "{{ inputs.golden_prompt }}"

  - id: analyze_response_quality
    type: io.kestra.plugin.scripts.python.Script
    description: "Parses model output JSON, verifies semantic intent keywords, and
      detects formatting drift."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      llm_response.txt: "{{ outputs.execute_model_inference.choices[0].message.content }}"
    outputFiles:
      - drift_report.json
    script: |
      import json
      import re
      from kestra import Kestra

      with open("llm_response.txt", "r", encoding="utf-8") as f:
          raw_text = f.read().strip()

      clean_json = re.sub(r"^```(?:json)?\s*", "", raw_text, flags=re.MULTILINE)
      clean_json = re.sub(r"\s*```$", "", clean_json, flags=re.MULTILINE).strip()

      parse_success = False
      intent_matched = False
      parsed_data = {}
      drift_reasons = []

      try:
          parsed_data = json.loads(clean_json)
          parse_success = True
      except Exception as err:
          drift_reasons.append(f"JSON schema parsing failed: {err}")

      expected_kw = "{{ inputs.expected_intent_keyword }}".lower()
      if parse_success:
          intent_val = str(parsed_data.get("intent", "")).lower()
          if expected_kw in intent_val:
              intent_matched = True
          else:
              drift_reasons.append(f"Intent mismatch: expected '{expected_kw}', got '{intent_val}'")

          if "sentiment" not in parsed_data:
              drift_reasons.append("Missing required 'sentiment' key in model output")

      drift_detected = (not parse_success) or (not intent_matched) or (len(drift_reasons) > 0)

      report = {
          "model": "{{ inputs.model_name }}",
          "raw_response": raw_text,
          "parsed_data": parsed_data,
          "parse_success": parse_success,
          "intent_matched": intent_matched,
          "drift_detected": drift_detected,
          "drift_reasons": drift_reasons
      }

      with open("drift_report.json", "w", encoding="utf-8") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "drift_detected": drift_detected,
          "parse_success": parse_success,
          "intent_matched": intent_matched,
          "failure_reason": drift_reasons[0] if drift_reasons else "None"
      })

  - id: evaluate_drift_alarm
    type: io.kestra.plugin.core.flow.If
    description: "Branches execution based on whether model output regression or
      formatting drift was detected."
    condition: "{{ outputs.analyze_response_quality.vars.drift_detected }}"
    then:
      - id: notify_slack_prompt_drift
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches high-urgency model regression card to AI engineering
          channel via Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*CRITICAL: AI Model Quality Regression and Output Drift Alert*\n• *Evaluated Model*: `{{ inputs.model_name }}`\n• *Golden Benchmark Result*: FAILED\n• *Primary Failure Reason*: `{{ outputs.analyze_response_quality.vars.failure_reason }}`\n• *JSON Schema Parse*: {{ outputs.analyze_response_quality.vars.parse_success ? 'PASSED' : 'FAILED' }}\n• *Intent Keyword Match*: {{ outputs.analyze_response_quality.vars.intent_matched ? 'PASSED' : 'FAILED' }}\n• *Operational Impact*: Silent model drift causes downstream automated pipelines and customer classifiers to fail with unhandled exceptions.\n• *Recommended Remediation*: Review provider release notes for `{{ inputs.model_name }}`, adjust system prompt constraints, or pin model to an immutable checkpoint snapshot."
          }

    else:
      - id: log_model_accuracy_compliant
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of model benchmark compliance."
        message: "Model {{ inputs.model_name }} benchmark compliant. Zero output drift
          or schema violations detected."

  - id: export_drift_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON evaluation manifest for AI monitoring dashboards."
    format: |
      {
        "pipeline": "ai-prompt-drift-and-quality-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "model": "{{ inputs.model_name }}",
        "parse_success": {{ outputs.analyze_response_quality.vars.parse_success }},
        "intent_matched": {{ outputs.analyze_response_quality.vars.intent_matched }},
        "status": "{{ outputs.analyze_response_quality.vars.drift_detected ? 'DRIFT_DETECTED' : 'COMPLIANT' }}"
      }