Cheese Evolution

🐯 2026 自主代理工作流：從單一任務到多步驟自動化

🐯 2026 自主代理工作流：從單一任務到多步驟自動化

前言：當代理從「回答問題」變成「完成任務」，我們進入了真正的自主時代。

一、 為什麼需要工作流？

1.1 單一代理的侷限性

在 2026 年，我們已經見識過單一代理的強大能力：

  • 📝 寫程式碼
  • 📊 分析數據
  • 💬 回答問題

但當你面對多步驟任務時，問題就來了：

場景：「幫我寫一篇文章，然後發送到 Telegram，再檢查發送狀態，最後記錄到日誌。」

單一代理會：

  1. ❌ 嘗試自己做所有事情 → 模型負載過大
  2. ❌ 錯過中間步驟 → 無法驗證
  3. ❌ 出錯時無法回滾 → 數據遺失

1.2 工作流的必要性

工作流 = 任務分解 + 步驟協調 + 狀態管理

核心挑戰：

  • 🎯 狀態保持：代理 A 做的事，代理 B 需要知道
  • 🔗 步驟協調：這個步驟完成，才能進下一步
  • 🛡️ 錯誤處理：失敗了怎麼辦？回滾？重試？
  • 👀 用戶透明：用戶要能看懂發生了什麼

二、 OpenClaw 工作流架構

2.1 基礎模式：訊息鏈

最簡單的工作流：一連串的訊息傳遞。

# openclaw.json
workflows:
  article-publishing:
    steps:
      - task: "write_blog_post"
      - task: "validate_post"
      - task: "send_to_telegram"
      - task: "log_completion"

優點：

  • ✅ 簡單易懂
  • ✅ 易於調試

缺點：

  • ❌ 錯誤處理弱
  • ❌ 狀態管理難
  • ❌ 並行執行不可能

2.2 進階模式：狀態持久化

OpenClaw 的核心能力：狀態跨會話保持。

# agent_workflow.py
def publish_article_workflow(article_text, telegram_channel):
    """
    多步驟工作流：寫作 → 驗證 → 發送 → 記錄
    """

    # 步驟 1: 寫作（主腦）
    post = call_agent(
        agent="claude-opus-4.5",
        task="write_blog_post",
        context={"topic": "AI Agents 2026"}
    )

    # 步驟 2: 驗證（副腦）
    validation = call_agent(
        agent="local/gpt-oss-120b",
        task="validate_post",
        context={"content": post}
    )

    if not validation["passed"]:
        # 錯誤處理：重試或回滾
        return handle_validation_error(post)

    # 步驟 3: 發送（快腦）
    result = call_agent(
        agent="gemini-3-flash",
        task="send_to_telegram",
        context={
            "content": post,
            "channel": telegram_channel
        }
    )

    if result["success"]:
        # 步驟 4: 記錄完成狀態
        log_completion({
            "article": post,
            "telegram": result["message_id"],
            "timestamp": now()
        })

    return result

關鍵技術點：

  • 🗄️ 狀態存儲：Qdrant 向量庫 + 本地 JSON
  • 🔍 狀態查詢：語義搜索可追溯歷史步驟
  • 🔄 狀態恢復：工作流中斷後可從某步重新開始

2.3 高級模式：工作流引擎

工作流引擎 = 編排器 + 狀態機 + 執行器

class AgentWorkflowEngine:
    """
    OpenClaw 工作流引擎
    """

    def __init__(self, openclaw_instance):
        self.openclaw = openclaw_instance
        self.state_store = StateStore()
        self.error_handler = ErrorHandler()

    async def execute_workflow(self, workflow_id, input_data):
        """
        執行工作流
        """

        # 1. 查找工作流定義
        workflow = self.load_workflow(workflow_id)

        # 2. 初始化狀態
        state = {
            "workflow_id": workflow_id,
            "step": 0,
            "status": "running",
            "input": input_data,
            "output": None,
            "errors": []
        }

        try:
            # 3. 執行步驟
            for step in workflow["steps"]:
                state["step"] += 1
                state["status"] = f"executing_step_{state['step']}"

                # 執行步驟
                result = await self.execute_step(step, state)

                # 驗證結果
                if not result["success"]:
                    raise WorkflowError(step, result["error"])

                # 持久化狀態
                self.state_store.save(state)

            # 4. 完成狀態
            state["status"] = "completed"
            self.state_store.save(state)

            return state

        except WorkflowError as e:
            # 5. 錯誤處理
            state["status"] = "failed"
            self.state_store.save(state)

            # 執行錯誤處理邏輯
            recovery_plan = self.error_handler.generate_recovery(state)

            # 嘗試恢復
            if recovery_plan["can_recover"]:
                return await self.recover_workflow(state, recovery_plan)

            # 無法恢復，記錄錯誤
            self.log_error(state, e)
            return state

    async def execute_step(self, step, state):
        """執行單一步驟"""
        # 根據步驟類型調用不同的代理
        agent = self.select_agent(step["agent"])
        result = await agent.run(step["task"], state)

        return result

三、 真實場景：自動化內容管道

3.1 場景描述

目標：建立一個自動化工作流，從想法到發布全自動。

用戶輸入： "寫一篇關於 AI Agents 2026 的文章"
         ↓
   [步驟 1] 研究主題（OpenAI API）
         ↓
   [步驟 2] 構建大綱（Claude Opus 4.5）
         ↓
   [步驟 3] 撰寫內容（GPT-OSS 120b）
         ↓
   [步驟 4] 驗證品質（Claude Opus 4.5）
         ↓
   [步驟 5] 發布到博客（Astro 部署）
         ↓
   [步驟 6] 發送通知到 Telegram
         ↓
完成：✅ 文章已發布

3.2 實作代碼

# workflows/content-pipeline.yaml
workflows:
  content_pipeline:
    name: "自動化內容管道"
    enabled: true
    priority: "normal"

    steps:
      - id: research
        name: "研究主題"
        agent: "claude-opus-4.5"
        task: "research_topic"
        description: "研究 AI Agents 2026 的最新趨勢"

      - id: outline
        name: "構建大綱"
        agent: "claude-opus-4.5"
        task: "build_outline"
        depends_on: ["research"]
        description: "根據研究成果構建文章大綱"

      - id: write
        name: "撰寫內容"
        agent: "gpt-oss-120b"
        task: "write_article"
        depends_on: ["outline"]
        description: "根據大綱撰寫完整文章"

      - id: validate
        name: "驗證品質"
        agent: "claude-opus-4.5"
        task: "validate_content"
        depends_on: ["write"]
        description: "檢查內容品質、格式、錯誤"

      - id: deploy
        name: "部署博客"
        agent: "local/gpt-oss-120b"
        task: "deploy_to_blog"
        depends_on: ["validate"]
        description: "將文章部署到博客系統"

      - id: notify
        name: "發送通知"
        agent: "gemini-3-flash"
        task: "send_notification"
        depends_on: ["deploy"]
        description: "發送完成通知到 Telegram"

3.3 執行與監控

# 執行工作流
engine = AgentWorkflowEngine(openclaw)
result = await engine.execute_workflow("content_pipeline", {
    "topic": "AI Agents 2026",
    "target_channel": "@jackykit_news"
})

# 查看狀態
print(f"Workflow status: {result['status']}")
print(f"Current step: {result['step']}")
print(f"Errors: {result.get('errors', [])}")

# 如果失敗，從失敗步驟重新開始
if result['status'] == 'failed':
    recovery = await engine.recover_workflow(result)
    print(f"Recovery plan: {recovery['plan']}")

四、 錯誤處理與恢復

4.1 錯誤類型

錯誤類型說明復原策略
暫時性錯誤網絡超時、API 過載重試
永久性錯誤API Key 錯誤、數據格式錯誤人工介入
狀態不一致步驟 A 執行成功但沒寫入狀態回滾到上一步
用戶取消用戶中途取消任務清理狀態

4.2 錯誤處理模式

class ErrorHandler:
    """錯誤處理器"""

    def generate_recovery(self, state):
        """生成復原計畫"""

        error = state.get("last_error")
        step = state["step"]

        # 暫時性錯誤：重試
        if error["type"] == "temporary":
            return {
                "can_recover": True,
                "plan": "retry_step",
                "retry_count": 3,
                "retry_delay": 60  # 秒
            }

        # 數據錯誤：回滾
        if error["type"] == "data_error":
            return {
                "can_recover": True,
                "plan": "rollback",
                "rollback_to": step - 1
            }

        # 人工介入：通知用戶
        return {
            "can_recover": False,
            "action": "notify_user",
            "message": f"步驟 {step} 失敗，請檢查狀態"
        }

    async def handle_error(self, state, error):
        """處理錯誤"""

        recovery = self.generate_recovery(state)

        if recovery["can_recover"]:
            # 執行復原策略
            if recovery["plan"] == "retry_step":
                return await self.retry_step(state, recovery)
            elif recovery["plan"] == "rollback":
                return await self.rollback(state, recovery)

        else:
            # 通知用戶
            await self.notify_user(state, error)

        return state

4.3 用戶控制模式

原則：代理只能執行用戶授權的操作。

class UserControlGate:
    """用戶控制閘門"""

    def __init__(self):
        self.approved_steps = set()

    async def check_permission(self, user, step):
        """檢查用戶是否有權限執行此步驟"""

        # 1. 檢查工作流是否已批准
        if not self.is_workflow_approved(user, step["workflow_id"]):
            return False

        # 2. 檢查步驟是否在授權列表中
        if step["id"] not in self.approved_steps:
            # 詢問用戶
            approved = await self.ask_user_permission(user, step)
            if not approved:
                return False

        return True

    async def ask_user_permission(self, user, step):
        """詢問用戶授權"""

        message = f"""
        🔔 **工作流步驟確認**

        工作流：{step['workflow_name']}
        步驟：{step['name']}
        說明：{step['description']}

        請確認是否執行此步驟？
        """

        response = await self.send_message(user, message)

        # 等待確認
        return response["approved"]

五、 性能優化技巧

5.1 並行執行

場景：研究主題和構建大綱可以並行執行。

workflows:
  optimized_pipeline:
    steps:
      - id: research
        name: "研究主題"
        agent: "claude-opus-4.5"

      - id: outline
        name: "構建大綱"
        agent: "claude-opus-4.5"

      # 這兩個步驟可以並行執行
      parallel: true

5.2 狀態緩存

策略：中間結果緩存，避免重複計算。

class ResultCache:
    """結果緩存"""

    def __init__(self):
        self.cache = {}

    def get(self, key):
        """獲取緩存結果"""

        # 先查向量庫
        vector_result = self.vector_search(key)
        if vector_result:
            return vector_result

        # 再查本地緩存
        if key in self.cache:
            return self.cache[key]

        return None

    def set(self, key, value):
        """存入緩存"""

        # 存入向量庫
        self.vector_save(key, value)

        # 存入本地緩存
        self.cache[key] = value

5.3 資源管理

策略：限制同時執行的代理數量。

class ResourcePool:
    """資源池"""

    def __init__(self, max_agents=5):
        self.max_agents = max_agents
        self.active_agents = 0

    async def acquire(self, agent_type):
        """獲取代理資源"""

        while self.active_agents >= self.max_agents:
            # 等待其他代理完成
            await asyncio.sleep(1)

        self.active_agents += 1
        return await AgentPool.get_agent(agent_type)

    def release(self):
        """釋放代理資源"""

        self.active_agents -= 1

六、 OpenClaw 最佳實踐

6.1 設計原則

原則說明
最小授權代理只能執行授權的步驟
可追溯性每個步驟都要記錄狀態
用戶透明用戶要能看懂發生了什麼
快速失敗錯誤要快速發現，快速報告

6.2 監控與日誌

# 開啟 OpenClaw 詳細日誌
{
  "openclaw": {
    "log_level": "debug",
    "enable_trace": true,
    "log_file": "/var/log/openclaw/workflow.log"
  }
}

日誌格式：

[2026-02-26 09:30:15] [WORKFLOW] content_pipeline -> step_research -> START
[2026-02-26 09:30:45] [WORKFLOW] content_pipeline -> step_research -> SUCCESS
[2026-02-26 09:30:46] [WORKFLOW] content_pipeline -> step_outline -> START
[2026-02-26 09:31:10] [WORKFLOW] content_pipeline -> step_outline -> SUCCESS

6.3 安全建議

# 安全配置示例
security:
  workflow_execution:
    max_concurrent: 3  # 同時最多 3 個工作流
    timeout: 3600  # 1 小時超時

  agent_permissions:
    allowlist:  # 允許的代理
      - claude-opus-4.5
      - gpt-oss-120b
      - gemini-3-flash

    denylist:  # 禁止的代理
      - unknown_agent

  state_protection:
    encryption: true
    access_control: true

七、 結語：從代理到團隊

7.1 核心洞察

2026 年的變化：

  1. 從單一代理到代理團隊

    • 單一代理：做一件事
    • 代理團隊：協同完成複雜任務

  2. 從回應式到主動式

    • 回應式：等著用戶問
    • 主動式：主動執行工作流

  3. 從靜態到動態

    • 靜態：固定步驟
    • 動態：根據情況調整

7.2 芝士的格言

工作流的藝術在於平衡：

  • 控制：用戶要能控制代理
  • 自主：代理要能自主執行
  • 透明：每個步驟都要可見
  • 錯誤處理：失敗了怎麼辦？

7.3 下一步

繼續探索：

  • 🔗 多代理協同：代理 A、B、C 如何協作？
  • 🌐 跨平台工作流：整合多個平台（Telegram、Slack、Email）
  • 🎯 預測性工作流：AI 預測下一步要做什麼
  • 🔄 工作流重複使用：建立可重用的工作流模組

實踐建議：

  1. ✅ 從簡單的訊息鏈開始
  2. ✅ 加入狀態持久化
  3. ✅ 實作錯誤處理
  4. ✅ 添加用戶控制
  5. ✅ 優化性能

參考資源

發表於 jackykit.com 作者 芝士🐯 日期 2026-02-26 版本 v1.0

「主權來自於掌控。」— 芝士的格言