TemporalとLakebaseで耐久性に優れたエージェントを構築

個人向けローンの審査エージェントは、証拠を収集し、ポリシーを適用し、審査担当者を数日間待つことがあります。その間、ワーカーが再起動したり、ツール呼び出しが失敗したりする可能性があります。アプリケーションは、完了した作業を保存し、実行を再開し、審査担当者が証拠を利用できるように維持する必要があります。

このリファレンス実装では、永続的な実行にTemporalを使用し、クエリ可能な運用状態にLakebase Postgresを使用しています。同期されたテーブルにより、Unity Catalogからの審査ポリシーがLakebaseで利用可能になります。Temporal Activitiesは、証拠、決定、メトリクスをLakebaseに書き込みます。有効化されると、Lakebase Change Data Feedはこれらの変更をUnity Catalogが管理するDelta履歴テーブルにパブリッシュできます。この組み合わせは、Databricksがすでにエージェントの入力と下流の分析を管理している場合に特に便利です。

長期実行されるクラウドエージェントの課題

クラウドエージェントは、それを開始したリクエスト、ワーカー、コンテナ、またはデプロイメントよりも長寿命になる可能性があります。ユーザーはセッションを開始し、翌日戻ってきて、別のワーカーで続行することができます。デプロイメントやプロセスの失敗は日常茶飯事であるため、エージェントの進行状況はそれを実行しているプロセスとは独立して存続する必要があります。復旧には、完了した操作の結果と、次に何が起こるかを決定するために必要な制御フローの状態の両方が必要です。

この審査エージェントにとって、これにより6つの要件が生じます。

1. 復旧： 代替ワーカーは、最後に完了したステップから再開する必要があります。
2. 再試行： ツールの呼び出しやデータベース操作は、副作用を重複させることなく、繰り返しの実行に耐える必要があります。
3. 長時間の待機： エージェントは、ワーカーを開いたままにすることなく、人や外部システムを待つ必要があります。
4. 運用の可視性： アプリケーションとオペレーターは、現在のステータス、証拠、再試行状態、および失敗の詳細を必要とします。
5. ランタイムガバナンス： ポリシーの更新はコードのデプロイなしで利用可能になる必要があり、アプリケーションはオープンなケースがいつそれらを採用するかを定義する必要があります。
6. 監査： システムは、各実行に関連付けられた証拠、ポリシー、推奨事項、および人間の決定を保持する必要があります。

会話の書き起こしは、この状態の一部しかカバーしていません。復旧には、どの操作がスケジュールされたか、どの結果が記録されたか、エージェントが何を待っているか、どのコマンドを受け入れたかという制御フローの履歴も必要です。

Temporalは分散システムの管理を簡素化します。Temporalを使用して構築する場合、Workflowは1回のエージェント実行のための永続的な制御フローです。Activityは、モデル、ツール、またはデータベースへの呼び出しであり、その結果はWorkflowのEvent Historyに記録されます。Activityは再試行可能です。Signalは、審査担当者の決定など、実行中のWorkflowに送信される非同期コマンドです。Temporal Lakebase AgentWorkflowリファレンス実装は、実行可能な個人向けローン審査エージェントです。いくつかのツールを呼び出し、管理されたポリシーを読み取り、推奨事項を生成し、審査担当者を待ちます。

Lakebase Postgresも開発者がこれらの問題を管理するのに役立ちますが、TemporalとLakebaseは異なるコンシューマー向けに異なる状態を保存します。TemporalのEvent Historyはリプレイを駆動します。Lakebaseは、現在の実行ステータス、メッセージ、証拠、審査状態、メトリクスなど、アプリケーション向けのビューを保存します。Unity Catalogは引き続きポリシーソースであり、同期されたテーブルによってそのポリシーがPostgresでクエリ可能になり、Change Data Feedが運用履歴の返送パスを提供します。システムはトランザクションを共有しません。Lakebaseの書き込みは、at-least-once（少なくとも1回）の実行下でTemporal Activitiesとして実行されます。決定論的な識別子、制約、保護された更新、およびPostgresのupsertにより、繰り返されるActivityの試行が同じ論理レコードをターゲットにすることが保証されます。

このアーキテクチャは、2つの管理システムとそれらの間のプロジェクション契約を追加します。これらは連携して、運用のオーバーヘッドを低く抑えながら、エージェントの回復力とスケーラビリティを向上させます。TemporalとLakebaseの組み合わせは、エージェントセッションがワーカーの交代を乗り切り、長時間の待機後に入力を受け入れ、リレーショナルな状態をアプリケーションに公開し、オープンな状態のままで管理されたデータを適用する必要がある場合に最も役立ちます。

審査のユースケース

ローン審査を選択したのは、同じ実行で証拠の収集、ポリシーの適用、推奨事項の作成、そして人の待機を行う必要があるためです。これらのステップのいずれかの間でワーカーが失敗する可能性があります。アプリケーションをデプロイすることなくポリシーを変更でき、UIはWorkflowが閉じる前に現在の証拠を必要とします。

模擬の申請者が実際の信用調査機関や収入プロバイダーに取って代わり、ツールのシーケンスは簡素化のために決定論的です。各リクエストには、ユーザーID、申請者ID、金額、目的、モデルの選択、およびターン制限が含まれます。FastAPIはrun_idを割り当て、LoanUnderwritingWorkflowを開始し、API、Temporalの実行、およびLakebaseの行全体で同じIDを使用します。

最初のターンで、credit_checkはスコア、取引履歴、延滞、および現在の債務を返します。income_verificationは収入と雇用の証拠を返します。debt_to_income_calcは返済負担率（DTI）を計算します。policy_lookupはローンの目的に応じたポリシーをロードし、承認、保留、および即時否決のしきい値に対して証拠を評価します。

サンプルの境界線上の申請者は、信用スコアが665、確認済みの年収が76,000ドル、月々の債務が2,400ドル、および重要ではない延滞フラグが1つあります。ポリシー結果には、すべてのルール、しきい値、実際の値、合否結果、ソース、推奨事項、および根拠が記録されます。モデルは推奨することはできますが、決定することはできません。審査担当者は、承認、否決、または追加情報の要求を行います。追加情報の要求は、別のユーザーメッセージとなり、別のエージェントのターンとなります。このケースでは、完了したツール呼び出し後のワーカーのクラッシュ、Activityの完了が失われたコミット済みのLakebase書き込み、数日間開いたままにされた審査、古いブラウザの決定、および実行中のポリシー変更をテストします。

アーキテクチャ

審査エージェントを実装するために、ReactとFastAPIがHTTPとUIの処理（実行の開始、証拠のレンダリング、ケースのリスト表示、審査決定の送信）を担当します。Temporal CloudはEvent Historyを保存し、Taskをディスパッチします。WorkerはWorkflowコードをリプレイし、モデル、ツール、およびLakebaseのActivityを実行します。ネットワークとデータベースのI/Oは、決定論的なWorkflowコードの外部に留まります。

FastAPIがWorkflowを開始すると、実行が始まります。ワーカーはActivityをスケジュールし、Temporalはその結果を記録し、エージェントは最終的にAWAITING_REVIEWに到達します。審査担当者の応答はSignalを介して返されます。承認または否決によって実行が終了し、追加情報の要求によってエージェントのループが再開されます。

Lakebaseは2つの運用スキーマを保持します。agent_opsには、FastAPIがSQLでクエリできる実行ステータス、メッセージ、ツール呼び出し、審査記録、イベント、およびメトリクスが含まれています。agent_policyには、policy_lookupによって使用される読み取り専用の同期されたポリシーが含まれています。各Activityは、Workflowで使用されるのと同じ決定論的な識別子をキーとするレコードを書き込むため、LakebaseをTemporalのリプレイメカニズムの一部にすることなく、再試行後にプロジェクションが追いつくことができます。

Unity Catalogは審査のしきい値のソースです。継続的に同期されるテーブルにより、実行中のエージェントがそれらを利用できるようになります。適用されたしきい値、証拠、およびその後の人間の決定はagent_opsに書き込まれます。Change Data Feedは、監査と分析のために、それらの変更をUnity Catalogが管理する履歴テーブルにパブリッシュできます。

ワーカーの失敗後に完了した作業を復旧する

Temporalは、別のワーカー上でWorkflowの状態を再構築するために必要な、順序付けられたEvent Historyを保持します。その履歴には、Activityのスケジュールと結果、タイマー、およびSignalsが含まれます。リプレイは、それらの記録されたEventに対してWorkflowコードを実行し、現在のターン、受け入れられた審査決定、トークン使用量、収集された証拠などの変数を再構築します。

記録されたActivityの結果は、Activityを再実行する代わりに、リプレイ中に返されます。完了した信用調査は完了したままであり、記録されたモデルの応答はその実行の応答のままです。ワーカーが失敗したときにActivityが実行中であり、Temporalがその完了を記録しなかった場合、Temporalは別の試行をスケジュールできます。エージェントの場合、これによりEvent Historyにすでに記録されているモデルの応答が保存されます。完了が記録されなかったモデル呼び出しは、プロバイダーが処理を完了していたとしても、再度実行される可能性があります。

リトライポリシーは個々の操作の粒度で割り当てられ、コード内で再利用できます。この例では、モデル呼び出しアクティビティは、3分間のschedule-to-closeタイムアウト内で最大4回まで試行できます。ツール呼び出しアクティビティは最大3回まで試行でき、60秒のstart-to-closeタイムアウトが設定されています。Lakebaseアクティビティは、15秒のstart-to-closeタイムアウトで最大5回まで試行できます。

外部への影響を安全に繰り返せるようにする

リスクの1つは、ワーカーがアクティビティの完了を報告する前に、Lakebaseのツール結果の書き込みがコミットされる可能性があることです。その隙に接続が切断された場合、Temporalには記録された結果がないため、別の試行がスケジュールされます。どちらの試行も、論理的には同じ書き込みを表します。

各Lakebaseレコードは、一意で安定した識別子（ID）を持っています。run_id は運用スキーマの基盤となります。message_id はメッセージを識別し、tool_call_id はツール呼び出し、event_id はマイルストーン、review_id はレビューラウンド、decision_id はレビュアーコマンドを識別します。Postgresの主キーとユニーク制約によって、これらの識別子（ID）の一意性が保証されます。

ツールの開始書き込みは、安定した識別子と終端状態ガードの両方を示しています：

リトライは同じ tool_call_id を対象とします。最後の述語は、既存の非終端行のみを「開始（started）」状態に書き戻すことを許可します。行がすでに成功または失敗している場合、PostgreSQLが影響を与える行数は0になります。エラーは発生しません。

呼び出し元は、影響を受けた行数が0である結果を検査する必要があります。LakebaseWriteResult は影響を受けた行数を返しますが、現在のアクティビティラッパーは0を失敗として扱いません。本番コードでは、保存された終端状態を確認した後にのみ、0を想定内のno-op（何もしない操作）として分類すべきです。そうでない場合は、競合を発生させるか記録する必要があります。同じルールが、ガードされた実行およびレビューの遷移にも適用されます。

同様のアップサート（upsert）が、メッセージ、ツール結果、およびイベントをカバーします。決定論的なIDによってリトライは同じ論理行に収束し、ガードされた各書き込みは、どの状態遷移が正当であるかを定義します。書き込みがリトライされている間、APIは一時的に古い状態を表示することがあります。アクティビティが成功した後は、受け入れられた行をクエリできるようになります。

副作用を伴うすべてのツールには、同等のコントラクト（契約）が必要です。決済APIはべき等性キーを受け入れ、メールサービスは呼び出し元が提供するメッセージIDを受け入れ、データベースはユニーク制約を受け入れるといった具合です。外部システムが重複排除メカニズムを提供していない場合、アクティビティ自体に独自の記録または照合プロセスが必要になります。Temporalがリトライのタイミングを決定します。アクティビティは、外部システムがそのリトライをどのように処理するかを決定します。

現在の状態と運用メトリクスの公開

イベント履歴は、実行セマンティクスとデバッグの詳細を提供します。アプリケーションは、現在の実行に対してインデックス付きのリレーショナルクエリを実行する必要があります。これには、ユーザーとステータスごとのケース一覧の表示、1つのトランスクリプトとそのエビデンスのロード、特定の担当者の承認待ちレビューの検索、実行全体での測定値の集計などが含まれます。

Lakebaseは、そのアプリケーションビューを正規化されたPostgresスキーマに保存します。agent_runs は、現在のステータス、ワークフローID、リクエスト、トークン合計、タイムスタンプ、および推奨メタデータを保持します。agent_messages はトランスクリプトを保存します。agent_tool_calls は、引数、ステータス、構造化された結果、エラー、およびタイミングを記録します。agent_review_decisions は、推奨事項を、安定したレビューID、レビュアーコマンド、根拠、および決定時間に関連付けます。

このスキーマは、ワークフロー、ターン、およびアクティビティ試行レベルで、名前付きイベントとメトリクスも記録します。FastAPIは、これらのテーブルをバックエンドとする run-detail、workflow-metrics、および retry-metrics エンドポイントを公開します。UIは、エビデンスを収集している実行、レビューを待っている別の実行、失敗したツールをリトライしている3つ目の実行を表示できます。オペレーターはSQLを使用して同じ行をクエリできます。

エビデンスは、ワークフローが完了する前でも利用可能です。policy_lookup が終了した後、その構造化された結果がツール呼び出しとともに保存されます。実行が AWAITING_REVIEW に達すると、査定担当者は、推奨事項を生成した信用スコア、DTI、しきい値、ルール結果、根拠、およびポリシーソースを確認できます。

人手によるレビューの耐久性を維持し、古いコマンドを拒否する

モデルが推奨事項を返すと、ワークフローは run_id と現在のターンから review_id を導出します。保留中のレビューをLakebaseに書き込み、agent.review_pending イベントを記録し、プロジェクションを AWAITING_REVIEW に設定して、workflow.wait_condition を呼び出します。Temporalは、ワーカープロセスを占有し続けることなく、開いているワークフローを保持します。

APIは、査定担当者のアクションをシグナル（Signal）として送信します。送信する前に、APIはLakebaseがレビュー待ちの実行を示していること、および送信された review_id が現在のラウンドと一致していることを確認します。いずれかのチェックが失敗した場合、APIは競合（conflict）を返します。ワークフローは独自に自身の状態に照らしてコマンドを検証し、古い決定や重複した決定を無視します。これにより、Lakebaseのプロジェクションに遅延が発生している場合でも実行が保護されます。

シグナルを受け入れた後、ワークフローはべき等なLakebaseアクティビティを通じて決定を永続化します。承認または否認によって実行が完了します。追加情報の要求は、プロジェクションをRUNNINGに戻し、レビュアーの根拠をユーザーメッセージとして追加して、次のターンを開始します。ターンが変更されたため、次の推奨事項は新しい review_id を受け取ります。

APIの202レスポンスは、Temporalがシグナルを受信したことを確認するものです。ビジネス上の受け入れはワークフロー内で非同期に行われるため、コマンドがAPIの事前チェックを通過しても、レビュー状態が変更されている場合は無視される可能性があります。クライアントはLakebaseのプロジェクションをリフレッシュして、結果の状態を観察します。

ワーカーを再デプロイせずにガバナンスされたポリシーを提供する

査定のしきい値は、ワーカーのコードとは独立して変更されます。Unity Catalog内のソーステーブルには、最低信用スコア、自動承認DTI、ハードデクライン（即時否認）しきい値、ポリシー名など、目的に応じた値が含まれています。

セットアップスクリプトは、agent_policy.underwriting_policy_limits. policy_lookup という名前の継続的なLakebase同期テーブルを作成し、正規化されたローン目的に従ってこの読み取り専用のPostgresコピーをクエリします。ポリシー所有者が Unity Catalog ソースを更新すると、同期パイプラインが変更を伝播し、その後の実行ではワーカーやAPIをデプロイすることなくその変更が読み込まれます。

ポリシー結果には、適用されたしきい値、各ルールの実際の値と合否結果、およびソースが含まれます。Lakebaseが無効になっているか、行が利用できない場合、デモはフィクスチャポリシーにフォールバックし、そのパスを fixture_fallback として記録します。規制されたワークフローは、代わりにフェイルクローズ（fail closed）になる場合があります。アプリケーションは、そのフォールバックの決定を明示的に行う必要があります。

運用の変更を Unity Catalog に戻す

リポジトリは、REPLICA IDENTITY FULL を設定することによって、各 agent_ops テーブルをLakebaseの変更データフィード（Change Data Feed）用に準備します。管理者は依然として、スキーマに対してこの機能を有効にする必要があります。その後、LakebaseはPostgresのライトアヘッドログ（WAL）から挿入、更新、削除をキャプチャし、それらをバッチ処理で、lb_<table>_history パターンで命名された Unity Catalog 管理のDelta履歴テーブルに書き込みます。

変更データフィード（Change Data Feed）は現在パブリックプレビュー中であり、およそ15秒ごとに変更をフラッシュします。この間隔は監査や分析に適していますが、UIは現在の運用状態を把握するためにLakebaseを直接クエリします。

履歴テーブルは、実行のポリシーソース、ツールのエビデンス、アクティビティの試行、レビュー待ち、推奨事項、および人手による決定を再構築できます。リポジトリはこのパス用にソーススキーマを構成しますが、実際に観測されたエンドツーエンドの変更データフィード（Change Data Feed）の実行は含まれていません。フィードの有効化と宛先テーブルの検証は、デプロイ手順として残されています。

システムの運用

デプロイでは、React/FastAPIがTemporalワーカーから分離されます。APIレプリカはリクエスト負荷に応じてスケールし、ワーカーはワークフローとアクティビティタスクのバックログ、および構成された同時実行数に応じてスケールします。Lakebaseの自動スケーリングは、プロジェクトの制限範囲内でデータベースのコンピューティングを調整します。

Temporal Cloudの価格設定は、アクションに加えて、アクティブおよび保持されているイベント履歴（Event History）ストレージに基づいているため、リトライの頻度や長期間開いている履歴もコストに影響します。チームは依然として、ワークロードに合わせてKubernetesレプリカ、タスクキュー（Task Queues）、接続プール制限、およびデータベースの境界を設定する必要があります。あるいは、最新のオープンソースリリースを使用して、独自のオープンソースTemporalサービスを立ち上げることもできます。

Lakebaseクライアントは、OAuthのM2M（Machine-to-Machine）認証を使用します。DatabricksのOAuthトークンと生成されたデータベース資格情報は期限切れになるため、クライアントは1時間のデータベース資格情報が期限切れになる前にSQLAlchemy接続プールをリフレッシュします。接続にはTLSを使用します。ローテーションを行わないと、長時間実行されているワーカーは予測可能なスケジュールでデータベースエラーに遭遇することになります。

オペレーターは、Temporalを使用してワークフローとアクティビティの履歴を検査し、Lakebaseを使用してアプリケーションの状態とメトリクスをクエリし、Kubernetesを使用してプロセスとデプロイの健全性を確認します。これにより、オペレーターは意図的なレビュー待ちを、アクティビティのリトライ、データベースへのアクセス失敗、またはツールの失敗と区別できます。

エビデンスと制限

テストスイートには、ワークフローのシーケンス制御、レビュー動作、OAuth接続の構築、べき等な永続化、メトリクス規約、APIワークフローの起動、およびワーカー設定に関する21個の合格テストが含まれています。クラッシュリカバリスクリプトは、決定論的プロバイダーを使用したプロセス障害のエクササイズを追加します。

申請者とプロバイダーのデータはフィクスチャです。このリポジトリは、融資モデル、規制コンプライアンス、本番環境のセキュリティ管理、地域的な可用性、または大規模なパフォーマンスの検証は行いません。ローカルでのクラッシュエクササイズはLakebaseを無効にして実行されたため、Temporalのリカバリを分離して検証します。Change Data Feedは、対象のDatabricks環境で有効化と検証を行う必要があります。

次のステップ

さらに詳しく知りたいですか？ご自身でデモを実行して、耐久性のある実行（durable execution）の動作を確認してください。リファレンス実装を実行し、実行途中でワーカーを停止して、エージェントが復旧する様子を確認してください。Lakebaseを接続して、各意思決定の背後にあるエビデンス、ポリシー、および人的レビューの状態を探索できます。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事