Vereinfachen Sie die Orchestrierung von KI-Agenten mit Lakebase Postgres

Einführung

Traditionell ist die Wirtschaftsprüfung ein mühsamer Prozess, der oft eine detaillierte Dokumentenprüfung und Informationsextraktion erfordert. Um diesen Prozess zu beschleunigen, hat CLA (CliftonLarsonAllen LLP), ein führendes Dienstleistungsunternehmen mit wachsender globaler Präsenz, gemeinsam mit dem Forward Deployed Engineering-Team von Databricks eine agentenbasierte Prüfungslösung entwickelt und in Betrieb genommen. Gemeinsam haben wir eine Anwendung zur Dokumentenverarbeitung entwickelt, die die Extraktionszeit von Stunden auf Minuten verkürzt – ohne Qualitätsverluste. Die Anwendung basiert vollständig auf Databricks und nutzt Lakebase Postgres, Databricks Apps, Lakeflow Jobs, MLflow und Unity Catalog Volumes. In diesem Blog konzentrieren wir uns auf eine Schlüsselkomponente dieses Systems: die von Lakebase unterstützte Orchestrierungsschicht.

Die Orchestrierungsschicht ist für die Koordinierung lang laufender Aufgaben, die Verwaltung von Wiederholungsversuchen, die Kostenzuordnung und die Bereitstellung von Echtzeittransparenz verantwortlich. Mit Lakebase und Databricks Apps machen wir eine separate Infrastruktur für Warteschlangen, Orchestrierung und Observability überflüssig.

Lakebase macht diese Architektur auch im großen Maßstab praxistauglich, indem es Speicher und Rechenleistung voneinander trennt. Im Gegensatz zu herkömmlichen Postgres-Bereitstellungen kann die Rechenleistung je nach Bedarf skaliert werden, während der Speicher beständig und unabhängig bleibt. Zusammen machen diese Funktionen Lakebase zu einer praktischen Grundlage für ein einfacheres, skalierbares Orchestrierungsmuster für lang laufende agentenbasierte Workloads auf Databricks.

Herausforderungen bei der Orchestrierung von agentenbasierten Workloads

Das Parsen von Dokumenten ist ein sehr häufiger, volumenstarker agentenbasierter Workload. Unternehmen aller Branchen müssen große Mengen an Verträgen, Rechnungen, Finanzberichten und anderen Dokumenten in strukturierte Daten umwandeln. Die Ausführung in großem Maßstab bringt fünf spezifische Probleme verteilter Systeme mit sich:

• Unvorhersehbare Latenz pro Aufgabe: Eine zweiseitige Rechnung kann in Sekundenschnelle verarbeitet werden, während ein zweihundertseitiger Vertrag mehrere Minuten dauern kann. Dies macht es schwierig vorherzusagen, wie lange eine einzelne Aufgabe dauern wird.
• Ratenbegrenzungsbewusste Drosselung: Endpunkte für LLM- und Vision-Modelle begrenzen die Anzahl der Anfragen und Token, die sie in einem bestimmten Zeitraum verarbeiten können. Das gleichzeitige Senden von Hunderten von Aufgaben kann diese Grenzwerte überschreiten, eine Drosselung auslösen und zu wiederholten Wiederholungsversuchen führen. Der Orchestrator muss laufende Arbeiten proaktiv begrenzen (durch die Anzahl gleichzeitiger Aufgaben, das Token-Budget oder beides), anstatt sich nur auf reaktive Wiederholungsversuche zu verlassen.
• Priorisierung von Workloads: Dringende Einreichungen sollten nicht hinter großen Batch-Verarbeitungen zurückgestellt werden. Die Priorität pro Aufgabe stellt sicher, dass Arbeit mit höherer Priorität (interaktive Einreichungen, Premium-Anfragen, vom Operator initiierte Neuverarbeitungen) zuerst bereitgestellt wird.
• Kostenzuordnung pro Aufgabe: Finanzteams müssen die Ausgaben bestimmten Aufgaben, Kunden und Agenten zuordnen können, aufgeschlüsselt nach AI-Token-Nutzung und Rechenleistungsverbrauch.
• Echtzeit-Fortschrittsanzeige: Benutzer, die Hunderte von Dokumenten hochladen, benötigen eine Live-Fortschrittsanzeige.

Viele Unternehmen kombinieren mehrere spezialisierte Systeme für Orchestrierung und Observability. Jedes System bringt seine eigene Infrastruktur, Authentifizierung, Überwachung und betriebliche Anforderungen mit sich, ebenso wie den Aufwand für deren Integration. Für lang laufende, unabhängige agentenbasierte Aufgaben steht dieser Aufwand in keinem Verhältnis zur tatsächlichen Komplexität der Zeitplanung.

Die von uns entwickelte Databricks-native Lösung erfüllt alle oben genannten Anforderungen mit Lakebase als Grundlage.

Lösungsarchitektur

Der gesamte Anwendungs-Stack besteht ausschließlich aus Databricks-Diensten:

• Webanwendung (Databricks Apps). Eine FastAPI-basierte Benutzeroberfläche, auf der Benutzer PDFs hochladen (gespeichert in Unity Catalog Volumes) und Parse-Anfragen senden. Anfragen werden direkt in die Lakebase-Task-Tabelle geschrieben.
• Lakebase. Eine automatisch skalierende Postgres-Datenbank, die den relationalen Zustand des Orchestrators über zusammenhängende Tabellen hinweg hostet: tasks (zu parsende Dokumente mit Status, Lease-Informationen und dem strukturierten Ergebnis) und task_attempts (eine Zeile pro Ausführungsversuch, die die Databricks Job-Run-ID, die MLflow-Trace-ID und die Kosten-Metadaten pro Versuch erfasst). Lakebase dient als einzige Quelle der Wahrheit für den Orchestrator-Zustand.
• Orchestrator (Databricks Apps). Ein lang laufender Worker-Daemon und ein Operator-Dashboard. Der Daemon entnimmt Aufgaben aus Lakebase, leitet sie an die AI-Agents-Schicht weiter und schreibt die Ergebnisse zurück. Das Dashboard liest dieselben Tabellen aus, um den Status in Echtzeit anzuzeigen.
• AI Agents (Lakeflow Jobs). Lakeflow Jobs führen die Parsing-Arbeit aus. Jeder Job liest ein PDF aus den Unity Catalog Volumes, verarbeitet es über Intelligent Document Processing und Vision-/LLM-Aufrufe, speichert die geparste Ausgabe in Lakebase und bestätigt dies dem Orchestrator über einen Webhook. MLflow Tracing erfasst Ausführungsdetails wie Modellaufrufe, Token-Nutzung, Latenz und Kosten-Metadaten.

Der Datenfluss zwischen den Komponenten erfolgt wie folgt: Die Web-App schreibt PDFs in Unity Catalog Volumes und Parse-Anfragen in Lakebase. Der Orchestrator entnimmt Aufgaben aus Lakebase und leitet Databricks Jobs an die AI-Agents-Schicht weiter. Die AI Agents verarbeiten Dokumente, schreiben die Ergebnisse zurück in Lakebase und senden Status-Updates an den Orchestrator zurück.

Aufgrund dieser integrierten Funktionen auf Databricks mussten wir uns nicht auf externe Message Broker (Kafka, Redis), separate Scheduler (Airflow, Temporal) oder dedizierte Caching-Schichten verlassen.

Implementierung der Aufgabenwarteschlange

Die Aufgabenwarteschlange wird durch zwei Postgres-Tabellen in Lakebase unterstützt. Die Tabelle tasks enthält eine Zeile pro logischer Arbeitseinheit und erfasst den aktuellen Status der Aufgabe, Lease-Informationen, die Agenten-Zuweisung, die übergeordnete Extraktion und das Endergebnis. Die Tabelle task_attempts enthält eine Zeile pro Ausführungsversuch und erfasst die Databricks Job-Run-ID, die MLflow-Trace-ID und die Kosten-Metadaten pro Versuch. Die Parent-Child-Beziehung unterstützt Wiederholungsversuche (eine einzelne Aufgabe kann mehrere Versuche haben) und bewahrt die Observability auf Versuchsebene für die Kostenzuordnung und das Debugging.

Zwei Postgres-Tabellen allein sind noch keine Aufgabenwarteschlange. Vier Postgres-native Muster verwandeln sie in eine robuste, hochgradig parallele, ausfallsichere und ratenbegrenzungsbewusste Warteschlange, die für lang laufende agentenbasierte Workloads geeignet ist.

Paralleles, prioritätsbewusstes Dequeuing

Eine einfache Dequeue-Abfrage wählt möglicherweise die nächste verfügbare Aufgabe mit WHERE status = 'enqueued' und LIMIT batch_size aus. Obwohl diese Abfrage eine eingereihte Aufgabe korrekt identifiziert, reicht sie nicht aus, wenn mehrere Worker gleichzeitig Aufgaben entnehmen. Ohne Zeilensperrung wählen möglicherweise mehrere Worker dieselbe Aufgabe aus, bevor der Status aktualisiert wird.

Das Hinzufügen von FOR UPDATE SKIP LOCKED macht das Dequeueing nebenläufigkeitssicher. Jeder Worker sperrt die von ihm ausgewählte Zeile, während andere Worker diese Zeile überspringen und mit der nächsten verfügbaren Aufgabe fortfahren. Darüber hinaus stellt eine ORDER BY priority DESC, created_at-Klausel sicher, dass Aufgaben mit höherer Priorität zuerst ausgewählt werden, während die FIFO-Reihenfolge innerhalb jeder Prioritätsstufe beibehalten wird.

Die vollständige nebenläufigkeitssichere und prioritätsstabile Anweisung lautet:

Absturzsicherung durch Lease-basiertes Sperren

Worker können mitten in einer Aufgabe aufgrund von VM-Eviction, Out-of-Memory-Bedingungen oder Deployment-Ereignissen abgebrochen werden. Wenn abgebrochene Aufgaben weiterhin als in Bearbeitung markiert bleiben, würden sie unbegrenzt blockiert. Die Lösung besteht darin, beim Dequeueing ein ablaufendes Lease zu erfassen:

Ein periodischer Sweeper reiht jede Aufgabe, deren lease_expires_at abgelaufen ist, wieder in die Warteschlange ein. Aufgaben, die von abgebrochenen Workern gehalten werden, werden innerhalb weniger Minuten automatisch wiederhergestellt, ohne dass ein externer Koordinationsdienst erforderlich ist.

Rate-Limit-bewusste Drosselung

LLM- und Vision-Modell-Endpunkte erzwingen typischerweise zwei verschiedene Kontingente: eine Obergrenze für Anfragen pro Sekunde und eine Obergrenze für Token pro Minute (TPM). Eine einzige Drosselungsstrategie passt selten für beides. Der Orchestrator unterstützt drei Modi, die pro Agent über die Konfiguration ausgewählt werden.

Obergrenze für Nebenläufigkeit. Ein MAX_CONCURRENT_TASKS-Parameter begrenzt die Anzahl der Aufgaben, die der Orchestrator gleichzeitig verteilt. Die Obergrenze wird beim Dequeueing erzwungen, indem die aktuellen PROCESSING-Zeilen in der Aufgabentabelle gezählt werden:

Wenn die Anzahl bei oder über der Obergrenze liegt, wird keine neue Aufgabe aus der Warteschlange entnommen. Die Verankerung der Prüfung auf der Anzahl der Datenbankzeilen anstelle der Warteschlangengröße des lokalen Executors sorgt dafür, dass die Obergrenze über Worker-Restarts, Lease-Wiederherstellungen und Multi-Replica-Deployments hinweg präzise bleibt. Dieser Modus eignet sich gut für Endpunkte, die durch Anfragen-pro-Sekunde-Limits eingeschränkt sind und bei denen der Token-Verbrauch jeder Aufgabe in etwa einheitlich ist.

Token-Budget. Ein MAX_TPM-Parameter begrenzt die prognostizierte Token-Rate über alle aktiven Aufgaben hinweg. Der Orchestrator schätzt die Token-Anzahl einer Aufgabe und summiert die prognostizierte Token-Rate über alle PROCESSING-Aufgaben. Eine neue Aufgabe wird nur dann aus der Warteschlange entnommen, wenn die Summe plus die prognostizierten Token der neuen Aufgabe in das Budget passt.

Kombinierte Obergrenze. Wenn sowohl MAX_CONCURRENT_TASKS als auch MAX_TPM konfiguriert sind, wendet der Orchestrator die jeweils strengere Einschränkung an. Dieser Modus bewältigt Workloads, die in einem Szenario durch die Nebenläufigkeit begrenzt sind (viele kurze, günstige Aufgaben) und in einem anderen durch Token (ein einzelnes, sehr langes Dokument, das das Minutenkontingent auslastet).

In allen drei Modi wird die Drosselungsentscheidung beim Dequeueing innerhalb derselben Transaktion wie FOR UPDATE SKIP LOCKED getroffen. Eine Aufgabe, die nicht in das aktuelle Kontingent passt, bleibt in der Warteschlange und wird beim nächsten Dequeue-Zyklus erneut berücksichtigt – kein separater Scheduling-Status, keine In-Memory-Warteschlange, keine Koordinationsschicht zwischen Worker-Replikaten.

Idempotente Webhook-Callbacks

Wenn die AI-Agents-Ebene eine Aufgabe abschließt, sendet sie einen Callback mit dem Ergebnis an den Orchestrator. Die Callback-Zustellung erfolgt nicht genau einmal (Exactly-Once): Databricks führt möglicherweise erneute Versuche durch, Netzwerke können unterbrochen werden und Proxys können Daten erneut zustellen. Der Callback-Handler ist so konzipiert, dass er idempotent ist: Er akzeptiert sowohl den Status PROCESSING als auch ENQUEUED und behandelt bereits abgeschlossene Aufgaben als No-Ops. Identische Payloads führen zu identischen Ergebnissen, wodurch das Risiko einer doppelten Abrechnung oder doppelten Verarbeitung ausgeschlossen wird.

Diese vier Muster zusammen ergeben eine Aufgabenwarteschlange, die bei Nebenläufigkeit korrekt, bei Abstürzen ausfallsicher, unter Last Rate-Limit-bewusst und bei Wiederholungsversuchen idempotent ist. Echtzeit-Einblick in das laufende System wird durch einen separaten Mechanismus ermöglicht, der im nächsten Abschnitt beschrieben wird.

Echtzeit-Operator-Dashboard

Wenn viele Dokumente aktiv sind, benötigen Operator eine klare Sicht auf die Agent-Performance, den Aufgabenstatus und die Workload-Kosten. Sie sollten nicht ständig den Orchestrator abfragen oder sich auf eine separate Metrikplattform verlassen müssen. Der Orchestrator integriert diese Funktionalität direkt in ein einziges Dashboard, das von derselben Databricks-App bereitgestellt wird, auf der auch der Worker-Daemon läuft.

Dashboard-Funktionen

Das für Operator konzipierte Dashboard stellt eine Reihe von Betriebskennzahlen bereit, die zusammen das laufende System beschreiben. Alle Metriken unterstützen die Filterung nach Datumsbereich, Aufgabenstatus und Agent.

• Aufgabensummen nach Status. Anzahl der Aufgaben in jedem Status (eingereiht, in Bearbeitung, abgeschlossen, fehlgeschlagen, abgebrochen), die in Echtzeit bei Statusübergängen aktualisiert werden.
• Input- und Output-Token. Token-Anzahl pro Aufgabe und aggregiert, bezogen aus MLflow Traces.
• LLM-Kosten. Sowohl die vom Modell ausgegebene Schätzung aus MLflow Traces (verfügbar innerhalb von Sekunden nach jedem Modellaufruf).
• Compute-Kosten. Compute-Kosten für Serverless-Jobs, die den Aufgabenläufen des Orchestrators zuzuordnen sind, bezogen aus system.billing.usage.
• Mediane Antwortzeit. Berechnet über abgeschlossene Aufgaben. Der Median wird anstelle des Durchschnitts verwendet, um Verzerrungen durch Ausreißer bei Wiederholungsversuchen (Retry-Backoff) und Latenzen am Ende der Warteschlange bei Auslastung zu vermeiden.
• Konfidenz. Konfidenzwerte pro Dokument, die von der AI-Agents-Ebene zurückgegeben und zusammen mit den Aufgabenergebnissen angezeigt werden.

Implementierung

Statusänderungen in der Aufgabentabelle lösen Postgres-LISTEN/NOTIFY-Ereignisse aus. Das Backend unterhält eine einzige LISTEN-Verbindung und verteilt Ereignisse über Server-Sent Events (SSE) an verbundene Dashboard-Clients. Browser öffnen eine EventSource-Verbindung und erhalten Live-Updates innerhalb von etwa einer Sekunde nach jeder relevanten Statusänderung. Die Implementierung erfordert kein Redis, keinen WebSocket-Server und keinen Message-Bus.

Polling wird als dauerhafter Fallback mit einem Standardintervall von zehn Sekunden beibehalten. Streaming-Verbindungen über Cloud-Ingress-Proxys können Bytes unbemerkt verwerfen, ohne clientseitige Fehlerereignisse auszulösen. Permanentes Polling stellt sicher, dass das Dashboard auch in solchen Fällen aktuell bleibt. Eine UI-Anzeige unterscheidet zwischen live (SSE aktiv) und polling (SSE nicht verfügbar) Kanälen.

Die Daten des Dashboards stammen aus drei Quellen mit unterschiedlichen Latenzcharakteristiken: Postgres (sofort), der Trace-API von MLflow (unter einer Sekunde) und Warehouse-Abfragen auf System-Abrechnungstabellen (gelegentlich einige zehn Sekunden). Die schnellen Abfragen speisen jeden Aktualisierungszyklus. Die langsamen Abfragen werden nur bei Benutzeraktionen ausgeführt und geben optimistisch einen Ladezustand zurück, bis Ergebnisse verfügbar sind.

Kostenzuordnung pro Anwendung

Die System-Abrechnungstabellen von Databricks sind auf Kontoebene definiert: Jeder Job, jeder Modellaufruf und jede andere Anwendung trägt zu denselben system.billing.usage-Zeilen bei. Ohne diese Eingrenzung würde eine Kachel für „OCR-Kosten“ auf Anwendungsebene die Nutzung aller Modellaufrufe im Workspace aggregieren.

Die Lösung besteht darin, aufzuzeichnen, welche Databricks-Job-Runs der Orchestrator übermittelt hat (nachverfolgt in tasks.locked_by und task_attempts.run_id) und die Abrechnungsabfrage auf diese Auswahl zu filtern. Ein einziges SQL-Warehouse kann mehrere Anwendungen unterstützen, und jedes Dashboard zeigt nur die eigenen Ausgaben an.

Dieselbe Abfragearchitektur lässt sich nahtlos mit benutzergesteuerten Filtern kombinieren. Kostenwerte können, zusammen mit allen anderen Dashboard-Metriken, weiter nach Datumsbereich, Task-Status oder Agent eingegrenzt werden. Dies unterstützt Fragen wie „Was haben fehlgeschlagene Tasks in den letzten sieben Tagen gekostet?“ oder „Wie hoch waren die medianen Ausgaben pro Task für Agent X in diesem Monat?“, ohne das Dashboard verlassen zu müssen.

Dadurch lassen sich Kostenwerte einfach überwachen, zuordnen und melden.

Lakebase als Orchestrierungs-Backbone

„Postgres-as-Queue“-Muster sind in der Data-Engineering-Community fest etabliert. Lakebase bietet die zusätzlichen betrieblichen Eigenschaften, die dieses Muster als Produktionsarchitektur auf Databricks tragfähig machen:

• Automatisch skalierende Rechenleistung. Lakebase skaliert Postgres-Recheneinheiten je nach Workload nach oben oder unten. So kann der Orchestrator die Datenbank nutzen, ohne rund um die Uhr für Spitzenkapazitäten zu zahlen.
• Über OAuth rotierte Authentifizierung. Lakebase verwendet kurzlebige OAuth-Token für die Verbindungsauthentifizierung. Verbindungspools aktualisieren Token automatisch, wodurch statische Anmeldedaten in der Anwendungskonfiguration und Rotations-Runbooks überflüssig werden.
• Integration in Unity Catalog. Lakebase teilt Identität, Berechtigungen und Governance mit dem Rest von Databricks. Der Service Principal des Orchestrators erhält explizite Freigaben für die Tabellen tasks und results; es ist keine separate IAM-Konfiguration erforderlich.
• Branching und Snapshots. Das Klonen einer Produktions-Task-Tabelle in eine Entwicklungsumgebung zum Debuggen ist ein Standardvorgang in Lakebase, der nativ unterstützt wird.

Diese Funktionen eliminieren den betrieblichen Aufwand, der Teams normalerweise dazu bewegt, verwaltete Message Broker anstelle von selbstgehostetem Postgres für das Task-Queueing einzusetzen.

Auswirkungen und Fazit

Bei CLA unterstützt das hier beschriebene Orchestrierungsmuster einen produktiven Dokumentenverarbeitungsworkflow, der die Extraktionszeit von Stunden auf Minuten verkürzt. Die Architektur nutzt Databricks-native Dienste mit Lakebase Postgres im Zentrum, um Queueing, Scheduling und Observability ohne externe Systeme zu verwalten. Dies reduziert den Integrationsaufwand und nutzt gleichzeitig alle Vorteile einer einheitlichen, auf Skalierbarkeit ausgelegten Plattform.

In der Produktion bietet dieses Muster ein robustes Task-Management, Prioritätssteuerung, ratenbegrenzungsbewusstes Scheduling, Echtzeit-Sichtbarkeit und Kostenverfolgung pro Task. Zusammen bieten diese Funktionen eine praktische Möglichkeit, agentenbasierte Workloads zu orchestrieren und gleichzeitig die umgebende Infrastruktur einfach zu halten.

Bereit, die Orchestrierung von KI-Agenten auf einer einzigen Plattform zu vereinfachen? Testen Sie die Databricks Free Edition, erstellen Sie Ihr erstes Lakebase Postgres-Projekt und eine Databricks App, und folgen Sie dann der 10-Minuten-Demo zu MLflow Tracing, um End-to-End-Observability in Ihren Agenten-Workflow zu integrieren.

(Dieser Blogbeitrag wurde mit KI-gestützten Tools übersetzt.) Originalbeitrag