RAG Synthetic Golden Dataset Generator with Kestra

id: rag-synthetic-golden-dataset-generator
namespace: company.ai

description: |
  Synthesize verified Q&A evaluation datasets from document chunks using
  LLM-as-a-Generator with structured JSON Schema output, and export
  a golden dataset artifact for automated RAG quality benchmarking.

triggers:
  - id: weekly_dataset_refresh
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic synthetic dataset generation run. Shipped disabled by default.
    cron: "0 2 * * 1"
    disabled: true

  - id: documentation_update_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook triggered by CI/CD when documentation or
      knowledge base changes.
    key: "{{ secret('WEBHOOK_KEY') }}"

inputs:
  - id: model_name
    type: STRING
    defaults: openai/gpt-oss-120b
    description: Model identifier used for synthetic test generation
      (OpenAI-compatible Groq endpoint).

  - id: min_total_cases
    type: INT
    defaults: 3
    description: Minimum required synthesized test cases before promoting the golden
      dataset artifact.

  - id: document_chunks
    type: JSON
    defaults:
      - id: chunk_kestra_001
        title: Kestra Architecture and Flow Lifecycle
        content: |
          Kestra is an event-driven declarative orchestrator built on a stateless executor model.
          Workflows are defined in YAML where each flow belongs to a dot-separated namespace.
          Tasks execute inside isolated task runners such as Docker, Kubernetes, or native processes.
          When an execution starts, Kestra emits state transitions to an internal event queue,
          ensuring exactly-once execution semantics and persistent audit history.

      - id: chunk_postgres_002
        title: PostgreSQL Write-Ahead Logging and Crash Recovery
        content: |
          PostgreSQL uses Write-Ahead Logging (WAL) to ensure data integrity during unexpected crashes.
          Before changes are flushed to permanent data heap pages, the transactional changes
          must first be recorded sequentially in WAL buffers and written to disk.
          During recovery, PostgreSQL replays WAL records from the last valid checkpoint (REDO phase)
          to bring all table pages to a fully consistent state.

      - id: chunk_k8s_003
        title: Kubernetes Horizontal Pod Autoscaler Mechanics
        content: |
          The Kubernetes Horizontal Pod Autoscaler (HPA) automatically adjusts the number of replica
          pods in a deployment based on observed CPU utilization, memory pressure, or custom metrics.
          HPA operates on an evaluation loop (default every 15 seconds) managed by the kube-controller-manager.
          It queries the Metrics API (metrics.k8s.io) and scales pod replicas according to the ratio
          between current metric value and target metric threshold.
    description: |
      Array of raw document chunks to synthesize test cases from.
      Each item must contain id, title, and content fields.

tasks:
  - id: log_generation_start
    type: io.kestra.plugin.core.log.Log
    description: Log generation parameters and input chunk count.
    message: |
      ======================================================
      STARTING RAG SYNTHETIC GOLDEN DATASET GENERATION
      ======================================================
      Target Model: {{ inputs.model_name }}
      Input Chunks: {{ inputs.document_chunks | length }}
      Minimum Acceptance Threshold: {{ inputs.min_total_cases }} cases
      ======================================================

  - id: generate_synthetic_cases
    type: io.kestra.plugin.core.flow.Loop
    description: Concurrently process each document chunk and synthesize Q&A
      evaluation pairs.
    values: "{{ inputs.document_chunks }}"
    concurrencyLimit: 2
    fetchType: FETCH

    tasks:
      - id: synthesize_qa
        type: io.kestra.plugin.ai.completion.ChatCompletion
        description: Call LLM with strict JSON schema to generate synthetic evaluation
          cases.

        provider:
          type: io.kestra.plugin.ai.provider.OpenAI
          apiKey: "{{ secret('GROQ_API_KEY') }}"
          baseUrl: https://api.groq.com/openai/v1
          modelName: "{{ inputs.model_name }}"

        messages:
          - type: SYSTEM
            content: |
              You are a senior AI quality engineer specializing in synthetic dataset generation for RAG evaluation.
              Your task is to extract unambiguous, factually grounded Question-and-Answer evaluation pairs
              from the supplied reference text.

              Guidelines for each test case:
              1. question: Formulate a realistic, domain-specific query that can ONLY be answered from the provided text.
              2. expected: The concise, objective, ground-truth answer based exclusively on the context.
              3. context: Exact excerpt from the input text confirming the answer.
              4. difficulty: Classify difficulty as "easy" (direct lookup), "medium" (multi-sentence synthesis), or "hard" (technical nuances).
              5. topic: Short technical category describing the subject.

              Return ONLY valid JSON matching the specified schema.

          - type: USER
            content: |
              Document Chunk ID: {{ fromJson(item.value).id }}
              Document Title: {{ fromJson(item.value).title }}

              Document Content:
              """
              {{ fromJson(item.value).content }}
              """

              Generate one challenging, realistic evaluation test case for this chunk.

        configuration:
          responseFormat:
            type: JSON
            jsonSchema:
              type: object
              properties:
                id:
                  type: string
                question:
                  type: string
                expected:
                  type: string
                context:
                  type: string
                difficulty:
                  type: string
                  enum:
                    - easy
                    - medium
                    - hard
                topic:
                  type: string
              required:
                - id
                - question
                - expected
                - context
                - difficulty
                - topic

    outputs:
      - id: test_case
        type: JSON
        value: "{{ outputs.synthesize_qa.jsonOutput }}"

  - id: aggregate_dataset
    type: io.kestra.plugin.core.output.OutputValues
    description: Aggregate synthesized evaluation cases and compute summary metrics.
    values:
      total_chunks: "{{ inputs.document_chunks | length }}"
      all_test_cases: "{{ loopOutputs(outputs.generate_synthetic_cases.outputs,
        'test_case') }}"

  - id: calculate_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Extract count and distribution of synthesized cases.
    values:
      total_cases: "{{ outputs.aggregate_dataset.values.all_test_cases | length }}"

  - id: export_golden_dataset
    type: io.kestra.plugin.core.storage.Write
    description: Save the synthesized golden dataset into internal Kestra storage as
      a JSON artifact.
    content: "{{ outputs.aggregate_dataset.values.all_test_cases | json }}"
    extension: .json

  - id: dataset_quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Validate that generated dataset satisfies minimum count requirements.
    condition: "{{ outputs.calculate_metrics.values.total_cases >=
      inputs.min_total_cases }}"
    then:
      - id: log_dataset_summary
        type: io.kestra.plugin.core.log.Log
        description: Log dataset creation summary and artifact link.
        message: |
          ======================================================
          SYNTHETIC GOLDEN DATASET CREATED SUCCESSFULLY
          ======================================================
          Total Source Chunks: {{ outputs.aggregate_dataset.values.total_chunks }}
          Synthesized Test Cases: {{ outputs.calculate_metrics.values.total_cases }}
          Golden Dataset Artifact URI: {{ outputs.export_golden_dataset.uri }}
          Target Threshold Met: {{ outputs.calculate_metrics.values.total_cases }} >= {{ inputs.min_total_cases }}
          ======================================================

      - id: notify_slack_success
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post optional completion summary to Slack on-call or ML channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *Synthetic Golden Dataset Generated* :sparkles:
          *Model:* `{{ inputs.model_name }}`
          *Chunks Processed:* `{{ outputs.aggregate_dataset.values.total_chunks }}`
          *Total Test Cases:* `{{ outputs.calculate_metrics.values.total_cases }}`
          *Artifact:* `{{ outputs.export_golden_dataset.uri }}`
          Ready for automated LLM regression evaluation.
        disabled: true
    else:
      - id: log_insufficient_cases
        type: io.kestra.plugin.core.log.Log
        description: Log warning when generated case count does not satisfy minimum
          threshold.
        message: |
          WARNING: Generated {{ outputs.calculate_metrics.values.total_cases }} test cases,
          which is below the required threshold of {{ inputs.min_total_cases }}.

      - id: fail_insufficient_cases
        type: io.kestra.plugin.core.execution.Fail
        description: Mark flow as failed when synthesized dataset is incomplete.
        errorMessage: |
          Synthetic dataset failed quality gate: only {{ outputs.calculate_metrics.values.total_cases }}
          cases generated, minimum required is {{ inputs.min_total_cases }}.

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert engineering channel if dataset generation crashes.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :rotating_light: *RAG Synthetic Golden Dataset Generation FAILED*
      *Flow:* `{{ flow.id }}`
      *Execution:* `{{ execution.id }}`
      *Error:* {{ error.message }}
    disabled: true

Evaluating Large Language Model (LLM) and Retrieval-Augmented Generation (RAG) pipelines requires a robust "Golden Dataset" composed of realistic questions, ground-truth answers, and reference context chunks. Manually authoring hundreds of domain-specific evaluation pairs is cost-prohibitive, tedious, and difficult to keep in sync with evolving knowledge bases.

This blueprint automates synthetic golden dataset creation by orchestrating LLM-as-a-Generator across raw document chunks using Kestra's native AI plugin with strict JSON Schema outputs, verifying test case coverage, and exporting a versioned golden dataset artifact.

How it works

log_generation_start (io.kestra.plugin.core.log.Log) logs the synthesis run parameters and total input chunks.
generate_synthetic_cases (io.kestra.plugin.core.flow.Loop) iterates through each document chunk in parallel with configurable concurrency.
synthesize_qa (io.kestra.plugin.ai.completion.ChatCompletion) sends each chunk to an OpenAI-compatible LLM endpoint with a strict JSON Schema, enforcing extraction of question, ground-truth expected answer, verbatim context excerpt, difficulty rating, and technical topic.
aggregate_dataset (io.kestra.plugin.core.output.OutputValues) aggregates synthesized test cases across all loop outputs into a single flattened collection.
calculate_metrics (io.kestra.plugin.core.output.OutputValues) measures total generated evaluation cases.
export_golden_dataset (io.kestra.plugin.core.storage.Write) persists the consolidated golden dataset into Kestra internal storage as a downloadable JSON artifact.
dataset_quality_gate (io.kestra.plugin.core.flow.If) verifies that the generated test count meets or exceeds min_total_cases. If successful, it logs the artifact URI and sends an optional Slack notification; otherwise, it triggers fail_insufficient_cases (io.kestra.plugin.core.execution.Fail).
The top-level errors handler captures runtime failures and sends an immediate alert to Slack.

What you get

Automated generation of domain-specific Q&A evaluation datasets from raw documentation.
Strict JSON Schema validation guaranteeing test case structure (id, question, expected, context, difficulty, topic).
Parallel chunk processing with controlled concurrency using Kestra's native Loop.
Golden dataset persisted directly to Kestra internal storage as a versioned JSON artifact.
Minimum coverage quality gate preventing downstream evaluation on incomplete datasets.
Seamless compatibility with downstream evaluation blueprints such as llm-evaluation-regression.
Proactive Slack notifications for successful generation and execution errors.

Who it's for

AI and ML Engineers building and evaluating production RAG pipelines.
Platform teams creating continuous evaluation CI/CD loops for knowledge bases.
Data Engineers responsible for keeping evaluation datasets aligned with new documentation releases.
Teams adopting LLM-as-a-Judge workflows who need automated ground-truth generation.

Why orchestrate this with Kestra

Generating synthetic datasets over large document corpora involves concurrency control, rate-limit handling, structured output validation, state persistence, and error alerting. Writing ad-hoc Python scripts leaves teams with untracked files, unhandled API timeouts, and no audit trail.

Kestra provides declarative orchestration, native concurrency throttling, built-in internal storage management, execution history, and seamless integration between AI generation tasks and notification systems—all defined in readable YAML.

Prerequisites

A running Kestra instance.
An OpenAI-compatible LLM API key (Groq, OpenAI, or self-hosted vLLM).
An optional Slack incoming webhook URL for notifications.

Secrets

GROQ_API_KEY: API key for accessing the LLM endpoint (default uses Groq's OpenAI-compatible API).
SLACK_WEBHOOK_URL: Webhook URL for Slack alerts (optional, enabled by removing disabled: true).
WEBHOOK_KEY: Secret authentication key for the inbound webhook trigger.

Quick start

Configure GROQ_API_KEY (and optionally SLACK_WEBHOOK_URL) in your Kestra namespace.
Provide your document chunks via the document_chunks input, or connect an upstream ingestion task.
Run the workflow manually from the Kestra UI to verify synthesis output.
Download the generated golden_dataset.json artifact from the execution outputs, or pass its URI directly into llm-evaluation-regression.
Enable weekly_dataset_refresh or wire your documentation build pipeline to documentation_update_webhook.

How to extend

Ingest document chunks directly from an AWS S3 bucket, Git repository, or vector database before generation.
Feed the resulting export_golden_dataset.uri directly into the llm-evaluation-regression blueprint via a subflow execution.
Add multi-language synthesis prompts to generate multilingual RAG evaluation sets.
Incorporate hard negative generation (synthesizing unanswerable questions to test hallucination resistance).

RAG Synthetic Golden Dataset Generator for LLM Evaluation

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

RAG Synthetic Golden Dataset Generator for LLM Evaluation

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Use HuggingFace Inference API to classify customer message

Qdrant Vector Index Health and Memory Optimizer

Hugging Face Model Safetensors Security and Serialization Guard

GraphRAG Knowledge Graph from GitHub Docs with Neo4j and Claude

Kubernetes AI-powered incident monitoring, analysis, and alerting

Measure a Presidio PII scrub, then block the load when residual PII survives

More Data Blueprints

Query data from Druid and load it to Redshift

DuckDB Index and Storage Growth Gate

Fetch data from Apache Druid and transform it in Python with Pandas

Query data from Dremio and process it in Polars with a Python script

Druid Segment Availability Gate

Download a Parquet file from Databricks DBFS and process it with Python

New to Kestra?