Skill 的組合與編排：讓 Skill 呼叫 Skill

上一篇我們用「程式碼審查」為例，設計了一個完整的 Skill。

但真實世界的任務，很少能靠單一 Skill 完成。

寫一份技術報告，需要研究、分析、撰寫、審查四個 Skill 協同工作。
開發一個功能，需要需求分析、程式生成、測試、部署多個 Skill 接力。

這一篇，我們要談 Skill 的下一個層次：組合與編排（Composition & Orchestration）。
我們會解釋為什麼需要組合、常見的組合模式、依賴管理、錯誤傳播，並用 Python 實作一個 Skill 編排系統。

一、為什麼需要 Skill 組合？

單一 Skill 的天花板

一個 Skill 再怎麼設計，也只能完成一類任務。
但真實世界的工作，往往是多個能力的串聯。

舉個例子：

幫我寫一份「2026 年 AI Agent 市場趨勢」的技術報告。

這個任務需要：

• 研究 Skill：蒐集市場資訊、論文、產品動態
• 分析 Skill：整理趨勢、找出模式、評估影響
• 撰寫 Skill：根據分析結果撰寫報告
• 審查 Skill：檢查事實正確性、邏輯一致性、表達清晰度
• 格式化 Skill：排版、加入圖表、產出最終文件

如果你試圖把這些全部塞進一個 Skill，會發生什麼事？

• Prompt 爆炸：一個 Skill 要包含研究、分析、寫作、審查的指令，幾千字跑不掉。
• 職責混亂：同一個 Skill 既要發散思考又要嚴格審查，兩種能力互相衝突。
• 難以測試：一個巨大的 Skill，出錯時很難定位是哪個環節的問題。
• 無法重用：這個 Skill 只能做「寫報告」，不能做其他需要研究的任務。
• 難以迭代：改進「審查」環節時，可能意外影響「研究」環節。

這就是單一 Skill 的天花板。

組合的價值

把任務拆成多個 Skill，每個 Skill 專注一件事，然後把它們組合起來：

好處	說明
職責單一	每個 Skill 只做一件事，Prompt 精簡
易於測試	每個 Skill 可以獨立測試
可重用	「研究」Skill 可以用在報告、簡報、決策等各種場景
易於迭代	改進某個 Skill 不會影響其他 Skill
易於除錯	出錯時能快速定位是哪個 Skill 的問題
可組合	不同的 Skill 組合可以完成不同的任務

打個比方：

• 單一 Skill 像一把瑞士刀，什麼都能做，但什麼都不精。
• 多個 Skill 組合 像一個工具箱，每個工具專精一件事，組合起來能完成複雜工程。

二、常見的 Skill 組合模式

Skill 的組合有多種模式，以下是最常見的幾種：

模式一：順序式（Sequential / Pipeline）

Skill 按照固定順序執行，前一個的輸出是後一個的輸入。

研究 Skill → 分析 Skill → 撰寫 Skill → 審查 Skill

優點：簡單、可控、易除錯。
缺點：缺乏彈性，無法根據中間結果動態調整。

適用場景：流程明確的任務，如報告生成、程式碼審查。

def sequential_pipeline(topic: str) -> str:
    research = research_skill.execute(topic)
    analysis = analysis_skill.execute(research)
    draft = writing_skill.execute(analysis)
    final = review_skill.execute(draft)
    return final

模式二：協調者式（Orchestrator）

有一個「協調者」負責決定何時呼叫哪個 Skill，並整合結果。

                ┌→ 研究 Skill ─┐
協調者 ──────────┼→ 分析 Skill ─┼→ 協調者 → 最終輸出
                └→ 撰寫 Skill ─┘

優點：靈活、可動態決策、易於擴展。
缺點：協調者可能成為瓶頸，需要良好的任務分解能力。

適用場景：複雜任務、需要動態決策的場景。

def orchestrator(task: str) -> str:
    plan = planner_skill.execute(task)
    results = {}
    for step in plan["steps"]:
        skill = get_skill(step["skill"])
        results[step["id"]] = skill.execute(step["input"], context=results)
    return synthesizer_skill.execute(task, results)

模式三：平行式（Parallel）

多個 Skill 同時執行，最後合併結果。

                ┌→ 研究 Skill A ─┐
協調者 ──────────┼→ 研究 Skill B ─┼→ 合併 → 最終輸出
                └→ 研究 Skill C ─┘

優點：速度快、能同時處理多個角度。
缺點：需要合併邏輯，可能有衝突。

適用場景：需要多來源、多角度的任務。

import asyncio

async def parallel_execution(topic: str) -> str:
    tasks = [
        research_skill_a.execute_async(topic),
        research_skill_b.execute_async(topic),
        research_skill_c.execute_async(topic),
    ]
    results = await asyncio.gather(*tasks)
    return merge_skill.execute(results)

模式四：條件式（Conditional）

根據中間結果，決定下一步要呼叫哪個 Skill。

分析 Skill
    ↓
    ├── 如果發現問題 → 修正 Skill
    └── 如果沒問題 → 輸出 Skill

優點：能根據實際情況動態調整。
缺點：邏輯複雜，需要明確定義條件。

適用場景：需要根據結果分支的任務。

def conditional_execution(code: str) -> str:
    review = code_review_skill.execute(code)
    if review["summary"]["score"] < 8:
        fixed = code_fix_skill.execute(code, review["issues"])
        return code_review_skill.execute(fixed)
    return review

模式五：迭代式（Iterative）

重複執行某個 Skill 組合，直到達到終止條件。

草稿 → 審查 → 修正 → 審查 → 修正 → ... → 最終版本

優點：能持續改進品質。
缺點：可能花費較多資源，需要終止條件。

適用場景：需要高品質輸出的任務。

def iterative_refinement(topic: str, max_iterations: int = 3) -> str:
    draft = writing_skill.execute(topic)
    for i in range(max_iterations):
        review = review_skill.execute(draft)
        if review["summary"]["score"] >= 9:
            break
        draft = revision_skill.execute(draft, review["issues"])
    return draft

模式六：階層式（Hierarchical）

多層編排結構，上層負責策略，下層負責執行。

                總協調者
             /     |     \
         研究組  分析組  撰寫組
         /  \    /  \    /  \
        A    B  C    D  E    F

優點：可處理超大規模任務、職責分明。
缺點：架構複雜、通訊開銷大。

適用場景：大型專案、企業級自動化。

三、Skill 的依賴管理

當 Skill 開始互相呼叫，就會產生依賴關係。
良好的依賴管理，是 Skill 組合的基礎。

依賴的類型

1. 資料依賴

Skill B 需要 Skill A 的輸出才能執行。

# Skill B 依賴 Skill A 的輸出
research_result = research_skill.execute(topic)
analysis_result = analysis_skill.execute(research_result)  # 需要 research_result

2. 順序依賴

Skill B 必須在 Skill A 之後執行，但不一定需要它的輸出。

# 必須先驗證再執行
auth_skill.execute(user)
action_skill.execute(action)  # 必須在 auth 之後

3. 資源依賴

多個 Skill 共享同一個資源（如資料庫連線、API 額度）。

# 共享資料庫連線
db = DatabaseConnection()
skill_a = SkillA(db=db)
skill_b = SkillB(db=db)

依賴圖（Dependency Graph）

用有向圖表示 Skill 之間的依賴關係：

research ──→ analysis ──→ writing ──→ review
    │                        ↑
    └────────────────────────┘

在程式碼中，可以用 DAG（有向無環圖）來表示：

from dataclasses import dataclass, field

@dataclass
class SkillNode:
    name: str
    skill: object
    depends_on: list = field(default_factory=list)
    input_mapping: dict = field(default_factory=dict)

# 定義依賴圖
skills = {
    "research": SkillNode(
        name="research",
        skill=research_skill,
        depends_on=[],
    ),
    "analysis": SkillNode(
        name="analysis",
        skill=analysis_skill,
        depends_on=["research"],
        input_mapping={"research_result": "research"},  # 從 research 的輸出取資料
    ),
    "writing": SkillNode(
        name="writing",
        skill=writing_skill,
        depends_on=["research", "analysis"],
        input_mapping={
            "research": "research",
            "analysis": "analysis",
        },
    ),
    "review": SkillNode(
        name="review",
        skill=review_skill,
        depends_on=["writing"],
        input_mapping={"draft": "writing"},
    ),
}

拓撲排序

要執行依賴圖，需要先做拓撲排序，確保每個 Skill 在它的依賴之後執行。

def topological_sort(skills: dict) -> list:
    """拓撲排序，回傳執行順序"""
    visited = set()
    order = []

    def visit(name):
        if name in visited:
            return
        visited.add(name)
        for dep in skills[name].depends_on:
            visit(dep)
        order.append(name)

    for name in skills:
        visit(name)

    return order

# 執行順序：research → analysis → writing → review
execution_order = topological_sort(skills)

循環依賴檢測

如果 Skill A 依賴 B，B 依賴 C，C 依賴 A，就會形成循環，無法執行。
必須在執行前檢測。

def detect_cycle(skills: dict) -> bool:
    """檢測是否有循環依賴"""
    visited = set()
    rec_stack = set()

    def has_cycle(name):
        visited.add(name)
        rec_stack.add(name)

        for dep in skills[name].depends_on:
            if dep not in visited:
                if has_cycle(dep):
                    return True
            elif dep in rec_stack:
                return True

        rec_stack.remove(name)
        return False

    for name in skills:
        if name not in visited:
            if has_cycle(name):
                return True
    return False

四、錯誤傳播與處理

當多個 Skill 串聯執行時，一個 Skill 的錯誤會傳播到後續的 Skill。

錯誤傳播的模式

1. 直接傳播

Skill A 失敗 → Skill B 無法執行 → 整個流程失敗。

research (失敗) → analysis (無法執行) → writing (無法執行)

2. 間接傳播

Skill A 回傳錯誤的結果 → Skill B 基於錯誤結果產出錯誤答案 → 錯誤累積。

research (回傳錯誤資料) → analysis (基於錯誤資料分析) → 最終報告錯誤

3. 靜默失敗

Skill A 失敗，但沒有明確回報，後續 Skill 誤以為成功。

research (靜默失敗，回傳空結果) → analysis (基於空結果分析) → 無意義輸出

錯誤處理策略

策略一：快速失敗（Fail Fast）

任何 Skill 失敗，立即終止整個流程。

def fail_fast_pipeline(topic: str) -> dict:
    result = research_skill.execute(topic)
    if not result.get("success"):
        return {"success": False, "error": f"研究失敗：{result.get('error')}"}

    analysis = analysis_skill.execute(result["data"])
    if not analysis.get("success"):
        return {"success": False, "error": f"分析失敗：{analysis.get('error')}"}

    # ...
    return {"success": True, "data": ...}

優點：快速發現問題、不浪費資源。
缺點：一個小錯誤就讓整個流程失敗。

策略二：重試（Retry）

Skill 失敗時，自動重試。

def retry_execute(skill, input_data, max_retries: int = 3, backoff: float = 1.0):
    """帶重試的 Skill 執行"""
    for attempt in range(max_retries):
        result = skill.execute(input_data)
        if result.get("success"):
            return result

        if attempt < max_retries - 1:
            time.sleep(backoff * (2 ** attempt))  # 指數退避

    return {"success": False, "error": f"重試 {max_retries} 次後仍失敗"}

策略三：降級（Fallback）

Skill 失敗時，改用備用方案。

def fallback_execute(primary_skill, fallback_skill, input_data):
    """優先使用 primary，失敗時改用 fallback"""
    result = primary_skill.execute(input_data)
    if result.get("success"):
        return result

    print(f"主要 Skill 失敗，改用備用方案")
    return fallback_skill.execute(input_data)

策略四：容錯（Fault Tolerance）

某個 Skill 失敗時，跳過它，繼續執行其他部分。

def fault_tolerant_pipeline(topic: str) -> dict:
    results = {}

    for name, skill in [("research", research_skill),
                        ("analysis", analysis_skill),
                        ("writing", writing_skill)]:
        try:
            result = skill.execute(topic, context=results)
            results[name] = result
        except Exception as e:
            results[name] = {"error": str(e), "skipped": True}
            print(f"Skill {name} 失敗，跳過")

    return results

策略五：補償（Compensation）

如果某個 Skill 失敗，執行補償動作來回復之前的狀態。

class SagaOrchestrator:
    """Saga 模式的編排器：每個步驟都有對應的補償動作"""

    def __init__(self):
        self.steps = []  # [(action, compensation), ...]
        self.completed = []

    def add_step(self, action, compensation):
        self.steps.append((action, compensation))

    def execute(self):
        try:
            for action, compensation in self.steps:
                action()
                self.completed.append(compensation)
        except Exception as e:
            # 執行已完成的補償動作（反向）
            for compensation in reversed(self.completed):
                try:
                    compensation()
                except Exception as ce:
                    print(f"補償失敗：{ce}")
            raise

錯誤傳播的設計原則

1. 每個 Skill 都應該回傳統一格式：包含 success、data、error。
2. 錯誤要明確：不要靜默失敗。
3. 錯誤要可追溯：記錄是哪個 Skill、哪一步失敗。
4. 錯誤要可恢復：設計重試、降級、補償機制。
5. 關鍵步驟要驗證：在 Skill 之間加入驗證，避免錯誤累積。

五、實作：Skill 編排系統

現在讓我們實作一個完整的 Skill 編排系統。

核心資料結構

from dataclasses import dataclass, field
from typing import Any
from enum import Enum


class StepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class StepResult:
    """單一步驟的執行結果"""
    step_name: str
    status: StepStatus
    data: Any = None
    error: str = None
    duration: float = 0.0
    retries: int = 0


@dataclass
class OrchestrationResult:
    """整個編排的執行結果"""
    success: bool
    final_output: Any = None
    steps: list = field(default_factory=list)
    total_duration: float = 0.0
    error: str = None

Skill 編排器

import time
import json
from collections import defaultdict


class SkillOrchestrator:
    """Skill 編排器：負責按依賴順序執行多個 Skill"""

    def __init__(self):
        self.skills = {}           # name -> skill object
        self.dependencies = {}     # name -> [dep_names]
        self.input_mappings = {}   # name -> {param: source_step}
        self.error_handlers = {}   # name -> handler function
        self.max_retries = {}
        self.timeouts = {}

    def register(
        self,
        name: str,
        skill,
        depends_on: list = None,
        input_mapping: dict = None,
        max_retries: int = 0,
        timeout: float = None,
        error_handler=None,
    ):
        """註冊一個 Skill"""
        self.skills[name] = skill
        self.dependencies[name] = depends_on or []
        self.input_mappings[name] = input_mapping or {}
        self.max_retries[name] = max_retries
        self.timeouts[name] = timeout
        if error_handler:
            self.error_handlers[name] = error_handler

    def _validate(self):
        """驗證依賴圖"""
        # 檢查所有依賴都存在
        for name, deps in self.dependencies.items():
            for dep in deps:
                if dep not in self.skills:
                    raise ValueError(f"Skill {name} 依賴不存在的 {dep}")

        # 檢查是否有循環
        if self._detect_cycle():
            raise ValueError("依賴圖中存在循環依賴")

    def _detect_cycle(self) -> bool:
        """檢測循環依賴"""
        visited = set()
        rec_stack = set()

        def has_cycle(name):
            visited.add(name)
            rec_stack.add(name)

            for dep in self.dependencies.get(name, []):
                if dep not in visited:
                    if has_cycle(dep):
                        return True
                elif dep in rec_stack:
                    return True

            rec_stack.remove(name)
            return False

        for name in self.skills:
            if name not in visited:
                if has_cycle(name):
                    return True
        return False

    def _topological_sort(self) -> list:
        """拓撲排序，回傳執行順序"""
        visited = set()
        order = []

        def visit(name):
            if name in visited:
                return
            visited.add(name)
            for dep in self.dependencies.get(name, []):
                visit(dep)
            order.append(name)

        for name in self.skills:
            visit(name)

        return order

    def _build_input(self, name: str, results: dict, initial_input: dict) -> dict:
        """根據 input_mapping 組裝 Skill 的輸入"""
        mapping = self.input_mappings.get(name, {})

        if not mapping:
            # 沒有 mapping，使用初始輸入
            return initial_input

        skill_input = {}
        for param, source in mapping.items():
            if source in results:
                skill_input[param] = results[source].data
            elif source in initial_input:
                skill_input[param] = initial_input[source]

        return skill_input

    def _execute_with_retry(self, name: str, input_data: dict) -> StepResult:
        """帶重試的 Skill 執行"""
        skill = self.skills[name]
        max_retries = self.max_retries.get(name, 0)

        last_error = None

        for attempt in range(max_retries + 1):
            start = time.time()
            try:
                data = skill.execute(**input_data)
                duration = time.time() - start

                # 檢查是否成功
                if isinstance(data, dict) and not data.get("success", True):
                    last_error = data.get("error", "未知錯誤")
                    if attempt < max_retries:
                        time.sleep(1.0 * (2 ** attempt))
                        continue
                    return StepResult(
                        step_name=name,
                        status=StepStatus.FAILED,
                        error=last_error,
                        duration=duration,
                        retries=attempt,
                    )

                return StepResult(
                    step_name=name,
                    status=StepStatus.SUCCESS,
                    data=data,
                    duration=duration,
                    retries=attempt,
                )

            except Exception as e:
                duration = time.time() - start
                last_error = str(e)
                if attempt < max_retries:
                    time.sleep(1.0 * (2 ** attempt))
                    continue
                return StepResult(
                    step_name=name,
                    status=StepStatus.FAILED,
                    error=last_error,
                    duration=duration,
                    retries=attempt,
                )

        return StepResult(
            step_name=name,
            status=StepStatus.FAILED,
            error=last_error or "未知錯誤",
        )

    def execute(
        self,
        initial_input: dict,
        fail_fast: bool = True,
        verbose: bool = True,
    ) -> OrchestrationResult:
        """執行整個編排"""
        self._validate()

        start_time = time.time()
        order = self._topological_sort()
        results = {}
        step_results = []

        if verbose:
            print(f"[編排] 執行順序：{' → '.join(order)}")

        for name in order:
            if verbose:
                print(f"\n[執行] {name}")

            # 檢查依賴是否都成功
            deps = self.dependencies.get(name, [])
            failed_deps = [
                d for d in deps
                if d in results and results[d].status != StepStatus.SUCCESS
            ]

            if failed_deps:
                step_result = StepResult(
                    step_name=name,
                    status=StepStatus.SKIPPED,
                    error=f"依賴失敗：{failed_deps}",
                )
                results[name] = step_result
                step_results.append(step_result)

                if fail_fast:
                    return OrchestrationResult(
                        success=False,
                        steps=step_results,
                        error=f"Skill {name} 因依賴失敗而跳過",
                        total_duration=time.time() - start_time,
                    )
                continue

            # 組裝輸入
            input_data = self._build_input(name, results, initial_input)

            # 執行
            step_result = self._execute_with_retry(name, input_data)
            results[name] = step_result
            step_results.append(step_result)

            if verbose:
                status_icon = "✓" if step_result.status == StepStatus.SUCCESS else "✗"
                print(f"  {status_icon} {step_result.status.value} "
                      f"({step_result.duration:.2f}s)")

            if step_result.status == StepStatus.FAILED:
                # 呼叫錯誤處理器
                if name in self.error_handlers:
                    handler = self.error_handlers[name]
                    handler(step_result, results)

                if fail_fast:
                    return OrchestrationResult(
                        success=False,
                        steps=step_results,
                        error=f"Skill {name} 失敗：{step_result.error}",
                        total_duration=time.time() - start_time,
                    )

        # 取得最終輸出（最後一個 Skill 的結果）
        final_name = order[-1] if order else None
        final_output = results.get(final_name, StepResult("", StepStatus.PENDING)).data

        return OrchestrationResult(
            success=True,
            final_output=final_output,
            steps=step_results,
            total_duration=time.time() - start_time,
        )

使用範例：報告生成流程

# ========== 定義 Skill 介面 ==========

class BaseSkill:
    """所有 Skill 的基礎類別"""
    name = "base"

    def execute(self, **kwargs) -> dict:
        raise NotImplementedError


class ResearchSkill(BaseSkill):
    name = "research"

    def execute(self, topic: str) -> dict:
        # 模擬研究
        return {
            "success": True,
            "data": {
                "topic": topic,
                "findings": [
                    "多 Agent 協作成為主流",
                    "工具使用標準化",
                    "自主規劃能力提升",
                ],
                "sources": ["source_a", "source_b"],
            },
        }


class AnalysisSkill(BaseSkill):
    name = "analysis"

    def execute(self, research_result: dict) -> dict:
        # 模擬分析
        findings = research_result.get("findings", [])
        return {
            "success": True,
            "data": {
                "trends": findings,
                "insights": [
                    f"趨勢「{f}」值得關注" for f in findings
                ],
                "confidence": 0.85,
            },
        }


class WritingSkill(BaseSkill):
    name = "writing"

    def execute(self, research_result: dict, analysis: dict) -> dict:
        # 模擬撰寫
        topic = research_result.get("topic", "未知主題")
        trends = analysis.get("trends", [])

        report = f"# {topic} 報告\n\n"
        report += "## 主要趨勢\n\n"
        for i, trend in enumerate(trends, 1):
            report += f"{i}. {trend}\n"
        report += "\n## 結論\n\n根據以上分析..."

        return {
            "success": True,
            "data": {
                "report": report,
                "word_count": len(report),
            },
        }


class ReviewSkill(BaseSkill):
    name = "review"

    def execute(self, draft: dict) -> dict:
        # 模擬審查
        report = draft.get("report", "")
        issues = []

        if len(report) < 500:
            issues.append({
                "severity": "medium",
                "title": "報告過短",
                "description": "報告字數不足，建議補充更多細節。",
            })

        score = 10.0 - len(issues) * 1.0

        return {
            "success": True,
            "data": {
                "report": report,
                "score": score,
                "issues": issues,
                "approved": score >= 8.0,
            },
        }


# ========== 建立編排器 ==========

orchestrator = SkillOrchestrator()

# 註冊 Skill 與依賴關係
orchestrator.register(
    "research",
    ResearchSkill(),
    depends_on=[],
)

orchestrator.register(
    "analysis",
    AnalysisSkill(),
    depends_on=["research"],
    input_mapping={"research_result": "research"},
)

orchestrator.register(
    "writing",
    WritingSkill(),
    depends_on=["research", "analysis"],
    input_mapping={
        "research_result": "research",
        "analysis": "analysis",
    },
)

orchestrator.register(
    "review",
    ReviewSkill(),
    depends_on=["writing"],
    input_mapping={"draft": "writing"},
)

# ========== 執行 ==========

result = orchestrator.execute(
    initial_input={"topic": "2026 年 AI Agent 市場趨勢"},
    fail_fast=True,
    verbose=True,
)

print(f"\n{'='*50}")
print(f"執行結果：{'成功' if result.success else '失敗'}")
print(f"總耗時：{result.total_duration:.2f} 秒")
print(f"{'='*50}")

if result.success:
    print(result.final_output["report"])
    print(f"\n審查分數：{result.final_output['score']}")

執行結果

[編排] 執行順序：research → analysis → writing → review

[執行] research
  ✓ success (0.00s)

[執行] analysis
  ✓ success (0.00s)

[執行] writing
  ✓ success (0.00s)

[執行] review
  ✓ success (0.00s)

==================================================
執行結果：成功
總耗時：0.00 秒
==================================================

# 2026 年 AI Agent 市場趨勢 報告

## 主要趨勢

1. 多 Agent 協作成為主流
2. 工具使用標準化
3. 自主規劃能力提升

## 結論

根據以上分析...

審查分數：10.0

六、進階：動態編排

上面的例子是靜態編排：依賴關係在執行前就固定了。
但有些任務需要動態編排：根據中間結果決定下一步。

動態編排的設計

class DynamicOrchestrator:
    """動態編排器：用 LLM 決定下一步要執行哪個 Skill"""

    def __init__(self, skills: dict, llm_client):
        self.skills = skills
        self.llm = llm_client
        self.history = []

    def execute(self, task: str, max_steps: int = 10) -> dict:
        """動態執行任務"""
        context = {"task": task, "results": {}}

        for step in range(max_steps):
            # 1. 讓 LLM 決定下一步
            decision = self._decide_next_step(context)

            if decision["action"] == "finish":
                return {
                    "success": True,
                    "output": decision["output"],
                    "steps": step,
                }

            # 2. 執行 Skill
            skill_name = decision["skill"]
            skill_input = decision["input"]

            if skill_name not in self.skills:
                context["results"]["error"] = f"找不到 Skill {skill_name}"
                continue

            try:
                result = self.skills[skill_name].execute(**skill_input)
                context["results"][skill_name] = result
                self.history.append({
                    "step": step,
                    "skill": skill_name,
                    "input": skill_input,
                    "result": result,
                })
            except Exception as e:
                context["results"]["error"] = str(e)

        return {
            "success": False,
            "error": "達到最大步數",
            "steps": max_steps,
        }

    def _decide_next_step(self, context: dict) -> dict:
        """讓 LLM 決定下一步"""
        skill_descriptions = "\n".join(
            f"- {name}: {skill.description}"
            for name, skill in self.skills.items()
        )

        prompt = f"""你是一個任務規劃器。請根據當前狀態，決定下一步要執行哪個 Skill。

任務：{context['task']}

已完成的結果：
{json.dumps(context['results'], ensure_ascii=False, indent=2)}

可用的 Skill：
{skill_descriptions}

請以 JSON 格式輸出：
{{"action": "execute" 或 "finish", "skill": "skill_name", "input": {{...}}, "output": "如果 action 是 finish，這裡是最終回答"}}

只輸出 JSON，不要其他文字。"""

        result = self.llm([{"role": "user", "content": prompt}])

        try:
            content = result.choices[0].message.content
            json_match = re.search(r"\{.*\}", content, re.DOTALL)
            if json_match:
                return json.loads(json_match.group(0))
        except (json.JSONDecodeError, AttributeError):
            pass

        return {"action": "finish", "output": "無法決定下一步"}

七、編排的最佳實踐

1. 明確定義每個 Skill 的介面

class Skill:
    name: str
    description: str
    input_schema: dict
    output_schema: dict

    def execute(self, **kwargs) -> dict:
        """永遠回傳 {"success": bool, "data": ..., "error": ...}"""
        ...

2. 使用統一的結果格式

# 所有 Skill 都回傳相同結構
{
    "success": True,
    "data": {...},
    "error": None,
    "metadata": {
        "skill_name": "...",
        "duration": 1.23,
        "tokens_used": 1500,
    },
}

3. 加入驗證步驟

在關鍵 Skill 之間加入驗證，避免錯誤累積。

orchestrator.register(
    "validate_research",
    ValidationSkill(schema=RESEARCH_SCHEMA),
    depends_on=["research"],
    input_mapping={"data": "research"},
)

4. 記錄完整的執行日誌

def log_execution(step_result: StepResult):
    logger.info(json.dumps({
        "step": step_result.step_name,
        "status": step_result.status.value,
        "duration": step_result.duration,
        "error": step_result.error,
        "retries": step_result.retries,
    }, ensure_ascii=False))

5. 設定超時與資源限制

orchestrator.register(
    "expensive_skill",
    ExpensiveSkill(),
    timeout=60.0,       # 最多執行 60 秒
    max_retries=2,      # 失敗時重試 2 次
)

6. 支援平行執行

對於沒有依賴關係的 Skill，可以平行執行。

async def parallel_execute(self, skill_names: list, input_data: dict):
    """平行執行多個獨立的 Skill"""
    tasks = [
        asyncio.create_task(self._execute_async(name, input_data))
        for name in skill_names
    ]
    return await asyncio.gather(*tasks)

7. 提供可觀測性

class OrchestratorMetrics:
    def __init__(self):
        self.skill_durations = defaultdict(list)
        self.skill_failures = defaultdict(int)

    def record(self, step_result: StepResult):
        self.skill_durations[step_result.step_name].append(step_result.duration)
        if step_result.status == StepStatus.FAILED:
            self.skill_failures[step_result.step_name] += 1

    def report(self) -> dict:
        return {
            "avg_duration": {
                name: sum(durations) / len(durations)
                for name, durations in self.skill_durations.items()
            },
            "failures": dict(self.skill_failures),
        }

八、常見的編排陷阱

1. 過度依賴單一 Skill

# 不好的例子：所有邏輯都在一個 Skill
class MegaSkill:
    def execute(self, task):
        # 研究 + 分析 + 寫作 + 審查...
        ...

# 好的例子：拆成多個 Skill
class ResearchSkill: ...
class AnalysisSkill: ...
class WritingSkill: ...

2. 忽略錯誤處理

# 錯誤例子：沒有錯誤處理
result = research_skill.execute(topic)
analysis = analysis_skill.execute(result)  # result 可能是 None

# 正確例子：檢查每個步驟
result = research_skill.execute(topic)
if not result.get("success"):
    return {"error": "研究失敗"}
analysis = analysis_skill.execute(result["data"])

3. 循環依賴

# 錯誤例子：A 依賴 B，B 依賴 A
skills["A"].depends_on = ["B"]
skills["B"].depends_on = ["A"]

# 正確例子：明確定義單向依賴
skills["A"].depends_on = []
skills["B"].depends_on = ["A"]

4. 沒有超時控制

# 錯誤例子：Skill 可能永遠執行下去
result = skill.execute(input)

# 正確例子：設定超時
result = execute_with_timeout(skill, input, timeout=30)

5. 沒有記錄中間結果

# 錯誤例子：只記錄最終結果
final = pipeline(topic)
print(final)

# 正確例子：記錄每一步
results = {}
for step in steps:
    results[step.name] = step.execute(...)
    log(step.name, results[step.name])

6. 硬編碼執行順序

# 錯誤例子：硬編碼
research()
analysis()
writing()
review()

# 正確例子：用依賴圖
orchestrator.register("research", ..., depends_on=[])
orchestrator.register("analysis", ..., depends_on=["research"])

九、總結：從零件到組裝線

讓我們回顧這一篇的核心：

• 為什麼需要組合：單一 Skill 有天花板，組合能處理複雜任務。
• 常見組合模式：順序式、協調者式、平行式、條件式、迭代式、階層式。
• 依賴管理：資料依賴、順序依賴、資源依賴，用 DAG 表示。
• 拓撲排序：確保 Skill 按正確順序執行。
• 循環檢測：避免無限循環。
• 錯誤傳播：直接傳播、間接傳播、靜默失敗。
• 錯誤處理：快速失敗、重試、降級、容錯、補償。
• 編排系統實作：註冊、驗證、拓撲排序、執行、重試。
• 動態編排：用 LLM 決定下一步。
• 最佳實踐：統一介面、標準化輸出、加入驗證、記錄日誌、設定超時、支援平行、提供可觀測性。
• 常見陷阱：過度依賴、忽略錯誤、循環依賴、沒有超時、沒有記錄、硬編碼順序。

Skill 讓 Agent 有能力，組合讓 Agent 能完成複雜任務。
從零件到組裝線，從單一技能到協同工作，這就是 Skill 編排的價值。

不過，設計了 Skill、組合了 Skill，下一個問題是：你怎麼知道這些 Skill 真的好用？
當你修改了某個 Skill 的 Prompt，你怎麼知道它有沒有變好？
當你有多個 Skill 時，你怎麼比較它們的優劣？

這就是下一個系列要談的主題：Agent Harness。

---

下一篇預告

《Skill 的版本管理與迭代》

我們會談如何為 Skill 建立版本控制、如何設計迭代流程、如何進行 A/B 測試，以及如何管理多個版本的 Skill 在生產環境中共存。