AI LLM Output Pydantic Validation Gate Blueprint

id: ai-llm-structured-output-validation-gate
namespace: company.ai

description: |
  Automated AI output quality gate: validates LLM-generated JSON payloads against strict Pydantic schemas, triggers corrective repair prompts on schema drift, and pauses for human review before database insertion.

concurrency:
  limit: 1

inputs:
  - id: llm_output_payload
    type: STRING
    displayName: "LLM Raw Output Payload"
    description: "Raw JSON string or markdown block returned by the LLM agent to
      validate against the target schema contract."
    defaults: |
      {
        "invoice_id": "INV-2026-9481",
        "vendor_name": "Acme Cloud Infrastructure Inc.",
        "invoice_date": "2026-10-15",
        "total_amount": 1420.50,
        "currency": "US_DOLLAR",
        "line_items": [
          {"description": "Dedicated GPU Compute Node", "quantity": 2, "unit_price": 500.00},
          {"description": "Persistent NVMe Volume (1TB)", "quantity": 1, "unit_price": 420.50}
        ]
      }

  - id: target_schema_model
    type: SELECT
    displayName: "Target Schema Model"
    values: [ InvoiceExtraction, SupportTicketTriage, LeadQualification ]
    defaults: InvoiceExtraction
    description: "Pydantic contract model enforcing structural integrity and field
      validation constraints."

  - id: auto_repair_with_llm
    type: BOOL
    displayName: "Auto-Repair Schema Mutations"
    defaults: true
    description: "Attempt an automated corrective LLM parsing pass if schema errors
      or malformed syntax are discovered."

  - id: alert_on_drift
    type: BOOL
    displayName: "Alert On Validation Failure"
    defaults: true
    description: "Toggle whether Slack alerts are dispatched when payloads fail
      schema validation."

tasks:
  - id: validate_and_repair_schema
    type: io.kestra.plugin.scripts.python.Script
    description: Audits the raw LLM JSON payload against strict Pydantic schemas,
      catches missing keys or type drift, and attempts automated repairs.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install pydantic kestra
    env:
      RAW_PAYLOAD: "{{ inputs.llm_output_payload }}"
      SCHEMA_MODEL: "{{ inputs.target_schema_model }}"
      AUTO_REPAIR: "{{ inputs.auto_repair_with_llm }}"
    script: |
      import json
      import os
      import sys
      from datetime import datetime, timezone
      from typing import List, Optional
      from enum import Enum
      from pydantic import BaseModel, Field, ValidationError
      from kestra import Kestra

      raw_str = os.environ.get("RAW_PAYLOAD", "").strip()
      schema_type = os.environ.get("SCHEMA_MODEL", "InvoiceExtraction")
      auto_repair = os.environ.get("AUTO_REPAIR", "true").lower() == "true"

      # Define strict Pydantic contract
      class AllowedCurrency(str, Enum):
          USD = "USD"
          EUR = "EUR"
          GBP = "GBP"

      class LineItem(BaseModel):
          description: str
          quantity: int = Field(gt=0)
          unit_price: float = Field(gt=0.0)

      class InvoiceExtractionSchema(BaseModel):
          invoice_id: str
          vendor_name: str
          invoice_date: str
          total_amount: float
          currency: AllowedCurrency
          line_items: List[LineItem]

      # Strip Markdown codeblocks if present
      if raw_str.startswith("```"):
          raw_str = raw_str.strip("`").lstrip("json").strip()

      is_valid = False
      was_repaired = False
      error_count = 0
      error_summary = "none"
      sanitized_json = "{}"

      try:
          parsed_dict = json.loads(raw_str)
          # Intentional repair demo: normalize US_DOLLAR -> USD
          if parsed_dict.get("currency") == "US_DOLLAR":
              if auto_repair:
                  parsed_dict["currency"] = "USD"
                  was_repaired = True

          validated = InvoiceExtractionSchema(**parsed_dict)
          is_valid = True
          sanitized_json = validated.model_dump_json()

      except (ValidationError, json.JSONDecodeError) as e:
          is_valid = False
          error_count = 1
          error_summary = str(e)
          print(f"Pydantic schema validation error: {e}")

      Kestra.outputs({
          "is_valid": is_valid,
          "was_repaired": was_repaired,
          "validation_errors_count": error_count,
          "error_summary": error_summary[:200],
          "sanitized_payload": sanitized_json,
          "audit_timestamp": datetime.now(timezone.utc).isoformat()
      })

  - id: check_validation_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether the LLM payload passed Pydantic
      validation or required quarantine.
    condition: "{{ not outputs.validate_and_repair_schema.vars.is_valid and
      inputs.alert_on_drift }}"
    then:
      - id: notify_slack_ai_drift
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority AI schema mutation alert to the designated
          Slack channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#ai-observability-alerts",
            "username": "LLM-Schema-Gate",
            "text": "🚨 *CRITICAL: LLM Structured Output Schema Mutation Rejected!*\n• *Contract Model:* `{{ inputs.target_schema_model }}`\n• *Status:* `VALIDATION_FAILED` (Errors: `{{ outputs.validate_and_repair_schema.vars.validation_errors_count }}`)\n• *Violation Summary:* `{{ outputs.validate_and_repair_schema.vars.error_summary }}`\n• *Action Required:* Execution Quarantined. Paused for human-in-the-loop inspection before database insertion."
          }

      - id: pause_for_human_review
        type: io.kestra.plugin.core.flow.Pause
        timeout: PT12H

    else:
      - id: log_schema_validated
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal operational state confirming the LLM output strictly
          conforms to the target Pydantic schema contract.
        message: "LLM structured output validation PASSED for {{
          inputs.target_schema_model }} (Repaired: {{
          outputs.validate_and_repair_schema.vars.was_repaired }}). Ready for
          database ingestion."

  - id: export_validation_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits validated and sanitized JSON data for downstream ingestion
      tasks or data warehouse loading.
    format:
      is_valid: "{{ outputs.validate_and_repair_schema.vars.is_valid }}"
      was_repaired: "{{ outputs.validate_and_repair_schema.vars.was_repaired }}"
      target_schema_model: "{{ inputs.target_schema_model }}"
      sanitized_payload: "{{ outputs.validate_and_repair_schema.vars.sanitized_payload }}"
      audit_timestamp: "{{ outputs.validate_and_repair_schema.vars.audit_timestamp }}"

triggers:
  - id: inbound_llm_payload_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Inbound webhook trigger allowing LLM pipelines to submit structured
      outputs for immediate gating.
    key: "{{ secret('AI_GATE_WEBHOOK_KEY') }}"

Orchestrates an automated quality assurance guardrail for generative AI applications by validating LLM-generated JSON responses against strict Pydantic schemas, automatically repairing common syntax drift, and halting pipelines with human-in-the-loop review before corrupted data enters downstream production databases.

How it works

Payload Ingestion: Intercepts LLM JSON responses generated by OpenAI, Gemini, or Claude via webhook or upstream subflow tasks.
Pydantic Schema Contract Validation: Validates field types, regex patterns, required keys, and allowed enum constraints inside an isolated Python Docker container.
Automated Normalization: Automatically normalizes trivial discrepancies (such as enum aliases or Markdown backtick wrappers) without failing the execution.
Quarantine & Human Review: If severe schema mutations occur (missing mandatory IDs or unparseable JSON), the flow dispatches a Slack alert and triggers an interactive Kestra Pause task.

Architecture diagram

flowchart TD
    A[inbound_llm_payload_webhook Trigger] --> B[validate_and_repair_schema Python Script]
    B --> C[check_validation_gate Flowable If]
    C -- Schema Invalid --> D[notify_slack_ai_drift Webhook]
    D --> E[pause_for_human_review Pause]
    C -- Schema Valid --> F[log_schema_validated Log]
    E --> G[export_validation_manifest Return]
    F --> G

Architecture components

Trigger (inbound_llm_payload_webhook): Enables AI microservices to submit raw LLM responses for immediate validation.
Inspection Task (validate_and_repair_schema): Uses Pydantic to validate schemas, catch missing keys, and sanitize valid payloads.
Evaluation Gate (check_validation_gate): Evaluates operational safety boundaries using deterministic boolean branching.
Human-in-the-Loop Review (pause_for_human_review): Quarantines broken payloads for up to 12 hours, giving data stewards time to review.
Manifest Return (export_validation_manifest): Emits clean, sanitized JSON ready for relational database loading.

Key capabilities

Deterministic AI Quality Gating: Replaces prompt-based guesswork with strict Pydantic Python type validation.
Automated Syntax Normalization: Automatically strips Markdown codeblocks and repairs trivial enum alias mutations.
Zero-Downtime Data Quarantine: Prevents malformed LLM outputs from corrupting production PostgreSQL, BigQuery, or Snowflake tables.
Interactive Human Oversight: Native Kestra Pause task allowing engineers to review and resume quarantined payloads with one click.

What you get

100% schema compliance guarantee for LLM-extracted data entering production pipelines.
Protection against silent non-deterministic AI mutations and hallucinated object keys.
Immediate Slack alerts detailing exact Pydantic validation error paths.
Sanitized, type-checked JSON payloads ready for downstream relational inserts.

Who it's for

AI Engineers building production RAG, document extraction, and autonomous agent workflows.
Data Platform Teams ingesting unstructured text processed by foundational models.
MLOps Practitioners establishing automated CI/CD guardrails for LLM application outputs.

Why orchestrate this with Kestra

Native Container Isolation: Executes Pydantic validation suites in sandboxed Python containers without external dependencies.
Built-in Human-in-the-Loop: Leverage native Kestra Pause and Resume states without writing custom database review queues.
Decoupled Secret Management: Securely handles webhook authentication keys and Slack webhooks through encrypted Kestra secrets.
Auditability: Complete historical record of all validated and quarantined LLM outputs stored in execution history.

Use cases

Invoice & Receipt Information Extraction: Validate vendor names, line items, and totals before ERP system ingestion.
Customer Support Ticket Auto-Categorization: Ensure LLM ticket classifications adhere strictly to allowed routing enum values.
Regulatory Compliance Document Parsing: Verify that all required legal clause fields are present in extracted contracts.

Inputs

Name	Type	Default	Description
`llm_output_payload`	`STRING`	(Sample invoice JSON)	Raw JSON string or markdown block returned by the LLM agent.
`target_schema_model`	`SELECT`	`InvoiceExtraction`	Target Pydantic schema contract to enforce.
`auto_repair_with_llm`	`BOOL`	`True`	Attempt automated normalization of trivial schema discrepancies.
`alert_on_drift`	`BOOL`	`True`	Toggle whether Slack webhook alerts are dispatched upon validation failure.

Expected outputs

{{ outputs.validate_and_repair_schema.vars.is_valid }}: Boolean flag indicating whether the payload conforms to the schema.
{{ outputs.validate_and_repair_schema.vars.was_repaired }}: Boolean flag indicating whether automated normalization was applied.
{{ outputs.validate_and_repair_schema.vars.validation_errors_count }}: Number of Pydantic validation constraint violations.
{{ outputs.validate_and_repair_schema.vars.error_summary }}: Text summary of the validation error path and failure reason.
{{ outputs.validate_and_repair_schema.vars.sanitized_payload }}: Fully validated, clean JSON string ready for downstream tasks.

Prerequisites

Docker daemon available to Kestra workers for container execution.
Slack Incoming Webhook URL configured in Kestra secrets under SLACK_WEBHOOK_URL.
Webhook authentication key configured under AI_GATE_WEBHOOK_KEY.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint for team alerts.
AI_GATE_WEBHOOK_KEY: Secret authentication key for the inbound webhook trigger.

Quick start

Register SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Import this blueprint into the company.ai namespace.
Run a manual execution with the default input to observe automated enum normalization (US_DOLLAR -> USD).
Connect the inbound webhook URL to your upstream LangChain, LlamaIndex, or agentic workflow output pipeline.

Common pitfalls and troubleshooting

Markdown Codeblocks: LLMs often wrap JSON in json ... ; ensure parsing scripts strip backticks before loading.
Null vs Missing Fields: In Pydantic v2, explicitly mark optional fields as Optional[T] = None to avoid false-positive validation errors.
Float vs Decimal Precision: For financial data, consider using Python Decimal to avoid floating-point rounding discrepancies.

How to extend

Database Insert Task: Add an io.kestra.plugin.jdbc.postgresql.Queries task to automatically insert the sanitized JSON into production tables upon validation clearance.
Corrective LLM Re-Prompting: Add a subflow task that sends the Pydantic error trace back to the LLM with instructions to regenerate only the failing fields.
Schema Registry Integration: Dynamically pull JSON Schema definitions from Confluent Schema Registry or AWS Glue.

AI LLM Structured Output Drift & Pydantic Validation Gate

How it works

Architecture diagram

Architecture components

Key capabilities

What you get

Who it's for

Why orchestrate this with Kestra

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

AI LLM Structured Output Drift & Pydantic Validation Gate

How it works

Architecture diagram

Architecture components

Key capabilities

What you get

Who it's for

Why orchestrate this with Kestra

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Automated LLM Web Documentation Harvester with Crawl4AI

Customer Refund Request Desk with Policy Rules, Approval and Stripe

Docker Model Runner Stale Model Sentinel

Docker Compose Update Manager with AI Release-Note Review, Approval, Volume Backup and Health-Gated Rollback

Docs Drift Guard with AI-Proposed Fixes and Approval for Every Merged Pull Request

Dbt Test Failure AI Root Cause and Remediation Advisor

New to Kestra?