Développez des agents durables avec Temporal et Lakebase

Un agent de souscription de prêts personnels rassemble des preuves, applique des politiques et peut attendre plusieurs jours l'avis d'un réviseur. Pendant ce temps, les workers peuvent redémarrer et les appels d'outils peuvent échouer. L'application doit préserver le travail accompli, reprendre l'exécution et maintenir les preuves à la disposition du réviseur.

Cette implémentation de référence utilise Temporal pour une exécution durable et Lakebase Postgres pour un état opérationnel interrogeable. Une table synchronisée rend la politique de souscription de Unity Catalog disponible dans Lakebase. Les activités Temporal écrivent les preuves, les décisions et les métriques dans Lakebase ; une fois activé, le Change Data Feed de Lakebase peut publier ces modifications dans les tables d'historique Delta gérées par Unity Catalog. Cette combinaison est particulièrement utile lorsque Databricks gère déjà les entrées de l'agent et l'analyse en aval.

Le défi des agents cloud à exécution longue

Un agent cloud peut survivre à la requête, au worker, au conteneur ou au déploiement qui l'a démarré. Un utilisateur peut commencer une session, revenir le lendemain et continuer sur un autre worker. Les déploiements et les défaillances de processus sont courants, de sorte que la progression de l'agent doit survivre indépendamment du processus qui l'exécute. La récupération nécessite à la fois les résultats des opérations terminées et l'état du flux de contrôle nécessaire pour déterminer la suite.

Pour cet agent de souscription, cela crée six exigences :

1. Récupération : Un worker de remplacement doit reprendre à partir de la dernière étape terminée.
2. Tentatives : Les appels d'outils et les opérations de base de données doivent tolérer une exécution répétée sans dupliquer les effets secondaires.
3. Longues attentes : L'agent doit attendre des personnes ou des systèmes externes sans maintenir un worker ouvert.
4. Visibilité opérationnelle : Les applications et les opérateurs ont besoin de l'état actuel, des preuves, de l'état des tentatives et des détails des échecs.
5. Gouvernance au moment de l'exécution : Les mises à jour des politiques doivent être disponibles sans déploiement de code, et l'application doit définir quand un dossier ouvert les adopte.
6. Audit : Le système doit conserver les preuves, la politique, la recommandation et la décision humaine associées à chaque exécution.

Une transcription de conversation ne couvre qu'une partie de cet état. La récupération nécessite également l'historique du flux de contrôle : quelles opérations ont été planifiées, quels résultats ont été enregistrés, ce que l'agent attend et quelles commandes il a acceptées.

Temporal simplifie la gestion des systèmes distribués. Lors de la création avec Temporal, un Workflow est le flux de contrôle durable pour une exécution d'agent. Une Activity est un appel à un modèle, un outil ou une base de données dont le résultat est enregistré dans l'historique des événements du Workflow ; les Activities peuvent être réessayées. Un Signal est une commande asynchrone envoyée à un Workflow en cours d'exécution, comme la décision d'un souscripteur. L'implémentation de référence Temporal Lakebase AgentWorkflow est un agent de souscription de prêts personnels exécutable. Il appelle plusieurs outils, lit la politique régie, produit une recommandation et attend un souscripteur.

Lakebase Postgres aide également les développeurs à gérer ces problèmes, mais Temporal et Lakebase stockent des états différents pour des consommateurs différents. L'historique des événements de Temporal pilote la relecture. Lakebase stocke la vue orientée application : statut de l'exécution en cours, messages, preuves, état de la révision et métriques. Unity Catalog reste la source des politiques ; une table synchronisée rend cette politique interrogeable dans Postgres, et Change Data Feed fournit le chemin de retour pour l'historique opérationnel. Les systèmes ne partagent pas de transaction. Les écritures Lakebase s'exécutent en tant qu'Activities Temporal sous une exécution au moins une fois. Des identifiants déterministes, des contraintes, des mises à jour protégées et des upserts Postgres garantissent que les tentatives d'Activity répétées ciblent le même enregistrement logique.

Cette architecture ajoute deux systèmes gérés et un contrat de projection entre eux. Ensemble, ils améliorent la résilience et l'évolutivité de l'agent tout en maintenant des coûts opérationnels bas. L'association de Temporal et Lakebase est particulièrement utile lorsqu'une session d'agent doit survivre au remplacement d'un worker, accepter des entrées après de longues attentes, exposer un état relationnel à une application et appliquer des données régies tant qu'elle reste ouverte.

Le cas d'usage de la souscription

J'ai choisi la souscription de prêts car une même exécution doit rassembler des preuves, appliquer des politiques, produire une recommandation et attendre l'intervention d'une personne. Un worker peut échouer entre n'importe laquelle de ces étapes. La politique peut changer sans déploiement d'application, et l'UI a besoin des preuves actuelles avant la fermeture du Workflow.

Des demandeurs fictifs remplacent les véritables bureaux de crédit et fournisseurs de revenus, et la séquence d'outils est déterministe par souci de simplicité. Chaque requête contient un ID utilisateur, un ID demandeur, un montant, un objectif, un choix de modèle et une limite de tours. FastAPI attribue run_id, démarre LoanUnderwritingWorkflow et utilise ce même ID à travers l'API, l'exécution Temporal et les lignes Lakebase.

Au premier tour, credit_check renvoie le score, les lignes de crédit, les impayés et la dette actuelle. income_verification renvoie les preuves de revenus et d'emploi. debt_to_income_calc calcule le ratio d'endettement. policy_lookup charge la politique correspondant à l'objectif du prêt et évalue les preuves par rapport aux seuils d'approbation, d'orientation et de refus catégorique.

Le demandeur limite de l'échantillon a un score de crédit de 665, un revenu annuel vérifié de 76 000 $, une dette mensuelle de 2 400 $ et un signalement d'impayé non significatif. Le résultat de la politique enregistre chaque règle, seuil, valeur réelle, résultat de réussite/échec, source, recommandation et justification. Le modèle peut recommander mais ne peut pas décider. Un souscripteur approuve, refuse ou demande des informations complémentaires. Une demande d'informations complémentaires devient un autre message de l'utilisateur et un autre tour d'agent. Ce cas teste un plantage de worker après des appels d'outils terminés, une écriture Lakebase validée dont la confirmation d'Activity est perdue, une révision laissée ouverte pendant des jours, une décision de navigateur obsolète et un changement de politique en cours d'exécution.

Architecture

Pour implémenter l'agent de souscription, React et FastAPI gèrent le travail HTTP et d'UI : démarrage des exécutions, rendu des preuves, liste des dossiers et soumission des décisions de révision. Temporal Cloud stocke l'historique des événements et distribue les Tasks. Les workers rejouent le code du Workflow et exécutent les Activities de modèle, d'outil et de Lakebase ; les I/O réseau et de base de données restent en dehors du code déterministe du Workflow.

Une exécution commence lorsque FastAPI démarre un Workflow. Le worker planifie les Activities, Temporal enregistre leurs résultats, et l'agent finit par atteindre AWAITING_REVIEW. La réponse du souscripteur est renvoyée via un Signal. L'approbation ou le refus clôture l'exécution ; une demande d'informations complémentaires relance la boucle de l'agent.

Lakebase contient deux schémas opérationnels. agent_ops contient le statut de l'exécution, les messages, les appels d'outils, les enregistrements de révision, les événements et les métriques que FastAPI peut interroger avec SQL. agent_policy contient la politique synchronisée en lecture seule utilisée par policy_lookup. Chaque Activity écrit des enregistrements indexés par les mêmes identifiants déterministes que ceux utilisés par le Workflow, so the projection can catch up after a retry without making Lakebase part of Temporal’s replay mechanism.

Unity Catalog est la source des seuils de souscription. Une table synchronisée en continu les met à la disposition de l'agent en cours d'exécution. Les seuils appliqués, les preuves et la décision humaine qui en découle sont écrits dans agent_ops. Change Data Feed peut publier ces modifications dans des tables d'historique gérées par Unity Catalog à des fins d'audit et d'analyse.

Récupérer le travail accompli après une défaillance de worker

Temporal conserve l'historique des événements ordonné requis pour reconstruire l'état du Workflow sur un autre worker. Cet historique comprend la planification et les résultats des Activities, les minuteurs et les Signals. La relecture exécute le code du Workflow par rapport à ces événements enregistrés et reconstruit des variables telles que le tour actuel, les décisions de révision acceptées, l'utilisation des jetons et les preuves collectées.

Un résultat d'Activity enregistré est renvoyé pendant la relecture au lieu d'exécuter à nouveau l'Activity. Une vérification de solvabilité terminée reste terminée, et une réponse de modèle enregistrée reste la réponse pour cette exécution. Si une Activity était en cours d'exécution lorsque le worker a échoué et que Temporal n'a jamais enregistré sa finalisation, Temporal peut planifier une autre tentative. Pour un agent, cela préserve les réponses du modèle déjà enregistrées dans l'historique des événements. Un appel de modèle dont la finalisation n'a pas été enregistrée peut tout de même être réexécuté, même si le fournisseur a fini de le traiter.

Les politiques de réessai sont attribuées à l'échelle des opérations individuelles et peuvent être réutilisées dans le code. Dans cet exemple, les activités d'appel de modèle autorisent jusqu'à quatre tentatives dans la limite d'un délai d'expiration schedule-to-close de trois minutes. Les activités d'appel d'outils autorisent jusqu'à trois tentatives et ont un délai d'expiration start-to-close de 60 secondes. Les activités Lakebase autorisent jusqu'à d'un délai d'expiration start-to-close de 15 secondes.

Rendre les effets externes répétables en toute sécurité

L'un des risques est qu'une écriture de résultat d'outil Lakebase soit validée avant que le Worker ne signale la fin de l'activité. Si la connexion est coupée pendant ce laps de temps, Temporal n'a aucun résultat enregistré et planifie une autre tentative. Les deux tentatives représentent la même écriture logique.

Chaque enregistrement Lakebase possède une identité stable. run_id ancre le schéma opérationnel. message_id identifie un message, tool_call_id une invocation d'outil, event_id un jalon, review_id un cycle de révision et decision_id une commande de réviseur. Les clés primaires et les contraintes d'unicité Postgres appliquent ces identités.

L'écriture de début d'outil montre à la fois l'identité stable et la protection de l'état terminal :

Un réessai cible le même tool_call_id. Le prédicat final permet uniquement de réécrire une ligne non terminale existante à l'état démarré. Si la ligne est déjà en succès ou en échec, PostgreSQL affecte zéro ligne. Cela ne génère pas d'erreur.

L'appelant doit inspecter un résultat de zéro ligne. LakebaseWriteResult renvoie le nombre de lignes affectées, mais le wrapper d'activité actuel ne transforme pas un zéro en échec. Le code de production ne doit classer zéro comme une opération sans effet attendue qu'après avoir confirmé l'état terminal stocké ; sinon, il doit lever ou enregistrer un conflit. La même règle s'applique aux transitions d'exécution et de révision sécurisées.

Des upserts similaires couvrent les messages, les résultats d'outils et les événements. Des ID déterministes permettent aux réessais de converger vers la même ligne logique, tandis que chaque écriture sécurisée définit quelles transitions d'état sont autorisées. L'API peut brièvement afficher un état plus ancien pendant qu'une écriture est réessayée. Une fois que l'activité réussit, la ligne acceptée peut être interrogée.

Chaque outil ayant des effets secondaires nécessite un contrat équivalent. Une API de paiement peut accepter une clé d'idempotence, un service de messagerie un ID de message fourni par l'appelant, et une base de données une contrainte d'unicité. Si le système externe ne fournit aucun mécanisme de déduplication, l'activité a besoin de son propre enregistrement ou d'un processus de réconciliation. Temporal détermine quand réessayer. L'activité détermine comment le système externe gère ce réessai.

Exposer l'état actuel et les métriques opérationnelles

L'historique des événements fournit la sémantique d'exécution et les détails de débogage. L'application a besoin de requêtes relationnelles indexées sur l'exécution actuelle : lister les cas par utilisateur et par statut, charger une transcription avec ses preuves, trouver les révisions en attente d'une personne et agréger les mesures sur l'ensemble des exécutions.

Lakebase stocke cette vue de l'application dans un schéma Postgres normalisé. agent_runs contient le statut actuel, le Workflow ID, la requête, le total des jetons, les horodatages et les métadonnées de recommandation. agent_messages conserve la transcription. agent_tool_calls enregistre les arguments, le statut, le résultat structuré, l'erreur et le timing. agent_review_decisions associe la recommandation à un ID de révision stable, à une commande de réviseur, à une justification et à l'heure de la décision.

Le schéma enregistre également les événements nommés et les métriques au niveau du Workflow, du tour et de la tentative d'activité. FastAPI expose les points de terminaison run-detail, workflow-metrics et retry-metrics adossés à ces tables. L'UI peut afficher une exécution collectant des preuves, une autre en attente de révision et une troisième réessayant un outil en échec. Les opérateurs peuvent interroger les mêmes lignes avec SQL.

Les preuves sont disponibles avant la fin du Workflow. Une fois que policy_lookup est terminé, son résultat structuré est stocké avec l'appel d'outil. Lorsque l'exécution atteint AWAITING_REVIEW, le souscripteur peut voir le score de crédit, le DTI, les seuils, les résultats des règles, la justification et la source de la politique qui ont produit la recommandation.

Garantir la durabilité de la révision humaine et rejeter les commandes obsolètes

Lorsque le modèle renvoie une recommandation, le Workflow dérive review_id de run_id et du tour actuel. Il écrit la révision en attente dans Lakebase, enregistre un événement agent.review_pending, définit la projection sur AWAITING_REVIEW et appelle workflow.wait_condition. Temporal conserve le Workflow ouvert sans maintenir un processus Worker occupé.

L'API envoie l'action du souscripteur sous forme de Signal. Avant de l'envoyer, l'API vérifie que Lakebase indique que l'exécution est en attente de révision et que le review_id soumis correspond au cycle actuel. Si l'une de ces vérifications échoue, l'API renvoie un conflit. Le Workflow valide indépendamment la commande par rapport à son propre état et ignore les décisions obsolètes ou en double, protégeant ainsi l'exécution même lorsque la projection Lakebase est en retard.

Après avoir accepté le Signal, le Workflow pérennise la décision via une activité Lakebase idempotente. L'approbation ou le refus met fin à l'exécution. Une demande d'informations complémentaires fait repasser la projection à RUNNING, ajoute la justification du réviseur sous forme de message utilisateur et démarre le tour suivant. Comme le tour a changé, la recommandation suivante reçoit un nouveau review_id.

La réponse 202 de l'API confirme que Temporal a reçu le Signal. L'acceptation métier se fait de manière asynchrone dans le Workflow, de sorte qu'une commande peut passer le contrôle préalable de l'API et être tout de même ignorée si l'état de la révision a changé. Le client actualise la projection Lakebase pour observer l'état résultant.

Servir une politique gouvernée sans redéployer les Workers

Les seuils de souscription changent indépendamment du code du Worker. La table source dans Unity Catalog contient des valeurs spécifiques à l'usage, telles que le score de crédit minimum, le DTI d'approbation automatique, les seuils de refus ferme et le nom de la politique.

Le script de configuration crée une table synchronisée Lakebase continue nommée agent_policy.underwriting_policy_limits. policy_lookup qui interroge cette copie Postgres en lecture seule par objectif de prêt normalisé. Les propriétaires de politiques mettent à jour la source Unity Catalog ; le pipeline de synchronisation propage la modification, et une exécution ultérieure la lit sans déploiement de Worker ou d'API.

Le résultat de la politique contient les seuils appliqués, la valeur réelle et le résultat de réussite/échec de chaque règle, ainsi que la source. La démo peut se rabattre sur une politique fictive lorsque Lakebase est désactivé ou que la ligne n'est pas disponible, et enregistre ce chemin sous le nom de fixture_fallback. Un Workflow réglementé peut à la place échouer de manière sécurisée. L'application doit prendre cette décision de repli de manière explicite.

Renvoyer les modifications opérationnelles à Unity Catalog

Le dépôt prépare chaque table agent_ops pour Lakebase Change Data Feed en définissant REPLICA IDENTITY FULL. Un administrateur doit tout de même activer la fonctionnalité pour le schéma. Lakebase capture ensuite les insertions, mises à jour et suppressions du journal d'écriture anticipée Postgres et les écrit par lots dans des tables d'historique Delta gérées par Unity Catalog et nommées selon le modèle lb_<table>_history.

Change Data Feed est actuellement en Public Preview et vide les modifications environ toutes les 15 secondes. Cet intervalle convient à l'audit et à l'analyse, tandis que l'UI interroge directement Lakebase pour connaître l'état opérationnel actuel.

Les tables d'historique peuvent reconstruire la source de la politique d'une exécution, les preuves d'outils, les tentatives d'activité, l'attente de révision, la recommandation et la décision humaine. Le dépôt configure le schéma source pour ce chemin mais n'inclut pas d'exécution Change Data Feed de bout en bout observée. L'activation du flux et la vérification des tables de destination restent des étapes de déploiement.

Exploiter le système

Le déploiement sépare React/FastAPI du Worker Temporal. Les réplicas d'API s'adaptent à la charge des requêtes ; les Workers s'adaptent au backlog de tâches de Workflow et d'activité ainsi qu'à la simultanéité configurée. L'autoscaling de Lakebase ajuste la puissance de calcul de la base de données dans les limites du projet.

La tarification de Temporal Cloud est basée sur les Actions ainsi que sur le stockage actif et conservé de l'historique des événements, de sorte que la fréquence des réessais et les historiques ouverts de longue durée affectent également le coût. L'équipe définit toujours les réplicas Kubernetes, les files d'attente de tâches, les limites du pool de connexions et les limites de la base de données pour sa charge de travail. Alternativement, vous pouvez configurer votre propre service Temporal open source en utilisant la dernière version open source.

Le client Lakebase utilise l'authentification machine-to-machine OAuth. Les jetons OAuth Databricks et les identifiants de base de données générés expirent, le client actualise donc son pool de connexions SQLAlchemy avant l'expiration d'une heure des identifiants de base de données. Les connexions utilisent TLS. Sans rotation, un Worker à exécution longue rencontrerait des pannes de base de données selon un calendrier prévisible.

Les opérateurs utilisent Temporal pour inspecter l'historique des Workflows et des activités, Lakebase pour interroger l'état et les métriques de l'application, et Kubernetes pour vérifier la santé des processus et des déploiements. Un opérateur peut alors distinguer une attente de révision délibérée d'un réessai d'activité, d'un échec d'accès à la base de données ou d'un outil en échec.

Preuves et limites

La suite de tests contient 21 tests réussis pour le séquençage de Workflow, le comportement de révision, la construction de connexions OAuth, la persistance idempotente, les contrats de métriques, le démarrage de Workflow par API et les paramètres de Worker. Le script de récupération après plantage ajoute un exercice de défaillance de processus avec le fournisseur déterministe.

Les données du demandeur et du fournisseur sont des fixtures. Le dépôt ne valide pas les modèles de prêt, la conformité réglementaire, les contrôles de sécurité de production, la disponibilité régionale ou les performances à grande échelle. L'exercice de plantage local a été exécuté avec Lakebase désactivé, ce qui permet d'isoler la récupération Temporal. Change Data Feed nécessite toujours une activation et une vérification dans l'environnement Databricks cible.

Prochaines étapes

Vous voulez en savoir plus ? Essayez d'exécuter la démo par vous-même et découvrez l'exécution durable en action. Lancez l'implémentation de référence, arrêtez un worker en cours d'exécution et observez la récupération de l'agent. Connectez Lakebase pour explorer les preuves, les politiques et l'état de la révision humaine derrière chaque décision.

(Cet article de blog a été traduit à l'aide d'outils basés sur l'intelligence artificielle) Article original