long-horizon-workflows

长时程工作流

由 googleadsagent.ai™ 推出的 Agent Skills™ 项目的一部分

描述

长时程工作流使 AI Agent 能够执行多小时、多阶段的自主流程，远超单次对话轮次的范围。这类工作流借鉴了 DeerFlow 架构以及 googleadsagent.ai™ 的生产实践模式，将复杂目标分解为分阶段的执行计划，并包含检查点、进度跟踪、错误恢复，以及在关键决策点设置的人机协作关卡。长时程工作流可以自主运行数小时，分析整个 Google Ads 账户（数十个广告系列、数千个关键词）、生成全面的优化报告，并准备可直接实施的变更集。

长时程执行的核心挑战在于可靠性。一个需要 3 小时的工作流，如果在第 2.5 小时失败且无法恢复，非但毫无用处，还会浪费时间和算力。检查点管理可确保在失败前完成的工作得到保留，并可在恢复后继续执行。进度跟踪让你了解 Agent 正在做什么以及进展如何。人机协作关卡允许人工在 Agent 继续执行前审核关键决策（例如预算变更），避免无人值守运行时发生灾难性错误。

DeerFlow 对这一模式的贡献是层次化任务分解的概念——规划 Agent 将目标拆分为多个阶段，每个阶段再拆分为任务，每个任务进一步拆分为原子操作。层次结构的每一层都有自己的检查点、超时时间和错误处理策略。由此形成的稳健执行模型可以应对单项任务失败，而不会丢失更广泛的工作流状态。

适用场景

• 目标所需的计算量超出单次 Agent 对话的承载能力
• 需要在无人值守的情况下运行多广告系列或多账户分析
• 工作流具有多个自然阶段，且各阶段所需的工具和数据不同
• 关键决策需要在继续执行前获得人工批准
• 长时间运行的工作流必须能够经受中断并顺利恢复
• 需要审计记录来展示 Agent 在各阶段执行、决策和产出的内容

工作原理

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

规划 Agent 接收一个高层目标，并将其分解为有序阶段。每个阶段负责执行工作流中不同的部分（数据收集、分析、生成建议、撰写报告）。阶段之间的检查点会持久化中间状态，使工作流在中断后能够恢复。人机协作关卡会在高风险决策点暂停执行，等待明确批准后再继续。进度跟踪器可实时显示工作流状态。每个阶段的错误恢复机制可以重试失败的任务、跳过非关键任务，或将问题升级给人工操作员处理。

实现

工作流定义：

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. 记录每次阶段转换 — 完整记录 Agent 执行的操作、执行时间和所用数据，对于合规审查和调试至关重要。
7. 明确测试恢复路径 — 在每个检查点中止工作流，并验证其能否正确恢复；恢复错误隐蔽且常常未经测试。

平台兼容性

功能	Claude Code	Cursor	Codex	Gemini CLI
多阶段工作流	✅ 子 Agent	✅ 后台任务	✅ 异步	✅ 异步
检查点	✅ 基于文件	✅ 基于文件	✅ 基于文件	✅ 基于文件
人工审核关卡	✅ 权限提示	✅ UI 提示	✅ CLI 提示	✅ CLI 提示
进度跟踪	✅ 状态更新	✅ 状态栏	✅ 标准输出	✅ 标准输出
超时管理	✅ 完整支持	✅ 完整支持	✅ 完整支持	✅ 完整支持

相关技能

• 并行 Agent 编排 - 长周期工作流中的阶段可以分派并行子 Agent，以加快执行速度
• 记忆持久化 - 持久化检查点数据，可在工作流中断后恢复执行
• 创建 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