Local RAG knowledge base with MinIO, Tika, Ollama and Qdrant

id: local-rag-knowledge-base-sync
namespace: company.ai

variables:
  minio_endpoint: http://localhost:9000
  bucket: knowledge-base
  qdrant_url: http://localhost:6333
  qdrant_host: localhost
  qdrant_grpc_port: 6334
  collection: knowledge_base
  ollama_endpoint: http://localhost:11434
  embedding_model: nomic-embed-text
  embedding_dimension: 768

concurrency:
  limit: 1
  behavior: QUEUE

tasks:
  - id: ensure_bucket
    type: io.kestra.plugin.minio.CreateBucket
    endpoint: "{{ vars.minio_endpoint }}"
    accessKeyId: "{{ secret('MINIO_ACCESS_KEY') }}"
    secretKeyId: "{{ secret('MINIO_SECRET_KEY') }}"
    bucket: "{{ vars.bucket }}"

  - id: collection_exists
    type: io.kestra.plugin.core.http.Request
    uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}/exists"
    headers:
      api-key: "{{ secret('QDRANT_API_KEY') }}"

  - id: create_collection_if_missing
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not (outputs.collection_exists.body | jq('.result.exists') | first) }}"
    then:
      - id: create_collection
        type: io.kestra.plugin.core.http.Request
        method: PUT
        uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}"
        headers:
          api-key: "{{ secret('QDRANT_API_KEY') }}"
        contentType: application/json
        body: |
          {"vectors": {"size": {{ vars.embedding_dimension }}, "distance": "Cosine"}}

      - id: index_source_field
        type: io.kestra.plugin.core.http.Request
        description: A keyword index on `source` makes per-document deletes and the
          facet query cheap.
        method: PUT
        uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}/index?wait=true"
        headers:
          api-key: "{{ secret('QDRANT_API_KEY') }}"
        contentType: application/json
        body: |
          {"field_name": "source", "field_schema": "keyword"}

  - id: list_bucket
    type: io.kestra.plugin.minio.List
    endpoint: "{{ vars.minio_endpoint }}"
    accessKeyId: "{{ secret('MINIO_ACCESS_KEY') }}"
    secretKeyId: "{{ secret('MINIO_SECRET_KEY') }}"
    bucket: "{{ vars.bucket }}"
    filter: FILES

  - id: indexed_sources
    type: io.kestra.plugin.core.http.Request
    method: POST
    uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}/facet"
    headers:
      api-key: "{{ secret('QDRANT_API_KEY') }}"
    contentType: application/json
    body: |
      {"key": "source", "limit": 100000, "exact": true}

  - id: plan_sync
    type: io.kestra.plugin.core.output.OutputValues
    description: |
      Compares the bucket with the index. `deleted` lists sources with chunks in Qdrant but no file in the bucket, which the MinIO trigger cannot see.
      `to_index` lists the files the trigger reported as created or updated, plus any file still missing from the index, so a failed run heals on the next one.
    values:
      deleted: "{{ {'indexed': outputs.indexed_sources.body | jq('[.result.hits[] |
        select(.count > 0) | .value]') | first, 'present':
        outputs.list_bucket.objects | jq('[.[].key]') | first} | jq('.indexed -
        .present') | first | toJson }}"
      to_index: "{{ {'objects': outputs.list_bucket.objects, 'changed':
        (trigger.objects ?? []) | jq('[.[].key]') | first, 'indexed':
        outputs.indexed_sources.body | jq('[.result.hits[] | select(.count > 0)
        | .value]') | first} | jq('. as $d | [$d.objects[] | select(.key as $k |
        ($d.changed | index($k)) != null or ($d.indexed | index($k)) == null) |
        {key, lastModified}]') | first | toJson }}"

  - id: purge_deleted_documents
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.plan_sync.values.deleted) | length > 0 }}"
    then:
      - id: delete_orphaned_chunks
        type: io.kestra.plugin.core.http.Request
        method: POST
        uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection
          }}/points/delete?wait=true"
        headers:
          api-key: "{{ secret('QDRANT_API_KEY') }}"
        contentType: application/json
        body: |
          {"filter": {"must": [{"key": "source", "match": {"any": {{ outputs.plan_sync.values.deleted }}}}]}}

      - id: log_purged
        type: io.kestra.plugin.core.log.Log
        message: "Removed chunks of documents no longer in the bucket: {{
          outputs.plan_sync.values.deleted }}"

  - id: index_documents
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.plan_sync.values.to_index }}"
    concurrencyLimit: 1
    tasks:
      - id: download
        type: io.kestra.plugin.minio.Download
        endpoint: "{{ vars.minio_endpoint }}"
        accessKeyId: "{{ secret('MINIO_ACCESS_KEY') }}"
        secretKeyId: "{{ secret('MINIO_SECRET_KEY') }}"
        bucket: "{{ vars.bucket }}"
        key: "{{ fromJson(item.value).key }}"

      - id: extract_text
        type: io.kestra.plugin.tika.Parse
        description: A corrupt file only skips itself, so one bad upload cannot block
          the rest of the bucket.
        allowFailure: true
        from: "{{ outputs.download.uri }}"
        contentType: TEXT
        store: false

      - id: has_text
        type: io.kestra.plugin.core.flow.If
        description: Scanned PDFs without OCR and unreadable files yield no text, and an
          empty document would silently replace the previous version.
        condition: "{{ (outputs.extract_text.result.content ?? '') | trim | length > 0 }}"
        then:
          - id: delete_previous_chunks
            type: io.kestra.plugin.core.http.Request
            description: An updated document is re-indexed from scratch, so its old chunks
              must not linger next to the new ones.
            method: POST
            uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection
              }}/points/delete?wait=true"
            headers:
              api-key: "{{ secret('QDRANT_API_KEY') }}"
            contentType: application/json
            body: |
              {"filter": {"must": [{"key": "source", "match": {"value": {{ fromJson(item.value).key | toJson }}}}]}}

          - id: embed_and_store
            type: io.kestra.plugin.ai.rag.IngestDocument
            retry:
              type: exponential
              interval: PT10S
              maxInterval: PT2M
              maxAttempts: 4
            provider:
              type: io.kestra.plugin.ai.provider.Ollama
              endpoint: "{{ vars.ollama_endpoint }}"
              modelName: "{{ vars.embedding_model }}"
            embeddings:
              type: io.kestra.plugin.ai.embeddings.Qdrant
              host: "{{ vars.qdrant_host }}"
              port: "{{ vars.qdrant_grpc_port }}"
              apiKey: "{{ secret('QDRANT_API_KEY') }}"
              collectionName: "{{ vars.collection }}"
            documentSplitter:
              splitter: RECURSIVE
              maxSegmentSizeInChars: 1000
              maxOverlapSizeInChars: 150
            fromDocuments:
              - content: "{{ outputs.extract_text.result.content }}"
                metadata:
                  source: "{{ fromJson(item.value).key }}"
                  last_modified: "{{ fromJson(item.value).lastModified }}"

          - id: count_stored_chunks
            type: io.kestra.plugin.core.http.Request
            method: POST
            uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}/points/count"
            headers:
              api-key: "{{ secret('QDRANT_API_KEY') }}"
            contentType: application/json
            body: |
              {"exact": true, "filter": {"must": [{"key": "source", "match": {"value": {{ fromJson(item.value).key | toJson }}}}]}}

          - id: assert_document_indexed
            type: io.kestra.plugin.core.execution.Assert
            conditions:
              - "{{ (outputs.count_stored_chunks.body | jq('.result.count') |
                first) > 0 }}"

        else:
          - id: skip_document_without_text
            type: io.kestra.plugin.core.log.Log
            level: WARN
            message: "{{ fromJson(item.value).key }} was not indexed because no text could
              be extracted from it. A scanned PDF needs OCR, and a corrupt file
              needs replacing."

  - id: index_snapshot
    type: io.kestra.plugin.core.http.Request
    method: POST
    uri: "{{ vars.qdrant_url }}/collections/{{ vars.collection }}/facet"
    headers:
      api-key: "{{ secret('QDRANT_API_KEY') }}"
    contentType: application/json
    body: |
      {"key": "source", "limit": 100000, "exact": true}

  - id: publish_manifest
    type: io.kestra.plugin.core.kv.Set
    key: knowledge_base_manifest
    kvType: JSON
    value: |
      {
        "collection": "{{ vars.collection }}",
        "updatedAt": "{{ now() }}",
        "documents": {{ outputs.index_snapshot.body | jq('[.result.hits[] | select(.count > 0) | {source: .value, chunks: .count}]') | first | toJson }}
      }

errors:
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Knowledge base sync failed in {{ errorLogs()[0].taskId ?? 'an unknown
      task' }}: {{ errorLogs()[0].message ?? 'see the execution logs' }}.
      Documents that failed are retried on the next run."

triggers:
  - id: on_document_change
    type: io.kestra.plugin.minio.Trigger
    endpoint: "{{ vars.minio_endpoint }}"
    accessKeyId: "{{ secret('MINIO_ACCESS_KEY') }}"
    secretKeyId: "{{ secret('MINIO_SECRET_KEY') }}"
    bucket: "{{ vars.bucket }}"
    filter: FILES
    on: CREATE_OR_UPDATE
    action: NONE
    interval: PT30S

  - id: hourly_reconciliation
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"

Most RAG demos index a folder once and never look at it again. Documents change, get deleted and sometimes fail to parse, and the index quietly drifts from the source. This flow keeps a Qdrant collection in sync with a MinIO bucket, so dropping, replacing or removing a file is all it takes to update the knowledge base. Text extraction (Apache Tika), embeddings (Ollama) and storage all run on your machine, so documents never leave it.

Pair it with local-rag-knowledge-base-ask, which answers questions over the same collection with a local LLM and refuses the ones your documents do not cover.

This blueprint was created by kruxarth.

How it works

on_document_change (io.kestra.plugin.minio.Trigger) polls the bucket every 30 seconds and fires for files that are new or whose ETag changed (on: CREATE_OR_UPDATE). Files stay in the bucket (action: NONE), because the bucket is the source of truth. hourly_reconciliation (Schedule) runs the same flow every hour without any file event.
ensure_bucket and create_collection_if_missing (If) create the bucket, the Qdrant collection (768 dimensions, cosine) and a keyword index on source on the first run, so there is nothing to set up by hand.
plan_sync compares list_bucket with the sources that have chunks in Qdrant (indexed_sources, a facet query) and outputs two lists:
- deleted: sources with chunks but no file. The MinIO trigger cannot see deletions, so this is how they are caught.
- to_index: files reported by the trigger, plus any file still missing from the index.
purge_deleted_documents (If) removes the chunks of deleted files in one Qdrant call.
index_documents (Loop) processes one file at a time:
- download (io.kestra.plugin.minio.Download) and extract_text (io.kestra.plugin.tika.Parse) turn PDF, DOCX, Markdown and anything else Tika reads into plain text.
- has_text (If) skips files with no text, such as scanned PDFs or corrupt files, with a WARN log, so an empty or broken file never replaces a good version.
- delete_previous_chunks removes the old version of the file, embed_and_store (io.kestra.plugin.ai.rag.IngestDocument) splits it into 1,000-character chunks and embeds them with nomic-embed-text, tagging each chunk with its source key, and assert_document_indexed checks that chunks really landed in Qdrant.
publish_manifest writes the indexed files and their chunk counts to the knowledge_base_manifest KV entry.

What you get

A vector index that follows the bucket: new and replaced files are indexed within about 30 seconds, and deleted files are purged by the next run, at the latest by the hourly one.
Self-healing: a file that failed to index (Ollama down, a crash mid-run) is picked up again by the next run, with no manual restart.
One bad file never blocks the others: it is skipped with a log line that names it.
Retries with exponential backoff on embedding, an ERROR log naming the failing task, and a manifest you can read from the KV Store.

Who it's for

AI practitioners who want a RAG index that stays correct without re-indexing everything.
Platform teams who cannot send internal documents to a hosted embedding API.

Why orchestrate this with Kestra

The trigger keeps per-file state, so only changed files are processed, and concurrency queues runs so two of them never index the same file at once.
Every file is its own loop iteration with its own logs, so you can see exactly which document failed and why.
Retries, scheduling and alerting are declared in YAML instead of being re-implemented in a script.

Prerequisites

MinIO, Qdrant and Ollama, reachable from Kestra:

docker run -d --name minio -p 9000:9000 -p 9001:9001 -e MINIO_ROOT_USER=minioadmin -e MINIO_ROOT_PASSWORD=minioadmin cgr.dev/chainguard/minio server /data --console-address :9001
docker run -d --name qdrant -p 6333:6333 -p 6334:6334 -e QDRANT__SERVICE__API_KEY=change-me qdrant/qdrant
docker run -d --name ollama -p 11434:11434 ollama/ollama
docker exec ollama ollama pull nomic-embed-text

The flow's variables point at localhost. If Kestra itself runs in Docker, replace localhost with host.docker.internal (add extra_hosts: ["host.docker.internal:host-gateway"] to the Kestra service on Linux).

Secrets

MINIO_ACCESS_KEY and MINIO_SECRET_KEY: credentials with read and list access to the bucket, and permission to create it on the first run.
QDRANT_API_KEY: the value of QDRANT__SERVICE__API_KEY (the Qdrant plugin requires one).

Quick start

Start the services above and add the three secrets.
Execute the flow once. It creates the knowledge-base bucket and the knowledge_base collection, and publishes an empty manifest. Until then, the trigger logs that the bucket does not exist.
Upload a few PDF, DOCX or Markdown files to the bucket, for example from the MinIO console at http://localhost:9001. Within 30 seconds an execution indexes them: open KV Store in the company.ai namespace to see knowledge_base_manifest with one entry per file.
Replace a file, then delete another, and watch the manifest follow.

Things to know

Files without extractable text are logged as WARN on every run until they are fixed or removed. To read scanned PDFs, install Tesseract on the worker and set ocrOptions on extract_text.
An updated file is missing from the index for the few seconds between delete_previous_chunks and embed_and_store.
Changing embedding_model means a different vector size: update embedding_dimension, delete the collection and let the next run rebuild it.

How to extend

Watch only part of the bucket with prefix or regexp on the trigger, or run one collection per team.
Send the log_failure message to Slack or email instead of the logs.
Swap Ollama for a hosted embedding provider, or Qdrant for another io.kestra.plugin.ai.embeddings store.

Self-refreshing local RAG knowledge base from a MinIO bucket

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

Things to know

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

Self-refreshing local RAG knowledge base from a MinIO bucket

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

Things to know

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More AI Blueprints

Gate AI Agent Outputs for Leaked Secrets and PII

Trigger the Right Automation with an AI Agent Based on User Intent

Generate AI Blog Posts in Your Brand Voice and Publish to WordPress as Draft

Summarize Text with an AI Agent, Generate Multilingual Summaries in Short, Medium, or Long Format

AI Website Accessibility Auditor

Firewall AI Agent Inputs Against Prompt Injection

New to Kestra?