Costruisci agenti durevoli con Temporal e Lakebase

Un agente di underwriting di prestiti personali raccoglie prove, applica le policy e può attendere giorni per un revisore. Durante questo periodo, i worker possono riavviarsi e le chiamate agli strumenti possono fallire. L'applicazione deve preservare il lavoro completato, riprendere l'esecuzione e mantenere le prove a disposizione del revisore.

Questa implementazione di riferimento utilizza Temporal per l'esecuzione durevole e Lakebase Postgres per lo stato operativo interrogabile. Una tabella sincronizzata rende la policy di underwriting di Unity Catalog disponibile in Lakebase. Le Temporal Activities scrivono prove, decisioni e metriche su Lakebase; una volta abilitato, il Change Data Feed di Lakebase può pubblicare tali modifiche nelle tabelle della cronologia Delta gestite da Unity Catalog. Questa combinazione è particolarmente utile quando Databricks gestisce già gli input dell'agente e l'analisi a valle.

La sfida degli agenti cloud a lunga esecuzione

Un agente cloud può sopravvivere alla richiesta, al worker, al container o al deployment che lo ha avviato. Un utente può iniziare una sessione, tornare il giorno successivo e continuare su un altro worker. I deployment e i guasti dei processi sono all'ordine del giorno, quindi il progresso dell'agente deve sopravvivere indipendentemente dal processo che lo esegue. Il ripristino richiede sia i risultati delle operazioni completate sia lo stato del flusso di controllo necessario per determinare cosa accadrà in seguito.

Per questo agente di underwriting, ciò comporta sei requisiti:

1. Ripristino: un worker sostitutivo deve riprendere dall'ultimo passaggio completato.
2. Tentativi (retries): le chiamate agli strumenti e le operazioni del database devono tollerare l'esecuzione ripetuta senza duplicare gli effetti collaterali.
3. Lunghe attese: l'agente deve attendere le persone o i sistemi esterni senza mantenere aperto un worker.
4. Visibilità operativa: le applicazioni e gli operatori hanno bisogno dello stato attuale, delle prove, dello stato dei tentativi e dei dettagli sui guasti.
5. Governance a runtime: gli aggiornamenti delle policy devono diventare disponibili senza un deployment del codice e l'applicazione deve definire quando un caso aperto li adotta.
6. Audit: il sistema deve conservare le prove, la policy, la raccomandazione e la decisione umana associate a ciascuna esecuzione.

La trascrizione di una conversazione copre solo una parte di questo stato. Il ripristino richiede anche la cronologia del flusso di controllo: quali operazioni sono state pianificate, quali risultati sono stati registrati, cosa sta aspettando l'agente e quali comandi ha accettato.

Temporal semplifica la gestione dei sistemi distribuiti. Quando si sviluppa con Temporal, un Workflow è il flusso di controllo durevole per un'esecuzione dell'agente. Un' Activity è una chiamata a un modello, a uno strumento o a un database il cui risultato viene registrato nella Event History del Workflow; le Activity possono essere riprovate. Un Signal è un comando asincrono inviato a un Workflow in esecuzione, come la decisione di un addetto all'underwriting. L'implementazione di riferimento di Temporal Lakebase AgentWorkflow è un agente di underwriting di prestiti personali eseguibile. Chiama diversi strumenti, legge la policy regolamentata, produce una raccomandazione e attende un addetto all'underwriting.

Anche Lakebase Postgres aiuta gli sviluppatori a gestire questi problemi, ma Temporal e Lakebase memorizzano stati diversi per consumatori diversi. L'Event History di Temporal guida il replay. Lakebase memorizza la vista rivolta all'applicazione: stato dell'esecuzione corrente, messaggi, prove, stato della revisione e metriche. Unity Catalog rimane la fonte delle policy; una tabella sincronizzata rende tale policy interrogabile in Postgres e il Change Data Feed fornisce il percorso di ritorno per la cronologia operativa. I sistemi non condividono una transazione. Le scritture di Lakebase vengono eseguite come Temporal Activities con un'esecuzione "at-least-once" (almeno una volta). Identificatori deterministici, vincoli, aggiornamenti protetti e upsert di Postgres garantiscono che i tentativi ripetuti di Activity abbiano come target lo stesso record logico.

Questa architettura aggiunge due sistemi gestiti e un contratto di proiezione tra di essi. Insieme, migliorano la resilienza e la scalabilità dell'agente mantenendo basso il sovraccarico operativo. Temporal più Lakebase è particolarmente utile quando una sessione dell'agente deve sopravvivere alla sostituzione del worker, accettare input dopo lunghe attese, esporre lo stato relazionale a un'applicazione e applicare dati regolamentati mentre rimane aperta.

Il caso d'uso dell'underwriting

Ho scelto l'underwriting dei prestiti perché la stessa esecuzione deve raccogliere prove, applicare la policy, produrre una raccomandazione e attendere una persona. Un worker può fallire tra uno qualsiasi di questi passaggi. La policy può cambiare senza un deployment dell'applicazione e la UI ha bisogno delle prove correnti prima della chiusura del Workflow.

I richiedenti fittizi (mocked) sostituiscono le centrali rischi reali e i fornitori di reddito, e la sequenza degli strumenti è deterministica per semplicità. Ogni richiesta contiene un ID utente, un ID richiedente, l'importo, lo scopo, la scelta del modello e il limite di turni. FastAPI assegna run_id, avvia LoanUnderwritingWorkflow e utilizza lo stesso ID per l'API, l'esecuzione di Temporal e le righe di Lakebase.

Al primo turno, credit_check restituisce il punteggio, le linee di credito, le insolvenze e il debito corrente. income_verification restituisce le prove di reddito e occupazione. debt_to_income_calc calcola il rapporto debito/reddito. policy_lookup carica la policy per lo scopo del prestito e valuta le prove rispetto alle soglie di approvazione, rinvio e rifiuto categorico.

Il richiedente borderline dell'esempio ha un punteggio di credito di 665, 76.000 $ di reddito annuo verificato, 2.400 $ di debito mensile e una segnalazione di insolvenza non rilevante. Il risultato della policy registra ogni regola, soglia, valore effettivo, risultato positivo/negativo, fonte, raccomandazione e motivazione. Il modello può consigliare ma non può decidere. Un addetto all'underwriting approva, rifiuta o richiede ulteriori informazioni. Una richiesta di ulteriori informazioni diventa un altro messaggio dell'utente e un altro turno dell'agente. Il caso simula un crash del worker dopo chiamate agli strumenti completate, una scrittura su Lakebase confermata la cui notifica di completamento dell'Activity è andata persa, una revisione lasciata aperta per giorni, una decisione del browser obsoleta e una modifica della policy durante l'esecuzione.

Architettura

Per implementare l'agente di underwriting, React e FastAPI gestiscono il lavoro HTTP e UI: avvio delle esecuzioni, rendering delle prove, elenco dei casi e invio delle decisioni di revisione. Temporal Cloud memorizza l'Event History e distribuisce i Task. I worker eseguono il replay del codice del Workflow ed eseguono le Activity del modello, dello strumento e di Lakebase; l'I/O di rete e del database rimane al di fuori del codice deterministico del Workflow.

Un'esecuzione inizia quando FastAPI avvia un Workflow. Il worker pianifica le Activity, Temporal registra i loro risultati e l'agente alla fine raggiunge AWAITING_REVIEW. La risposta dell'addetto all'underwriting ritorna tramite un Signal. L'approvazione o il rifiuto chiude l'esecuzione; una richiesta di ulteriori informazioni riprende il loop dell'agente.

Lakebase contiene due schemi operativi. agent_ops contiene lo stato dell'esecuzione, i messaggi, le chiamate agli strumenti, i record di revisione, gli eventi e le metriche che FastAPI può interrogare con SQL. agent_policy contiene la policy sincronizzata in sola lettura utilizzata da policy_lookup. Ciascuna Activity scrive record identificati dagli stessi identificatori deterministici utilizzati dal Workflow, in modo che la proiezione possa allinearsi dopo un tentativo senza rendere Lakebase parte del meccanismo di replay di Temporal.

Unity Catalog è la fonte per le soglie di underwriting. Una tabella sincronizzata continua le rende disponibili all'agente in esecuzione. Le soglie applicate, le prove e la successiva decisione umana vengono scritte in agent_ops. Il Change Data Feed può pubblicare tali modifiche nelle tabelle della cronologia gestite da Unity Catalog per scopi di audit e analisi.

Ripristinare il lavoro completato dopo un guasto del worker

Temporal conserva l'Event History ordinata richiesta per ricostruire lo stato del Workflow su un altro worker. Tale cronologia include la pianificazione e i risultati delle Activity, i timer e i Signal. Il replay esegue il codice del Workflow a fronte di tali eventi registrati e ricostruisce variabili quali il turno corrente, le decisioni di revisione accettate, l'utilizzo dei token e le prove raccolte.

Un risultato registrato dell'Activity viene restituito durante il replay invece di eseguire nuovamente l'Activity. Un controllo del credito completato rimane completato e una risposta del modello registrata rimane la risposta per tale esecuzione. Se un'Activity era in corso quando il worker ha subito un guasto e Temporal non ha mai registrato il suo completamento, Temporal può pianificare un altro tentativo. Per un agente, questo preserva le risposte del modello già registrate nell'Event History. Una chiamata al modello il cui completamento non è stato registrato potrebbe comunque essere eseguita di nuovo, anche se il provider ha terminato di elaborarla.

I criteri di ripetizione sono assegnati alla granularità delle singole operazioni e possono essere riutilizzati nel codice. Nell'esempio, le attività che chiamano il modello consentono fino a quattro tentativi entro un timeout schedule-to-close di tre minuti. Le attività che chiamano gli strumenti consentono fino a tre tentativi e hanno un timeout start-to-close di 60 secondi. Le attività Lakebase consentono fino a cinque tentativi con un timeout start-to-close di 15 secondi.

Rendere gli effetti esterni sicuri da ripetere

Un rischio è che la scrittura del risultato di uno strumento Lakebase possa essere confermata (commit) prima che il Worker segnali il completamento dell'Activity. Se la connessione si interrompe in questo intervallo, Temporal non ha alcun risultato registrato e pianifica un altro tentativo. Entrambi i tentativi rappresentano la stessa scrittura logica.

Ogni record di Lakebase ha un'identità stabile. run_id ancora lo schema operativo. message_id identifica un messaggio, tool_call_id un'invocazione di uno strumento, event_id una milestone, review_id un ciclo di revisione e decision_id un comando del revisore. Le chiavi primarie e i vincoli di unicità di Postgres applicano queste identità.

La scrittura di avvio dello strumento mostra sia l'identità stabile sia la protezione dello stato terminale:

Un tentativo ha come target lo stesso tool_call_id. Il predicato finale consente solo la riscrittura nello stato avviato di una riga non terminale esistente. Se la riga è già nello stato completato o non riuscito, PostgreSQL influisce su zero righe. Non genera un errore.

Il chiamante deve esaminare un risultato a zero righe. LakebaseWriteResult restituisce il numero di righe interessate, ma l'attuale wrapper dell'Activity non trasforma lo zero in un errore. Il codice di produzione dovrebbe classificare lo zero come un no-op previsto solo dopo aver confermato lo stato terminale memorizzato; in caso contrario, dovrebbe generare o registrare un conflitto. La stessa regola si applica alle transizioni protette di esecuzione e revisione.

Upsert simili coprono messaggi, risultati degli strumenti ed Eventi. Gli ID deterministici fanno sì che i tentativi convergano sulla stessa riga logica, mentre ogni scrittura protetta definisce quali transizioni di stato sono legali. L'API può mostrare brevemente uno stato precedente durante il tentativo di scrittura. Dopo che l'Activity ha avuto successo, la riga accettata è interrogabile.

Ogni strumento con effetti collaterali necessita di un contratto equivalente. Un'API di pagamento può accettare una chiave di idempotenza, un servizio di posta elettronica un ID messaggio fornito dal chiamante e un database un vincolo di unicità. Se il sistema esterno non fornisce alcun meccanismo di deduplicazione, l'Activity necessita di un proprio record o di un processo di riconciliazione. Temporal determina quando riprovare. L'Activity determina come il sistema esterno gestisce tale tentativo.

Esporre lo stato corrente e le metriche operative

La cronologia degli eventi (Event History) fornisce la semantica di esecuzione e i dettagli di debug. L'applicazione richiede query relazionali indicizzate sull'esecuzione corrente: elenca i casi per utente e stato, carica una trascrizione con le relative prove, trova le revisioni in attesa di una persona e aggrega le misurazioni tra le esecuzioni.

Lakebase memorizza la vista dell'applicazione in uno schema Postgres normalizzato. agent_runs contiene lo stato corrente, l'ID del Workflow, la richiesta, i totali dei token, i timestamp e i metadati dei consigli. agent_messages conserva la trascrizione. agent_tool_calls registra argomenti, stato, risultato strutturato, errore e tempistiche. agent_review_decisions collega il consiglio a un ID di revisione stabile, al comando del revisore, alla motivazione e al momento della decisione.

Lo schema registra anche Eventi denominati e metriche a livello di Workflow, turno e tentativo di Activity. FastAPI espone gli endpoint run-detail, workflow-metrics e retry-metrics supportati da queste tabelle. L'UI può mostrare un'esecuzione che raccoglie prove, un'altra in attesa di revisione e una terza che riprova uno strumento non riuscito. Gli operatori possono interrogare le stesse righe con SQL.

Le prove sono disponibili prima del completamento del Workflow. Al termine di policy_lookup, il suo risultato strutturato viene memorizzato con la chiamata dello strumento. Quando l'esecuzione raggiunge AWAITING_REVIEW, l'assicuratore (underwriter) può vedere il punteggio di credito, il DTI, le soglie, i risultati delle regole, la motivazione e la fonte della policy che ha generato il consiglio.

Garantire la durabilità della revisione umana e rifiutare i comandi obsoleti

Quando il modello restituisce un consiglio, il Workflow deriva review_id da run_id e dal turno corrente. Scrive la revisione in sospeso su Lakebase, registra un evento agent.review_pending, imposta la proiezione su AWAITING_REVIEW e chiama workflow.wait_condition. Temporal mantiene il Workflow aperto senza tenere occupato un processo Worker.

L'API invia l'azione dell'assicuratore come Signal. Prima di inviarla, l'API verifica che Lakebase mostri l'esecuzione in attesa di revisione e che il review_id inviato corrisponda al ciclo corrente. Se uno dei due controlli fallisce, l'API restituisce un conflitto. Il Workflow convalida in modo indipendente il comando rispetto al proprio stato e ignora le decisioni obsolete o duplicate, proteggendo l'esecuzione anche quando la proiezione di Lakebase è in ritardo.

Dopo aver accettato il Signal, the Workflow rende persistente la decisione tramite un'Activity Lakebase idempotente. L'approvazione o il rifiuto completano l'esecuzione. Una richiesta di ulteriori informazioni riporta la proiezione a RUNNING, aggiunge la motivazione del revisore come messaggio utente e avvia il turno successivo. Poiché il turno è cambiato, il consiglio successivo riceve un nuovo review_id.

La risposta 202 dell'API conferma che Temporal ha ricevuto il Signal. L'accettazione aziendale avviene in modo asincrono nel Workflow, quindi un comando può superare il pre-controllo dell'API ed essere comunque ignorato se lo stato della revisione è cambiato. Il client aggiorna la proiezione di Lakebase per osservare lo stato risultante.

Fornire policy governate senza ridistribuire i Worker

Le soglie di valutazione (underwriting) cambiano indipendentemente dal codice del Worker. La tabella di origine in Unity Catalog contiene valori specifici per lo scopo, come il punteggio di credito minimo, il DTI per l'approvazione automatica, le soglie di rifiuto definitivo e il nome della policy.

Lo script di configurazione crea una tabella sincronizzata continua di Lakebase denominata agent_policy.underwriting_policy_limits. policy_lookup che interroga questa copia Postgres di sola lettura in base allo scopo del prestito normalizzato. I proprietari delle policy aggiornano l'origine in Unity Catalog; la pipeline di sincronizzazione propaga la modifica e un'esecuzione successiva la legge senza richiedere la distribuzione di un Worker o di un'API.

Il risultato della policy contiene le soglie applicate, il valore effettivo di ogni regola, il risultato di superamento/fallimento e l'origine. La demo può ricorrere a una policy fittizia (fixture policy) quando Lakebase è disabilitato o la riga non è disponibile, e registra tale percorso come fixture_fallback. Un Workflow regolamentato potrebbe invece bloccarsi in modo sicuro (fail closed). L'applicazione deve prendere questa decisione di fallback in modo esplicito.

Restituire le modifiche operative a Unity Catalog

Il repository prepara ogni tabella agent_ops per il Change Data Feed di Lakebase impostando REPLICA IDENTITY FULL. Un amministratore deve comunque abilitare la funzionalità per lo schema. Lakebase acquisisce quindi inserimenti, aggiornamenti ed eliminazioni dal log write-ahead di Postgres e li scrive in batch nelle tabelle della cronologia Delta gestite da Unity Catalog, denominate con il pattern lb_<table>_history.

Il Change Data Feed è attualmente in Public Preview e scarica (flush) le modifiche all'incirca ogni 15 secondi. Questo intervallo è adatto per l'audit e l'analisi, mentre l'UI interroga direttamente Lakebase per lo stato operativo corrente.

Le tabelle della cronologia possono ricostruire l'origine della policy di un'esecuzione, le prove dello strumento, i tentativi di Activity, l'attesa della revisione, il consiglio e la decisione umana. Il repository configura lo schema di origine per questo percorso ma non include un'esecuzione end-to-end osservata del Change Data Feed. L'abilitazione del feed e la verifica delle tabelle di destinazione rimangono passaggi di distribuzione.

Gestire il sistema

La distribuzione separa React/FastAPI dal Worker di Temporal. Le repliche dell'API scalano con il carico delle richieste; i Worker scalano con il backlog dei task di Workflow e Activity e con la concorrenza configurata. L'autoscaling di Lakebase regola la capacità di calcolo del database entro i limiti del progetto.

I prezzi di Temporal Cloud si basano sulle Action più l'archiviazione attiva e conservata della Event History, quindi anche la frequenza dei tentativi e le cronologie aperte a lungo influiscono sui costi. Il team imposta comunque le repliche di Kubernetes, le Task Queue, i limiti del pool di connessioni e i limiti del database per il proprio carico di lavoro. In alternativa, è possibile configurare il proprio servizio Temporal open source utilizzando l'ultima versione open source.

Il client Lakebase utilizza l'autenticazione machine-to-machine OAuth. I token OAuth di Databricks e le credenziali del database generate scadono, quindi il client aggiorna il proprio pool di connessioni SQLAlchemy prima della scadenza di un'ora delle credenziali del database. Le connessioni utilizzano TLS. Senza rotazione, un Worker a lungo termine riscontrerebbe errori di database secondo una pianificazione prevedibile.

Gli operatori utilizzano Temporal per esaminare la cronologia di Workflow e Activity, Lakebase per interrogare lo stato e le metriche dell'applicazione e Kubernetes per verificare lo stato del processo e della distribuzione. Un operatore può quindi distinguere un'attesa di revisione intenzionale da un tentativo di Activity, un errore di accesso al database o uno strumento non riuscito.

Prove e limiti

La suite di test contiene 21 test superati per il sequenziamento del Workflow, il comportamento di revisione, la creazione della connessione OAuth, la persistenza idempotente, i contratti delle metriche, l'avvio del Workflow tramite API e le impostazioni del Worker. Lo script di ripristino in caso di crash aggiunge un'esercitazione di errore del processo con il provider deterministico.

I dati del richiedente e del provider sono fixture. Il repository non convalida i modelli di prestito, la conformità normativa, i controlli di sicurezza di produzione, la disponibilità a livello di area o le prestazioni su scala. L'esercitazione di crash locale è stata eseguita con Lakebase disabilitato, quindi isola il ripristino di Temporal. Change Data Feed richiede comunque l'abilitazione e la verifica nell'ambiente Databricks di destinazione.

Passaggi successivi

Vuoi saperne di più? Prova a eseguire la demo di persona e guarda l'esecuzione durevole in azione. Esegui l'implementazione di riferimento, arresta un worker a metà esecuzione e osserva l'agente ripristinarsi. Connetti Lakebase per esplorare le prove, le policy e lo stato di revisione umana alla base di ogni decisione.

(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale