long-horizon-workflows

長時間実行ワークフロー

googleadsagent.ai™ による Agent Skills™ の一部です

説明

長時間実行ワークフローを使うと、AI エージェントは単一の会話ターンの範囲を大幅に超える、複数時間・複数フェーズにわたる自律的なパイプラインを実行できます。googleadsagent.ai™ の本番環境でのパターンと DeerFlow アーキテクチャに着想を得たこのワークフローでは、複雑な目標を段階的な実行計画に分解し、チェックポイント、進捗追跡、エラー復旧、重要な判断ポイントでの人間参加型の承認ゲートを組み込みます。たとえば、長時間実行ワークフローでは、Google Ads アカウント全体（数十のキャンペーン、数千のキーワード）を分析し、包括的な最適化レポートを生成して、実装可能な変更セットを準備するまでを、数時間にわたって自律的に実行できます。

長時間実行における最大の課題は信頼性です。3 時間かかるワークフローが、復旧手段なしに 2 時間半後に失敗するのは、役に立たないどころか有害です。時間と計算リソースを無駄にするためです。チェックポイント管理により、失敗前に完了した作業を保持し、そこから再開できます。進捗追跡により、エージェントが実行している処理とその進み具合を確認できます。人間参加型の承認ゲートを使えば、エージェントの処理を進める前に、人間が重要な判断（予算変更など）を検証できるため、無人運用での致命的なエラーを防止できます。

DeerFlow がこのパターンにもたらしたのは、階層的なタスク分解という考え方です。プランナーエージェントが目標をフェーズに分解し、各フェーズをタスクに、各タスクを不可分な操作に分解します。階層の各レベルには、それぞれ固有のチェックポイント、タイムアウト、エラー処理ポリシーがあります。これにより、個々のタスクが失敗しても、ワークフロー全体の状態を失わずに済む堅牢な実行モデルが実現します。

使用する場面

• 目標の達成に必要な計算量が、単一のエージェント会話で実行できる範囲を超える場合
• 複数のキャンペーンまたはアカウントの分析を無人で実行する必要がある場合
• ワークフローに、必要なツールやデータが異なる自然なフェーズがある場合
• 重要な判断を進める前に、人間の承認が必要な場合
• 長時間実行ワークフローを中断後も維持し、確実に再開する必要がある場合
• 各段階でエージェントが実行、判断、生成した内容を示す監査証跡が必要な場合

仕組み

graph TD
    A[Objective] --> B[Planner Agent]
    B --> C[Phase 1: Data Collection]
    B --> D[Phase 2: Analysis]
    B --> E[Phase 3: Recommendations]
    B --> F[Phase 4: Report Generation]
    
    C --> G[Checkpoint 1]
    G --> D
    D --> H[Checkpoint 2]
    H --> I{Human Gate}
    I -->|Approved| E
    I -->|Rejected| J[Revise Analysis]
    J --> D
    E --> K[Checkpoint 3]
    K --> F
    F --> L[Final Output]
    
    M[Progress Tracker] --> C
    M --> D
    M --> E
    M --> F
    
    N[Error Recovery] --> C
    N --> D
    N --> E
    N --> F

プランナーエージェントは高レベルの目標を受け取り、順序付けられたフェーズに分解します。各フェーズでは、ワークフローの異なる部分（データ収集、分析、推奨事項の生成、レポートの作成）を実行します。フェーズ間のチェックポイントに中間状態を保持するため、中断後もワークフローを再開できます。人間参加型の承認ゲートは、重要度の高い判断ポイントで実行を一時停止し、明示的な承認が得られるまで先に進みません。進捗トラッカーにより、ワークフローの状況をリアルタイムで確認できます。各フェーズのエラー復旧では、失敗したタスクの再試行、重要でないタスクのスキップ、人間のオペレーターへのエスカレーションが可能です。

実装

ワークフロー定義：

interface WorkflowPhase {
  id: string;
  name: string;
  tasks: WorkflowTask[];
  checkpoint: boolean;
  humanGate: boolean;
  timeout_ms: number;
  onError: "retry" | "skip" | "abort" | "escalate";
}

interface WorkflowTask {
  id: string;
  name: string;
  execute: (context: WorkflowContext) => Promise<TaskResult>;
  retries: number;
  critical: boolean;
}

interface WorkflowContext {
  objective: string;
  checkpointData: Record<string, unknown>;
  progress: ProgressTracker;
  humanApprovals: Map<string, boolean>;
}

const accountAuditWorkflow: WorkflowPhase[] = [
  {
    id: "data_collection",
    name: "Collect Account Data",
    tasks: [
      { id: "fetch_campaigns", name: "Fetch all campaigns", execute: fetchCampaigns, retries: 3, critical: true },
      { id: "fetch_keywords", name: "Fetch keyword performance", execute: fetchKeywords, retries: 3, critical: true },
      { id: "fetch_ads", name: "Fetch ad creative data", execute: fetchAds, retries: 2, critical: false },
    ],
    checkpoint: true,
    humanGate: false,
    timeout_ms: 30 * 60 * 1000,
    onError: "retry",
  },
  {
    id: "analysis",
    name: "Analyze Performance",
    tasks: [
      { id: "campaign_analysis", name: "Campaign-level analysis", execute: analyzeCampaigns, retries: 2, critical: true },
      { id: "keyword_analysis", name: "Keyword opportunity analysis", execute: analyzeKeywords, retries: 2, critical: true },
      { id: "competitor_analysis", name: "Competitive positioning", execute: analyzeCompetitors, retries: 1, critical: false },
    ],
    checkpoint: true,
    humanGate: true,
    timeout_ms: 60 * 60 * 1000,
    onError: "escalate",
  },
];

ワークフローエンジン：

class WorkflowEngine:
    def __init__(self, checkpoint_store, notifier):
        self.checkpoint_store = checkpoint_store
        self.notifier = notifier

    async def execute(self, workflow: list, context: dict) -> dict:
        start_phase = await self.find_resume_point(workflow, context["workflow_id"])

        for phase in workflow[start_phase:]:
            context["progress"].update_phase(phase["id"], "running")
            self.notifier.notify(f"Starting phase: {phase['name']}")

            for task in phase["tasks"]:
                result = await self.execute_task(task, context)
                if not result["success"] and task["critical"]:
                    if phase["onError"] == "abort":
                        return {"status": "aborted", "phase": phase["id"], "task": task["id"]}
                    elif phase["onError"] == "escalate":
                        await self.notifier.escalate(f"Critical task failed: {task['name']}")
                        return {"status": "escalated", "phase": phase["id"]}

            if phase.get("checkpoint"):
                await self.save_checkpoint(context["workflow_id"], phase["id"], context)

            if phase.get("humanGate"):
                approved = await self.wait_for_approval(context["workflow_id"], phase["id"])
                if not approved:
                    return {"status": "rejected", "phase": phase["id"]}

            context["progress"].update_phase(phase["id"], "completed")

        return {"status": "completed", "context": context}

    async def execute_task(self, task: dict, context: dict) -> dict:
        for attempt in range(task.get("retries", 1) + 1):
            try:
                result = await task["execute"](context)
                context["checkpointData"][task["id"]] = result
                return {"success": True, "result": result}
            except Exception as e:
                if attempt < task.get("retries", 1):
                    await asyncio.sleep(2 ** attempt)
                    continue
                return {"success": False, "error": str(e)}

    async def find_resume_point(self, workflow, workflow_id):
        checkpoint = await self.checkpoint_store.get(workflow_id)
        if not checkpoint:
            return 0
        for i, phase in enumerate(workflow):
            if phase["id"] == checkpoint["last_completed_phase"]:
                return i + 1
        return 0

    async def save_checkpoint(self, workflow_id, phase_id, context):
        await self.checkpoint_store.put(workflow_id, {
            "last_completed_phase": phase_id,
            "data": context["checkpointData"],
            "timestamp": time.time(),
        })

進捗トラッカー：

class ProgressTracker:
    def __init__(self, workflow_id: str, total_phases: int):
        self.workflow_id = workflow_id
        self.total_phases = total_phases
        self.phases = {}
        self.start_time = time.time()

    def update_phase(self, phase_id: str, status: str):
        self.phases[phase_id] = {
            "status": status,
            "timestamp": time.time(),
        }

    def summary(self) -> dict:
        completed = sum(1 for p in self.phases.values() if p["status"] == "completed")
        elapsed = time.time() - self.start_time
        return {
            "workflow_id": self.workflow_id,
            "progress_pct": (completed / self.total_phases) * 100,
            "completed_phases": completed,
            "total_phases": self.total_phases,
            "elapsed_seconds": elapsed,
            "estimated_remaining": (elapsed / max(completed, 1)) * (self.total_phases - completed),
            "phase_details": self.phases,
        }

ベストプラクティス

1. 再開可能なフェーズに分割する — 各フェーズで自己完結型のチェックポイントを作成し、前のフェーズを再実行せずに再開できるようにします。
2. タスクにクリティカルまたは非クリティカルの印を付ける — 非クリティカルなタスクの失敗（競合分析など）によって、ワークフロー全体が中断されないようにします。
3. フェーズごとにタイムアウトを設定する — 想定時間を超過したフェーズは停止している可能性が高く、タイムアウトによって無期限の待機ではなくエスカレーションが行われます。
4. 不可逆な操作の前に人間による承認ゲートを設ける — 無人ワークフローでは、予算の変更、広告の一時停止、入札単価の変更には、必ず人間による承認を求めます。
5. 進捗を可視化する — 長時間実行されるワークフローでは、進捗状況を伝える必要があります。3時間にわたって何も通知しないプロセスは、ユーザーの不安やサポートへの問い合わせにつながります。
6. すべてのフェーズ遷移をログに記録する — エージェントが実行した操作、実行時刻、使用したデータを網羅した監査証跡は、コンプライアンスとデバッグに不可欠です。
7. 再開パスを明示的にテストする — 各チェックポイントでワークフローを強制終了し、正しく再開されることを確認します。再開に関するバグは発見しにくく、テストされていないこともよくあります。

プラットフォームの互換性

機能	Claude Code	Cursor	Codex	Gemini CLI
複数フェーズのワークフロー	✅ サブエージェント	✅ バックグラウンドタスク	✅ 非同期	✅ 非同期
チェックポイント	✅ ファイルベース	✅ ファイルベース	✅ ファイルベース	✅ ファイルベース
人間による承認ゲート	✅ 権限の確認プロンプト	✅ UI プロンプト	✅ CLI プロンプト	✅ CLI プロンプト
進捗の追跡	✅ ステータス更新	✅ ステータスバー	✅ 標準出力	✅ 標準出力
タイムアウト管理	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応

関連スキル

• 並列エージェントのオーケストレーション - 長期ワークフロー内のフェーズでは、並列サブエージェントを起動して処理を高速化できます
• メモリの永続化 - チェックポイントデータの永続化により、中断後にワークフローを再開できます
• MCP サーバーの作成 - MCP ツールを使用すると、ワークフローの各フェーズで外部 API に標準化された方法でアクセスできます
• Google Ads の監査 - 20 のカテゴリにわたるアカウント全体の監査は、長期ワークフローの代表的な例です

キーワード

long-horizon-workflows, multi-phase-pipelines, checkpointing, human-in-the-loop, progress-tracking, task-decomposition, error-recovery, deerflow, autonomous-execution, agent-skills

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License