Automated RAG Vector Index Drift and Hallucination Circuit Breaker

id: rag-vector-drift-hallucination-guard
namespace: company.ai.platform

labels:
  domain: ai-engineering
  subsystem: rag-vector-ops
  reliability_tier: mission-critical
  orchestration_pattern: blue-green-circuit-breaker

# =============================================================================
# WORKFLOW INPUT PARAMETERS (15 Configurable Quality & SLA Guardrails)
# =============================================================================
inputs:
  - id: vector_collection
    type: STRING
    defaults: "prod-enterprise-knowledge-v4"
    description: "Production vector database collection alias serving live RAG traffic"

  - id: candidate_snapshot_id
    type: STRING
    defaults: "snap_2026_10_17_blue_094"
    description: "Newly built candidate vector index snapshot awaiting validation
      and promotion"

  - id: baseline_snapshot_id
    type: STRING
    defaults: "snap_2026_10_10_green_093"
    description: "Currently active, known-good baseline vector snapshot used for
      instant rollback"

  - id: embedding_model_id
    type: STRING
    defaults: "text-embedding-3-large-3072d"
    description: "Identifier and dimensionality of the embedding model used to
      encode chunks"

  - id: observed_mean_cosine
    type: FLOAT
    defaults: 0.842
    description: "Observed mean top-K cosine similarity across golden evaluation
      queries (0.0 to 1.0)"

  - id: min_cosine_threshold
    type: FLOAT
    defaults: 0.780
    description: "Minimum acceptable mean cosine similarity before flagging semantic
      embedding drift"

  - id: observed_recall_at_5
    type: FLOAT
    defaults: 0.915
    description: "Observed Recall@5 score against the golden ground-truth retrieval
      dataset (0.0 to 1.0)"

  - id: min_recall_at_5_target
    type: FLOAT
    defaults: 0.850
    description: "Minimum required Recall@5 threshold required to certify candidate
      snapshot promotion"

  - id: observed_hallucination_rate
    type: FLOAT
    defaults: 0.018
    description: "Observed fraction of synthetic evaluation answers containing
      ungrounded citations"

  - id: max_hallucination_ceiling
    type: FLOAT
    defaults: 0.040
    description: "Maximum allowable hallucination/unverified citation rate before
      tripping circuit breaker"

  - id: stale_document_ratio
    type: FLOAT
    defaults: 0.032
    description: "Ratio of indexed vector chunks whose source document hash is older
      than the freshness window"

  - id: max_staleness_tolerance
    type: FLOAT
    defaults: 0.080
    description: "Maximum tolerated ratio of stale document chunks before blocking
      index promotion"

  - id: p95_retrieval_latency_ms
    type: INT
    defaults: 118
    description: "Observed 95th-percentile Approximate Nearest Neighbor (ANN) query
      latency in milliseconds"

  - id: max_retrieval_sla_ms
    type: INT
    defaults: 220
    description: "Hard SLA ceiling for p95 vector retrieval latency in milliseconds"

  - id: auto_rollback_on_breach
    type: BOOL
    defaults: true
    description: "Whether to automatically quarantine the candidate snapshot and pin
      baseline on SLA breach"

# =============================================================================
# WORKFLOW TRIGGERS (Scheduled Sweep + Event-Driven Index Build Webhook)
# =============================================================================
triggers:
  - id: six_hour_rag_quality_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Run continuous RAG retrieval drift and staleness verification
      every 6 hours"
    cron: "0 */6 * * *"

  - id: knowledge_sync_completed_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: "Trigger immediate blue-green verification when upstream document
      ingestion completes"
    key: "rag-index-build-webhook-token"

# =============================================================================
# ORCHESTRATION TASKS (8 Architectural Stages / 25 Total Tasks)
# =============================================================================
tasks:
  # ---------------------------------------------------------------------------
  # STAGE 1: Governance Session & Snapshot Context Initialization
  # ---------------------------------------------------------------------------
  - id: init_rag_governance_session
    type: io.kestra.plugin.core.log.Log
    description: "Initialize RAG governance context, snapshot identifiers, and
      target thresholds"
    level: INFO
    message: |
      [STAGE 1/8] Initializing RAG Vector Drift & Hallucination Circuit Breaker
      Collection Alias   : {{ inputs.vector_collection }}
      Candidate Snapshot : {{ inputs.candidate_snapshot_id }}
      Baseline Snapshot  : {{ inputs.baseline_snapshot_id }}
      Embedding Model    : {{ inputs.embedding_model_id }}
      Guardrails         : MinCosine={{ inputs.min_cosine_threshold }} | MinRecall@5={{ inputs.min_recall_at_5_target }} | MaxHallucination={{ inputs.max_hallucination_ceiling }}

  # ---------------------------------------------------------------------------
  # STAGE 2: Parallel Control-Plane & Inference Subsystem Pre-Flight Probes
  # ---------------------------------------------------------------------------
  - id: parallel_control_plane_probes
    type: io.kestra.plugin.core.flow.Parallel
    description: "Concurrently probe vector index nodes, embedding gateway,
      cross-encoder reranker, and doc store"
    tasks:
      - id: probe_vector_shard_primary
        type: io.kestra.plugin.core.http.Request
        description: "Verify primary HNSW vector index shard read/write readiness"
        uri: "https://jsonplaceholder.typicode.com/posts/1"
        method: GET

      - id: probe_vector_shard_replica
        type: io.kestra.plugin.core.http.Request
        description: "Verify read-replica vector shard synchronization health"
        uri: "https://jsonplaceholder.typicode.com/posts/2"
        method: GET

      - id: probe_embedding_inference_api
        type: io.kestra.plugin.core.http.Request
        description: "Verify dense embedding model inference gateway responsiveness"
        uri: "https://jsonplaceholder.typicode.com/todos/1"
        method: GET

      - id: probe_cross_encoder_reranker
        type: io.kestra.plugin.core.http.Request
        description: "Verify stage-2 cross-encoder semantic reranker endpoint health"
        uri: "https://jsonplaceholder.typicode.com/todos/2"
        method: GET

      - id: log_control_plane_readiness
        type: io.kestra.plugin.core.log.Log
        description: "Confirm all RAG retrieval subsystems responded before running
          DuckDB analytics"
        level: INFO
        message: "All 4 RAG control-plane subsystems (Primary Shard, Replica, Embedding
          API, Reranker) healthy."

  # ---------------------------------------------------------------------------
  # STAGE 3: In-Memory DuckDB Multi-CTE Vector Partition & Benchmark Audit
  # ---------------------------------------------------------------------------
  - id: audit_vector_partition_telemetry
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: "Run multi-CTE analytical SQL over vector partitions to audit chunk
      density, orphan rates, and cosine distributions"
    sql: |
      WITH partition_inventory AS (
        SELECT
          'ns-security-policies'   AS namespace_id,
          'tier-0-critical'        AS criticality,
          18420                    AS chunk_count,
          512                      AS avg_tokens,
          0.864                    AS mean_cosine,
          0.932                    AS recall_at_5,
          14                       AS orphan_chunks,
          92                       AS p95_ms
        UNION ALL
        SELECT
          'ns-api-reference-docs'  AS namespace_id,
          'tier-1-product'         AS criticality,
          42150                    AS chunk_count,
          480                      AS avg_tokens,
          0.838                    AS mean_cosine,
          0.910                    AS recall_at_5,
          29                       AS orphan_chunks,
          124                      AS p95_ms
        UNION ALL
        SELECT
          'ns-billing-runbooks'    AS namespace_id,
          'tier-0-critical'        AS criticality,
          9840                     AS chunk_count,
          445                      AS avg_tokens,
          0.851                    AS mean_cosine,
          0.925                    AS recall_at_5,
          4                        AS orphan_chunks,
          88                       AS p95_ms
        UNION ALL
        SELECT
          'ns-compliance-soc2'     AS namespace_id,
          'tier-0-critical'        AS criticality,
          15300                    AS chunk_count,
          530                      AS avg_tokens,
          0.829                    AS mean_cosine,
          0.898                    AS recall_at_5,
          11                       AS orphan_chunks,
          115                      AS p95_ms
        UNION ALL
        SELECT
          'ns-incident-postmortem' AS namespace_id,
          'tier-2-internal'        AS criticality,
          11290                    AS chunk_count,
          610                      AS avg_tokens,
          0.819                    AS mean_cosine,
          0.889                    AS recall_at_5,
          18                       AS orphan_chunks,
          134                      AS p95_ms
        UNION ALL
        SELECT
          'ns-sdk-code-examples'   AS namespace_id,
          'tier-1-product'         AS criticality,
          28900                    AS chunk_count,
          390                      AS avg_tokens,
          0.855                    AS mean_cosine,
          0.921                    AS recall_at_5,
          19                       AS orphan_chunks,
          108                      AS p95_ms
      ),
      golden_query_benchmarks AS (
        SELECT
          'q_auth_token_rotation'  AS benchmark_id,
          'ns-security-policies'   AS namespace_id,
          0.891                    AS top1_score,
          1                        AS grounded_citation
        UNION ALL
        SELECT
          'q_webhook_retry_spec'   AS benchmark_id,
          'ns-api-reference-docs'  AS namespace_id,
          0.844                    AS top1_score,
          1                        AS grounded_citation
        UNION ALL
        SELECT
          'q_dunning_grace_period' AS benchmark_id,
          'ns-billing-runbooks'    AS namespace_id,
          0.872                    AS top1_score,
          1                        AS grounded_citation
        UNION ALL
        SELECT
          'q_fips_encryption_key'  AS benchmark_id,
          'ns-compliance-soc2'     AS namespace_id,
          0.815                    AS top1_score,
          1                        AS grounded_citation
      ),
      namespace_rollup AS (
        SELECT
          p.namespace_id,
          p.criticality,
          p.chunk_count,
          p.avg_tokens,
          p.mean_cosine,
          p.recall_at_5,
          p.orphan_chunks,
          p.p95_ms,
          ROUND(CAST(p.orphan_chunks AS DOUBLE) / p.chunk_count, 6) AS orphan_ratio,
          COUNT(g.benchmark_id)                                     AS benchmark_queries_run
        FROM partition_inventory p
        LEFT JOIN golden_query_benchmarks g
          ON p.namespace_id = g.namespace_id
        GROUP BY
          p.namespace_id,
          p.criticality,
          p.chunk_count,
          p.avg_tokens,
          p.mean_cosine,
          p.recall_at_5,
          p.orphan_chunks,
          p.p95_ms
      )
      SELECT
        namespace_id,
        criticality,
        chunk_count,
        avg_tokens,
        mean_cosine,
        recall_at_5,
        orphan_ratio,
        p95_ms,
        benchmark_queries_run
      FROM namespace_rollup
      ORDER BY chunk_count DESC;
    fetchType: FETCH

  # ---------------------------------------------------------------------------
  # STAGE 4: Python RAG Drift, Grounding & Hallucination Scoring Engine
  # ---------------------------------------------------------------------------
  - id: evaluate_rag_drift_and_grounding
    type: io.kestra.plugin.scripts.python.Script
    description: "Compute composite RAG quality index, cosine drift delta,
      hallucination risk, and blue-green promotion decision"
    beforeCommands:
      - pip install kestra
    script: |
      from kestra import Kestra

      # 1. Load runtime inputs and DuckDB partition rollup count
      audited_namespaces = {{ outputs.audit_vector_partition_telemetry.size }}
      collection = "{{ inputs.vector_collection }}"
      candidate_snap = "{{ inputs.candidate_snapshot_id }}"
      baseline_snap = "{{ inputs.baseline_snapshot_id }}"

      mean_cosine = float({{ inputs.observed_mean_cosine }})
      min_cosine = float({{ inputs.min_cosine_threshold }})
      recall_5 = float({{ inputs.observed_recall_at_5 }})
      min_recall_5 = float({{ inputs.min_recall_at_5_target }})
      hallucination_rate = float({{ inputs.observed_hallucination_rate }})
      max_hallucination = float({{ inputs.max_hallucination_ceiling }})
      stale_ratio = float({{ inputs.stale_document_ratio }})
      max_stale = float({{ inputs.max_staleness_tolerance }})
      p95_ms = int({{ inputs.p95_retrieval_latency_ms }})
      max_sla_ms = int({{ inputs.max_retrieval_sla_ms }})

      # 2. Compute composite RAG Grounding & Retrieval Quality Score (0 to 100 scale)
      #    Weights: Cosine Similarity (35%), Recall@5 (35%), Citation Grounding (20%), Freshness (10%)
      cosine_component = min(1.0, mean_cosine / 0.90) * 35.0
      recall_component = min(1.0, recall_5 / 0.95) * 35.0
      grounding_component = max(0.0, (1.0 - (hallucination_rate / max_hallucination))) * 20.0
      freshness_component = max(0.0, (1.0 - (stale_ratio / max_stale))) * 10.0
      quality_index = round(
          cosine_component + recall_component + grounding_component + freshness_component,
          2
      )

      # 3. Evaluate multi-dimensional SLA and hallucination circuit breaker gates
      breaches = []
      if mean_cosine < min_cosine:
          breaches.append(
              f"Cosine similarity drift ({mean_cosine:.3f} < {min_cosine:.3f})"
          )
      if recall_5 < min_recall_5:
          breaches.append(
              f"Retrieval Recall@5 drop ({recall_5:.3f} < {min_recall_5:.3f})"
          )
      if hallucination_rate > max_hallucination:
          breaches.append(
              f"Unverified citation/hallucination spike ({hallucination_rate:.1%} > {max_hallucination:.1%})"
          )
      if stale_ratio > max_stale:
          breaches.append(
              f"Document embedding staleness exceeded ({stale_ratio:.1%} > {max_stale:.1%})"
          )
      if p95_ms > max_sla_ms:
          breaches.append(
              f"Vector ANN p95 latency breach ({p95_ms}ms > {max_sla_ms}ms)"
          )

      index_healthy = len(breaches) == 0

      # 4. Determine blue-green deployment verdict and traffic routing action
      if index_healthy and quality_index >= 85.0:
          deployment_verdict = "PROMOTE_CANDIDATE_SNAPSHOT"
          routing_action = (
              f"Alias {collection} -> {candidate_snap} (100% production traffic)"
          )
      elif index_healthy:
          deployment_verdict = "PROMOTE_WITH_CANARY_MONITORING"
          routing_action = (
              f"Shift 80/20 canary traffic to {candidate_snap} and increase eval sampling"
          )
      else:
          deployment_verdict = "TRIP_BREAKER_ROLLBACK_BASELINE"
          routing_action = (
              f"Abort {candidate_snap}; pin production alias to {baseline_snap}"
          )

      exec_summary = (
          f"[{deployment_verdict}] Score={quality_index}/100 | "
          f"Namespaces={audited_namespaces} | Cosine={mean_cosine:.3f} | "
          f"Recall@5={recall_5:.3f} | Hallucination={hallucination_rate:.1%} | "
          f"Staleness={stale_ratio:.1%} | p95={p95_ms}ms"
      )

      print(exec_summary)
      print(f"Routing Decision: {routing_action}")
      if breaches:
          print("Breach Ledger: " + " | ".join(breaches))

      Kestra.outputs({
          'index_healthy': index_healthy,
          'deployment_verdict': deployment_verdict,
          'quality_index': quality_index,
          'audited_namespaces': audited_namespaces,
          'routing_action': routing_action,
          'exec_summary': exec_summary
      })

  # ---------------------------------------------------------------------------
  # STAGE 5: Sequential Hybrid Retrieval & Cache Pipeline Synchronization
  # ---------------------------------------------------------------------------
  - id: sequential_retrieval_pipeline_sync
    type: io.kestra.plugin.core.flow.Sequential
    description: "Sequentially synchronize semantic prompt caches, hybrid BM25
      lexical indexes, and reranker weights"
    tasks:
      - id: sync_hybrid_bm25_inverted_index
        type: io.kestra.plugin.core.log.Log
        description: "Align sparse BM25 inverted index terms with dense vector partitions"
        level: INFO
        message: |
          [PIPELINE SYNC 1/4] Synchronized sparse BM25 lexical index across {{ outputs.evaluate_rag_drift_and_grounding.vars.audited_namespaces }} namespaces.

      - id: calibrate_cross_encoder_thresholds
        type: io.kestra.plugin.core.log.Log
        description: "Update stage-2 cross-encoder minimum relevance score cutoff"
        level: INFO
        message: |
          [PIPELINE SYNC 2/4] Calibrated cross-encoder reranker cutoff for Quality Index: {{ outputs.evaluate_rag_drift_and_grounding.vars.quality_index }}/100.

      - id: refresh_semantic_query_cache
        type: io.kestra.plugin.core.log.Log
        description: "Update semantic prompt cache generation watermarks"
        level: INFO
        message: |
          [PIPELINE SYNC 3/4] Updated semantic query cache watermarks for snapshot {{ inputs.candidate_snapshot_id }}.

      - id: record_lineage_attestation
        type: io.kestra.plugin.core.log.Log
        description: "Log cryptographic model and chunker lineage metadata"
        level: INFO
        message: |
          [PIPELINE SYNC 4/4] Recorded embedding provenance attestation for model {{ inputs.embedding_model_id }}.

  # ---------------------------------------------------------------------------
  # STAGE 6: Conditional Blue-Green Alias Promotion vs. Circuit Breaker Rollback
  # ---------------------------------------------------------------------------
  - id: route_blue_green_index_decision
    type: io.kestra.plugin.core.flow.If
    description: "Branch between promoting the validated candidate vector snapshot
      or rolling back to the baseline snapshot"
    condition: "{{ outputs.evaluate_rag_drift_and_grounding.vars.index_healthy == true }}"
    then:
      - id: atomic_swap_collection_alias
        type: io.kestra.plugin.core.log.Log
        description: "Atomically repoint production collection alias to validated
          candidate snapshot"
        level: INFO
        message: |
          [BLUE-GREEN PROMOTION 1/4] Atomically swapped collection alias [{{ inputs.vector_collection }}] -> {{ inputs.candidate_snapshot_id }}.

      - id: warm_hnsw_graph_memory
        type: io.kestra.plugin.core.log.Log
        description: "Pre-load HNSW entry points and upper graph layers into vector node
          RAM"
        level: INFO
        message: |
          [BLUE-GREEN PROMOTION 2/4] Pre-warmed HNSW vector graph pages in memory (p95 target: {{ inputs.p95_retrieval_latency_ms }}ms).

      - id: mark_baseline_snapshot_standby
        type: io.kestra.plugin.core.log.Log
        description: "Retain previous green snapshot in warm standby for instant rollback"
        level: INFO
        message: |
          [BLUE-GREEN PROMOTION 3/4] Retained previous snapshot {{ inputs.baseline_snapshot_id }} as warm rollback standby.

      - id: emit_promotion_compliance_certificate
        type: io.kestra.plugin.core.log.Log
        description: "Emit formal RAG quality compliance record for audit trail"
        level: INFO
        message: |
          [BLUE-GREEN PROMOTION 4/4] RAG Index Certified for Production Traffic.
          Summary: {{ outputs.evaluate_rag_drift_and_grounding.vars.exec_summary }}
    else:
      - id: pin_traffic_to_baseline_snapshot
        type: io.kestra.plugin.core.log.Log
        description: "Prevent alias swap and lock production traffic to verified
          baseline snapshot"
        level: WARN
        message: |
          [CIRCUIT BREAKER ROLLBACK 1/4] Halted candidate promotion! Traffic pinned to safe baseline {{ inputs.baseline_snapshot_id }}.

      - id: quarantine_drifted_candidate_index
        type: io.kestra.plugin.core.log.Log
        description: "Move drifted candidate snapshot into isolated diagnostic namespace"
        level: WARN
        message: |
          [CIRCUIT BREAKER ROLLBACK 2/4] Quarantined drifted snapshot {{ inputs.candidate_snapshot_id }} for offline chunk inspection.

      - id: purge_poisoned_semantic_cache
        type: io.kestra.plugin.core.log.Log
        description: "Invalidate any canary semantic cache keys populated by candidate
          snapshot"
        level: WARN
        message: |
          [CIRCUIT BREAKER ROLLBACK 3/4] Purged semantic cache entries linked to candidate snapshot {{ inputs.candidate_snapshot_id }}.

      - id: escalate_rag_quality_incident
        type: io.kestra.plugin.core.log.Log
        description: "Dispatch high-priority alert with breach diagnostics to AI
          Platform on-call"
        level: WARN
        message: |
          [CIRCUIT BREAKER ROLLBACK 4/4] Dispatched AI Platform quality incident alert.
          Diagnostic Summary: {{ outputs.evaluate_rag_drift_and_grounding.vars.exec_summary }}

  # ---------------------------------------------------------------------------
  # STAGE 7: Parallel Post-Deployment SLO Telemetry & Registry Sync
  # ---------------------------------------------------------------------------
  - id: parallel_post_deployment_telemetry
    type: io.kestra.plugin.core.flow.Parallel
    description: "Concurrently emit post-decision observability metrics and audit
      trail snapshots"
    tasks:
      - id: publish_rag_slo_metrics
        type: io.kestra.plugin.core.log.Log
        description: "Ship composite RAG quality index and Recall@5 metrics to
          observability sink"
        level: INFO
        message: "Published RAG SLO metrics (QualityIndex={{
          outputs.evaluate_rag_drift_and_grounding.vars.quality_index }}) to
          observability sink."

      - id: update_governance_catalog_state
        type: io.kestra.plugin.core.log.Log
        description: "Record snapshot routing state in enterprise AI governance catalog"
        level: INFO
        message: "Updated AI Governance Registry -> Action: {{
          outputs.evaluate_rag_drift_and_grounding.vars.routing_action }}"

  # ---------------------------------------------------------------------------
  # STAGE 8: Final Execution Ledger Closure
  # ---------------------------------------------------------------------------
  - id: finalize_rag_circuit_breaker_run
    type: io.kestra.plugin.core.log.Log
    description: "Close audit run and output final blue-green deployment verdict"
    level: INFO
    message: |
      [STAGE 8/8] Completed 25-task RAG Vector Drift & Hallucination Circuit Breaker (Execution: {{ execution.id }})
      Final Verdict: {{ outputs.evaluate_rag_drift_and_grounding.vars.deployment_verdict }}