Temporal活用事例ブログ深掘りまとめ + 導入設計

概要

Temporalはワークフローオーケストレーションエンジンであり、耐障害性・状態管理・可視性を備えた長時間実行プロセスを構築できる。本ノートでは海外の主要ブログ記事から活用パターンを整理し、自身のユースケース（Claude Codeスキルの自動化）に適用する設計をまとめる。

---

Part 1: 主要ブログ記事の深掘りまとめ

記事1: Platform Engineering × Temporal — 5つの実用ユースケース

URL: https://temporal.io/blog/enabling-platform-engineering-with-temporal-five-practical-use-cases

ユースケース	課題	Temporalでの解決	効果・事例
インシデント対応・ランブック自動化	手動プレイブック依存、属人化	インシデントロジックをコード化、人的承認は一時停止、全ステップ履歴記録	Datadog: 大規模マイグレーション自動化
証明書ローテーション	cronジョブでは失敗検知不可、無声障害	耐障害ワークフロー、自動リトライ+指数バックオフ、UI進捗可視化	クレデンシャル放置を防止
マルチクラウド・ハイブリッド環境管理	マルチステップ処理中断→リソース不整合	AWS/GCP/Azure横断のプロビジョニングオーケストレーション、ドリフト検知・修正	完全なライフサイクル管理と可視化
CI/CDパイプライン	数時間後の失敗で全プロセス再開	ステートフルワークフローで失敗ポイントから再開、カナリアデプロイ管理	Netflix: SpinnakerをTemporal基盤に再構築
内部開発プラットフォーム（IDP）	開発者がインフラ詳細を理解する必要	複雑タスクを隠蔽、シンプルなトリガーで実行、「ゴールデンパス」提供	Humana: 開発者がビジネスロジックに集中

アーキテクチャの要点: - ワークフローは「コードとしての運用ロジック」であり、脆弱なスクリプトを置き換える - Human-in-the-loop: ワークフローは人間の入力を無期限に待機できる - ワークフロー履歴による完全な可観測性（監査・ポストモーテム）

---

記事2: 生成AI × Temporal — 9つのユースケース

URL: https://temporal.io/blog/temporal-use-case-roundup-generative-ai

#	ユースケース	企業	Temporal活用
1	ビデオコンテンツ処理・翻訳	Descript	AI翻訳・字幕ワークフローのオーケストレーション
2	会話型AI・コールセンター文字起こし	複数社	日々数千件の通話処理パイプライン（Python SDK）
3	合成データ生成	—	データ生成→匿名化→処理のパイプライン
4	予測分析・顧客インサイト	—	データ収集→モデル学習→予測のワークフロー
5	市場インテリジェンス	Messari	暗号資産の大規模データストリームの取り込み・充実化
6	AI搭載HRワークフロー	—	オンボーディング・福利厚生処理の自動化
7	AI基盤・リソース最適化	—	GPUセットアップ〜監視〜削除の全プロセス
8	生成AI状態追跡	—	チャットボットの会話状態・セッション管理
9	AI研究ワークフロー	—	実験の反復・スケーリング・本番デプロイ

Docker社もTemporal Generative AI Catalogにエージェントオーケストレーション機能を掲載。

Temporalが生成AIに適する理由: - 同期/非同期プロセスの統合ハンドリング → レイテンシ削減 - 状態追跡 → チャットボット・会話AIに不可欠 - コスト最適化 → 必要なデータのみLLMに通す設計が可能 - スケーラビリティ → 高負荷AI処理に対応

---

記事3: TemporalでのAIエージェント構築

URL: https://temporal.io/blog/of-course-you-can-build-dynamic-ai-agents-with-temporal

核心: 「LLMは非決定論的だからTemporalは使えない」は誤り。OpenAI Codex、Replit Agent 3もTemporal上で動作している。

アーキテクチャの分離原則

┌─────────────────────────────────────┐
│  Workflow層（決定論的）               │
│  - オーケストレーションの骨格         │
│  - LLM決定の記録・管理               │
│  - クラッシュ復旧（イベント履歴から） │
└──────────────┬──────────────────────┘
               │ execute_activity()
┌──────────────▼──────────────────────┐
│  Activity層（非決定論的）             │
│  - LLM呼び出し                       │
│  - ツール実行                         │
│  - 外部API呼び出し                   │
└─────────────────────────────────────┘

• 決定論的に実行される（同じコンテキスト・同じLLM決定なら同じパスを辿る）
• 事前決定ではない（実行時までLLMが何を決定するかは不明）
• クラッシュ復旧時はイベント履歴からリプレイ（LLMを再呼び出ししない）

コード例: AIエージェントワークフロー

@workflow.defn
class AIAgentWorkflow:
    @workflow.run
    async def run(self, user_goal: str) -> str:
        conversation_history = []
        while not self.is_goal_achieved(conversation_history):
            # 非決定論的: LLMが次のアクションを決定
            next_action = await workflow.execute_activity(
                llm_decide_next_action,
                LLMRequest(
                    goal=user_goal,
                    history=conversation_history,
                    available_tools=self.get_available_tools()
                ),
                start_to_close_timeout=timedelta(seconds=30),
            )
            # 決定論的: 選択されたツールを実行
            result = await workflow.execute_activity(
                next_action.tool,
                next_action.params,
                start_to_close_timeout=timedelta(seconds=30)
            )
            conversation_history.append({
                "action": next_action, "result": result
            })
        return self.format_final_result(conversation_history)

メリット: 1. 耐久性 — クラッシュ時にイベント履歴から正確に再開 2. 一貫性 — 同じLLM決定なら同じパスを辿る 3. 効率性 — LLM呼び出しの重複実行が不要

---

記事4: マルチエージェント × Temporal

URL: https://temporal.io/blog/using-multi-agent-architectures-with-temporal

なぜマルチエージェントか

• 単一エージェントはツール過多・コンテキスト過多で性能低下
• エージェントごとにアクセス制限、特化モデルの使用が可能
• 特化エージェントは汎用エージェントより評価が容易

設計パターン

1. エージェントルーティング: タスクに応じた専門エージェントへの振り分け 2. タスク委譲: オーケストレータが各専門エージェントにサブタスクを委譲

DAPERパターン（問題解決の再利用可能パターン）

フェーズ	説明
Detect	問題の検知
Analyze	問題の性質を分析
Plan	修正計画の作成
Execute	計画の実行
Report	解決結果の分析・報告

エージェント種別とTemporal実装の対応

エージェント種別	実装層	理由
会話型エージェント	Workflow	対話的、長時間実行、オーケストレーション
LLM呼び出し	Activity	外部システム呼び出し
単純自動化エージェント	Activity	短命、対話なし
長時間実行エージェント	Workflow	長時間、対話あり
プロアクティブエージェント	Workflow	対話的、長時間、能動的

信頼度ベース意思決定

if self.planning_confidence_score <= 0.95:
    # 95%未満: 人間に通知し承認を待つ
    await self.execute_notification()
    await workflow.wait_condition(
        lambda: self.approved or self.rejected,
        timeout=timedelta(hours=12),
    )
else:
    # 95%以上: 自動承認
    self.approved = True

MCP連携

• Model Context Protocolを通じて会話型エージェントとTemporalワークフローを統合可能
• Temporal Workflow Queriesで状態をMCPツールに公開
• goose（Block社）、Slack、Claude、VSCode等のMCP互換エージェントと連携

---

Part 2: Temporal導入設計

対象ユースケース

現在のClaude Codeスキル/日常タスクをTemporal化する:

優先度	ワークフロー	現状	Temporal化のメリット
高	企業リサーチ	/company-researchスキル（手動実行）	定期実行、複数企業の並列処理、結果の差分検知
高	文献レビュー	/literature-reviewスキル（手動実行）	新着論文の自動検知・レビュー、Zotero連携
中	研究日誌更新	/research-journal-updateスキル	スケジュール実行、進捗の自動トラッキング
中	求人情報モニタリング	手動WebSearch	定期的に候補企業の求人変更を検知・通知

推奨アーキテクチャ

┌─────────────────────────────────────┐
│           Temporal Server            │
│        (Docker / Temporal Cloud)     │
└──────────┬──────────────────────────┘
           │
┌──────────▼──────────────────────────┐
│          Python Worker               │
│  ┌─────────────────────────────┐    │
│  │ Workflows (決定論的)         │    │
│  │  - CompanyResearchWorkflow  │    │
│  │  - LiteratureReviewWorkflow │    │
│  │  - JobMonitorWorkflow       │    │
│  └─────────────────────────────┘    │
│  ┌─────────────────────────────┐    │
│  │ Activities (非決定論的)      │    │
│  │  - web_search()             │    │
│  │  - web_fetch()              │    │
│  │  - call_claude_api()        │    │
│  │  - write_obsidian_note()    │    │
│  │  - search_zotero()          │    │
│  └─────────────────────────────┘    │
└─────────────────────────────────────┘
           │
    ┌──────▼──────┐
    │  Obsidian   │  ← ノート自動生成
    │  Vault      │
    └─────────────┘

技術選定

項目	選定	理由
言語	Python	Anthropic SDK、既存スキルのロジック移植が容易
Temporal Server	Docker Compose（ローカル開発）→ Temporal Cloud（本番）	個人利用ならローカルで十分
LLM呼び出し	Anthropic Python SDK	Claude APIで既存スキルと同等の処理
スケジュール	Temporal Schedules	cron相当の定期実行機能が組み込み
リポジトリ	~/temporal-workflows/（新規）	mcp-serversと同様に独立リポジトリ

環境準備（前提条件）

注意: 現在Dockerが未インストールのため、事前にインストールが必要

1. Docker Desktop のインストール: https://www.docker.com/products/docker-desktop/
2. Python 3.12+ の準備（現在のシステムPythonは3.9.6）: uv経由で管理

実装ステップ

Step 1: 環境構築

• ~/temporal-workflows/ リポジトリ作成
• Docker Compose で Temporal Server + Web UI 起動
• Python Worker のプロジェクト構造セットアップ（uv使用）
• temporalio Python SDK インストール

Step 2: CompanyResearchWorkflow（最初のワークフロー）

現在の /company-research スキルをTemporal化:

@workflow.defn
class CompanyResearchWorkflow:
    @workflow.run
    async def run(self, input: CompanyResearchInput) -> CompanyResearchOutput:
        # Phase 1: 情報収集（並列Activity）
        blog_data, company_info, job_info = await asyncio.gather(
            workflow.execute_activity(fetch_tech_blog, ...),
            workflow.execute_activity(search_company_info, ...),
            workflow.execute_activity(fetch_job_listings, ...),
        )
        # Phase 2: Claude APIで分析・レポート生成
        report = await workflow.execute_activity(
            generate_research_report,
            AnalysisInput(blog_data, company_info, job_info),
        )
        # Phase 3: Obsidianノートに書き出し
        await workflow.execute_activity(write_obsidian_note, report)
        return report

Step 3: スケジュール設定

# 毎週月曜9:00に候補企業3社のリサーチを自動実行
await client.create_schedule(
    "weekly-company-research",
    Schedule(spec=ScheduleSpec(cron_expressions=["0 9 * * MON"])),
    action=ScheduleActionStartWorkflow(
        CompanyResearchWorkflow.run,
        CompanyResearchInput(companies=["Datachain", "Securitize", "Progmat"]),
    ),
)

Step 4: 追加ワークフローの段階的実装

• LiteratureReviewWorkflow（Zotero MCP連携）
• JobMonitorWorkflow（求人変更検知→通知）
• ResearchJournalWorkflow（研究日誌の自動更新）

ディレクトリ構成

~/temporal-workflows/
├── docker-compose.yml          # Temporal Server
├── pyproject.toml
├── src/
│   ├── workflows/
│   │   ├── __init__.py
│   │   ├── company_research.py
│   │   ├── literature_review.py
│   │   └── job_monitor.py
│   ├── activities/
│   │   ├── __init__.py
│   │   ├── web.py              # web_search, web_fetch
│   │   ├── claude.py           # Claude API呼び出し
│   │   ├── obsidian.py         # ノート読み書き
│   │   └── zotero.py           # Zotero API
│   ├── models/
│   │   ├── __init__.py
│   │   └── schemas.py          # Input/Output dataclasses
│   └── worker.py               # Worker起動エントリポイント
├── schedules/
│   └── setup_schedules.py      # スケジュール登録スクリプト
└── README.md

検証方法

1. docker compose up で Temporal Server 起動を確認
2. Worker プロセスを起動し、Temporal Web UI（localhost:8233）でWorker登録を確認
3. CompanyResearchWorkflow を手動トリガーし、Obsidianノートが生成されることを確認
4. スケジュール登録後、Web UI でスケジュール一覧と次回実行時刻を確認
5. 意図的にActivity失敗（ネットワーク切断等）を起こし、自動リトライを確認

---

参考リンク

• Temporal公式ドキュメント
• Temporal Python SDK
• Temporal AI Agent Bundle
• Platform Engineering × Temporal
• 生成AI × Temporal
• AIエージェント × Temporal
• マルチエージェント × Temporal

---

作成: 2026-03-29 ／ 最終更新: 2026-03-29