06. Temporal との統合パターン¶

ゴール¶

ユーザーの既存 ~/temporal-workflows に LangGraph agent をActivity として組み込む
「durable な外殻 = Temporal」「agent の中身 = LangGraph」の責務分担を体に染み込ませる
落ちないし、人間レビューで数日寝かせても OK な構成を作る

全体像¶

┌────────────────────────────────────────────┐
│ Temporal Workflow                           │
│   - 入力: ユーザーリクエスト                  │
│   - durable timer / retry / signal          │
│   - human review は Signal で待つ            │
│                                             │
│   ┌──────────────────────────────────┐      │
│   │ Activity: run_agent_step()        │      │
│   │   ┌─────────────────────────┐    │      │
│   │   │ LangGraph .invoke()      │    │      │
│   │   │   - tool loop            │    │      │
│   │   │   - in-memory checkpoint │    │      │
│   │   └─────────────────────────┘    │      │
│   └──────────────────────────────────┘      │
└────────────────────────────────────────────┘

役割分担:

層	責務	なぜそこか
Temporal Workflow	全体の流れ、retry、timeout、signal、schedule、durable timer	数時間〜数日にまたがる信頼性は Temporal の得意分野
Activity	LangGraph の1ステップ実行（または1回の invoke）	I/O 境界。LLM 呼び出しは Activity で必ず包む
LangGraph	agent loop の中身、tool 選択、state machine	「どう考えるか」のロジック

ベストプラクティス¶

1. Workflow 内で LLM を直接呼ばない¶

# NG: Workflow code で LLM 呼ぶ
@workflow.defn
class BadAgent:
    @workflow.run
    async def run(self):
        result = await llm.ainvoke(...)  # ← determinism 違反

LLM 呼び出しは非決定的。必ず Activity に隔離する。

# OK
@workflow.defn
class GoodAgent:
    @workflow.run
    async def run(self, task: str):
        result = await workflow.execute_activity(
            run_langgraph_step,
            task,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

2. LangGraph の checkpoint は Activity 内に閉じる¶

Workflow が durable なので、agent 全体の state は Workflow state（普通の Python 変数）に持てばよい
LangGraph 側 checkpointer は MemorySaver（in-memory）で十分。Activity 内で短命に使い捨てる
永続化は Temporal が event history でやってくれる

3. Tool 呼び出しを別 Activity に切り出す¶

@activity.defn
async def call_tool(name: str, args: dict) -> str:
    return TOOLS[name](**args)

@workflow.defn
class AgentWorkflow:
    @workflow.run
    async def run(self, task):
        messages = [("user", task)]
        for _ in range(MAX_STEPS):
            # 1. LLM に次のアクションを聞く
            ai_msg = await workflow.execute_activity(
                llm_decide, messages, start_to_close_timeout=timedelta(minutes=2)
            )
            messages.append(ai_msg)
            if not ai_msg.get("tool_calls"):
                return ai_msg["content"]
            # 2. ツールを並列実行
            results = await asyncio.gather(*[
                workflow.execute_activity(
                    call_tool, c["name"], c["args"],
                    start_to_close_timeout=timedelta(minutes=10),
                    retry_policy=RetryPolicy(maximum_attempts=5),
                )
                for c in ai_msg["tool_calls"]
            ])
            messages.extend(results)

これで tool ごとに独立した retry / timeout がかかる。LangGraph のグラフ構造を Workflow code に書き直したような形。

4. Human-in-the-loop は Signal で¶

@workflow.defn
class AgentWorkflow:
    def __init__(self):
        self.review_result = None

    @workflow.signal
    def submit_review(self, decision: dict):
        self.review_result = decision

    @workflow.run
    async def run(self, task):
        plan = await workflow.execute_activity(make_plan, task, ...)
        # 人間レビュー待ち（数日寝かせても OK）
        await workflow.wait_condition(lambda: self.review_result is not None)
        if self.review_result["approve"]:
            await workflow.execute_activity(execute_plan, plan, ...)

LangGraph の interrupt() と思想は同じ。実装が Workflow 側に寄っただけ。

いつ LangGraph をそのまま Activity 内で使うか¶

Workflow にロジックを書き下すか、LangGraph をそのまま 1 Activity で呼ぶかは選択。

選択肢	向く状況
Workflow にロジック展開	tool ごとに retry / timeout を厳密制御したい、人間レビューが複数回入る、ステップ数が多い
LangGraph をそのまま 1 Activity	agent が短時間で完結、グラフ構造を流用したい、LangGraph Studio でデバッグしたい

ハイブリッドも可: 「大枠は Workflow、各サブ問題は LangGraph subgraph を Activity 内で呼ぶ」。

最小サンプル: LangGraph を 1 Activity で呼ぶ¶

# activities.py
from temporalio import activity
from my_langgraph_agent import graph

@activity.defn
async def run_agent(task: str) -> str:
    result = await graph.ainvoke({"messages": [("user", task)]})
    return result["messages"][-1].content

# workflows.py
from temporalio import workflow
from datetime import timedelta
from temporalio.common import RetryPolicy

@workflow.defn
class SimpleAgentWorkflow:
    @workflow.run
    async def run(self, task: str) -> str:
        return await workflow.execute_activity(
            run_agent, task,
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

Worker と Client の組み立ては既存の ~/temporal-workflows のコードに準じる。

参考実装¶

temporal-community/temporal-ai-agent — multi-turn agent を Workflow に格納したリファレンス
temporalio/openai-agents-python — activity_as_tool ヘルパー
knowledgestack/ks-cookbook — Temporal / OpenAI Agents / LangGraph 横断 32 サンプル
Temporal blog: Of course you can build dynamic AI agents with Temporal

Try¶

02章の ReAct agent を Activity でラップして Temporal Workflow から呼ぶ
tool 呼び出しを Workflow code に展開し、tool ごとに違う retry policy を設定
Signal で人間レビューを差し込み、temporal workflow signal CLI で承認を送る
Continue-As-New で 1000 ターンの長期 agent を組む
Schedule で「毎朝9時に research agent を回す」を作る（既存 ~/temporal-workflows に近い）

学んだこと¶

LLM 呼び出しは必ず Activity に隔離（determinism 違反を避ける）
LangGraph checkpoint は in-memory で OK、durable 性は Temporal が担保
Tool ごとに retry / timeout を独立に設定できるのが Temporal 化のうま味
HITL は Workflow Signal で実現。LangGraph interrupt と等価
既存の ~/temporal-workflows に自然に組み込める

卒業課題¶

00_index のチェックリストを全部埋める。最後の項目「LangGraph agent を Temporal Activity から呼び出すサンプルが動く」が動けば卒業。

次のステップは 99_resources のリンクから興味あるものを深掘り。実プロジェクトに何か投入してみる（research-orchestrator や ai-business-lab で使い道がありそう）。

作成: 2026-05-16 ／最終更新: 2026-05-16