Ordnung statt Dubletten: Paperless-ngx mit Windmill, Ollama und LlamaIndex automatisieren
Wie ich Dokumente lokal klassifiziere, bestehende Korrespondenten und Dokumenttypen wiederverwende, schlechte OCR per Vision rette und nur neue Fälle per Telegram freigebe.
Ein LLM dazu zu bringen, eine Stromrechnung als Stromrechnung zu erkennen, ist inzwischen der einfache Teil. Der schwierige Teil beginnt direkt danach: Soll der Absender als „Stadtwerke Hogwarts“, „Stadtwerke Hogwarts GmbH“ oder „SWM“ gespeichert werden? Ist das Dokument eine „Rechnung“, eine „Energierechnung“ oder eine „Stromabrechnung“? Und was passiert, wenn der passende Korrespondent in Paperless längst existiert, nur unter einer leicht anderen Schreibweise?
Genau daran sind meine bisherigen Versuche mit fertigen Paperless-Erweiterungen gescheitert. Die Klassifikation war meistens inhaltlich richtig, das Archiv wurde mit der Zeit aber unordentlicher. Ein LLM ist ausgesprochen gut darin, einen plausiblen Namen zu formulieren. Es hat von sich aus jedoch keinen Grund, genau den Namen wiederzuverwenden, den ich vor drei Jahren einmal in Paperless angelegt habe.
Mein Ziel war deshalb nicht einfach „Dokumente mit KI taggen“. Ich wollte einen Agenten, der sich in eine bestehende Ordnung einfügt:
- Vorhandene Korrespondenten, Dokumenttypen und Tags haben Vorrang.
- Neue Korrespondenten und Dokumenttypen sind eine Ausnahme, kein normaler Output.
- Das Modell macht Vorschläge, aber der Anwendungscode entscheidet, was gültig ist.
- Nur wirklich neue Einträge brauchen eine manuelle Bestätigung.
- Schlechte OCR wird gezielt mit einem Vision-Modell nachbearbeitet, nicht grundsätzlich jedes Dokument.
- Ein Ausfall von Ollama oder ein verlorener Webhook darf kein Dokument verschwinden lassen.
Daraus ist ein kleiner AI-Inbox-Agent für paperless-ngx entstanden. Die Orchestrierung läuft bei mir in Windmill, die Dokumentanalyse als LlamaIndex-Workflow und die Modelle lokal über Ollama.
Der interessante Teil ist dabei nicht Windmill selbst. Die gleiche Architektur lässt sich auch als normaler Python-Dienst mit SQLite, einem Cronjob oder einem kleinen Webhook-Endpunkt betreiben. Dazu später mehr.
Was am Ende automatisch passiert
Ein neues Dokument landet wie gewohnt in Paperless. Ein Paperless-Workflow versieht es mit dem Tag inbox und ruft einen Windmill-Webhook auf. Von dort läuft die Verarbeitung weitgehend ohne Eingriff:
- Ollama wird auf Erreichbarkeit geprüft.
- Das Dokument wird gegen parallele Verarbeitung geclaimt.
- Der vorhandene OCR-Text wird auf Brauchbarkeit geprüft.
- Falls nötig, werden die Seiten gerendert und mit einem Vision-Modell erneut gelesen.
- Das Dokument wird gegen die bereits vorhandenen Korrespondenten, Dokumenttypen und Tags klassifiziert.
- Titel und Dokumentdatum werden in einem separaten Schritt extrahiert.
- Bestehende Zuordnungen werden direkt übernommen.
- Nur wenn ein neuer Korrespondent oder Dokumenttyp nötig scheint, bekomme ich eine Telegram-Nachricht mit „Anlegen“ und „Ablehnen“.
- Das Dokument wird über die Paperless-API aktualisiert, verliert den Tag
inboxund erhältai-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]Die zentrale Designentscheidung steckt bereits im ersten Kasten: Der Tag inbox in Paperless ist die Quelle der Wahrheit. Die Windmill-Queue dient nur dazu, neue Dokumente sofort zu verarbeiten. Selbst wenn ein Webhook verloren geht oder eine Queue-Variable beschädigt wird, findet der regelmässige Schedule-Lauf alle Dokumente mit inbox wieder.
Warum ich nicht einfach paperless-ai verwende
Mit paperless-ai und dem früher als paperless-ai-next bekannten Projekt Zettelrobbe gibt es bereits fertige Lösungen. Beide sind deutlich komplettere Anwendungen als mein Flow: Sie bringen eine Einrichtung im Browser, mehrere Modellanbieter, eine Historie und manuelle Verarbeitung mit. Zettelrobbe hat ausserdem die OCR- und Vision-Unterstützung stark ausgebaut. Wer eine möglichst schnell installierte Komplettlösung sucht, sollte sich diese Projekte zuerst ansehen.
Zum Stand Juli 2026 ist das ursprüngliche paperless-ai allerdings nicht mehr aktiv gepflegt. Zettelrobbe ist die weiterentwickelte Alternative und unterstützt sowohl vorhandene Metadaten als Prompt-Kontext als auch Optionen, die Ausgabe auf bestehende Werte zu beschränken.
Mein Problem liegt daher nicht darin, dass diese Projekte vorhandene Korrespondenten grundsätzlich ignorieren würden. Der Unterschied ist die Richtung, in der die Klassifikation funktioniert.
Die dort verwendeten Analyse-Schemata lassen das Modell Namen zurückgeben:
{
"correspondent": "Stadtwerke Hogwarts GmbH",
"document_type": "Stromrechnung",
"tags": ["Energie", "Haus"],
"document_date": "2026-07-14"
}Anschliessend versucht die Anwendung, diese Namen in Paperless wiederzufinden. In der aktuellen Implementierung geschieht das bei Korrespondenten und Dokumenttypen letztlich über einen exakten, nicht zwischen Gross- und Kleinschreibung unterscheidenden Namensvergleich. Wird kein passender Eintrag gefunden und ist die Ausgabe nicht strikt auf bestehende Werte begrenzt, kann ein neuer Eintrag angelegt werden.
Das ist bequem, verschiebt aber die Hoheit über die Taxonomie zum Modell. Für ein frisches Archiv kann das gut funktionieren. In einem über Jahre gewachsenen Archiv führen Rechtsformen, Abkürzungen, Bindestriche und unterschiedliche Granularität schnell zu Dubletten.
Mein Workflow dreht das um. Das Modell bekommt nicht nur Namen, sondern stabile IDs und muss zuerst aus diesen IDs wählen:
{
"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."
}Nur wenn wirklich kein vorhandener Eintrag passt, darf es zusätzlich einen neuen Namen vorschlagen:
{
"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."
}Dieser zweite Fall wird nicht automatisch ausgeführt, sondern landet bei mir in Telegram.
Ein konkretes Dubletten-Beispiel
Angenommen, Paperless enthält bereits:
[
{"id": 17, "name": "Stadtwerke Hogwarts"},
{"id": 23, "name": "Stadt Hogwarts"},
{"id": 31, "name": "Energieversorgung Mittelrhein"}
]Auf dem Briefkopf steht dagegen:
Stadtwerke Hogwarts GmbH
Max-Strom-Strasse 1
12345 HogwartsEin frei formulierendes Modell liefert mit hoher Wahrscheinlichkeit Stadtwerke Hogwarts GmbH. Das ist semantisch korrekt, aber kein exakter Treffer auf Stadtwerke Hogwarts. Mein Prompt stellt deshalb ausdrücklich klar, dass Rechtsformen, Adresszusätze und Schreibvarianten keinen neuen Korrespondenten rechtfertigen. Weil das Modell die ID 17 zurückgeben kann, muss es den vorhandenen Namen nicht exakt reproduzieren.
Noch wichtiger ist die nachgelagerte Validierung: Eine ID, die nicht in der übergebenen Liste existiert, wird verworfen. Liefert das Modell gleichzeitig eine gültige ID und einen neuen Namen, gewinnt immer die ID. Die entscheidende Regel steht also nicht nur im Prompt, sondern zusätzlich im Code.
Vergleich der Ansätze
| Eigenschaft | paperless-ai | Zettelrobbe, früher paperless-ai-next | Mein Workflow |
|---|---|---|---|
| Hauptziel | Komplettlösung mit UI, Klassifikation und RAG | Weiterentwickelte Komplettlösung mit UI und besserer OCR | Kontrollierte Automatisierung meines Inbox-Prozesses |
| Modellanbieter | Ollama und mehrere kompatible APIs | Ollama, OpenAI-kompatible APIs, Mistral OCR und weitere | Ollama, über LlamaIndex leicht austauschbar |
| Analyse | Im Wesentlichen ein gemeinsamer Analyse-Request | Im Wesentlichen ein gemeinsamer Analyse-Request | Getrennte Schritte für OCR-Prüfung, Vision, Klassifikation und Metadaten |
| Vorhandene Werte | Optional als Kontext oder Restriktion | Optional als Kontext oder Restriktion | Immer primärer Auswahlraum |
| Modell-Output | Namen | Namen | Bestehende IDs plus getrennte Vorschläge für neue Namen |
| Umgang mit Dubletten | Abhängig von Prompt, Restriktion und Namensabgleich | Abhängig von Prompt, Restriktion und Namensabgleich | ID-Validierung im Code, neue Werte nur als Ausnahme |
| Neue Korrespondenten und Typen | Je nach Einstellung automatisch möglich | Je nach Einstellung automatisch möglich | Nur nach expliziter Freigabe |
| Vorhandene Paperless-Zuordnung | Kein expliziter Stabilitätsanker im betrachteten Ollama-Prompt | Kein expliziter Stabilitätsanker im betrachteten Ollama-Prompt | Wird dem Modell mitgegeben und bei vertretbarer Zuordnung beibehalten |
| OCR-Fallback | Kein eigener mehrstufiger Qualitätsentscheid im betrachteten Ollama-Pfad | Umfangreiche lokale und externe Vision-OCR | Vision nur nach negativem OCR-Qualitätscheck |
| Oberfläche | Eigene Weboberfläche | Eigene Weboberfläche | Windmill für Betrieb, Telegram für Ausnahmen |
| RAG und Dokument-Chat | Ja | Bewusst entfernt | Nicht Teil dieses Flows |
| Wartung | Derzeit nicht aktiv gepflegt | Aktiv weiterentwickelt | Auf meinen Anwendungsfall zugeschnitten |
Das ist kein allgemeiner Benchmark und auch kein Urteil darüber, welches Projekt für jeden besser ist. Für mein Ziel ist die Sache aber ziemlich eindeutig: Ich möchte keine neue Dokumentenverwaltung neben Paperless und keinen Chat über mein Archiv. Ich möchte, dass meine bestehende Ablage zuverlässig gepflegt wird. Dafür funktioniert ein restriktiver, ID-basierter Workflow bei mir deutlich besser.
Nebenbei hat paperless-ngx inzwischen selbst optionale KI-Funktionen für Vorschläge und Dokument-Chat. Die Vorschläge werden jedoch bewusst pro Dokument angefordert. Mein Flow löst ein anderes Problem: eine automatische, fehlertolerante Verarbeitung mit eigener Freigabelogik.
Zwei Trigger, aber nur eine Wahrheit
Der schnellste Pfad beginnt in Paperless selbst. Unter Einstellungen → Workflows lässt sich ein Workflow mit dem Trigger „Dokument hinzugefügt“ und einer Webhook-Aktion anlegen. Aktuelle Paperless-Versionen stellen dafür unter anderem {{doc_id}} und {{doc_url}} bereit.
Mein Handler akzeptiert bewusst beides. Damit funktioniert er auch mit älteren Installationen oder Konfigurationen, in denen nur die Dokument-URL zuverlässig im Payload ankam:
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 };
}Der Webhook analysiert selbst nichts. Er schreibt nur die ID in eine kleine Queue und startet den eigentlichen Flow asynchron. Dadurch antwortet er schnell und muss nicht offen bleiben, während ein lokales Modell mehrere Seiten verarbeitet.
Als zweiter Trigger läuft alle 30 Minuten ein Schedule:
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/ZurichDer Schedule ist keine zweite unabhängige Verarbeitungsschiene. Beide Trigger starten denselben Flow. Dieser vereinigt zwei Kandidatenquellen:
- IDs aus der Webhook-Queue
- alle Dokumente, die in Paperless aktuell den Tag
inboxtragen
Die Queue verkürzt die Wartezeit. Der Tag verhindert Datenverlust.
Das Prinzip ist absichtlich langweilig. Eine Queue kann einen Eintrag verlieren. Ein Webhook kann während eines Deployments fehlschlagen. Ein lokaler Ollama-Rechner kann gerade schlafen. Solange das Dokument in Paperless inbox trägt, ist es beim nächsten Lauf wieder sichtbar.
Ollama prüfen und Dokumente claimen
Bevor der Flow ein Dokument lädt, prüft er /api/tags am Ollama-Server. Ist Ollama nicht erreichbar, wird kein Fehleralarm ausgelöst. Der Schritt liefert stattdessen ein normales Ergebnis wie dieses:
{
"ok": false,
"reason": "ollama unreachable: connection refused"
}Ollama-Downtime ist in meinem Homelab kein aussergewöhnlicher Fehlerzustand. Sie bedeutet nur: jetzt nicht, beim nächsten Schedule erneut.
Danach werden Queue und inbox-Dokumente zusammengeführt und bereits geclaimte IDs entfernt:
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 };
}Ein Claim ist hier ein Eintrag wie:
{
"4711": "2026-07-28T08:42:15.527Z"
}Nach sieben Tagen gilt er als verwaist und wird ignoriert. Das deckt abgestürzte Flows und nicht beantwortete Freigaben ab.
Eine ehrliche Grenze dieser Lösung
Die Windmill-Variablen werden per Read-Modify-Write aktualisiert. Das ist kein echtes transaktionales Lock. Zwei exakt gleichzeitig gestartete Flow-Läufe könnten theoretisch denselben alten Stand lesen und dasselbe Dokument claimen. In meinem privaten Archiv mit sequenzieller Verarbeitung ist das Risiko klein, und der inbox-Tag verhindert zumindest verlorene Dokumente. Die meisten Updates sind zudem wiederholbar.
Wer hohe Parallelität oder harte Exactly-once-Garantien braucht, sollte die Claims in einer Datenbank halten, beispielsweise mit einer Unique Constraint, SELECT ... FOR UPDATE SKIP LOCKED oder einem PostgreSQL Advisory Lock. In der Variante ohne Windmill zeige ich weiter unten eine kleine SQLite-Lease, die diesen Teil tatsächlich atomar ausführt.
Der Windmill-Flow bleibt absichtlich einfach
Nach dem Prerequisite-Schritt besteht der Flow aus einer Verzweigung und einer sequenziellen Schleife. Pro Dokument gibt es drei mögliche Aktionen:
- analysieren
- bei Bedarf auf eine Freigabe warten
- Paperless aktualisieren
Der relevante Ausschnitt aus flow.yaml sieht so aus:
- 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_documentparallel: false ist kein Versehen. Mein Ollama-Server soll nicht gleichzeitig mehrere lange Dokumente verarbeiten, und die Zahl neu eingehender Dokumente rechtfertigt keine Parallelisierung. skip_failures: true auf der Schleife sorgt dagegen dafür, dass ein defektes Dokument nicht den gesamten Batch blockiert.
Die sequenzielle Schleife hat allerdings eine Konsequenz: check_prereqs claimt zunächst den gesamten Batch. Braucht ein frühes Dokument eine Freigabe, warten auch die späteren Dokumente dieses Batches auf meine Telegram-Antwort. Für mein geringes Volumen und kurze Reaktionszeiten ist das vertretbar. Für einen allgemeinen Dienst würde ich immer nur das nächste Dokument claimen oder die Freigabe in einen separaten Flow auslagern.
Schlägt die Analyse eines Dokuments fehl, wird es nicht gepatcht. Damit behält es inbox und taucht nach Ablauf des Claims wieder auf.
Das Herzstück: ein LlamaIndex-Workflow
Die Dokumentanalyse ist kein autonomer Agent, der sich selbst Aufgaben ausdenkt. Das ist Absicht. Sie ist ein deterministischer, ereignisgesteuerter Workflow mit klaren Übergängen:
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 passen hier gut, weil jeder @step genau einen Ereignistyp konsumiert und einen neuen erzeugt. Die Übergänge ergeben sich aus den Typannotationen. Gleichzeitig kann ich gemeinsame Daten wie OCR-Text, Referenzlisten und Dateibytes im Context halten, ohne sie durch jedes Event zu kopieren.
Für eine eigenständige Python-Installation braucht man mindestens:
llama-index-core
llama-index-llms-ollama
llama-index-workflows
pymupdf
requestsIm produktiven Setup sollten diese Abhängigkeiten natürlich auf konkrete Versionen gepinnt werden. Bei Windmill gibt es dabei eine Besonderheit: Meine LlamaIndex-Imports liegen absichtlich innerhalb von build_workflow(). Dadurch lassen sich die Validierungsfunktionen ohne installierten LlamaIndex-Stack testen. Der Windmill-Metadaten-Generator erkennt diese dynamischen Imports aber nicht vollständig. Das Lockfile für dieses Script muss deshalb die transitive LlamaIndex-Abhängigkeitskette explizit enthalten.
Die Ereignisse und der Workflow-Kontext
Der Workflow braucht nur drei eigene 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: dictBeim ersten Schritt werden die Eingaben im Context abgelegt:
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 ist praktisch, um problematische Dokumente gezielt erneut zu testen. Im normalen Betrieb entscheidet jedoch zuerst ein günstiger Text-Request, ob Vision überhaupt nötig ist.
Prompt 1: Ist der OCR-Text wirklich brauchbar?
Eine reine Mindestlänge reicht für diese Entscheidung nicht. Ein Formular kann tausende korrekt erkannte Zeichen enthalten, obwohl gerade die handschriftlich ausgefüllten Felder fehlen. Deshalb lasse ich das Modell die semantische Brauchbarkeit beurteilen:
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]}"""Typische Antworten sind:
{
"ok": true,
"reason": "Absender, Rechnungsnummer, Betrag und Rechnungsdatum sind trotz kleiner OCR-Fehler eindeutig erkennbar."
}oder:
{
"ok": false,
"reason": "Es wurden fast nur die gedruckten Formularbeschriftungen erkannt; Name, Datum und handschriftliche Messwerte fehlen."
}Das Modell muss hier nichts klassifizieren. Es beantwortet nur eine eng begrenzte Frage. Das macht den Output erheblich stabiler als ein Prompt, der gleichzeitig OCR bewerten, Absender auswählen, Tags bestimmen und das Datum extrahieren soll.
Vision-Rescue statt Vision für alles
Ist der OCR-Text brauchbar, wird er unverändert übernommen:
@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 hierNur im negativen Fall lade ich die Originaldatei, rendere PDF-Seiten als PNG und schicke sie einzeln an das Vision-Modell. Die Grenzen sind bewusst konservativ:
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 pagesDer Vision-Prompt soll weder erklären noch zusammenfassen:
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 kann Text- und Bildblöcke gemeinsam an Ollama übergeben:
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()Das spart auf normalen PDFs viel Rechenzeit. Ein digital erzeugter Versicherungsbrief muss nicht erst in Bilder zerlegt werden, nur weil ein Vision-Modell verfügbar ist. Gleichzeitig bleiben handschriftlich ausgefüllte Formulare nicht dauerhaft unklassifiziert.
Prompt 2: Klassifizieren gegen eine vorhandene Taxonomie
Vor der Analyse lädt der Agent über die Paperless-API:
- das aktuelle Dokument
- alle Korrespondenten
- alle Dokumenttypen
- alle Tags
- die Originaldatei
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,
}Aus den Listen werden kompakte JSON-Zeilen gebaut. IDs sind wichtiger als Namen, weil genau sie später an die Paperless-API zurückgeschrieben werden:
corr_list = "\n".join(
f' {{"id": {item["id"]}, "name": "{item["name"]}"}}'
for item in correspondents
) or " (leer)"Der vollständige Klassifikations-Prompt ist der wichtigste Teil des Systems:
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]}"""Drei Details sind für die Trefferqualität entscheidend.
Erstens bekommt das Modell die aktuelle Paperless-Zuordnung mit. Paperless verfügt bereits über eigenes regel- und klassifikatorbasiertes Matching. Wenn dessen Ergebnis inhaltlich vertretbar ist, gibt es keinen Grund, bei jedem AI-Lauf eine neue Entscheidung zu erzwingen. Stabilität ist in einem Archiv oft wertvoller als eine marginal „schönere“ Bezeichnung.
Zweitens ist null kein Fehler. Es bedeutet: Unter den vorhandenen Einträgen gibt es keinen sicheren Treffer. Dieser Zustand ist besser als eine erfundene ID oder eine halbpassende Zuordnung.
Drittens sind neue und vorhandene Werte getrennte Felder. Damit kann der Code eindeutig unterscheiden, ob das Modell einen bestehenden Datensatz auswählen oder einen neuen vorschlagen wollte.
Der Workflow-Schritt selbst bleibt dadurch klein:
@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: Titel und Dokumentdatum separat extrahieren
Titel und Datum gehören bewusst nicht in den Klassifikations-Prompt. Besonders das Datum ist fehleranfällig, weil auf einem Dokument viele plausible Datumswerte stehen können: Geburtstag, Vertragsbeginn, Zahlungsfrist, Gültigkeitsende oder Lieferdatum.
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]}"""Die explizite Negativliste ist wichtiger als die Formulierung „finde das Dokumentdatum“. Ohne sie verwendet ein Modell bei einer Versicherungskarte gerne das Gültigkeitsdatum oder bei einer Mahnung das Zahlungsziel.
Der letzte Workflow-Schritt führt Klassifikation und Metadaten zusammen:
@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",
),
}
)Vier kleine Prompts sind zuverlässiger als ein grosser
Damit benötigt ein normales Dokument drei Modellaufrufe:
- OCR-Qualität
- Klassifikation
- Titel und Datum
Bei schlechtem OCR kommen pro gerenderter Seite Vision-Aufrufe hinzu. Das ist mehr als ein einzelner Mega-Prompt, aber jeder Aufruf hat eine eindeutigere Aufgabe und ein kleineres Ausgabeschema.
Für lokale Modelle ist das oft der bessere Tausch. Ein Modell kann gleichzeitig sehr gut lesen und trotzdem bei einem grossen JSON-Schema ein Feld vergessen. Durch die Trennung kann ich ausserdem unterschiedliche Textlängen verwenden:
MAX_OCR_CHECK_CHARS = 6000
MAX_CLASSIFY_CHARS = 24000
MAX_METADATA_CHARS = 12000Das OCR-Urteil braucht nicht das gesamte Dokument. Für die Klassifikation ist mehr Kontext sinnvoll. Für das Datum möchte ich wiederum vermeiden, dass ein langer Anhang mit weiteren Datumswerten die eigentliche Briefseite überlagert.
Meine Ollama-Konfiguration verwendet eine niedrige Temperatur:
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"],
}Bei einer kreativen Aufgabe wäre 0.1 unnötig restriktiv. Bei der Auswahl einer bestehenden Datenbank-ID ist Kreativität eher ein Fehler.
JSON aus einem lokalen Modell robust lesen
Auch mit der Anweisung „ausschliesslich JSON“ liefern Modelle gelegentlich Markdown-Codeblöcke oder einen einleitenden Satz. Für diesen begrenzten Fall reicht ein defensiver Extraktor:
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])Das ersetzt keine fachliche Validierung. Es sorgt nur dafür, dass die Verarbeitung nicht an einem Code-Fence scheitert. Ob die enthaltenen IDs zulässig sind, wird danach separat geprüft.
Das LLM schlägt vor, der Code entscheidet
Der wichtigste Code im gesamten Projekt ist weder der Ollama-Aufruf noch der Prompt, sondern die Normalisierung danach:
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",
"",
),
}Damit gelten unabhängig vom Modell folgende Invarianten:
- Eine Korrespondenten-ID muss wirklich in der zuvor geladenen Liste existieren.
- Eine Dokumenttyp-ID muss wirklich existieren.
- Tag-IDs müssen aus Paperless stammen.
- Eine gültige ID schlägt immer einen gleichzeitig vorgeschlagenen neuen Namen.
- Neue Tags werden case-insensitiv gegen bestehende Tags geprüft.
- Pro Dokument entstehen höchstens zwei neue Tags.
Ein Prompt kann die Wahrscheinlichkeit eines Fehlers reduzieren. Eine solche Funktion kann bestimmte Fehler ausschliessen. Genau diese Trennung macht für mich den Unterschied zwischen einer interessanten LLM-Demo und einer Automatisierung, der ich mein Archiv anvertraue.
Für die Datumsvalidierung verwende ich im aktuellen Code einen Formatcheck und falle bei einem ungültigen Ergebnis auf das Eingangsdatum zurück. Eine noch strengere Variante prüft zusätzlich, ob das Datum tatsächlich existiert:
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 fallbackSo würde auch ein formal passendes, aber unmögliches Datum wie 2026-19-42 abgefangen.
Den Workflow aus einem synchronen Script starten
LlamaIndex Workflows sind asynchron. Windmill ruft mein Python-Script dagegen über eine normale main()-Funktion auf. Der Event-Loop muss deshalb sauber gekapselt werden:
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,
}Wichtig ist, workflow.run() innerhalb der Coroutine aufzurufen. Der Workflow legt beim Start selbst Tasks an. Ein Ausdruck wie asyncio.run(workflow.run(...)) kann je nach Umgebung zu Problemen führen, weil der Aufruf bereits vor dem sauber gestarteten Event-Loop ausgewertet wird.
Human in the Loop, aber nur für Ausnahmen
Eine Freigabe für jedes Dokument wäre keine Automatisierung, sondern nur eine neue Inbox. Deshalb fragt der Flow ausschliesslich dann nach, wenn new_correspondent oder new_document_type gesetzt ist.
Die Nachricht enthält Titel, Vorschlag und Begründung:
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 kann einen Flow an einem Schritt suspendieren und über eine geheime Resume-URL später fortsetzen. Mein Telegram-Baustein erzeugt für jede Auswahl eine kurze Callback-ID und speichert dahinter die Resume-URL:
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,
};
}Ein separater Bot-Webhook löst die Callback-ID auf und ruft die Resume-URL mit folgendem Payload auf:
{
"choice": "approve"
}oder:
{
"choice": "reject"
}Der nächste Windmill-Schritt kann den Wert direkt als resume.choice lesen. Der Flow wartet höchstens sieben Tage. Bleibt die Antwort aus, wird das Dokument nicht aktualisiert und bleibt über seinen inbox-Tag auffindbar.
Telegram ist hier nur mein bevorzugter Kanal. Der gleiche Mechanismus funktioniert mit E-Mail, Matrix, Slack, einer kleinen Weboberfläche oder schlicht einer Windmill-Approval-Seite.
Erst beim Update dürfen neue Datensätze entstehen
Der Analyse-Schritt schreibt nichts nach Paperless. Erst update_document.py verarbeitet den freigegebenen Vorschlag.
Neue Korrespondenten und Dokumenttypen werden nur bei approve angelegt:
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() behandelt ausserdem eine kleine Race Condition. Zwei Dokumente könnten im selben Batch denselben neuen Absender vorschlagen. Wenn der zweite POST fehlschlägt, wird noch einmal case-insensitiv nach einem inzwischen existierenden Eintrag gesucht:
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"]
raiseDanach werden die Tags vereinigt, inbox entfernt und ai-processed ergänzt:
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)Das eigentliche PATCH bleibt unspektakulär:
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,
)Erst nach einem erfolgreichen PATCH wird der Claim entfernt. Bricht vorher etwas ab, bleibt er bis zu seinem TTL bestehen und das Dokument behält inbox.
Fehlerfälle sind Teil der Architektur
Bei einer Automatisierung, die dauerhaft läuft, sind Fehler keine Ausnahme. Sie sind normaler Betrieb. Deshalb hat jeder erwartbare Fall einen definierten Ausgang:
| Situation | Verhalten |
|---|---|
| Ollama ist nicht erreichbar | Flow endet ohne Alarm; Schedule versucht es später erneut |
| Webhook geht verloren | Schedule findet das Dokument über inbox |
| Zwei Trigger laufen nahezu gleichzeitig | Claims reduzieren Doppelverarbeitung; Datenbank-Lock wäre die strengere Variante |
| Ein Dokument schlägt in der Analyse fehl | Andere Dokumente im Batch laufen weiter |
| Vision liefert keinen Text | Dokument bleibt unverändert und damit in inbox |
| Telegram wird nicht beantwortet | Suspend läuft aus; Claim verfällt, Dokument bleibt auffindbar |
| Ein neuer Korrespondent wurde parallel angelegt | create_named() sucht nach fehlgeschlagenem POST erneut |
| Modell halluziniert eine ID | Validierung setzt sie auf null |
| Modell schlägt vorhandenen Tag erneut vor | Case-insensitiver Abgleich entfernt den Vorschlag |
| Paperless-PATCH schlägt fehl | Claim wird nicht aktiv freigegeben; Dokument wird später erneut versucht |
Diese Fehlersemantik ist ein grösserer Teil des Projekts als der eigentliche Modellaufruf. Genau deshalb verwende ich dafür den Begriff AI Engineering und nicht nur Prompt Engineering.
Nachbauen mit Windmill
Die Architektur lässt sich mit wenigen klar getrennten Bausteinen nachbauen. Für ein eigenes Setup sind die folgenden Schritte entscheidend.
1. Voraussetzungen
Benötigt werden:
- eine laufende paperless-ngx-Instanz
- ein Paperless-API-Token
- eine Windmill-Instanz mit einem Python-Worker
- ein erreichbarer Ollama-Server
- ein Textmodell und bei Bedarf ein Vision-Modell
- optional ein Telegram-Bot
Paperless und Ollama müssen nicht auf demselben Host laufen. Der Windmill-Worker muss beide erreichen können. Ollama sollte dabei nicht ungeschützt aus dem Internet erreichbar sein.
2. Tags in Paperless anlegen
Mindestens diese beiden Tags werden erwartet:
inbox
ai-processedinbox markiert offene Arbeit. ai-processed ist kein technisches Muss, macht aber Filter und Fehlersuche deutlich einfacher.
3. Paperless-API-Token erzeugen
Der Token wird bei jedem API-Aufruf als Header gesendet:
Authorization: Token <PAPERLESS_API_TOKEN>In Windmill gehört er in eine Secret Variable, nicht direkt in ein öffentlich sichtbares Resource-YAML.
4. Windmill-Resources anlegen
Eine Ollama-Resource kann so aussehen:
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_configFür Paperless:
description: Paperless-API
value:
base_url: http://paperless.internal:8000
api_token: $var:f/paperless/config/paperless-api-token
resource_type: paperless_configDie Namen der Modelle sind austauschbar. Wichtig ist nur, dass vision_model tatsächlich Bilder verarbeiten kann. Auf kleineren Systemen kann ein kompaktes Textmodell für Klassifikation und ein separates Vision-Modell sinnvoller sein.
5. Scripts und Flow deployen
Die Windmill-Umsetzung gliedert sich in folgende Bausteine:
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.tsZusätzlich verwendet mein Flow den generischen Telegram-Baustein unter f/inbox/notify/.
Nach dem Import sollten zuerst die drei Scripts einzeln mit einem Testdokument ausgeführt werden. So lassen sich API-, Modell- und Dependency-Probleme voneinander trennen.
6. Paperless-Workflow konfigurieren
In Paperless:
- Workflow mit Trigger „Dokument hinzugefügt“ erstellen.
- Optional auf Dokumente mit
inboxfiltern oderinboxüber eine vorherige Workflow-Aktion setzen. - Als Aktion einen Webhook auf die Windmill-Webhook-URL eintragen.
- JSON als Encoding wählen.
- Folgenden Body senden:
{
"doc_id": {{doc_id}},
"doc_url": "{{doc_url}}"
}Die URL ist nur als Fallback enthalten. In einer aktuellen Paperless-Version genügt normalerweise doc_id.
Falls Paperless interne HTTP-Ziele nicht aufrufen darf, müssen die Webhook-Einstellungen geprüft werden. Relevant sind unter anderem die erlaubten Schemes, Ports und internen Ziele.
7. Drei Testfälle verwenden
Ein einzelnes Test-PDF sagt wenig über die Qualität aus. Ich würde mindestens diese Fälle prüfen:
Fall A: vorhandener Korrespondent und vorhandener Dokumenttyp
Beispiel: eine weitere Rechnung eines bekannten Energieversorgers.
Erwartung:
{
"correspondent_id": 17,
"new_correspondent": null,
"document_type_id": 4,
"new_document_type": null,
"needs_approval": false
}Fall B: bekannter Korrespondent in abweichender Schreibweise
Beispiel: In Paperless steht Stadtwerke Hogwarts, im Dokument Stadtwerke Hogwarts GmbH.
Erwartung: vorhandene ID, kein neuer Vorschlag.
Fall C: tatsächlich neuer Absender
Erwartung:
{
"correspondent_id": null,
"new_correspondent": "Solarwerk Bodensee",
"needs_approval": true
}Zusätzlich lohnt sich ein handschriftlich ausgefülltes Formular, um sicherzustellen, dass ocr_rescued auf true wechselt.
Die gleiche Lösung ohne Windmill
Windmill nimmt mir einige Dinge ab: Webhooks, Schedules, Secrets, Logs, Flow-Visualisierung und Suspend/Resume. Keine dieser Funktionen ist für die Dokumentanalyse selbst zwingend.
Die einfachste eigenständige Architektur besteht aus vier Bausteinen:
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)]Auch hier gilt: Die Queue ist nur der schnelle Pfad. Der Worker fragt bei jedem Lauf zusätzlich alle Dokumente mit inbox ab.
Variante A: Nur Polling und Cron
Für ein privates Archiv reicht häufig ein Script, das alle fünf Minuten läuft:
*/5 * * * * /opt/paperless-agent/.venv/bin/python /opt/paperless-agent/worker.py --onceDer Webhook entfällt vollständig. Das erhöht die Latenz auf höchstens fünf Minuten, reduziert die Infrastruktur aber auf ein einziges Python-Programm.
Der Ablauf in worker.py ist:
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,
)Ohne Telegram kann mark_for_review() einfach einen Tag wie ai-review setzen. Man prüft diese wenigen Fälle direkt in Paperless und startet sie danach erneut. Das ist weniger elegant als Suspend/Resume, aber sehr robust und leicht verständlich.
Atomare Claims mit SQLite
SQLite eignet sich für diese Aufgabe überraschend gut. Eine Tabelle mit doc_id als Primary Key verhindert, dass zwei Worker dieselbe ID gleichzeitig übernehmen.
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 reserviert die Schreibtransaktion, bevor der alte Claim gelöscht und der neue eingefügt wird. Die Primary Key Constraint macht den Claim auch bei mehreren Prozessen eindeutig.
Eine dauerhafte Queue für den Webhook
Wer die sofortige Verarbeitung behalten möchte, ergänzt einen kleinen FastAPI-Endpunkt. Dieser startet die Analyse nicht im HTTP-Request, sondern schreibt nur in 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,
}Ein lang laufender Worker kann die Queue alle paar Sekunden prüfen. Ein zusätzlicher Cronjob bleibt als Backstop bestehen und liest weiterhin den inbox-Tag aus Paperless.
Wichtig ist, keine FastAPI-BackgroundTasks als einzige Queue zu verwenden. Sie sind bequem, aber nicht dauerhaft. Bei einem Prozessneustart geht der Auftrag verloren. SQLite, Redis, PostgreSQL oder ein richtiger Message Broker sind dafür besser geeignet.
Ein einfacher eigenständiger Worker
Die Kandidatenmenge entsteht genau wie in der Windmill-Version aus Queue und 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 einer reinen Review-Tag-Variante sollte der Claim nach dem Speichern der Approval-Anfrage entweder bewusst bestehen bleiben oder durch einen separaten Status ersetzt werden. Sonst würde der nächste Poll das Dokument sofort erneut analysieren. Eine klare State Machine ist besser als mehrere implizite Tags:
queued -> processing -> pending_approval -> completed
\-> failed/retrySQLite kann diesen Status direkt in einer jobs-Tabelle halten. Der Paperless-Tag bleibt trotzdem die übergeordnete Absicherung gegen verlorenen internen Zustand.
Telegram ohne Windmill
Auch Telegram lässt sich ohne Windmill nachbauen. Benötigt werden:
- eine
approvals-Tabelle mit zufälligem, einmal verwendbarem Token - Inline-Buttons mit
callback_data=<token>:approve - ein Telegram-Webhook, der Token und Entscheidung prüft
- ein Worker, der die gespeicherte Entscheidung verarbeitet
Ein Datensatz könnte so aussehen:
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
);Der Bot-Handler darf nicht einfach der Dokument-ID vertrauen. Das zufällige Token sollte kurzlebig und nach der ersten Verwendung ungültig sein. Nach approve ruft der Worker denselben update_document()-Code wie die Windmill-Version auf.
Für ein privates Setup ist die Alternative mit ai-review oft ausreichend. Der Telegram-Teil lohnt sich, wenn man neue Dokumente auch unterwegs freigeben möchte.
Betrieb als systemd-Service
Ein dauerhafter Worker lässt sich ohne Container als systemd-Service betreiben:
[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.targetDie Secrets liegen in /etc/paperless-agent.env mit restriktiven Dateirechten:
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-a3bDamit bleibt vom Windmill-Setup am Ende nur normaler Python-Code übrig. Der LlamaIndex-Workflow kann unverändert übernommen werden.
Tests, die sich wirklich lohnen
LLM-Aufrufe selbst sind nicht deterministisch genug für klassische Unit-Tests. Die fachlichen Schranken drumherum lassen sich dagegen sehr gut testen.
Eine gültige ID gewinnt gegen einen neuen Namen
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 NoneHalluzinierte IDs werden verworfen
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"] == []Neue Tags werden begrenzt und dedupliziert
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",
]Zusätzlich teste ich JSON-Extraktion, Datums-Fallback, Tag-Handling beim Update und das Freigeben des Claims. Diese Tests brauchen weder Ollama noch LlamaIndex und laufen deshalb schnell in jeder CI-Pipeline.
Wie man die Qualität sinnvoll misst
„Funktioniert bei mir besser“ ist zunächst eine Betriebserfahrung, kein wissenschaftlicher Benchmark. Wer verschiedene Modelle oder Prompts vergleichen möchte, sollte ein festes Set bereits korrekt klassifizierter Dokumente verwenden und mindestens folgende Werte erfassen:
| Metrik | Aussage |
|---|---|
| Korrespondent korrekt | Wurde die richtige vorhandene ID gewählt? |
| Dokumenttyp korrekt | Wurde die richtige vorhandene ID gewählt? |
| Wiederverwendungsquote | Wie oft konnte ein vorhandener Eintrag genutzt werden? |
| Falsche Neuvorschläge | Wie oft wurde trotz vorhandenem Treffer ein neuer Name vorgeschlagen? |
| Approval-Rate | Wie viele Dokumente brauchen einen Menschen? |
| Titel brauchbar | Ist der Titel kurz, suchbar und inhaltlich richtig? |
| Dokumentdatum korrekt | Wurde wirklich das Ausstellungsdatum gewählt? |
| Vision-Erfolgsquote | Wie viele OCR-Problemfälle konnten gerettet werden? |
| Fehler pro 100 Dokumente | Wie oft endet der technische Ablauf ohne Update? |
Für meinen Anwendungsfall sind die falschen Neuvorschläge fast wichtiger als die reine Klassifikationsgenauigkeit. Ein falsch zugeordneter Eintrag fällt bei einer Stichprobe auf. Zehn fast identische Korrespondenten verschlechtern dagegen dauerhaft Suche, Filter und zukünftige Klassifikation.
Wo ich die Lösung noch ausbauen würde
Der aktuelle Stand funktioniert für mein privates Dokumentaufkommen gut, ist aber nicht das Ende der möglichen Optimierungen.
Kandidaten vorfiltern
Bei einigen Dutzend oder wenigen hundert Korrespondenten kann die vollständige Liste direkt in den Prompt. Bei mehreren tausend Einträgen wird das teuer und unübersichtlich. Dann würde ich vor dem LLM-Aufruf Kandidaten bilden:
- Normalisierung von Rechtsformen und Satzzeichen
- Fuzzy Matching auf erkannte Absenderzeilen
- Embeddings über Namen und bekannte Dokumentbeispiele
- Übergabe nur der besten 20 bis 50 IDs an das Modell
Das Modell bleibt Entscheider innerhalb einer kontrollierten Kandidatenmenge. Der Code validiert weiterhin gegen die echten Paperless-IDs.
Konfidenz explizit modellieren
Derzeit entscheidet null gegen ID. Zusätzlich könnte das Modell eine grobe Konfidenz liefern:
{
"correspondent_id": 17,
"correspondent_confidence": 0.93,
"document_type_id": 4,
"document_type_confidence": 0.88
}Ich würde eine solche Zahl aber nie ungeprüft als echte Wahrscheinlichkeit behandeln. Sinnvoll wäre sie erst nach Kalibrierung auf einem eigenen Testset. Bis dahin ist ein diskreter Zustand wie certain, plausible oder uncertain häufig ehrlicher.
Approval differenzierter gestalten
Statt nur „Anlegen“ oder „Ablehnen“ könnte Telegram zusätzlich die drei ähnlichsten vorhandenen Korrespondenten anbieten. Das wäre besonders nützlich, wenn das Modell keinen sicheren Treffer findet, aber der Mensch sofort erkennt, welcher bestehende Eintrag gemeint ist.
Claims in eine echte Datenbank verschieben
Die Windmill-Variablen sind für mein Volumen ausreichend, aber kein transaktionales Queue-System. Eine kleine PostgreSQL-Tabelle oder Windmill Data Table würde parallele Worker und atomare Leases sauberer abbilden. Dabei würde ich auch das Claimen vom Batch auf das einzelne Dokument verschieben. Eine wartende Freigabe blockiert dann keine bereits geclaimten Folgedokumente.
Neue Tags bei Bedarf ebenfalls freigeben
Neue Tags werden in meinem aktuellen Flow ohne Telegram-Rückfrage angelegt. Die Zahl ist auf zwei begrenzt und bestehende Namen werden case-insensitiv herausgefiltert. Sollte die Tag-Liste trotzdem unruhig werden, lässt sich dieselbe Approval-Regel wie bei Korrespondenten und Dokumenttypen anwenden.
Observability ergänzen
Der nächste sinnvolle Schritt wäre eine kleine Auswertung über:
- Laufzeit je Workflow-Schritt
- Token- oder Prompt-Länge
- Anteil der Vision-Fallbacks
- Approval-Rate
- häufigste neue Vorschläge
- technische Fehler nach Kategorie
Damit lässt sich erkennen, ob ein Modellwechsel wirklich hilft oder nur anders klingt.
Fazit
Die eigentliche Herausforderung bei der KI-gestützten Dokumentenablage ist nicht das Erkennen einer Rechnung. Es ist die kontrollierte Integration in eine bestehende Ordnung.
Mein Paperless-Agent behandelt Korrespondenten, Dokumenttypen und Tags deshalb nicht als frei formulierbare Texte, sondern als vorhandene Referenzdaten. Das Modell wählt bevorzugt stabile IDs. Neue Namen werden getrennt vorgeschlagen. Der Code verwirft ungültige IDs, begrenzt neue Tags und lässt neue Stammdaten erst nach einer Freigabe entstehen.
Windmill macht die Orchestrierung angenehm: Webhook, Schedule, Secrets, Schleifen und Telegram-Suspend lassen sich übersichtlich zusammensetzen. Die tragenden Ideen sind aber unabhängig davon:
- Der
inbox-Tag ist die Quelle der Wahrheit. - Eine Queue beschleunigt, darf aber nie der einzige Zustand sein.
- OCR wird zuerst bewertet und nur bei Bedarf durch Vision ersetzt.
- Kleine, getrennte Prompts sind zuverlässiger als ein Alleskönner-Prompt.
- Das LLM liefert Vorschläge, nicht ungeprüfte Datenbankbefehle.
- Ein Mensch sieht nur die Fälle, in denen tatsächlich neue Stammdaten entstehen würden.
- Fehlerpfade und Wiederholbarkeit gehören von Anfang an zum Design.
Damit ist aus einer einfachen Dokumentklassifikation ein kleiner, aber vollständiger AI-Engineering-Workflow geworden. Und vor allem einer, der mein Archiv nach der Verarbeitung ordentlicher hinterlässt als vorher.