GraphRAG Docs Knowledge Graph with Neo4j

id: graphrag-docs-to-neo4j-knowledge-graph
namespace: company.team
description: |
  Turn a folder of Markdown docs in a GitHub repository into a Neo4j knowledge
  graph and answer questions from it. New or changed files are detected by their
  Git blob SHA, Claude extracts entities and relationships as JSON, and the rows
  are merged into Neo4j with parameterized Cypher. The run ends by answering a
  question from the graph, with the source documents cited.

triggers:
  - id: on_docs_push
    type: io.kestra.plugin.core.trigger.Webhook
    description: Point a GitHub push webhook here so the graph updates when the docs
      change. Only new or changed files are processed.
    key: "{{ secret('GRAPHRAG_WEBHOOK_KEY') }}"

inputs:
  - id: repo
    type: STRING
    displayName: GitHub repository
    description: Public repository holding the docs, as owner/name.
    defaults: kestra-io/docs
  - id: branch
    type: STRING
    displayName: Branch
    defaults: main
  - id: docs_path
    type: STRING
    displayName: Docs folder
    description: Folder to ingest. Every .md and .mdx file below it is included.
    defaults: src/contents/docs/05.workflow-components
  - id: max_files
    type: INT
    displayName: Max files per run
    description: Caps how many new or changed files are sent to the model in one
      run. The rest are picked up by the next run.
    defaults: 10
  - id: question
    type: STRING
    displayName: Question
    description: Answered from the graph at the end of the run.
    defaults: Which tasks can a flow use to wait for a condition or for a person
      before it continues?

variables:
  claude_model: claude-opus-5-5

tasks:
  - id: ensure_constraint
    type: io.kestra.plugin.neo4j.Query
    description: Make entity keys unique, so concurrent or repeated runs can't
      create duplicate nodes. Safe to run every time.
    url: "{{ secret('NEO4J_URL') }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: CREATE CONSTRAINT entity_key IF NOT EXISTS FOR (e:Entity) REQUIRE e.key
      IS UNIQUE

  - id: list_files
    type: io.kestra.plugin.core.http.Request
    description: List every file under the docs folder, with its Git blob SHA, in a
      single GitHub API call.
    uri: "https://api.github.com/repos/{{ inputs.repo }}/git/trees/{{ inputs.branch
      }}:{{ inputs.docs_path }}?recursive=1"
    headers:
      Accept: application/vnd.github+json

  - id: known_docs
    type: io.kestra.plugin.neo4j.Query
    description: Read which files are already in the graph and at which SHA. The
      graph itself is the record of what was ingested.
    url: "{{ secret('NEO4J_URL') }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: |
      MATCH (d:Document {repo: {{ inputs.repo | toJson }}})
      RETURN d {.path, .sha} AS doc
    storeType: FETCH

  - id: select_docs
    type: io.kestra.plugin.scripts.python.Script
    description: Pick the Markdown files that are new or whose SHA changed, up to
      `max_files`.
    dependencies:
      - kestra
    inputFiles:
      tree.json: "{{ outputs.list_files.body }}"
      known.json: "{{ outputs.known_docs.rows | toJson }}"
    script: |
      import json
      from kestra import Kestra

      prefix = {{ inputs.docs_path | toJson }}
      max_files = int({{ inputs.max_files }})

      with open("tree.json") as f:
          tree = json.load(f)
      with open("known.json") as f:
          known = {row["path"]: row["sha"] for row in json.load(f)}

      docs = [
          entry for entry in tree.get("tree", [])
          if entry["type"] == "blob" and entry["path"].endswith((".md", ".mdx"))
      ]
      changed = [d for d in docs if known.get(f"{prefix}/{d['path']}") != d["sha"]]
      selected = changed[:max_files]

      Kestra.outputs({
          "paths": [d["path"] for d in selected],
          "shas": {d["path"]: d["sha"] for d in selected},
          "total_docs": len(docs),
          "changed_docs": len(changed),
          "selected_docs": len(selected),
      })

  - id: ingest_changed
    type: io.kestra.plugin.core.flow.If
    description: Skip extraction entirely when nothing changed, so an unchanged
      repository costs no model calls.
    condition: "{{ outputs.select_docs.vars.selected_docs > 0 }}"
    then:
      - id: ingest_docs
        type: io.kestra.plugin.core.flow.Loop
        description: Process each new or changed file. One at a time, so merges never
          race each other.
        values: "{{ outputs.select_docs.vars.paths }}"
        concurrencyLimit: 1
        tasks:
          - id: download_doc
            type: io.kestra.plugin.core.http.Download
            description: Fetch the raw Markdown. Raw downloads don't count against the
              GitHub API rate limit.
            uri: "https://raw.githubusercontent.com/{{ inputs.repo }}/{{ inputs.branch }}/{{
              inputs.docs_path }}/{{ item.value }}"

          - id: extract
            type: io.kestra.plugin.ai.completion.ChatCompletion
            description: Extract entities and relationships from the document as JSON, with
              a short evidence quote for each relationship.
            provider:
              type: io.kestra.plugin.ai.provider.Anthropic
              apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
              modelName: "{{ vars.claude_model }}"
            configuration:
              maxToken: 16000
            messages:
              - type: SYSTEM
                content: |
                  You build a knowledge graph from technical documentation. Respond with JSON only: no code fences, no commentary.
              - type: USER
                content: |
                  Extract the important entities and the relationships between them from the document below.

                  Return exactly this shape:
                  {"entities": [{"name": "...", "type": "...", "description": "..."}],
                   "relationships": [{"source": "...", "target": "...", "type": "...", "evidence": "..."}]}

                  Rules:
                  - Entity types: Concept, Task, Trigger, Plugin, Property, Feature, Tool or Other.
                  - Use one canonical, singular name per entity and reuse it exactly in relationships, for example "Pause" rather than "the Pause task".
                  - Relationship types: USES, PART_OF, CONFIGURES, TRIGGERS, DEPENDS_ON, ALTERNATIVE_TO or RELATED_TO.
                  - Every relationship source and target must also appear in entities.
                  - evidence is a short quote of at most 20 words from the document that supports the relationship.
                  - Only include facts stated in the document.

                  Document path: {{ inputs.docs_path }}/{{ item.value }}

                  {{ read(outputs.download_doc.uri) }}

          - id: entity_rows
            type: io.kestra.plugin.core.storage.Write
            description: The entities as a JSON array, tolerating a model reply wrapped in
              code fences.
            extension: .json
            content: "{{ outputs.extract.textOutput | replace({'```json': ''}) |
              replace({'```': ''}) | jq('.entities // []') | first | toJson }}"

          - id: relationship_rows
            type: io.kestra.plugin.core.storage.Write
            description: The relationships as a JSON array.
            extension: .json
            content: "{{ outputs.extract.textOutput | replace({'```json': ''}) |
              replace({'```': ''}) | jq('.relationships // []') | first | toJson
              }}"

          - id: entities_ion
            type: io.kestra.plugin.serdes.json.JsonToIon
            description: Convert the entity array to Ion rows, the format `neo4j.Batch`
              reads.
            from: "{{ outputs.entity_rows.uri }}"
            newLine: false

          - id: relationships_ion
            type: io.kestra.plugin.serdes.json.JsonToIon
            description: Convert the relationship array to Ion rows.
            from: "{{ outputs.relationship_rows.uri }}"
            newLine: false

          - id: clear_previous_facts
            type: io.kestra.plugin.neo4j.Query
            description: Remove the relationships and mentions this document produced last
              time, so a re-ingested file replaces its old facts instead of
              piling up.
            url: "{{ secret('NEO4J_URL') }}"
            username: "{{ secret('NEO4J_USERNAME') }}"
            password: "{{ secret('NEO4J_PASSWORD') }}"
            query: |
              OPTIONAL MATCH ()-[r:RELATES_TO {source_doc: {{ (inputs.docs_path ~ '/' ~ item.value) | toJson }}}]->()
              DELETE r
              WITH count(*) AS removed
              OPTIONAL MATCH (:Entity)-[m:MENTIONED_IN]->(:Document {repo: {{ inputs.repo | toJson }}, path: {{ (inputs.docs_path ~ '/' ~ item.value) | toJson }}})
              DELETE m

          - id: load_entities
            type: io.kestra.plugin.neo4j.Batch
            description: Merge entities on a normalized key and link each one to the
              document that mentions it. Values arrive as `$props`, never as
              Cypher text.
            url: "{{ secret('NEO4J_URL') }}"
            username: "{{ secret('NEO4J_USERNAME') }}"
            password: "{{ secret('NEO4J_PASSWORD') }}"
            from: "{{ outputs.entities_ion.uri }}"
            query: |
              UNWIND $props AS row
              WITH row WHERE row.name IS NOT NULL AND trim(row.name) <> ''
              MERGE (e:Entity {key: toLower(trim(row.name))})
                ON CREATE SET e.name = trim(row.name)
              SET e.type = coalesce(row.type, e.type),
                  e.description = coalesce(row.description, e.description)
              MERGE (d:Document {repo: {{ inputs.repo | toJson }}, path: {{ (inputs.docs_path ~ '/' ~ item.value) | toJson }}})
              MERGE (e)-[:MENTIONED_IN]->(d)

          - id: load_relationships
            type: io.kestra.plugin.neo4j.Batch
            description: Merge one edge per relationship per source document, with its type
              and evidence quote, so every fact keeps its own citation.
            url: "{{ secret('NEO4J_URL') }}"
            username: "{{ secret('NEO4J_USERNAME') }}"
            password: "{{ secret('NEO4J_PASSWORD') }}"
            from: "{{ outputs.relationships_ion.uri }}"
            query: |
              UNWIND $props AS row
              WITH row WHERE row.source IS NOT NULL AND row.target IS NOT NULL
              MERGE (a:Entity {key: toLower(trim(row.source))})
                ON CREATE SET a.name = trim(row.source)
              MERGE (b:Entity {key: toLower(trim(row.target))})
                ON CREATE SET b.name = trim(row.target)
              MERGE (a)-[r:RELATES_TO {type: toUpper(trim(coalesce(row.type, 'RELATED_TO'))), source_doc: {{ (inputs.docs_path ~ '/' ~ item.value) | toJson }}}]->(b)
              SET r.evidence = row.evidence

          - id: mark_ingested
            type: io.kestra.plugin.neo4j.Query
            description: Record the file's SHA on its Document node only after both loads
              succeed, so a failed file is retried on the next run.
            url: "{{ secret('NEO4J_URL') }}"
            username: "{{ secret('NEO4J_USERNAME') }}"
            password: "{{ secret('NEO4J_PASSWORD') }}"
            query: |
              MERGE (d:Document {repo: {{ inputs.repo | toJson }}, path: {{ (inputs.docs_path ~ '/' ~ item.value) | toJson }}})
              SET d.sha = {{ outputs.select_docs.vars.shas[item.value] | toJson }},
                  d.url = {{ ('https://github.com/' ~ inputs.repo ~ '/blob/' ~ inputs.branch ~ '/' ~ inputs.docs_path ~ '/' ~ item.value) | toJson }},
                  d.ingested_at = datetime()

  - id: graph_stats
    type: io.kestra.plugin.neo4j.Query
    description: Count what the graph holds after this run.
    url: "{{ secret('NEO4J_URL') }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: |
      MATCH (d:Document {repo: {{ inputs.repo | toJson }}})
      WITH count(d) AS documents
      MATCH (e:Entity)
      WITH documents, count(e) AS entities
      OPTIONAL MATCH ()-[r:RELATES_TO]->()
      WITH documents, entities, count(r) AS relationships
      RETURN {documents: documents, entities: entities, relationships: relationships} AS stats
    storeType: FETCHONE

  - id: question_terms
    type: io.kestra.plugin.ai.completion.ChatCompletion
    description: Pull the key terms out of the question, which become the entry
      points into the graph.
    provider:
      type: io.kestra.plugin.ai.provider.Anthropic
      apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
      modelName: "{{ vars.claude_model }}"
    configuration:
      maxToken: 4000
    messages:
      - type: SYSTEM
        content: Respond with a JSON array of strings only, no code fences.
      - type: USER
        content: |
          List up to 6 short key terms from this question that are likely to be entity names in a knowledge graph about technical documentation. Use singular forms.

          Question: {{ inputs.question }}

  - id: retrieve_subgraph
    type: io.kestra.plugin.neo4j.Query
    description: Find entities matching the terms and return their relationships
      with evidence and source documents. Terms are slugified first, so only
      letters, digits and hyphens ever reach the Cypher text.
    url: "{{ secret('NEO4J_URL') }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    query: |
      WITH [{% for term in fromJson(outputs.question_terms.textOutput | replace({'```json': ''}) | replace({'```': ''})) %}"{{ term | slugify }}", {% endfor %}"-no-term-"] AS terms
      MATCH (e:Entity)
      WHERE any(t IN terms WHERE replace(e.key, ' ', '-') CONTAINS t)
      MATCH (e)-[r:RELATES_TO]-(n:Entity)
      RETURN {entity: e.name, relationship: r.type, neighbor: n.name,
              outgoing: startNode(r) = e, evidence: r.evidence, source: r.source_doc} AS fact
      LIMIT 80
    storeType: FETCH

  - id: answer
    type: io.kestra.plugin.ai.completion.ChatCompletion
    description: Answer the question using only the retrieved graph facts, citing
      the source documents.
    provider:
      type: io.kestra.plugin.ai.provider.Anthropic
      apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
      modelName: "{{ vars.claude_model }}"
    configuration:
      maxToken: 8000
    messages:
      - type: SYSTEM
        content: |
          You answer questions using only the knowledge graph facts provided. Each fact has an evidence quote and a source document. Cite sources by path. If the facts don't answer the question, say what is missing instead of guessing.
      - type: USER
        content: |
          Question: {{ inputs.question }}

          Graph facts (entity, relationship, neighbor, direction, evidence, source):
          {{ outputs.retrieve_subgraph.rows | toJson }}

outputs:
  - id: answer
    type: STRING
    displayName: Answer
    description: The answer to the question, grounded in graph facts.
    value: "{{ outputs.answer.textOutput }}"
  - id: graph_size
    type: JSON
    displayName: Graph size
    description: Documents, entities and relationships in the graph after this run.
    value: "{{ outputs.graph_stats.row }}"
  - id: ingestion
    type: JSON
    displayName: Ingestion summary
    description: How many files were found, changed and processed in this run.
    value: |
      {"total_docs": {{ outputs.select_docs.vars.total_docs }}, "changed_docs": {{ outputs.select_docs.vars.changed_docs }}, "processed_docs": {{ outputs.select_docs.vars.selected_docs }}}

Vector search finds passages that sound similar to a question. It struggles with questions about how things connect, such as which tasks depend on a feature, or which alternatives exist for a concept. GraphRAG answers those by storing entities and their relationships in a graph and walking it. The hard part is not the query. It's keeping the graph built and current as the documentation changes. This blueprint automates that pipeline. It reads a folder of Markdown docs from GitHub, uses Claude to extract entities and relationships, merges them into Neo4j with safe, parameterized Cypher, re-processes only files that changed, and answers a question from the graph with the source documents cited.

How it works

ensure_constraint (io.kestra.plugin.neo4j.Query) creates a uniqueness constraint on entity keys, so repeated or overlapping runs never create duplicate nodes.
list_files (io.kestra.plugin.core.http.Request) lists every file under docs_path with one call to GitHub's git trees API. Each file comes with its Git blob SHA, so no repository clone is needed, even for very large repositories.
known_docs (io.kestra.plugin.neo4j.Query) reads the SHA stored on each Document node. The graph itself is the record of what was ingested, so there is no separate state to keep in sync.
select_docs (io.kestra.plugin.scripts.python.Script) keeps the .md and .mdx files that are new or whose SHA changed, capped at max_files per run. ingest_changed (io.kestra.plugin.core.flow.If) skips extraction entirely when nothing changed, so an unchanged repository costs no model calls.
ingest_docs (io.kestra.plugin.core.flow.Loop) processes each selected file, one at a time:
- download_doc (io.kestra.plugin.core.http.Download) fetches the raw Markdown. Raw downloads don't count against the GitHub API rate limit.
- extract (io.kestra.plugin.ai.completion.ChatCompletion with Claude) returns entities with a type and description, and relationships with a type and a short evidence quote, as JSON.
- entity_rows and relationship_rows (io.kestra.plugin.core.storage.Write) turn the reply into JSON arrays, tolerating code fences, and io.kestra.plugin.serdes.json.JsonToIon converts them to rows.
- clear_previous_facts removes the edges and mentions this document produced last time, so a changed file replaces its old facts instead of piling up.
- load_entities and load_relationships (io.kestra.plugin.neo4j.Batch) merge the rows with UNWIND $props. Extracted text always arrives as query parameters and never becomes Cypher. Entities merge on a lower-cased key, and each relationship is stored once per source document, so every fact keeps its own citation.
- mark_ingested records the file's SHA only after both loads succeed, so a failed file is retried on the next run.
graph_stats counts the documents, entities and relationships in the graph.
question_terms asks Claude for the key terms in question. retrieve_subgraph slugifies each term, so only letters, digits and hyphens reach the Cypher text, then returns the matching entities' relationships with evidence and source documents.
answer (Claude) answers the question using only those graph facts and cites the source documents. If the graph doesn't hold the answer, it says what's missing instead of guessing.
The on_docs_push trigger (io.kestra.plugin.core.trigger.Webhook) lets a GitHub push webhook start an incremental update whenever the docs change.

What you get

A Neo4j knowledge graph of the entities in your docs and how they relate, with every relationship carrying an evidence quote and its source document.
Incremental updates driven by Git SHAs. Unchanged files are never sent to the model again, and changed files replace their previous facts.
Injection-safe loading. Model output reaches Neo4j only as $props parameters, and question terms are slugified before they touch Cypher.
A grounded answer to a question, with citations, as the answer flow output, plus graph_size and ingestion outputs for monitoring.
Bounded cost per run through max_files, so large documentation sets are ingested over several runs.

Who it's for

Teams building GraphRAG or knowledge-graph search over product docs, runbooks or internal wikis kept in Git.
Developer relations and documentation teams who want to see how concepts in their docs connect.
Data and AI engineers who need a repeatable, observable pipeline to keep a graph in sync with a source.

Why orchestrate this with Kestra

Extracting entities with an LLM is one call. Running it as a pipeline is what's hard: deciding what changed, keeping each document's facts replaceable, loading safely, retrying a failed file without reprocessing everything, and capping cost. Kestra handles the change detection and per-file Loop, keeps secrets out of the flow, runs Neo4j loads as parameterized batches, records each run's inputs and outputs, and lets a GitHub webhook or a schedule keep the graph current. If one file fails, the next run picks it up automatically because its SHA was never recorded.

Prerequisites

A Neo4j 5 database reachable from Kestra, for example docker run -p 7474:7474 -p 7687:7687 -e NEO4J_AUTH=neo4j/<password> neo4j:5, or Neo4j AuraDB.
An Anthropic API key, created inside a workspace in the Claude Console. Kestra can't send the anthropic-workspace-id header that keys without a workspace require.
A public GitHub repository with Markdown docs. No GitHub token is needed. The flow makes one GitHub API call per run, well inside the unauthenticated limit of 60 per hour.
A Kestra worker with Docker available, for the Python selection step.

Secrets

NEO4J_URL: Bolt URL of the database, for example bolt://neo4j:7687 or neo4j+s://<id>.databases.neo4j.io for AuraDB.
NEO4J_USERNAME: Neo4j user.
NEO4J_PASSWORD: Neo4j password.
ANTHROPIC_API_KEY: Anthropic API key used for extraction, question terms and the answer.
GRAPHRAG_WEBHOOK_KEY: secret key in the webhook URL that GitHub calls.

Inputs

repo (STRING, default kestra-io/docs): public repository as owner/name.
branch (STRING, default main): branch to read.
docs_path (STRING, default src/contents/docs/05.workflow-components): folder to ingest. Every .md and .mdx file below it is included.
max_files (INT, default 10): new or changed files sent to the model per run. Set it to 0 to only answer a question from the existing graph.
question (STRING, default Which tasks can a flow use to wait for a condition or for a person before it continues?): answered from the graph at the end of the run.

Outputs

{{ outputs.answer }} (flow output) / {{ outputs.answer.textOutput }}: the cited answer.
{{ outputs.graph_size }} (flow output) / {{ outputs.graph_stats.row }}: documents, entities and relationships in the graph.
{{ outputs.ingestion }} (flow output): total, changed and processed files for the run.
{{ outputs.select_docs.vars.paths }}: the files processed in this run.
{{ outputs.retrieve_subgraph.rows }}: the graph facts used for the answer, each with evidence and source.

Quick start

Start Neo4j and create the secrets above.
Save the flow and run it with the defaults. The first run ingests 10 files from the Kestra workflow components docs and answers the default question.
Run it again to ingest the next batch. Files already in the graph are skipped.
Open the Neo4j Browser at http://localhost:7474 and run MATCH (e:Entity)-[r:RELATES_TO]->(n) RETURN e, r, n LIMIT 100 to explore the graph.
To keep the graph current, add a GitHub webhook on the docs repository for push events, pointing at /api/v1/main/executions/webhook/company.team/graphrag-docs-to-neo4j-knowledge-graph/<GRAPHRAG_WEBHOOK_KEY>.

How to extend

Point repo and docs_path at your own docs, runbooks or architecture decision records.
Ingest websites with io.kestra.plugin.scrapy.CLI, PDFs with io.kestra.plugin.tika.Parse, or use Apify for managed scraping, then feed the text to extract.
Control cost by setting claude_model to claude-sonnet-5-5 or claude-haiku-4-5 for extraction, or swap in another provider by changing provider.
Run the extraction loop with a higher concurrencyLimit once the uniqueness constraint is in place, and add a retry on the load tasks.
Add a scheduled io.kestra.plugin.core.trigger.Schedule as a fallback to the webhook, or a cleanup step that removes Document nodes for files deleted from the repository.
Use FalkorDB or Memgraph instead of Neo4j through a script task, keeping the same JSON-to-MERGE design.

GraphRAG Knowledge Graph from GitHub Docs with Neo4j and Claude

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

GraphRAG Knowledge Graph from GitHub Docs with Neo4j and Claude

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More AI Blueprints

Qdrant Vector Index Health and Memory Optimizer

Hugging Face Model Safetensors Security and Serialization Guard

Extract, Transform, and Store HackerNews Top Stories with AI-Generated Daily Digest

Use HuggingFace Inference API to classify customer message

RAG Synthetic Golden Dataset Generator for LLM Evaluation

Kubernetes AI-powered incident monitoring, analysis, and alerting

More Data Blueprints

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

Upsert a DocumentDB record from an external source

DuckDB Index and Storage Growth Gate

Report the four DORA metrics per service every week and alert when delivery degrades

DocumentDB analytics report with Slack digest

Ingest a public JSON API into DocumentDB

New to Kestra?