ワーカーをキルしても、ワークフローは死なない — Temporal + Pydantic AI

こんにちは、次世代システム研究室のN.M.です。

この記事のために、小さなデモを作りました。Pydantic AI 製のターミナルチャットエージェントで、Temporal についての質問に答えます——そして Temporal の上で動いています。このデモの締めくくりは、いつも決まってこうです。

回答の途中で、ワーカーを kill -9 します。チャットエージェントは、よりによって「Temporal のワーカーが死ぬとどうなるか」を説明している最中です。

誰も受け取る者のいない虚空へ、さらにメッセージを送信します。それから新しいワーカーを起動して見守ると——中断された回答は最後まで生成され、待機していたメッセージにも返事が届き、会話履歴は無傷のまま。しかもエージェントは私の名前を覚えています。

種も仕掛けもありません。会話そのものを Temporal のワークフローにすると、こうなるのです。

なぜこれが重要なのか。クラウドは長時間実行プロセスにとって過酷な環境だからです。ワーカーは kill されては再スケジュールされ、マシンは入れ替わり、デプロイのたびにすべてが作業の途中で再起動されます。

ステートレスなリクエストハンドラなら平気です。しかしエージェントの実行はそうはいきません。長く、高価で、状態を持つ——何分ぶんもの LLM 呼び出しとツール実行の結果が、まさにクラウドが今にも回収しようとしているメモリの中に積み上がっていきます。プロセスを失えば、会話も失われます。

Durable Execution を1分で

Temporal の売り文句を一行で言えば「クラウドを生き延びるワークフロー」です。アプリケーションを2つの役割に分割します。

Temporal サーバーは「永続」を担う側です。ワークフローが実行する全ステップをイベント履歴として記録します。スケジュールされたアクティビティ、返ってきた結果、発火したタイマーのすべてです。

ワーカーは「使い捨て」の側で、実際の Python コードを実行します。サーバーはコードを一切実行しません。スケジュールし、記録し、リトライし、タイムアウトさせる——オーケストレーションに徹します。

つまり生存の正体はリプレイです。ワークフローのコードは決定的(deterministic)でなければならず、副作用のあるもの——LLM 呼び出し、DB 書き込み、ファイル読み取り——はすべてアクティビティに置かれ、その結果が履歴に記録されます。

ワーカーが死ぬと、別のワーカーがそのワークフローを引き継ぎ、履歴を先頭からリプレイします。完了済みのアクティビティは再実行されず、記録済みの結果を返すだけです。

数秒のうちに、新しいワーカーは死んだワーカーが失ったメモリ上の状態を寸分違わず再構築し、未完了だった最初のステップから実行を再開します。冒頭のシーンの種明かしは、これがすべてです。

会話そのものをワークフローにする

実際にキルできる対象を用意するため、Temporal のドキュメント(markdown)をもとに Temporal についての質問に答えるチャットエージェントを作りました。動作基盤も Temporal です。この自己言及が狙いで、「ワーカーが死ぬとどうなる?」と質問しておいて、その場で実演できます。

アーキテクチャは一文で説明できます。クライアントはただの端末で、メッセージをシグナルとして送り込み、会話履歴をクエリでポーリングして表示するだけです。

本物の状態——あなたのメッセージ、エージェントの記憶、会話履歴の全文——はすべてワークフローの中にあります。だからワーカー(やクライアント)を kill しても何も失われません。チャットの各概念は、Temporal のプリミティブと1対1で対応します:

チャットの概念	Temporal のプリミティブ	kill -9 を生き延びる?
メッセージを送る	シグナル	✅ ワーカーが1つもいなくてもサーバー上にキューされる
会話履歴を読む	クエリ	✅ 永続化された状態を読むだけ
エージェントの記憶	ワークフローの状態	✅ イベント履歴から再構築される
LLM / ツール呼び出し	アクティビティ	✅ 一度だけ記録され、中断時はリトライされる

会話ループの全体がこちらです。擬似コードではなく、実際に動くコードです:

from temporalio import workflow
from pydantic_ai.durable_exec.temporal import PydanticAIWorkflow

from .agents import temporal_chat_agent  # a pydantic-ai Agent wrapped for Temporal


@workflow.defn
class ChatWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [temporal_chat_agent]

    def __init__(self) -> None:
        self._inbox: list[str] = []
        self._transcript: list[dict] = []
        self._history = None  # pydantic-ai message list — the agent's memory
        self._ended = False

    @workflow.run
    async def run(self) -> str:
        while True:
            await workflow.wait_condition(lambda: bool(self._inbox) or self._ended)
            if self._ended and not self._inbox:
                break

            user_text = self._inbox.pop(0)
            self._transcript.append({"role": "user", "content": user_text})
            result = await temporal_chat_agent.run(
                user_text, message_history=self._history
            )
            self._history = result.all_messages()
            self._transcript.append({"role": "assistant", "content": result.output})

        turns = sum(1 for m in self._transcript if m["role"] == "user")
        return f"chat ended after {turns} user turns"

    @workflow.signal
    def user_message(self, text: str) -> None:
        self._inbox.append(text)

    @workflow.signal
    def end_chat(self) -> None:
        self._ended = True

    @workflow.query
    def transcript(self) -> list[dict]:
        return self._transcript

run() が何を受け取っていないかに注目してください。あなたのメッセージです。ワークフローの入力引数は開始時に固定されますが、チャットのメッセージは会話開始の時点ではまだ存在せず、その後何分にもわたって届き続けます。

実行中のワークフローに新しいデータを渡す方法はシグナルだけです。これは回避策ではなく、そういう設計なのです。

メッセージがワーカーの死をこうもあっさり生き延びる理由も同じです。シグナルの届け先はワーカーではなくサーバー。まず永続的なイベント履歴に記録され、ワーカー(どの個体でもかまいません)が次に生きているときに処理されます。

TemporalAgent の正体

Pydantic AI 側の話は意外なほど小さくまとまっています。pydantic_ai.durable_exec.temporal は pydantic-ai 本体に同梱される公式インテグレーションで、その仕事は普通の Agent をワークフローのコード内で「合法」にすることです。

エージェントはいつも通りに作り、それをラップするだけです:

from datetime import timedelta

from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalAgent
from temporalio.common import RetryPolicy
from temporalio.workflow import ActivityConfig

chat_agent = Agent(
    "google:gemini-2.5-flash",
    name="chat_agent",
    toolsets=[docs_tools],  # search_docs / read_doc over a folder of markdown
    instructions="You answer questions about Temporal using its documentation. "
                 "Use search_docs, then read_doc for detail.",
)

temporal_chat_agent = TemporalAgent(
    chat_agent,
    # LLM round-trips: generous timeout, exponential backoff
    model_activity_config=ActivityConfig(
        start_to_close_timeout=timedelta(seconds=120),
        retry_policy=RetryPolicy(
            initial_interval=timedelta(seconds=1),
            backoff_coefficient=2.0,
        ),
    ),
    # Tool calls: fast local I/O, quick retries
    activity_config=ActivityConfig(
        start_to_close_timeout=timedelta(seconds=60),
    ),
)

このラッパーの仕事はただ1つ、しかし決定的に重要です。すべてのモデルリクエストとツール呼び出しが、自動的に Temporal のアクティビティになります。

ワークフローのコードは決定的でなければなりませんが、エージェントの実行は非決定性の塊です——ネットワーク呼び出し、LLM のサンプリング、ツールの I/O。TemporalAgent はその非決定性をまるごとアクティビティへ移し、結果は記録され、リトライされます。

ワークフローに残るのは決定的なオーケストレーションのループだけです。設定はステップの種類ごとに分かれており、LLM 呼び出しは model_activity_config、ツール呼び出しは activity_config。いずれもサーバーが強制します。

ワークフロー側に必要なのは、先ほどの ChatWorkflow にあった PydanticAIWorkflow 基底クラスと __pydantic_ai_agents__ の宣言だけです。あとは素の pydantic-ai と同じように temporal_chat_agent.run(...) を呼ぶだけで、message_history もそのまま使えます。

この継ぎ目は後述の Web UI のスクリーンショットでも確認できます。agent__chat_agent__model_request_stream や agent__chat_agent__toolset__docs_tools__call_tool といったアクティビティ名は、まさにラッパーの仕事ぶりです——エージェントの1ステップが、1つの永続アクティビティになっています。

kill の一部始終

実際に行った手順は次の通りです:

1. 自己紹介する。エージェントの最初の回答は普通に返ってくる。
2. ドキュメント検索が必要な質問を投げ、回答の途中でワーカーを kill -9 する。回答は中途半端なところで止まる。今回は search_docs ツール呼び出しの直後、LLM リクエストの最中に死んだ。
3. ワーカーが1つもいない状態でさらにメッセージを打つ。何もクラッシュしない。シグナルはサーバーにキューされ、クライアントのステータスバーが「誰もいない」と告げるだけ。
4. 新しいワーカーを起動する。イベント履歴がリプレイされ、中断された回答が完成し、キューされていたメッセージにも回答が届く。
5. 駄目押しにクライアントも終了し、ワークフロー ID を指定して新しいクライアントを接続する。会話履歴の全文が戻ってくる——ステップ1で名乗った名前も、エージェントは覚えている。

これが犯行直前の瞬間です。質問は入力済みで、あとは送信するだけ:

Enter を押し、エージェントがツールのループに入るまで数秒待ってから、ワーカーを kill -9。クライアントは異変に気づきますが、サーバーは我関せず。構わず次のメッセージを打ちます:

一方 Web UI を見ると、「サーバー上は安全」の意味がわかります。ワークフローは依然 Running、イベントはすべて記録済み、実行中だった LLM アクティビティは「Attempt 1 / ∞」で待機し、虚空に送ったメッセージも user_message シグナルとして記録されています:

そして新しいワーカーを起動します。最初に出力されるのが、この一部始終の要約です:

⟲ [workflow] REPLAY: rebuilding chat session from event history —
  full conversation memory restored, no APIs re-called

チャットクライアントには、中断されたことに気づいてすらいない会話が表示されます。死にかけていた回答は完成し、虚空へのメッセージにも返事が付きました:

このオチは仕込みではありません。ワーカーがキルされたとき答えていた質問は「ワーカーのプロセスが実行中に死んだら、私のワークフローはどうなりますか?」。そして蘇生後に完成した回答には、一字一句こう書かれていました:

The Cluster then automatically re-adds this uncompleted Workflow Task back to the Task Queue. Another available worker will then pick up this task and, by replaying the workflow’s “Event History,” reconstruct the workflow’s state to the point of failure.

(訳:クラスターは未完了の Workflow Task を自動的に Task Queue へ戻します。別の稼働中のワーカーがこのタスクを拾い、ワークフローの「イベント履歴」をリプレイすることで、障害発生時点までの状態を再構築します。)

システムは、答えを生成しながらその答えを実演したのです。

ステップ5は地味ですが、「プロセスには何も住んでいない」という主張に決着をつけます。クライアントを終了すると、戻り方まで教えてくれます:

(detached — the chat workflow keeps running; re-attach with: uv run demo-chat docs-chat)

まっさらなクライアントプロセスが、ワークフロー ID を頼りに transcript クエリ1回で会話全体を再構築します。エージェントの記憶も無傷です。そもそも記憶はクライアントにも(ワーカーにも)なかったのですから:

説得力には両方の画面が必要です。Web UI は機構を見せます——すべてのシグナル、すべての LLM アクティビティ、ワーカーが消えた瞬間、リプレイ。チャットクライアントは、それがユーザーには一切関係なかったことを見せます。

UI だけでは抽象的すぎ、アプリだけでは手品にしか見えません。

もう1つ、エージェント開発者が最も気にする点を。リプレイはモデルを再呼び出ししません。

完了済みの LLM 呼び出しは記録済みのアクティビティです。再起動後、その結果は履歴から返ってきます——API 呼び出しなし、課金なし。デモのワーカーは本物の API リクエストのたびに 💸 マーカーを出力しますが、リプレイ中には1つも現れません。

書かなくていいリトライ

ワーカーの死は派手な障害です。日常的なのは下流呼び出しの失敗——LLM プロバイダーが最悪のタイミングで返してくる 429 や 503 です。

たいていのエージェントのコードベースには、手書きの対策が生えてきます。全呼び出しを囲う try/except、バックオフのヘルパー、どこかに保存されるリトライカウンター。Temporal では、それはあなたのコードではありません。リトライポリシーを宣言すれば、リトライはサーバーが実行します。

デモでは、信頼できない API を模したアクティビティでこれを可視化します。最初の N 回はわざと失敗します:

@activity.defn
async def flaky_external_call(fail_times: int) -> str:
    """Simulates an unreliable downstream API: fails the first N attempts."""
    attempt = activity.info().attempt
    if attempt <= fail_times:
        raise RuntimeError(f"503 Service Unavailable (attempt {attempt})")
    return f"200 OK on attempt {attempt} (after {fail_times} failures)"

これを呼ぶワークフローには、リトライのロジックが一切ありません。あるのはポリシーだけです:

@workflow.defn
class RetryWorkflow:
    @workflow.run
    async def run(self, fail_times: int) -> str:
        return await workflow.execute_activity(
            flaky_external_call,
            fail_times,
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                backoff_coefficient=2.0,
                maximum_interval=timedelta(seconds=30),
                # no maximum_attempts: retry until it succeeds
            ),
        )

fail_times=6 で実行しました。Web UI の Pending Activities が、悪戦苦闘の一部始終を実況してくれます。試行回数、直近の失敗、次のリトライまでのカウントダウン:

その20秒後には "200 OK on attempt 7 (after 6 failures)"。リトライのループは誰も書いていません。

注目すべき点が2つあります。まず activity.info().attempt はサーバーがコードに渡してくる値です。つまりリトライの状態そのものもワーカーの死を生き延びます。試行3と4の間でワーカーを kill しても、次のワーカーは試行4から続きを実行します。

次に、これは TemporalAgent がチャットエージェントの LLM 呼び出しに宣言したのと同じ仕組みです。会話の途中でモデルプロバイダーが 429 を返しても、同じバックオフ処理が目に見えないところで行われます。

細則:永続性は契約である

ここまでの一切が成立するのは、Temporal があなたのアクティビティのコードに2つの前提を置いているからです。私たちはそれをよく知っています。両方とも破ったことがあるからです。ドキュメントは「長時間実行アクティビティにはハートビートを」「アクティビティは冪等に」と教えていました。どちらもやりませんでした。

その請求書は、1件のインシデントとして届きました。サーバーは長時間実行アクティビティがタイムアウトしたと判断し、リトライをスケジュールしました——最初の試行がまだ実行中だったにもかかわらず。

2つの試行が同じ行を INSERT。よりによって、この種の事故を防ぐために導入した仕組みの側から、一意キー制約違反が飛び出してきました。幸い、ステージング環境でした。

このインシデントを説明するメンタルモデルはこうです。Temporal サーバーはコントローラーであって、エグゼキューターではない。アクティビティのコードはワーカーの中で動いており、サーバーはそれを追跡しているだけです。

タイムアウトが発火しても、サーバーはアクティビティをキルしません——キルする手段を持っていないのです。履歴に ActivityTaskTimedOut と書いて先へ進むだけで、あなたのコードは何も知らずに動き続けます。

最初の試行は誰にも殺されていない。だからリトライは「置き換えたはず」の試行と並走しうる。契約条項が2つある理由はこれです:

• 長時間実行アクティビティにはハートビートを。 ハートビートは、サーバーの一方的なタイムアウトを双方向の対話に変えます。なければサーバーは「ワーカーが死んだ」と「アクティビティが遅いだけ」を区別できず、キャンセルも手遅れになるまでアクティビティに届きません。
• アクティビティは冪等に。 リトライは、置き換える対象の試行と並走したり、その副作用の後に実行されたりします。下流の状態をリトライから守れるのは冪等性だけです。

長寿命ワークフローに固有の条項がもう1つあります。イベント履歴は蓄積し続け、Temporal にはそのサイズに対する固い上限があります。

脱出口は continue-as-new です。ワークフローが現在の状態を、空の履歴で始まる新しい実行へ引き渡します。

今回のデモでは上限に近づくことすらありませんでしたが、プロダクションでは初日から設計に織り込みます。continue-as-new は履歴が爆発したときの緊急パッチではなく、「長寿命」という設計の一部なのです。

ここからどうするか

デモの全体は小さなものです。ワークフロークラス1つ、シグナル2つ、クエリ1つ、そして TemporalAgent でラップした pydantic-ai エージェント——Pydantic チームによる公式インテグレーションです。特殊なことは何ひとつしていません。それこそが要点です。

このパターンは、1リクエストに収まらない仕事をするあらゆるエージェントに一般化できます。リサーチのパイプライン、承認フロー、多段のツール使用、会話の向こう側に人間がいるすべてのもの。

クラウドはこれからもあなたのプロセスを殺し続けます。どれだけ丁寧に作っても、それは止められません。

変えるべきはアーキテクチャです。エージェントの状態を、たまたまそれを実行しているプロセスと運命共同体にするのはやめましょう。エージェントは、長寿命のワークフローとして作り始めるのです。

関連リンク

Temporal — durable execution プラットフォーム公式サイト

Pydantic AI — 本記事で使用したエージェントフレームワーク（Temporal integration）

次世代システム研究室では、グループ全体のインテグレーションを支援してくれるアーキテクトを募集しています。インフラ設計、構築経験者の方、次世代システム研究室にご興味を持って頂ける方がいらっしゃいましたら、ぜひ募集職種一覧からご応募をお願いします。