Getting an LLM to recognise an electricity bill as an electricity bill is the easy part by now. The difficult part starts immediately afterwards. Should the sender be stored as “Stadtwerke Hogwarts”, “Stadtwerke Hogwarts GmbH” or “SWM”? Is the document a “Rechnung”, an “Energierechnung” or a “Stromabrechnung”? And what happens when the correct correspondent already exists in Paperless under a slightly different spelling?

That is where my previous experiments with ready-made Paperless extensions fell short. Classification was usually correct in substance, but the archive became less tidy over time. An LLM is very good at inventing a plausible name. It has no inherent reason to reuse exactly the name I created in Paperless three years ago.

My goal was therefore not merely “tag documents with AI”. I wanted an agent that fits into an existing taxonomy:

• Existing correspondents, document types and tags take precedence.
• New correspondents and document types are exceptions, not normal output.
• The model makes suggestions, but application code decides what is valid.
• Only genuinely new entries require manual confirmation.
• Poor OCR is repaired selectively with a vision model instead of sending every document through vision.
• An Ollama outage or a lost webhook must never make a document disappear.

The result is a small AI inbox agent for paperless-ngx. I run the orchestration in Windmill, the document analysis as a LlamaIndex workflow and the models locally through Ollama.

Windmill itself is not the interesting part. The same architecture can run as a normal Python service with SQLite, a cron job or a small webhook endpoint. I will return to that later.

What happens automatically in the end

A new document arrives in Paperless as usual. A Paperless workflow adds the inbox tag and calls a Windmill webhook. From there, processing runs mostly without intervention:

1. Check whether Ollama is reachable.
2. Claim the document against parallel processing.
3. Check whether the existing OCR text is usable.
4. If necessary, render the pages and read them again with a vision model.
5. Classify the document against the correspondents, document types and tags that already exist.
6. Extract title and document date in a separate step.
7. Apply existing assignments directly.
8. Only if a new correspondent or document type appears necessary do I receive a Telegram message with “Create” and “Reject” choices.
9. Update the document through the Paperless API, remove inbox and add ai-processed.

flowchart TD
    A[Neues Paperless-Dokument<br/>mit Tag inbox] --> B[Paperless-Webhook]
    B --> C[Windmill-Queue<br/>nur als Beschleuniger]
    S[Schedule alle 30 Minuten] --> D[Ollama prüfen und Kandidaten suchen]
    C --> D
    D -->|Ollama nicht erreichbar| Z[Sauber beenden<br/>später erneut versuchen]
    D -->|Dokumente gefunden| E[Dokumente claimen]
    E --> F[OCR-Qualität prüfen]
    F -->|OCR brauchbar| G[Klassifikation]
    F -->|OCR unbrauchbar| V[PDF-Seiten rendern<br/>Vision-OCR]
    V --> G
    G --> H[Titel und Datum extrahieren]
    H --> I{Neuer Korrespondent<br/>oder Dokumenttyp?}
    I -->|Nein| J[Paperless aktualisieren]
    I -->|Ja| K[Telegram-Freigabe]
    K --> J
    J --> L[inbox entfernen<br/>ai-processed setzen]

The central design decision is already visible in the first box: the inbox tag in Paperless is the source of truth. The Windmill queue exists only to process new documents immediately. Even if a webhook is lost or a queue variable gets corrupted, the scheduled run still rediscovers every document carrying inbox.

Why I do not simply use paperless-ai

There are already complete projects such as paperless-ai and Zettelrobbe, formerly paperless-ai-next. Both are far more complete applications than my flow: they provide browser-based setup, several model providers, history and manual processing. Zettelrobbe has also expanded OCR and vision support substantially. Anyone looking for the fastest route to an installable all-in-one solution should look at those projects first.

As of July 2026, however, the original paperless-ai is no longer actively maintained. Zettelrobbe is the evolved alternative and supports both existing metadata as prompt context and options that restrict output to existing values.

My issue is therefore not that these projects ignore existing correspondents altogether. The difference is the direction in which classification works.

Their analysis schemas let the model return names:

{
  "correspondent": "Stadtwerke Hogwarts GmbH",
  "document_type": "Stromrechnung",
  "tags": ["Energie", "Haus"],
  "document_date": "2026-07-14"
}

The application then tries to find those names again in Paperless. In the current implementation, correspondents and document types are ultimately matched through an exact case-insensitive name comparison. If no matching entry is found and output is not restricted strictly to existing values, a new entry can be created.

That is convenient, but it shifts authority over the taxonomy towards the model. In a fresh archive that can work well. In an archive that has grown for years, legal suffixes, abbreviations, hyphens and differences in granularity quickly create duplicates.

My workflow reverses that direction. The model receives stable IDs in addition to names and has to choose from those IDs first:

{
  "correspondent_id": 17,
  "new_correspondent": null,
  "document_type_id": 4,
  "new_document_type": null,
  "tag_ids": [8, 21],
  "new_tags": [],
  "reasoning": "Absender und Inhalt entsprechen dem bestehenden Korrespondenten und dem Dokumenttyp Rechnung."
}

Only if no existing entry genuinely matches may it propose a new name as a separate field:

{
  "correspondent_id": null,
  "new_correspondent": "Solarwerk Bodensee",
  "document_type_id": 4,
  "new_document_type": null,
  "tag_ids": [8],
  "new_tags": ["Photovoltaik"],
  "reasoning": "Der Absender kommt in der vorhandenen Liste nicht vor; Rechnung ist bereits als Dokumenttyp vorhanden."
}

That second case is not executed automatically. It is sent to Telegram for approval.

A concrete duplicate example

Assume Paperless already contains:

[
  {"id": 17, "name": "Stadtwerke Hogwarts"},
  {"id": 23, "name": "Stadt Hogwarts"},
  {"id": 31, "name": "Energieversorgung Mittelrhein"}
]

The letterhead, however, says:

Stadtwerke Hogwarts GmbH
Max-Strom-Strasse 1
12345 Hogwarts

A model generating names freely will very likely return Stadtwerke Hogwarts GmbH. Semantically that is correct, but it is not an exact match for Stadtwerke Hogwarts. My prompt therefore states explicitly that legal forms, address additions and spelling variants do not justify a new correspondent. Because the model can return ID 17, it does not have to reproduce the existing name exactly.

The validation after the model call matters even more. An ID that does not exist in the supplied list is discarded. If the model returns both a valid ID and a new name, the ID always wins. The decisive rule therefore exists not only in the prompt but also in code.

Comparing the approaches

Property	paperless-ai	Zettelrobbe, formerly paperless-ai-next	My workflow
Primary goal	Complete application with UI, classification and RAG	Evolved complete application with UI and better OCR	Controlled automation of my inbox process
Model providers	Ollama and several compatible APIs	Ollama, OpenAI-compatible APIs, Mistral OCR and others	Ollama, easily replaceable through LlamaIndex
Analysis	Essentially one combined analysis request	Essentially one combined analysis request	Separate steps for OCR check, vision, classification and metadata
Existing values	Optional context or restriction	Optional context or restriction	Always the primary selection space
Model output	Names	Names	Existing IDs plus separate proposals for new names
Duplicate handling	Depends on prompt, restrictions and name matching	Depends on prompt, restrictions and name matching	ID validation in code, new values only as an exception
New correspondents and types	Can be created automatically depending on configuration	Can be created automatically depending on configuration	Only after explicit approval
Existing Paperless assignment	No explicit stability anchor in the Ollama prompt inspected	No explicit stability anchor in the Ollama prompt inspected	Supplied to the model and retained when the assignment is defensible
OCR fallback	No dedicated multistage quality decision in the Ollama path inspected	Extensive local and external vision OCR	Vision only after a negative OCR quality check
Interface	Own web UI	Own web UI	Windmill for operations, Telegram for exceptions
RAG and document chat	Yes	Deliberately removed	Not part of this flow
Maintenance	Currently not actively maintained	Actively developed	Tailored to my use case

This is not a general benchmark and not a judgement about which project is better for everyone. For my objective, though, the distinction is fairly clear. I do not want another document-management application beside Paperless and I do not need chat over my archive. I want the archive I already have to be maintained reliably. For that, a restrictive ID-based workflow works considerably better for me.

Paperless-ngx itself now also includes optional AI features for suggestions and document chat. Those suggestions are deliberately requested per document, however. My flow solves a different problem: automatic, fault-tolerant processing with its own approval logic.

Two triggers, but only one source of truth

The fastest path starts in Paperless itself. Under Settings → Workflows, a workflow can be created with the “Document added” trigger and a webhook action. Current Paperless versions expose values including {{doc_id}} and {{doc_url}} there.

My handler deliberately accepts both. That also keeps it compatible with older installations or configurations where only the document URL arrived reliably in the payload:

import * as wmill from "windmill-client";

const QUEUE_VAR = "f/paperless/state/pending_docs";
const FLOW_PATH = "f/paperless/process_inbox";

async function readQueue(): Promise<number[]> {
  try {
    const raw = (await wmill.getVariable(QUEUE_VAR)) as string;
    const parsed = JSON.parse(raw);
    return Array.isArray(parsed)
      ? parsed.filter((value) => Number.isInteger(value))
      : [];
  } catch {
    return [];
  }
}

function extractDocId(args: any): number | null {
  const candidate =
    args?.doc_id ??
    args?.document_id ??
    args?.id ??
    args?.document?.id ??
    args?.body?.doc_id;

  const directId = Number(candidate);
  if (Number.isInteger(directId) && directId > 0) {
    return directId;
  }

  const docUrl = args?.doc_url ?? args?.url ?? args?.body?.doc_url;
  if (typeof docUrl === "string") {
    const match = docUrl.match(/\/documents\/(\d+)\/?/);
    if (match) return Number(match[1]);
  }

  return null;
}

export async function main(
  doc_url?: string,
  doc_id?: number,
  body?: unknown,
): Promise<{ queued: boolean; doc_id?: number; note?: string }> {
  const docId = extractDocId({ doc_url, doc_id, body });
  if (docId === null) {
    return { queued: false, note: "keine Dokument-ID im Payload gefunden" };
  }

  const queue = await readQueue();
  if (!queue.includes(docId)) {
    queue.push(docId);
    await wmill.setVariable(QUEUE_VAR, JSON.stringify(queue));
  }

  await wmill.runFlowAsync(FLOW_PATH, {});
  return { queued: true, doc_id: docId };
}

The webhook performs no analysis itself. It merely writes the ID into a small queue and starts the actual flow asynchronously. That keeps the response fast and avoids holding an HTTP request open while a local model processes several pages.

A schedule runs every 30 minutes as the second trigger:

summary: Paperless inbox verarbeiten
args: {}
cron_version: v2
enabled: true
is_flow: true
no_flow_overlap: true
schedule: 0 */30 * * * *
script_path: f/paperless/process_inbox
timezone: Europe/Zurich

The schedule is not a second independent processing path. Both triggers start the same flow, which combines two sources of candidates:

• IDs from the webhook queue
• every document that currently carries the inbox tag in Paperless

The queue reduces latency. The tag prevents data loss.

That principle is intentionally boring. A queue can lose an entry. A webhook can fail during deployment. A local Ollama machine can be asleep. As long as the document still carries inbox in Paperless, the next run can see it again.

Checking Ollama and claiming documents

Before loading a document, the flow checks /api/tags on the Ollama server. If Ollama is unreachable, that does not trigger an alarm. The step simply returns a normal result such as:

{
  "ok": false,
  "reason": "ollama unreachable: connection refused"
}

Ollama downtime is not an exceptional condition in my homelab. It only means: not now, try again on the next schedule.

The flow then combines queued and inbox documents and removes IDs that have already been claimed:

const QUEUE_VAR = "f/paperless/state/pending_docs";
const CLAIMED_VAR = "f/paperless/state/claimed_docs";
const CLAIM_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const INBOX_TAG = "inbox";

type PrereqResult =
  | { ok: true; doc_ids: number[] }
  | { ok: false; reason: string };

export async function main(): Promise<PrereqResult> {
  const ollama = (await wmill.getResource(
    "f/paperless/config/ollama",
  )) as { base_url: string };

  try {
    const response = await fetch(`${ollama.base_url}/api/tags`, {
      signal: AbortSignal.timeout(5000),
    });
    if (!response.ok) {
      return { ok: false, reason: `ollama http ${response.status}` };
    }
  } catch (error: any) {
    return {
      ok: false,
      reason: `ollama unreachable: ${error?.message ?? error}`,
    };
  }

  const paperless = (await wmill.getResource(
    "f/paperless/config/paperless",
  )) as { base_url: string; api_token: string };

  const queued: number[] = (
    await readJsonVar(QUEUE_VAR, [])
  ).filter((value: unknown) => Number.isInteger(value));

  const tags = await paperlessGet(
    paperless,
    `/api/tags/?name__iexact=${encodeURIComponent(INBOX_TAG)}`,
  );
  const inboxTag = (tags.results ?? []).find(
    (tag: any) => tag.name.toLowerCase() === INBOX_TAG,
  );

  const inboxDocIds = inboxTag
    ? (
        await paperlessGet(
          paperless,
          `/api/documents/?tags__id__all=${inboxTag.id}&page_size=1000&fields=id`,
        )
      ).results.map((document: any) => document.id)
    : [];

  const claimedRaw: Record<string, string> = await readJsonVar(
    CLAIMED_VAR,
    {},
  );
  const now = Date.now();
  const activeClaims: Record<string, string> = {};

  for (const [id, timestamp] of Object.entries(claimedRaw)) {
    const claimedAt = Date.parse(timestamp);
    if (!Number.isNaN(claimedAt) && now - claimedAt < CLAIM_TTL_MS) {
      activeClaims[id] = timestamp;
    }
  }

  const candidates = [...new Set([...queued, ...inboxDocIds])]
    .filter((id) => !(String(id) in activeClaims))
    .sort((left, right) => left - right);

  if (candidates.length === 0) {
    await wmill.setVariable(CLAIMED_VAR, JSON.stringify(activeClaims));
    await wmill.setVariable(QUEUE_VAR, "[]");
    return { ok: false, reason: "queue empty" };
  }

  const timestamp = new Date().toISOString();
  for (const id of candidates) {
    activeClaims[String(id)] = timestamp;
  }

  await wmill.setVariable(CLAIMED_VAR, JSON.stringify(activeClaims));
  await wmill.setVariable(
    QUEUE_VAR,
    JSON.stringify(queued.filter((id) => !candidates.includes(id))),
  );

  return { ok: true, doc_ids: candidates };
}

A claim is an entry such as:

{
  "4711": "2026-07-28T08:42:15.527Z"
}

After seven days, it is considered orphaned and ignored. That covers crashed flows and approval requests that were never answered.

An honest limitation of this solution

Windmill variables are updated through read-modify-write. That is not a transactional lock. Two flow runs starting at exactly the same time could theoretically read the same old state and claim the same document. In my private archive with sequential processing, the risk is small, and the inbox tag at least prevents documents from being lost. Most updates are idempotent as well.

Anyone needing high concurrency or hard exactly-once guarantees should store claims in a database, for example with a unique constraint, SELECT ... FOR UPDATE SKIP LOCKED or a PostgreSQL advisory lock. The version without Windmill further down uses a small SQLite lease that performs this part atomically.

The Windmill flow is deliberately simple

After the prerequisite step, the flow consists of one branch and one sequential loop. For each document there are three possible actions:

1. analyse it
2. wait for approval when necessary
3. update Paperless

The relevant part of flow.yaml looks like this:

- id: check_prereqs
  summary: Ollama-Check und Kandidaten claimen
  value:
    type: script
    path: f/paperless/agent/check_prereqs

- id: route
  summary: Nur arbeiten, wenn Prerequisites erfüllt sind
  value:
    type: branchone
    branches:
      - expr: results.check_prereqs.ok === true
        modules:
          - id: process_docs
            value:
              type: forloopflow
              iterator:
                type: javascript
                expr: results.check_prereqs.doc_ids
              parallel: false
              skip_failures: true
              modules:
                - id: analyze
                  value:
                    type: script
                    input_transforms:
                      doc_id:
                        type: javascript
                        expr: flow_input.iter.value
                      ollama:
                        type: static
                        value: $res:f/paperless/config/ollama
                      paperless:
                        type: static
                        value: $res:f/paperless/config/paperless
                    path: f/paperless/agent/analyze_document

                - id: approval_route
                  value:
                    type: branchone
                    branches:
                      - expr: results.analyze.needs_approval
                        modules:
                          - id: ask_approval
                            value:
                              type: script
                              path: f/inbox/notify/prompt_with_choices
                            suspend:
                              required_events: 1
                              timeout: 604800

                          - id: update_with_choice
                            value:
                              type: script
                              input_transforms:
                                analysis:
                                  type: javascript
                                  expr: results.analyze
                                approval_choice:
                                  type: javascript
                                  expr: resume.choice
                                doc_id:
                                  type: javascript
                                  expr: flow_input.iter.value
                              path: f/paperless/agent/update_document
                    default:
                      - id: update_direct
                        value:
                          type: script
                          input_transforms:
                            analysis:
                              type: javascript
                              expr: results.analyze
                            approval_choice:
                              type: static
                              value: none
                            doc_id:
                              type: javascript
                              expr: flow_input.iter.value
                          path: f/paperless/agent/update_document

parallel: false is intentional. I do not want my Ollama server processing several long documents at the same time, and the number of newly arriving documents does not justify parallelisation. skip_failures: true on the loop, on the other hand, prevents one broken document from blocking the entire batch.

The sequential loop does have one consequence. check_prereqs claims the entire batch first. If an early document needs approval, all later documents in the batch wait for my Telegram response as well. For my low volume and short response times that is acceptable. For a general service I would claim only the next document or move approval into a separate flow.

If analysis of a document fails, the document is not patched. It therefore retains inbox and reappears after the claim expires.

The core: a LlamaIndex workflow

Document analysis is not an autonomous agent inventing its own tasks. That is deliberate. It is a deterministic, event-driven workflow with explicit transitions:

flowchart LR
    A[StartEvent] --> B[check_ocr]
    B --> C[OcrChecked]
    C --> D[ensure_text]
    D --> E[TextReady]
    E --> F[classify]
    F --> G[Classified]
    G --> H[extract_metadata]
    H --> I[StopEvent]

LlamaIndex Workflows fit this well because every @step consumes exactly one event type and produces another. The transitions follow directly from the type annotations. Shared data such as OCR text, reference lists and file bytes can live in the Context without being copied through every event.

A standalone Python installation needs at least:

llama-index-core
llama-index-llms-ollama
llama-index-workflows
pymupdf
requests

In production these dependencies should of course be pinned to concrete versions. Windmill adds one peculiarity: my LlamaIndex imports intentionally live inside build_workflow(). That lets the validation functions be tested without installing the complete LlamaIndex stack. Windmill’s metadata generator does not fully detect those dynamic imports, however, so the lockfile for this script has to include the transitive LlamaIndex dependency chain explicitly.

Events and workflow context

The workflow needs only three custom events:

from llama_index.core.workflow import (
    Context,
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)


class OcrChecked(Event):
    ocr_ok: bool
    reason: str


class TextReady(Event):
    pass


class Classified(Event):
    classification: dict

The first step stores the input in the context:

WORKFLOW_CONTEXT_KEYS = (
    "ocr_text",
    "file_bytes",
    "file_name",
    "added",
    "correspondents",
    "document_types",
    "tags",
    "current_correspondent",
    "current_document_type",
    "force_vision",
)


@step
async def check_ocr(
    self,
    ev: StartEvent,
    ctx: Context,
) -> OcrChecked:
    for key in WORKFLOW_CONTEXT_KEYS:
        await ctx.store.set(key, getattr(ev, key, None))

    if ev.force_vision:
        return OcrChecked(
            ocr_ok=False,
            reason="force_vision aktiviert",
        )

    ollama = Ollama(**llm_kwargs)
    response = await ollama.acomplete(
        build_ocr_prompt(ev.ocr_text),
    )
    parsed = extract_json(response.text)

    return OcrChecked(
        ocr_ok=bool(parsed.get("ok")),
        reason=str(parsed.get("reason", "")),
    )

force_vision is useful for selectively retesting problematic documents. In normal operation, though, an inexpensive text request decides first whether vision is needed at all.

Prompt 1: Is the OCR text actually usable?

A minimum length is not enough to make that decision. A form can contain thousands of correctly recognised characters while the handwritten fields that carry the actual information are missing. I therefore let the model judge semantic usability:

MAX_OCR_CHECK_CHARS = 6000


def build_ocr_prompt(text: str) -> str:
    return f"""Du bewertest die Qualität eines OCR-Textes aus einem eingescannten Dokument.

Beurteile, ob der Text inhaltlich brauchbar ist, d.h. ob man daraus Absender,
Inhalt und Datum des Dokuments zuverlässig ableiten kann.

WICHTIG: Formulare mit gedruckten Feldlabels, etwa Rechnungs- oder
Berichtsformulare, sind NICHT brauchbar, wenn die eigentlichen INHALTE
handschriftlich eingetragen sind und das OCR diese Einträge nicht erfassen
konnte. Die Labels allein tragen keine ausreichende Information.

Als NICHT brauchbar gilt:
- Zeichensalat, stark fragmentierte Wörter, systematisch verstümmelte Umlaute
- handschriftliche Inhalte, die das OCR nicht erfassen konnte
- fast leerer Text bei offensichtlich vorhandenem Inhalt

Kleine Fehler wie einzelne falsche Zeichen, Zeilenumbrüche oder leichte
Formatierungsartefakte sind in Ordnung.

Antworte AUSSCHLIESSLICH mit JSON:
{{"ok": true|false, "reason": "kurze Begründung auf Deutsch"}}

OCR-TEXT:
{text[:MAX_OCR_CHECK_CHARS]}"""

Typical responses are:

{
  "ok": true,
  "reason": "Absender, Rechnungsnummer, Betrag und Rechnungsdatum sind trotz kleiner OCR-Fehler eindeutig erkennbar."
}

or:

{
  "ok": false,
  "reason": "Es wurden fast nur die gedruckten Formularbeschriftungen erkannt; Name, Datum und handschriftliche Messwerte fehlen."
}

The model does not classify anything here. It answers one narrowly bounded question. That makes the output considerably more stable than a prompt that asks it to judge OCR quality, choose a sender, select tags and extract a date all at once.

Vision rescue instead of vision for everything

If the OCR text is usable, it is kept unchanged:

@step
async def ensure_text(
    self,
    ev: OcrChecked,
    ctx: Context,
) -> TextReady:
    if ev.ocr_ok:
        await ctx.store.set(
            "text",
            await ctx.store.get("ocr_text"),
        )
        await ctx.store.set("ocr_rescued", False)
        return TextReady()

    # Vision-Pfad folgt hier

Only in the negative case do I load the original file, render PDF pages as PNG and send them individually to the vision model. The limits are deliberately conservative:

MAX_VISION_PAGES = 10
VISION_DPI = 150


def render_pages(
    file_bytes: bytes,
    file_name: str,
) -> list[bytes]:
    if not file_name.lower().endswith(".pdf"):
        return [file_bytes]

    import pymupdf

    pages: list[bytes] = []
    with pymupdf.open(
        stream=file_bytes,
        filetype="pdf",
    ) as pdf:
        for page in pdf[:MAX_VISION_PAGES]:
            pixmap = page.get_pixmap(dpi=VISION_DPI)
            pages.append(pixmap.tobytes("png"))

    return pages

The vision prompt should neither explain nor summarise:

def build_vision_prompt() -> str:
    return """Extrahiere den vollständigen Text dieser Dokumentseite.

Regeln:
- Gib den Text so wieder, wie er auf der Seite steht, in natürlicher Lesereihenfolge.
- Übernimm Überschriften, Absätze, Tabelleninhalte als Textzeilen, Beträge und Daten.
- Keine Kommentare, keine Beschreibung des Layouts, keine Zusammenfassung.
- Gib NUR den extrahierten Text zurück."""

LlamaIndex can send text and image blocks together to Ollama:

from llama_index.core.llms import (
    ChatMessage,
    ImageBlock,
    TextBlock,
)
from llama_index.llms.ollama import Ollama


@step
async def ensure_text(
    self,
    ev: OcrChecked,
    ctx: Context,
) -> TextReady:
    if ev.ocr_ok:
        await ctx.store.set(
            "text",
            await ctx.store.get("ocr_text"),
        )
        await ctx.store.set("ocr_rescued", False)
        return TextReady()

    vision = Ollama(**vision_llm_kwargs)
    file_bytes = await ctx.store.get("file_bytes")
    file_name = await ctx.store.get("file_name")

    page_texts: list[str] = []
    for image in render_pages(file_bytes, file_name):
        message = ChatMessage(
            role="user",
            blocks=[
                ImageBlock(image=image),
                TextBlock(text=build_vision_prompt()),
            ],
        )
        response = await vision.achat([message])
        page_texts.append(response.message.content or "")

    merged = "\n\n".join(
        text.strip()
        for text in page_texts
        if text.strip()
    )
    if not merged:
        raise RuntimeError(
            "Vision-Rescue lieferte keinen Text",
        )

    await ctx.store.set("text", merged)
    await ctx.store.set("ocr_rescued", True)
    return TextReady()

That saves a great deal of compute on normal PDFs. A digitally generated insurance letter does not need to be converted into images simply because a vision model is available. At the same time, handwritten forms do not remain unclassified indefinitely.

Prompt 2: Classifying against an existing taxonomy

Before analysis, the agent loads through the Paperless API:

• the current document
• all correspondents
• all document types
• all tags
• the original file

def fetch_context(
    paperless: paperless_config,
    doc_id: int,
) -> dict:
    document = paperless_get(
        paperless,
        f"/api/documents/{doc_id}/",
    )
    correspondents = paperless_get(
        paperless,
        "/api/correspondents/",
        {"page_size": 10000},
    ).get("results", [])
    document_types = paperless_get(
        paperless,
        "/api/document_types/",
        {"page_size": 10000},
    ).get("results", [])
    tags = paperless_get(
        paperless,
        "/api/tags/",
        {"page_size": 10000},
    ).get("results", [])
    file_bytes = paperless_get_bytes(
        paperless,
        f"/api/documents/{doc_id}/download/",
    )

    return {
        "doc": document,
        "correspondents": correspondents,
        "document_types": document_types,
        "tags": tags,
        "file_bytes": file_bytes,
    }

The lists are turned into compact JSON lines. IDs matter more than names because those IDs are what will eventually be written back to the Paperless API:

corr_list = "\n".join(
    f'  {{"id": {item["id"]}, "name": "{item["name"]}"}}'
    for item in correspondents
) or "  (leer)"

The complete classification prompt is the most important part of the system:

MAX_CLASSIFY_CHARS = 24000


def build_classify_prompt(
    text: str,
    correspondents: list[dict],
    document_types: list[dict],
    tags: list[dict],
    current_correspondent: str | None = None,
    current_document_type: str | None = None,
) -> str:
    corr_list = "\n".join(
        f'  {{"id": {item["id"]}, "name": "{item["name"]}"}}'
        for item in correspondents
    ) or "  (leer)"

    type_list = "\n".join(
        f'  {{"id": {item["id"]}, "name": "{item["name"]}"}}'
        for item in document_types
    ) or "  (leer)"

    tag_list = "\n".join(
        f'  {{"id": {item["id"]}, "name": "{item["name"]}"}}'
        for item in tags
    ) or "  (leer)"

    current_section = ""
    if current_correspondent or current_document_type:
        current_section = f"""
AKTUELLE ZUORDNUNG DIESES DOKUMENTS:
- Korrespondent: {current_correspondent or "(keiner)"}
- Dokumenttyp: {current_document_type or "(keiner)"}

Diese Zuordnung stammt aus dem automatischen Matching von paperless und kann
falsch sein. Wenn sie inhaltlich vertretbar ist, übernimm sie. Stabilität hat
Vorrang. Wenn sie klar falsch ist, korrigiere sie nach den Regeln unten.
"""

    return f"""Du klassifizierst ein Dokument für ein
Dokumentenmanagementsystem auf Basis von paperless-ngx.

{current_section}
AUFGABE 1 - KORRESPONDENT, ALSO DER ABSENDER:
Wähle nach Möglichkeit einen BESTEHENDEN Korrespondenten aus der Liste.

STRIKTE REGELN GEGEN DUBLETTEN:
- Prüfe die Liste gründlich auf Schreibvarianten, Abkürzungen und Zusätze.
- "Stadtwerke Hogwarts", "Stadtwerke Hogwarts GmbH" und "SWM" können
  derselbe Korrespondent sein.
- Rechtsformen wie GmbH, AG oder e.V., Adresszusätze sowie Gross- und
  Kleinschreibung rechtfertigen KEINEN neuen Korrespondenten.
- Nur wenn wirklich kein Listeneintrag passt, setze correspondent_id auf null
  und schlage in new_correspondent einen kurzen, sauberen Namen ohne unnötige
  Rechtsform- oder Adresszusätze vor.

AUFGABE 2 - DOKUMENTTYP:
Wähle nach Möglichkeit einen BESTEHENDEN Dokumenttyp aus der Liste.
- Nur wenn keiner passt, setze document_type_id auf null und schlage einen
  kurzen, generischen Typnamen vor, etwa "Rechnung", "Vertrag" oder "Bescheid".

AUFGABE 3 - TAGS:
- Wähle passende Tags NUR aus der bestehenden Liste und gib deren IDs zurück.
- Zusätzlich darfst du höchstens zwei neue, generische Tags vorschlagen,
  wenn sie in der bestehenden Liste klar fehlen.
- Erzeuge keine Tags, die Korrespondent oder Dokumenttyp nur wiederholen.

Antworte AUSSCHLIESSLICH mit genau einem JSON-Objekt:
{{
  "correspondent_id": <ID aus der Liste oder null>,
  "new_correspondent": <string oder null>,
  "document_type_id": <ID aus der Liste oder null>,
  "new_document_type": <string oder null>,
  "tag_ids": [<IDs aus der Liste>],
  "new_tags": [<höchstens zwei Strings>],
  "reasoning": "ein bis zwei Sätze Begründung auf Deutsch"
}}

BESTEHENDE KORRESPONDENTEN:
{corr_list}

BESTEHENDE DOKUMENTTYPEN:
{type_list}

BESTEHENDE TAGS:
{tag_list}

DOKUMENTTEXT:
{text[:MAX_CLASSIFY_CHARS]}"""

Three details are decisive for classification quality.

First, the model receives the current Paperless assignment. Paperless already has its own rule- and classifier-based matching. If that result is defensible, there is no reason to force a new decision on every AI run. In an archive, stability is often more valuable than a marginally “nicer” label.

Second, null is not an error. It means that none of the existing entries is a safe match. That state is better than an invented ID or a barely plausible assignment.

Third, new and existing values live in separate fields. The code can therefore distinguish unambiguously between selecting an existing record and proposing a new one.

That keeps the workflow step small:

@step
async def classify(
    self,
    ev: TextReady,
    ctx: Context,
) -> Classified:
    ollama = Ollama(**llm_kwargs)
    response = await ollama.acomplete(
        build_classify_prompt(
            await ctx.store.get("text"),
            await ctx.store.get("correspondents"),
            await ctx.store.get("document_types"),
            await ctx.store.get("tags"),
            await ctx.store.get("current_correspondent"),
            await ctx.store.get("current_document_type"),
        )
    )

    return Classified(
        classification=extract_json(response.text),
    )

Prompt 3: Extracting title and document date separately

Title and date deliberately do not belong in the classification prompt. The date in particular is error-prone because a document can contain many plausible dates: date of birth, start of a contract, payment deadline, expiry date or delivery date.

MAX_METADATA_CHARS = 12000


def build_metadata_prompt(
    text: str,
    filename: str,
    added: str,
) -> str:
    return f"""Du extrahierst Titel und Erstellungsdatum eines Dokuments
für ein Archiv.

TITEL:
- Kurz, aussagekräftig und suchbar, ungefähr maximal 80 Zeichen.
- Schreibe den Titel auf Deutsch.
- Nenne den Inhalt und, falls sinnvoll, die Organisation.
- Keine langen Aktenzeichenlisten und kein Datum im Titel.

ERSTELLUNGSDATUM, FELD created:
- Gesucht ist das Datum, an dem das DOKUMENT erstellt oder ausgestellt wurde,
  also etwa Briefdatum, Rechnungsdatum oder Ausstellungsdatum.
- Achte auf den Kontext. Das Datum steht häufig im Briefkopf, bei der Anrede
  oder bei der Unterschrift.
- Verwende NIEMALS Geburtsdaten, Gültigkeitsdaten, Vertragsbeginn oder
  Vertragsende, Fristen, Zahlungsziele, Lieferdaten oder Werbedaten.
- Wenn mehrere Kandidaten vorhanden sind, wähle den Wert, der am ehesten das
  Ausstellungsdatum ist.
- Wenn kein plausibles Dokumentdatum vorhanden ist, setze created auf null.
- Format: YYYY-MM-DD

Antworte AUSSCHLIESSLICH mit JSON:
{{
  "title": "...",
  "created": "YYYY-MM-DD oder null",
  "reasoning": "kurze Begründung auf Deutsch"
}}

HINWEISE:
Dateiname: {filename or "unbekannt"}
Eingangsdatum im Archiv: {added}

DOKUMENTTEXT:
{text[:MAX_METADATA_CHARS]}"""

The explicit negative list matters more than the phrase “find the document date”. Without it, a model will happily use the expiry date on an insurance card or the payment deadline on a reminder.

The final workflow step combines classification and metadata:

@step
async def extract_metadata(
    self,
    ev: Classified,
    ctx: Context,
) -> StopEvent:
    ollama = Ollama(**llm_kwargs)
    response = await ollama.acomplete(
        build_metadata_prompt(
            await ctx.store.get("text"),
            await ctx.store.get("file_name"),
            await ctx.store.get("added"),
        )
    )

    return StopEvent(
        result={
            "classification": ev.classification,
            "metadata": extract_json(response.text),
            "ocr_rescued": await ctx.store.get(
                "ocr_rescued",
            ),
        }
    )

Four small prompts are more reliable than one large prompt

A normal document therefore needs three model calls:

1. OCR quality
2. classification
3. title and date

Poor OCR adds vision calls for each rendered page. That is more work than one mega-prompt, but each request has a clearer task and a smaller output schema.

For local models, that is often the better trade. A model can read very well and still forget a field in a large JSON schema. Splitting the tasks also lets me use different input lengths:

MAX_OCR_CHECK_CHARS = 6000
MAX_CLASSIFY_CHARS = 24000
MAX_METADATA_CHARS = 12000

The OCR judgement does not need the complete document. Classification benefits from more context. For the date, on the other hand, I want to avoid a long appendix full of additional dates overshadowing the actual letter page.

My Ollama configuration uses a low temperature:

llm_kwargs = {
    "model": ollama["model"],
    "base_url": ollama["base_url"],
    "temperature": 0.1,
    "request_timeout": float(timeout),
}

vision_llm_kwargs = {
    **llm_kwargs,
    "model": ollama["vision_model"],
}

For a creative task, 0.1 would be unnecessarily restrictive. When selecting an existing database ID, creativity is closer to a defect.

Reading JSON from a local model robustly

Even with an instruction to return “JSON only”, models occasionally wrap the object in a Markdown fence or add an introductory sentence. For this limited case, a defensive extractor is enough:

import json
import re


def extract_json(text: str) -> dict:
    cleaned = re.sub(r"```(?:json)?", "", text)
    start = cleaned.find("{")
    end = cleaned.rfind("}")

    if start == -1 or end == -1 or end <= start:
        raise ValueError(
            f"kein JSON-Objekt in LLM-Antwort: {text[:200]}"
        )

    return json.loads(cleaned[start : end + 1])

That is not a replacement for domain validation. It only prevents processing from failing because of a code fence. Whether the contained IDs are allowed is checked separately afterwards.

The LLM proposes; the code decides

The most important code in the entire project is neither the Ollama call nor the prompt, but the normalisation that follows:

def validate_classification(
    classification: dict,
    correspondents: list[dict],
    document_types: list[dict],
    tags: list[dict],
) -> dict:
    correspondent_ids = {
        item["id"] for item in correspondents
    }
    document_type_ids = {
        item["id"] for item in document_types
    }
    existing_tag_ids = {
        item["id"] for item in tags
    }

    correspondent_id = classification.get(
        "correspondent_id"
    )
    if correspondent_id not in correspondent_ids:
        correspondent_id = None

    new_correspondent = (
        classification.get("new_correspondent") or ""
    ).strip() or None

    if correspondent_id is not None:
        new_correspondent = None

    document_type_id = classification.get(
        "document_type_id"
    )
    if document_type_id not in document_type_ids:
        document_type_id = None

    new_document_type = (
        classification.get("new_document_type") or ""
    ).strip() or None

    if document_type_id is not None:
        new_document_type = None

    tag_ids = [
        tag_id
        for tag_id in classification.get("tag_ids") or []
        if tag_id in existing_tag_ids
    ]

    existing_tag_names = {
        item["name"].lower()
        for item in tags
    }
    new_tags = [
        str(name).strip()
        for name in classification.get("new_tags") or []
        if str(name).strip()
    ]
    new_tags = [
        name
        for name in new_tags
        if name.lower() not in existing_tag_names
    ][:2]

    return {
        "correspondent_id": correspondent_id,
        "new_correspondent": new_correspondent,
        "document_type_id": document_type_id,
        "new_document_type": new_document_type,
        "tag_ids": tag_ids,
        "new_tags": new_tags,
        "reasoning": classification.get(
            "reasoning",
            "",
        ),
    }

Regardless of the model, the following invariants now hold:

• A correspondent ID must actually exist in the list loaded beforehand.
• A document-type ID must really exist.
• Tag IDs must come from Paperless.
• A valid ID always overrides a simultaneously proposed new name.
• New tags are compared case-insensitively against existing tags.
• At most two new tags can be created per document.

A prompt can reduce the probability of an error. A function like this can make particular classes of error impossible. For me, that distinction is what separates an interesting LLM demo from automation I am willing to trust with my archive.

For date validation, my current code checks the format and falls back to the ingestion date for invalid output. A stricter version can additionally verify that the date actually exists:

from datetime import date


def normalize_created(
    value: object,
    fallback: str,
) -> str:
    if not isinstance(value, str):
        return fallback

    try:
        date.fromisoformat(value)
        return value
    except ValueError:
        return fallback

That would also reject a formally shaped but impossible date such as 2026-19-42.

Starting the workflow from synchronous code

LlamaIndex Workflows are asynchronous. Windmill calls my Python script through a normal main() function, so the event loop has to be encapsulated cleanly:

def main(
    doc_id: int,
    paperless: paperless_config,
    ollama: ollama_config,
    force_vision: bool = False,
) -> dict:
    context = fetch_context(paperless, doc_id)
    document = context["doc"]

    current_correspondent = next(
        (
            item["name"]
            for item in context["correspondents"]
            if item["id"] == document.get("correspondent")
        ),
        None,
    )
    current_document_type = next(
        (
            item["name"]
            for item in context["document_types"]
            if item["id"] == document.get("document_type")
        ),
        None,
    )

    workflow_input = {
        "ocr_text": document.get("content") or "",
        "file_bytes": context["file_bytes"],
        "file_name": document.get("original_file_name") or "",
        "added": (document.get("added") or "")[:10],
        "correspondents": context["correspondents"],
        "document_types": context["document_types"],
        "tags": context["tags"],
        "current_correspondent": current_correspondent,
        "current_document_type": current_document_type,
        "force_vision": force_vision,
    }

    workflow = build_workflow(
        llm_kwargs,
        vision_llm_kwargs,
    )

    async def run() -> dict:
        return await workflow.run(**workflow_input)

    result = asyncio.run(run())

    classification = validate_classification(
        result["classification"],
        context["correspondents"],
        context["document_types"],
        context["tags"],
    )
    metadata = validate_metadata(
        result["metadata"],
        workflow_input["added"],
    )

    return {
        "doc_id": doc_id,
        "current_title": document.get("title") or "",
        "current_tag_ids": document.get("tags") or [],
        "ocr_rescued": result["ocr_rescued"],
        "needs_approval": bool(
            classification["new_correspondent"]
            or classification["new_document_type"]
        ),
        **classification,
        **metadata,
    }

It is important to call workflow.run() inside the coroutine. The workflow creates tasks when it starts. An expression such as asyncio.run(workflow.run(...)) can cause problems in some environments because workflow.run(...) is evaluated before the event loop has been started cleanly.

Human in the loop, but only for exceptions

Approving every document would not be automation. It would simply create another inbox. The flow therefore asks only when new_correspondent or new_document_type is set.

The message contains the title, proposal and reasoning:

paperless: neuer Vorschlag (Dokument 4711)

Titel: Rechnung Solaranlage Juli 2026
Neuer Korrespondent: Solarwerk Bodensee

Der Absender kommt in der bestehenden Liste nicht vor;
Rechnung ist als Dokumenttyp bereits vorhanden.

[Anlegen] [Ablehnen]

Windmill can suspend a flow at a step and resume it later through a secret resume URL. My Telegram component generates a short callback ID for each option and stores the resume URL behind it:

import * as wmill from "windmill-client";

type Choice = {
  label: string;
  value: string;
};

export async function main(
  text: string,
  choices: Choice[],
): Promise<{
  telegram_message_id: number;
  cb_ids: string[];
}> {
  const telegram = (await wmill.getResource(
    "u/example/telegram-bot",
  )) as { apiKey: string };
  const chatId = (await wmill.getVariable(
    "f/general/config/telegram-chat-id",
  )) as string;

  const urls = await wmill.getResumeUrls(
    "paperless-approval",
  );

  const callbackIds: string[] = [];
  const keyboard: any[] = [];

  for (const choice of choices) {
    const id = crypto.randomUUID()
      .replace(/-/g, "")
      .slice(0, 10);

    await wmill.setVariable(
      `f/inbox/state/cb_${id}`,
      urls.resume,
    );

    callbackIds.push(id);
    keyboard.push([
      {
        text: choice.label,
        callback_data: `cb_${id}:${choice.value}`,
      },
    ]);
  }

  const response = await fetch(
    `https://api.telegram.org/bot${telegram.apiKey}/sendMessage`,
    {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        chat_id: chatId,
        text,
        parse_mode: "Markdown",
        reply_markup: {
          inline_keyboard: keyboard,
        },
      }),
    },
  );

  const result = await response.json();
  if (!response.ok || !result.ok) {
    throw new Error(
      `Telegram ${response.status}: ${JSON.stringify(result)}`,
    );
  }

  return {
    telegram_message_id: result.result.message_id,
    cb_ids: callbackIds,
  };
}

A separate bot webhook resolves the callback ID and calls the resume URL with the following payload:

{
  "choice": "approve"
}

or:

{
  "choice": "reject"
}

The next Windmill step can read that value directly as resume.choice. The flow waits for at most seven days. If nobody responds, the document is not updated and remains discoverable through its inbox tag.

Telegram is merely my preferred channel here. The same mechanism works with email, Matrix, Slack, a small web interface or a Windmill approval page.

New records may only be created during the update

The analysis step writes nothing to Paperless. Only update_document.py processes an approved proposal.

New correspondents and document types are created only for approve:

correspondent_id = analysis.get("correspondent_id")
if (
    correspondent_id is None
    and approval_choice == "approve"
    and analysis.get("new_correspondent")
):
    correspondent_id = create_named(
        paperless,
        "correspondents",
        analysis["new_correspondent"],
    )

document_type_id = analysis.get("document_type_id")
if (
    document_type_id is None
    and approval_choice == "approve"
    and analysis.get("new_document_type")
):
    document_type_id = create_named(
        paperless,
        "document_types",
        analysis["new_document_type"],
    )

create_named() also handles a small race condition. Two documents in the same batch can propose the same new sender. If the second POST fails, the code performs another case-insensitive lookup for an entry that may have appeared in the meantime:

def create_named(
    paperless: paperless_config,
    endpoint: str,
    name: str,
) -> int:
    try:
        return api(
            paperless,
            "POST",
            f"/api/{endpoint}/",
            json={"name": name},
        )["id"]
    except requests.HTTPError:
        results = api(
            paperless,
            "GET",
            f"/api/{endpoint}/",
            params={"name__iexact": name},
        ).get("results", [])

        for item in results:
            if item["name"].lower() == name.lower():
                return item["id"]
        raise

The tags are then merged, inbox is removed and ai-processed is added:

tag_ids: set[int] = set(
    analysis.get("current_tag_ids") or []
)
tag_ids.discard(
    find_tag_id(paperless, "inbox")
)
tag_ids.update(analysis.get("tag_ids") or [])

for name in analysis.get("new_tags") or []:
    tag_ids.add(
        find_tag_id(
            paperless,
            name,
            create=True,
        )
    )

processed_id = find_tag_id(
    paperless,
    "ai-processed",
    create=True,
)
if processed_id is not None:
    tag_ids.add(processed_id)

tag_ids.discard(None)

The actual PATCH remains uneventful:

payload: dict[str, Any] = {
    "tags": sorted(tag_ids),
}

if correspondent_id is not None:
    payload["correspondent"] = correspondent_id
if document_type_id is not None:
    payload["document_type"] = document_type_id
if analysis.get("title"):
    payload["title"] = analysis["title"]
if analysis.get("created"):
    payload["created"] = analysis["created"]

api(
    paperless,
    "PATCH",
    f"/api/documents/{doc_id}/",
    json=payload,
)

The claim is released only after a successful PATCH. If anything fails beforehand, the claim remains until its TTL expires and the document keeps inbox.

Failure cases are part of the architecture

In continuously running automation, errors are not exceptional. They are normal operation. Every expected case therefore has a defined outcome:

Situation	Behaviour
Ollama is unreachable	Flow ends without an alarm; schedule tries again later
Webhook is lost	Schedule finds the document through inbox
Two triggers start almost simultaneously	Claims reduce duplicate processing; a database lock would be stricter
Analysis fails for one document	Other documents in the batch continue
Vision returns no text	Document remains unchanged and therefore in inbox
Telegram is not answered	Suspend expires; claim eventually expires, document remains discoverable
A new correspondent was created concurrently	create_named() performs another lookup after the failed POST
Model hallucinates an ID	Validation resets it to null
Model proposes an existing tag again	Case-insensitive comparison removes the proposal
Paperless PATCH fails	Claim is not explicitly released; document will be retried later

This failure semantics is a larger part of the project than the model call itself. That is exactly why I call it AI Engineering rather than merely Prompt Engineering.

Rebuilding it with Windmill

The architecture can be reproduced from a handful of clearly separated components. The following steps are the important parts for a new setup.

1. Prerequisites

You need:

• a running paperless-ngx instance
• a Paperless API token
• a Windmill instance with a Python worker
• an accessible Ollama server
• a text model and, when required, a vision model
• optionally a Telegram bot

Paperless and Ollama do not need to run on the same host. The Windmill worker has to reach both. Ollama should not be exposed unprotected to the public internet.

2. Create tags in Paperless

At least these two tags are expected:

inbox
ai-processed

inbox marks open work. ai-processed is not technically required, but it makes filtering and troubleshooting considerably easier.

3. Create a Paperless API token

The token is sent with every API request as a header:

Authorization: Token <PAPERLESS_API_TOKEN>

In Windmill it belongs in a secret variable, not directly inside a publicly visible resource YAML file.

4. Create Windmill resources

An Ollama resource can look like this:

description: Lokaler Ollama-Server für Paperless
value:
  base_url: http://ollama.internal:11434
  model: qwen3.6:35b-a3b
  vision_model: qwen3.6:35b-a3b
  request_timeout_seconds: 600
resource_type: ollama_config

For Paperless:

description: Paperless-API
value:
  base_url: http://paperless.internal:8000
  api_token: $var:f/paperless/config/paperless-api-token
resource_type: paperless_config

The model names are interchangeable. What matters is that vision_model can actually process images. On smaller systems, using a compact text model for classification and a separate vision model can make more sense.

5. Deploy scripts and flow

The Windmill implementation is split into the following components:

f/paperless/
├── agent/
│   ├── analyze_document.py
│   ├── check_prereqs.ts
│   └── update_document.py
├── config/
│   ├── ollama.resource.yaml
│   └── paperless.resource.yaml
├── process_inbox__flow/
│   └── flow.yaml
├── process_inbox_schedule.schedule.yaml
└── webhook/
    └── inbox_webhook.ts

My flow additionally uses the generic Telegram component under f/inbox/notify/.

After importing, run the three scripts individually against a test document first. That isolates API, model and dependency problems from one another.

6. Configure the Paperless workflow

In Paperless:

1. Create a workflow with the “Document added” trigger.
2. Optionally filter to documents carrying inbox, or set inbox in a preceding workflow action.
3. Add a webhook action pointing to the Windmill webhook URL.
4. Select JSON encoding.
5. Send the following body:

{
  "doc_id": {{doc_id}},
  "doc_url": "{{doc_url}}"
}

The URL is included only as a fallback. On a current Paperless version, doc_id is normally sufficient.

If Paperless is not permitted to call internal HTTP targets, the webhook configuration has to be checked. Relevant controls include allowed schemes, ports and internal targets.

7. Use three test cases

One test PDF says little about quality. I would verify at least these cases.

Case A: existing correspondent and existing document type

Example: another invoice from a known energy supplier.

Expected:

{
  "correspondent_id": 17,
  "new_correspondent": null,
  "document_type_id": 4,
  "new_document_type": null,
  "needs_approval": false
}

Case B: known correspondent with different spelling

Example: Paperless contains Stadtwerke Hogwarts, while the document says Stadtwerke Hogwarts GmbH.

Expected result: existing ID, no new proposal.

Case C: genuinely new sender

Expected:

{
  "correspondent_id": null,
  "new_correspondent": "Solarwerk Bodensee",
  "needs_approval": true
}

It is also worth testing a handwritten form to verify that ocr_rescued changes to true.

The same solution without Windmill

Windmill handles several things for me: webhooks, schedules, secrets, logging, flow visualisation and suspend/resume. None of those capabilities is essential to the document analysis itself.

The simplest standalone architecture consists of four components:

flowchart TD
    A[Paperless-Webhook] --> Q[(SQLite Queue)]
    S[Cron oder systemd timer] --> W[Python Worker]
    Q --> W
    P[Paperless-Dokumente mit inbox] --> W
    W --> L[LlamaIndex und Ollama]
    L --> U[Paperless PATCH]
    L --> R[(Pending approvals)]

The same rule applies here: the queue is only the fast path. Every worker run additionally queries all Paperless documents carrying inbox.

Variant A: polling and cron only

For a private archive, a script running every five minutes is often sufficient:

*/5 * * * * /opt/paperless-agent/.venv/bin/python /opt/paperless-agent/worker.py --once

The webhook disappears completely. That increases latency to at most five minutes but reduces the infrastructure to one Python program.

The flow in worker.py is:

def run_once() -> None:
    if not ollama_is_ready():
        return

    for doc_id in find_inbox_document_ids():
        if not claims.try_acquire(doc_id):
            continue

        try:
            result = analyze_document(doc_id)

            if result["needs_approval"]:
                approvals.store(doc_id, result)
                mark_for_review(doc_id)
                continue

            update_document(
                doc_id=doc_id,
                analysis=result,
                approval_choice="none",
            )
            claims.release(doc_id)
        except Exception:
            logger.exception(
                "Dokument %s konnte nicht verarbeitet werden",
                doc_id,
            )

Without Telegram, mark_for_review() can simply add a tag such as ai-review. The few exceptional cases can then be reviewed directly in Paperless and started again afterwards. That is less elegant than suspend/resume, but extremely robust and easy to understand.

Atomic claims with SQLite

SQLite is surprisingly well suited to this task. A table with doc_id as its primary key prevents two workers from taking the same ID concurrently.

from __future__ import annotations

import sqlite3
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path

DB_PATH = Path("/var/lib/paperless-agent/state.db")
CLAIM_TTL = timedelta(days=7)


@contextmanager
def connection():
    conn = sqlite3.connect(
        DB_PATH,
        timeout=30,
        isolation_level=None,
    )
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


def initialize() -> None:
    DB_PATH.parent.mkdir(
        parents=True,
        exist_ok=True,
    )
    with connection() as conn:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS claims (
                doc_id INTEGER PRIMARY KEY,
                claimed_at TEXT NOT NULL
            )
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS queue (
                doc_id INTEGER PRIMARY KEY,
                queued_at TEXT NOT NULL
            )
            """
        )
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS approvals (
                doc_id INTEGER PRIMARY KEY,
                payload TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
            """
        )


def try_acquire(doc_id: int) -> bool:
    now = datetime.now(UTC)
    stale_before = now - CLAIM_TTL

    with connection() as conn:
        conn.execute("BEGIN IMMEDIATE")
        try:
            conn.execute(
                "DELETE FROM claims WHERE claimed_at < ?",
                (stale_before.isoformat(),),
            )
            cursor = conn.execute(
                """
                INSERT OR IGNORE INTO claims (
                    doc_id,
                    claimed_at
                ) VALUES (?, ?)
                """,
                (doc_id, now.isoformat()),
            )
            acquired = cursor.rowcount == 1
            conn.execute("COMMIT")
            return acquired
        except Exception:
            conn.execute("ROLLBACK")
            raise


def release(doc_id: int) -> None:
    with connection() as conn:
        conn.execute(
            "DELETE FROM claims WHERE doc_id = ?",
            (doc_id,),
        )

BEGIN IMMEDIATE reserves the write transaction before the stale claim is deleted and the new one inserted. The primary-key constraint makes the claim unique even across several processes.

A durable queue for the webhook

To retain immediate processing, add a small FastAPI endpoint. It does not start analysis inside the HTTP request; it only writes to SQLite:

from datetime import UTC, datetime
import re

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class PaperlessWebhook(BaseModel):
    doc_id: int | None = None
    doc_url: str | None = None


def extract_doc_id(payload: PaperlessWebhook) -> int | None:
    if payload.doc_id and payload.doc_id > 0:
        return payload.doc_id

    if payload.doc_url:
        match = re.search(
            r"/documents/(\d+)/?",
            payload.doc_url,
        )
        if match:
            return int(match.group(1))

    return None


@app.post("/webhooks/paperless/inbox")
def paperless_inbox(
    payload: PaperlessWebhook,
) -> dict:
    doc_id = extract_doc_id(payload)
    if doc_id is None:
        raise HTTPException(
            status_code=400,
            detail="keine Dokument-ID gefunden",
        )

    with connection() as conn:
        conn.execute(
            """
            INSERT INTO queue (doc_id, queued_at)
            VALUES (?, ?)
            ON CONFLICT(doc_id) DO UPDATE SET
                queued_at = excluded.queued_at
            """,
            (doc_id, datetime.now(UTC).isoformat()),
        )

    return {
        "queued": True,
        "doc_id": doc_id,
    }

A long-running worker can poll the queue every few seconds. A separate cron job remains as a backstop and continues reading the inbox tag from Paperless.

One important point is not to use FastAPI BackgroundTasks as the only queue. They are convenient but not durable. A process restart loses the task. SQLite, Redis, PostgreSQL or a proper message broker is a better fit.

A simple standalone worker

The set of candidates is constructed just like in the Windmill version, from both the queue and Paperless:

def next_candidates() -> list[int]:
    queued = read_queued_ids()
    inbox = paperless_inbox_ids()
    return sorted(set(queued) | set(inbox))


def process_batch() -> None:
    if not ollama_is_ready():
        logger.info("Ollama ist nicht erreichbar")
        return

    for doc_id in next_candidates():
        if not try_acquire(doc_id):
            continue

        try:
            remove_from_queue(doc_id)
            analysis = analyze_document_main(
                doc_id=doc_id,
                paperless=PAPERLESS_CONFIG,
                ollama=OLLAMA_CONFIG,
            )

            if analysis["needs_approval"]:
                save_pending_approval(
                    doc_id,
                    analysis,
                )
                add_tag(doc_id, "ai-review")
                continue

            update_document_main(
                doc_id=doc_id,
                analysis=analysis,
                approval_choice="none",
                paperless=PAPERLESS_CONFIG,
            )
            release(doc_id)
        except Exception:
            logger.exception(
                "Verarbeitung von Dokument %s fehlgeschlagen",
                doc_id,
            )

In a pure review-tag variant, the claim should deliberately remain after saving the approval request or be replaced with a separate status. Otherwise the next poll immediately analyses the document again. An explicit state machine is better than several implicit tags:

queued -> processing -> pending_approval -> completed
                      \-> failed/retry

SQLite can store this state directly in a jobs table. The Paperless tag should still remain as the higher-level safety net against lost internal state.

Telegram without Windmill

Telegram can also be recreated without Windmill. The required pieces are:

1. an approvals table containing a random single-use token
2. inline buttons with callback_data=<token>:approve
3. a Telegram webhook that validates token and decision
4. a worker that applies the stored decision

A row could look like this:

CREATE TABLE approvals (
    token TEXT PRIMARY KEY,
    doc_id INTEGER NOT NULL,
    payload TEXT NOT NULL,
    decision TEXT,
    expires_at TEXT NOT NULL,
    used_at TEXT
);

The bot handler must not simply trust the document ID. The random token should be short-lived and invalid after first use. After approve, the worker invokes the same update_document() code as the Windmill version.

For a private setup, the simpler ai-review alternative is often sufficient. Telegram becomes useful when new documents should also be approved while away from home.

Running it as a systemd service

A persistent worker can run without containers as a systemd service:

[Unit]
Description=Paperless AI Inbox Agent
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=paperless-agent
Group=paperless-agent
WorkingDirectory=/opt/paperless-agent
EnvironmentFile=/etc/paperless-agent.env
ExecStart=/opt/paperless-agent/.venv/bin/python -m agent.worker
Restart=on-failure
RestartSec=10
NoNewPrivileges=true
PrivateTmp=true

[Install]
WantedBy=multi-user.target

Secrets live in /etc/paperless-agent.env with restrictive file permissions:

PAPERLESS_URL=http://paperless.internal:8000
PAPERLESS_TOKEN=...
OLLAMA_URL=http://ollama.internal:11434
OLLAMA_MODEL=qwen3.6:35b-a3b
OLLAMA_VISION_MODEL=qwen3.6:35b-a3b

At that point the Windmill setup has become ordinary Python code. The LlamaIndex workflow can be reused unchanged.

Tests that are actually worth writing

LLM calls themselves are not deterministic enough for traditional unit tests. The domain constraints around them, however, are very easy to test.

A valid ID wins over a new name

def test_existing_correspondent_id_wins() -> None:
    result = validate_classification(
        {
            "correspondent_id": 17,
            "new_correspondent": "Stadtwerke Hogwarts GmbH",
            "document_type_id": 4,
            "new_document_type": None,
            "tag_ids": [],
            "new_tags": [],
        },
        correspondents=[
            {"id": 17, "name": "Stadtwerke Hogwarts"},
        ],
        document_types=[
            {"id": 4, "name": "Rechnung"},
        ],
        tags=[],
    )

    assert result["correspondent_id"] == 17
    assert result["new_correspondent"] is None

Hallucinated IDs are rejected

def test_unknown_ids_are_rejected() -> None:
    result = validate_classification(
        {
            "correspondent_id": 9999,
            "new_correspondent": None,
            "document_type_id": 8888,
            "new_document_type": None,
            "tag_ids": [7777],
            "new_tags": [],
        },
        correspondents=[
            {"id": 17, "name": "Stadtwerke Hogwarts"},
        ],
        document_types=[
            {"id": 4, "name": "Rechnung"},
        ],
        tags=[
            {"id": 8, "name": "Energie"},
        ],
    )

    assert result["correspondent_id"] is None
    assert result["document_type_id"] is None
    assert result["tag_ids"] == []

New tags are deduplicated and capped

def test_new_tags_are_deduplicated_and_limited() -> None:
    result = validate_classification(
        {
            "correspondent_id": None,
            "new_correspondent": None,
            "document_type_id": None,
            "new_document_type": None,
            "tag_ids": [],
            "new_tags": [
                "Energie",
                "Photovoltaik",
                "Förderung",
                "Dach",
            ],
        },
        correspondents=[],
        document_types=[],
        tags=[
            {"id": 8, "name": "energie"},
        ],
    )

    assert result["new_tags"] == [
        "Photovoltaik",
        "Förderung",
    ]

I additionally test JSON extraction, date fallback, tag handling during updates and claim release. None of these tests needs Ollama or LlamaIndex, so they run quickly in any CI pipeline.

How to measure quality sensibly

“Works better for me” is an operational observation, not a scientific benchmark. Anyone comparing models or prompts should use a fixed set of documents that have already been classified correctly and record at least the following metrics:

Metric	What it measures
Correct correspondent	Was the correct existing ID selected?
Correct document type	Was the correct existing ID selected?
Reuse rate	How often could an existing record be reused?
False new proposals	How often was a new name proposed despite an existing match?
Approval rate	How many documents require a human?
Usable title	Is the title concise, searchable and correct?
Correct document date	Was the actual issue date selected?
Vision success rate	How many OCR problem cases were rescued?
Errors per 100 documents	How often does the technical flow finish without an update?

For my use case, false new proposals are almost more important than raw classification accuracy. A wrongly assigned record is visible in a spot check. Ten nearly identical correspondents, on the other hand, permanently degrade search, filters and future classification.

Where I would extend the solution

The current state works well for my private document volume, but there are obvious areas for further improvement.

Prefilter candidates

With a few dozen or a few hundred correspondents, the complete list can be placed directly in the prompt. With several thousand entries, that becomes expensive and difficult to inspect. I would then construct candidates before the LLM call through:

• normalising legal forms and punctuation
• fuzzy matching against recognised sender lines
• embeddings over names and known document examples
• passing only the best 20 to 50 IDs to the model

The model remains the decision-maker inside a controlled candidate set. The code continues validating against the real Paperless IDs.

Model confidence explicitly

The current decision is null versus an ID. The model could additionally return a coarse confidence value:

{
  "correspondent_id": 17,
  "correspondent_confidence": 0.93,
  "document_type_id": 4,
  "document_type_confidence": 0.88
}

I would never treat such a number as a real probability without validation. It only becomes useful after calibration on a dedicated test set. Until then, a discrete state such as certain, plausible or uncertain is often more honest.

Make approval more informative

Instead of only “Create” and “Reject”, Telegram could additionally offer the three closest existing correspondents. That would be particularly useful when the model cannot make a safe selection but a human immediately recognises the intended existing record.

Move claims into a real database

Windmill variables are sufficient for my volume, but they are not a transactional queue system. A small PostgreSQL table or Windmill Data Table would represent parallel workers and atomic leases more cleanly. I would also change claiming from whole batches to one document at a time so that an approval waiting for input cannot block already claimed documents behind it.

Approve new tags when necessary

New tags in my current flow are created without a Telegram confirmation. Their number is limited to two and existing names are filtered case-insensitively. If the tag list becomes noisy anyway, the same approval rule used for correspondents and document types can be applied there too.

Add observability

The next useful step would be a small report covering:

• runtime per workflow step
• token or prompt length
• share of vision fallbacks
• approval rate
• most frequent new proposals
• technical failures by category

That would make it possible to tell whether changing models actually improves the system or merely changes how the output sounds.

Conclusion

The real challenge in AI-assisted document filing is not recognising an invoice. It is integrating the model into an existing taxonomy without letting that taxonomy drift.

My Paperless agent therefore treats correspondents, document types and tags not as freely generated text, but as existing reference data. The model prefers stable IDs. New names are proposed separately. The code rejects invalid IDs, limits new tags and creates new master data only after approval.

Windmill makes orchestration comfortable: webhook, schedule, secrets, loops and Telegram suspension fit together cleanly. The core ideas do not depend on it, though:

• The inbox tag is the source of truth.
• A queue reduces latency but must never be the only state.
• OCR is evaluated first and replaced with vision only when necessary.
• Small, separate prompts are more reliable than one prompt that tries to do everything.
• The LLM produces suggestions, not unchecked database commands.
• A human only sees cases that would genuinely create new master data.
• Failure paths and repeatability belong in the design from the beginning.

That turns simple document classification into a small but complete AI Engineering workflow. More importantly, it leaves my archive tidier after processing than it was before.

Further reading

• paperless-ngx: Workflows and Webhooks
• paperless-ngx: REST API
• paperless-ngx: optional AI features
• LlamaIndex Workflows
• LlamaIndex Ollama Integration
• Windmill: Suspend and Approval
• paperless-ai
• Zettelrobbe, formerly paperless-ai-next