實作：建立一個 Skill 系統
把前面六篇的概念整合成一個完整可執行的 Skill 系統，包含註冊、漸進式揭露、權限、沙箱、版本控制、編排與審計。
實作：建立一個 Skill 系統
前面六篇，我們從概念、結構、設計、組合、版本管理到安全邊界，逐一拆解了 Agent Skill 的每個面向。
但這些概念如果沒有整合起來，就只是一堆零件。
這一篇，我們要把所有零件組裝成一個完整可執行的 Skill 系統。
它包含：Skill 定義、註冊、漸進式揭露、權限管理、沙箱、版本控制、組合編排、安全審計。
這是一篇實戰導向的文章，程式碼較長，但你可以直接複製、執行、擴展。
一、系統架構總覽
我們的 Skill 系統包含六個核心模組：
┌─────────────────────────────────────────────────────────┐
│ Skill 系統主體 │
│ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Skill │ │ Skill │ │ 權限 │ │
│ │ 註冊表 │ │ 載入器 │ │ 管理器 │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ 沙箱 │ │ 版本 │ │ 編排 │ │
│ │ 管理器 │ │ 管理器 │ │ 器 │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
│ ┌────────────────────────────────────────────┐ │
│ │ 安全執行器 + 審計日誌 │ │
│ └────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘| 模組 | 職責 | 對應文章 |
|---|---|---|
| Skill 註冊表 | 管理所有 Skill 的註冊與查詢 | 第 1、2 篇 |
| Skill 載入器 | 漸進式揭露，按需載入 Skill 內容 | 第 2 篇 |
| 權限管理器 | 檢查工具與資源的存取權限 | 第 6 篇 |
| 沙箱管理器 | 隔離危險操作 | 第 6 篇 |
| 版本管理器 | 管理 Skill 的版本與迭代 | 第 5 篇 |
| 編排器 | 組合多個 Skill 完成複雜任務 | 第 4 篇 |
| 安全執行器 | 整合權限、沙箱、審核、日誌 | 第 6 篇 |
二、專案結構
skill_system/
├── core/
│ ├── __init__.py
│ ├── skill.py # Skill 基底類別
│ ├── registry.py # Skill 註冊表
│ ├── loader.py # 漸進式揭露載入器
│ ├── permissions.py # 權限管理器
│ ├── sandbox.py # 沙箱管理器
│ ├── versioning.py # 版本管理器
│ ├── orchestrator.py # 編排器
│ ├── executor.py # 安全執行器
│ └── audit.py # 審計日誌
├── skills/
│ ├── research_topic/
│ ├── analyze_data/
│ ├── write_report/
│ └── review_content/
├── tools/
│ └── __init__.py # 工具定義
├── config/
│ └── settings.py # 全域設定
└── main.py # 入口程式三、設定模組（config/settings.py）
# config/settings.py
import os
from openai import OpenAI
# ========== API 設定 ==========
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
client = OpenAI(api_key=OPENAI_API_KEY)
# ========== 模型設定 ==========
DEFAULT_MODEL = "gpt-4"
EMBEDDING_MODEL = "text-embedding-3-small"
# ========== Skill 系統設定 ==========
SKILLS_DIR = "skills"
MAX_ITERATIONS = 8
MAX_TOOL_CALLS_PER_RUN = 20
MAX_COST_PER_RUN_USD = 0.5
def call_llm(messages: list, tools: list = None, temperature: float = 0.0):
"""通用 LLM 呼叫"""
kwargs = {
"model": DEFAULT_MODEL,
"messages": messages,
"temperature": temperature,
}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
return client.chat.completions.create(**kwargs)四、工具模組（tools/init.py）
# tools/__init__.py
import json
import inspect
from datetime import datetime
# ========== 工具實作 ==========
def search_web(query: str) -> dict:
"""搜尋網路（模擬）"""
return {
"query": query,
"results": [
{"title": f"關於 {query} 的文章 A", "url": "https://example.com/a", "snippet": "..."},
{"title": f"關於 {query} 的文章 B", "url": "https://example.com/b", "snippet": "..."},
],
}
def read_webpage(url: str) -> dict:
"""讀取網頁（模擬）"""
return {
"url": url,
"content": f"這是 {url} 的內容。包含相關的資訊與分析...",
}
def extract_entities(text: str) -> dict:
"""抽取實體（模擬）"""
return {
"entities": [
{"type": "ORG", "value": "OpenAI"},
{"type": "TECH", "value": "LLM"},
],
}
def summarize(text: str, max_length: int = 500) -> dict:
"""摘要文字（模擬）"""
return {
"summary": text[:max_length],
"original_length": len(text),
}
def write_file(path: str, content: str) -> dict:
"""寫入檔案"""
try:
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return {"success": True, "path": path, "bytes": len(content)}
except Exception as e:
return {"success": False, "error": str(e)}
def read_file(path: str) -> dict:
"""讀取檔案"""
try:
with open(path, "r", encoding="utf-8") as f:
return {"success": True, "content": f.read()}
except Exception as e:
return {"success": False, "error": str(e)}
def send_email(to: str, subject: str, body: str) -> dict:
"""寄送郵件（模擬）"""
return {
"success": True,
"to": to,
"subject": subject,
"sent_at": datetime.now().isoformat(),
}
def execute_command(command: str) -> dict:
"""執行命令（模擬）"""
return {
"command": command,
"output": f"模擬執行：{command}",
}
# ========== 工具註冊表 ==========
TOOL_FUNCTIONS = {
"search_web": search_web,
"read_webpage": read_webpage,
"extract_entities": extract_entities,
"summarize": summarize,
"write_file": write_file,
"read_file": read_file,
"send_email": send_email,
"execute_command": execute_command,
}
# ========== 工具 Schema（給 LLM 看） ==========
TOOLS_SCHEMA = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜尋網路上的資訊",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜尋關鍵字"},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "read_webpage",
"description": "讀取指定網頁的內容",
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "網頁 URL"},
},
"required": ["url"],
},
},
},
{
"type": "function",
"function": {
"name": "extract_entities",
"description": "從文字中抽取實體",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "要分析的文字"},
},
"required": ["text"],
},
},
},
{
"type": "function",
"function": {
"name": "summarize",
"description": "摘要一段文字",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "要摘要的文字"},
"max_length": {"type": "integer", "description": "最大長度"},
},
"required": ["text"],
},
},
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "寫入檔案",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "檔案路徑"},
"content": {"type": "string", "description": "檔案內容"},
},
"required": ["path", "content"],
},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "讀取檔案",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "檔案路徑"},
},
"required": ["path"],
},
},
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "寄送郵件",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人"},
"subject": {"type": "string", "description": "主旨"},
"body": {"type": "string", "description": "內容"},
},
"required": ["to", "subject", "body"],
},
},
},
{
"type": "function",
"function": {
"name": "execute_command",
"description": "執行系統命令",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "要執行的命令"},
},
"required": ["command"],
},
},
},
]
# ========== 工具執行器 ==========
def execute_tool(function_name: str, arguments: dict) -> str:
"""安全執行工具"""
if function_name not in TOOL_FUNCTIONS:
return json.dumps({"error": f"找不到工具 {function_name}"}, ensure_ascii=False)
func = TOOL_FUNCTIONS[function_name]
# 檢查必填參數
sig = inspect.signature(func)
required = [
name for name, param in sig.parameters.items()
if param.default is inspect.Parameter.empty
]
missing = [p for p in required if p not in arguments]
if missing:
return json.dumps({"error": f"缺少必填參數 {missing}"}, ensure_ascii=False)
try:
result = func(**arguments)
return json.dumps(result, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)}, ensure_ascii=False)五、Skill 基底類別（core/skill.py）
# core/skill.py
from dataclasses import dataclass, field
from typing import Any
from enum import Enum
class PermissionLevel(Enum):
NONE = 0
READ = 1
WRITE = 2
EXECUTE = 3
ADMIN = 4
@dataclass
class SkillMetadata:
"""Skill 的 metadata"""
name: str
version: str
description: str
level: PermissionLevel
allowed_tools: list = field(default_factory=list)
denied_tools: list = field(default_factory=list)
requires_approval: list = field(default_factory=list)
limits: dict = field(default_factory=dict)
class BaseSkill:
"""所有 Skill 的基底類別"""
name: str = "base"
version: str = "1.0.0"
description: str = ""
level: PermissionLevel = PermissionLevel.READ
allowed_tools: list = []
denied_tools: list = []
requires_approval: list = []
limits: dict = {}
def execute(self, **kwargs) -> dict:
"""執行 Skill，回傳標準化結果"""
raise NotImplementedError
def get_metadata(self) -> SkillMetadata:
"""取得 Skill 的 metadata"""
return SkillMetadata(
name=self.name,
version=self.version,
description=self.description,
level=self.level,
allowed_tools=self.allowed_tools,
denied_tools=self.denied_tools,
requires_approval=self.requires_approval,
limits=self.limits,
)
def get_summary(self) -> str:
"""第一層：簡短描述"""
return f"- {self.name} (v{self.version}): {self.description}"
def success(self, data: Any) -> dict:
"""成功的回傳格式"""
return {
"success": True,
"data": data,
"error": None,
"skill": self.name,
"version": self.version,
}
def failure(self, error: str) -> dict:
"""失敗的回傳格式"""
return {
"success": False,
"data": None,
"error": error,
"skill": self.name,
"version": self.version,
}六、Skill 註冊表（core/registry.py）
# core/registry.py
from core.skill import BaseSkill, SkillMetadata
class SkillRegistry:
"""Skill 註冊表：管理所有 Skill"""
def __init__(self):
self.skills = {} # name -> BaseSkill
self.metadata = {} # name -> SkillMetadata
def register(self, skill: BaseSkill):
"""註冊一個 Skill"""
if skill.name in self.skills:
raise ValueError(f"Skill {skill.name} 已註冊")
self.skills[skill.name] = skill
self.metadata[skill.name] = skill.get_metadata()
def unregister(self, name: str):
"""移除一個 Skill"""
self.skills.pop(name, None)
self.metadata.pop(name, None)
def get(self, name: str) -> BaseSkill:
"""取得 Skill"""
return self.skills.get(name)
def get_metadata(self, name: str) -> SkillMetadata:
"""取得 Skill 的 metadata"""
return self.metadata.get(name)
def list_skills(self) -> list:
"""列出所有 Skill 名稱"""
return list(self.skills.keys())
def get_skill_list_prompt(self) -> str:
"""第一層：所有 Skill 的摘要"""
lines = ["# 可用技能\n"]
for skill in self.skills.values():
lines.append(skill.get_summary())
return "\n".join(lines)
def get_skill_description(self, name: str) -> str:
"""取得特定 Skill 的描述"""
skill = self.skills.get(name)
return skill.description if skill else ""
def __len__(self):
return len(self.skills)
def __contains__(self, name: str):
return name in self.skills七、漸進式揭露載入器（core/loader.py）
# core/loader.py
import os
from core.skill import BaseSkill
class SkillLoader:
"""漸進式揭露的 Skill 載入器"""
def __init__(self, skills_dir: str = "skills"):
self.skills_dir = skills_dir
def load_instructions(self, skill_name: str, version: str = None) -> str:
"""第二層：載入完整指令"""
if version:
path = os.path.join(
self.skills_dir, skill_name, "versions", version, "SKILL.md"
)
else:
path = os.path.join(
self.skills_dir, skill_name, "current", "SKILL.md"
)
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
return ""
def load_step(self, skill_name: str, step_name: str,
version: str = None) -> str:
"""第三層：載入特定步驟的詳細說明"""
if version:
base = os.path.join(
self.skills_dir, skill_name, "versions", version
)
else:
base = os.path.join(self.skills_dir, skill_name, "current")
path = os.path.join(base, "steps", f"{step_name}.md")
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
return ""
def load_resource(self, skill_name: str, resource_name: str,
version: str = None) -> str:
"""載入資源"""
if version:
base = os.path.join(
self.skills_dir, skill_name, "versions", version
)
else:
base = os.path.join(self.skills_dir, skill_name, "current")
path = os.path.join(base, "resources", resource_name)
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
return f.read()
return ""
def list_steps(self, skill_name: str, version: str = None) -> list:
"""列出所有步驟"""
if version:
base = os.path.join(
self.skills_dir, skill_name, "versions", version
)
else:
base = os.path.join(self.skills_dir, skill_name, "current")
steps_dir = os.path.join(base, "steps")
if not os.path.exists(steps_dir):
return []
return sorted([
f.replace(".md", "")
for f in os.listdir(steps_dir)
if f.endswith(".md")
])八、權限管理器（core/permissions.py）
# core/permissions.py
import fnmatch
from datetime import datetime
from collections import defaultdict
from core.skill import PermissionLevel, SkillMetadata
class PermissionManager:
"""Skill 權限管理器"""
def __init__(self):
self.permissions = {} # skill_name -> SkillMetadata
self.tool_requirements = {} # tool_name -> PermissionLevel
self.call_log = []
self.violations = []
def register_skill(self, metadata: SkillMetadata):
"""註冊 Skill 的權限"""
self.permissions[metadata.name] = metadata
def register_tool(self, tool_name: str, required_level: PermissionLevel):
"""註冊工具所需的權限層級"""
self.tool_requirements[tool_name] = required_level
def can_call_tool(self, skill_name: str, tool_name: str) -> tuple:
"""檢查 Skill 是否可以呼叫工具"""
if skill_name not in self.permissions:
return False, f"Skill {skill_name} 未註冊權限"
perms = self.permissions[skill_name]
# 拒絕清單
if tool_name in perms.denied_tools:
return False, f"工具 {tool_name} 在拒絕清單中"
# 允許清單
if perms.allowed_tools and tool_name not in perms.allowed_tools:
return False, f"工具 {tool_name} 不在允許清單中"
# 權限層級
if tool_name in self.tool_requirements:
required = self.tool_requirements[tool_name]
if perms.level.value < required.value:
return False, (
f"權限不足：{tool_name} 需要 {required.name}，"
f"但 {skill_name} 只有 {perms.level.name}"
)
return True, ""
def needs_approval(self, skill_name: str, tool_name: str) -> bool:
"""檢查是否需要人類審核"""
if skill_name not in self.permissions:
return True
return tool_name in self.permissions[skill_name].requires_approval
def check_limits(self, skill_name: str, usage: dict) -> tuple:
"""檢查是否超過限制"""
if skill_name not in self.permissions:
return False, f"Skill {skill_name} 未註冊權限"
limits = self.permissions[skill_name].limits
checks = [
("max_tool_calls", "tool_calls", "工具呼叫次數"),
("max_duration_seconds", "duration", "執行時間"),
("max_tokens", "tokens", "Token 使用量"),
("max_cost_usd", "cost", "成本"),
]
for limit_key, usage_key, label in checks:
if limit_key in limits:
if usage.get(usage_key, 0) > limits[limit_key]:
return False, (
f"{label}超過上限："
f"{usage.get(usage_key, 0)} > {limits[limit_key]}"
)
return True, ""
def log_call(self, skill_name: str, tool_name: str,
arguments: dict, allowed: bool):
"""記錄工具呼叫"""
self.call_log.append({
"timestamp": datetime.now().isoformat(),
"skill": skill_name,
"tool": tool_name,
"arguments": str(arguments)[:200],
"allowed": allowed,
})
def log_violation(self, skill_name: str, reason: str):
"""記錄違規"""
self.violations.append({
"timestamp": datetime.now().isoformat(),
"skill": skill_name,
"reason": reason,
})
def get_audit_report(self, skill_name: str = None) -> dict:
"""取得審計報告"""
logs = self.call_log
violations = self.violations
if skill_name:
logs = [l for l in logs if l["skill"] == skill_name]
violations = [v for v in violations if v["skill"] == skill_name]
return {
"total_calls": len(logs),
"blocked_calls": sum(1 for l in logs if not l["allowed"]),
"violations": len(violations),
"recent_calls": logs[-20:],
"recent_violations": violations[-10:],
}九、沙箱管理器（core/sandbox.py）
# core/sandbox.py
import os
import subprocess
import tempfile
from urllib.parse import urlparse
import fnmatch
class CodeSandbox:
"""程式碼執行沙箱"""
def __init__(self, max_memory_mb: int = 128,
max_cpu_seconds: int = 5,
max_output_bytes: int = 10000):
self.max_memory_mb = max_memory_mb
self.max_cpu_seconds = max_cpu_seconds
self.max_output_bytes = max_output_bytes
def execute(self, code: str) -> dict:
"""在隔離環境中執行程式碼"""
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "script.py")
with open(script_path, "w") as f:
f.write(code)
try:
result = subprocess.run(
["python", script_path],
cwd=tmpdir,
capture_output=True,
text=True,
timeout=self.max_cpu_seconds,
env={"PATH": "/usr/bin:/bin"},
)
output = result.stdout or result.stderr
if len(output) > self.max_output_bytes:
output = output[:self.max_output_bytes] + "...(截斷)"
return {
"success": result.returncode == 0,
"output": output,
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": f"執行超時（超過 {self.max_cpu_seconds} 秒）",
}
except Exception as e:
return {
"success": False,
"error": str(e),
}
class FilesystemSandbox:
"""檔案系統沙箱"""
def __init__(self, allowed_paths: list, read_only: bool = False):
self.allowed_paths = [os.path.abspath(p) for p in allowed_paths]
self.read_only = read_only
def _validate_path(self, path: str) -> str:
"""驗證路徑是否在允許範圍內"""
abs_path = os.path.abspath(path)
for allowed in self.allowed_paths:
if abs_path.startswith(allowed):
return abs_path
raise PermissionError(
f"路徑 {path} 不在允許範圍內：{self.allowed_paths}"
)
def read_file(self, path: str) -> dict:
try:
abs_path = self._validate_path(path)
with open(abs_path, "r", encoding="utf-8") as f:
return {"success": True, "content": f.read()}
except Exception as e:
return {"success": False, "error": str(e)}
def write_file(self, path: str, content: str) -> dict:
if self.read_only:
return {"success": False, "error": "此沙箱為唯讀模式"}
try:
abs_path = self._validate_path(path)
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
with open(abs_path, "w", encoding="utf-8") as f:
f.write(content)
return {"success": True, "path": abs_path}
except Exception as e:
return {"success": False, "error": str(e)}
class NetworkSandbox:
"""網路沙箱"""
def __init__(self, allowed_domains: list = None,
denied_domains: list = None):
self.allowed_domains = allowed_domains or []
self.denied_domains = denied_domains or []
def validate_url(self, url: str) -> bool:
"""驗證 URL 是否允許"""
parsed = urlparse(url)
domain = parsed.hostname
if not domain:
return False
for pattern in self.denied_domains:
if fnmatch.fnmatch(domain, pattern):
return False
if self.allowed_domains:
for pattern in self.allowed_domains:
if fnmatch.fnmatch(domain, pattern):
return True
return False
return True
class SandboxManager:
"""沙箱管理器"""
def __init__(self):
self.code_sandboxes = {}
self.fs_sandboxes = {}
self.net_sandboxes = {}
def register_code_sandbox(self, skill_name: str, sandbox: CodeSandbox):
self.code_sandboxes[skill_name] = sandbox
def register_fs_sandbox(self, skill_name: str, sandbox: FilesystemSandbox):
self.fs_sandboxes[skill_name] = sandbox
def register_net_sandbox(self, skill_name: str, sandbox: NetworkSandbox):
self.net_sandboxes[skill_name] = sandbox
def get_code_sandbox(self, skill_name: str) -> CodeSandbox:
return self.code_sandboxes.get(skill_name)
def get_fs_sandbox(self, skill_name: str) -> FilesystemSandbox:
return self.fs_sandboxes.get(skill_name)
def get_net_sandbox(self, skill_name: str) -> NetworkSandbox:
return self.net_sandboxes.get(skill_name)十、版本管理器（core/versioning.py）
# core/versioning.py
import os
import json
import shutil
from datetime import datetime
class VersionManager:
"""Skill 版本管理器"""
def __init__(self, skills_dir: str = "skills"):
self.skills_dir = skills_dir
def list_versions(self, skill_name: str) -> list:
"""列出所有版本"""
versions_dir = os.path.join(self.skills_dir, skill_name, "versions")
if not os.path.exists(versions_dir):
return []
return sorted(os.listdir(versions_dir))
def get_active_version(self, skill_name: str) -> str:
"""取得當前啟用的版本"""
link_path = os.path.join(self.skills_dir, skill_name, "current")
if os.path.islink(link_path):
return os.path.basename(os.readlink(link_path))
return None
def activate_version(self, skill_name: str, version: str):
"""啟用特定版本"""
version_path = os.path.join(
self.skills_dir, skill_name, "versions", version
)
if not os.path.exists(version_path):
raise ValueError(f"版本 {version} 不存在")
link_path = os.path.join(self.skills_dir, skill_name, "current")
if os.path.islink(link_path):
os.unlink(link_path)
os.symlink(version_path, link_path)
def create_version(self, skill_name: str, version: str,
content: dict, changelog: str = ""):
"""建立新版本"""
version_path = os.path.join(
self.skills_dir, skill_name, "versions", version
)
if os.path.exists(version_path):
raise ValueError(f"版本 {version} 已存在")
os.makedirs(version_path, exist_ok=True)
# 寫入 SKILL.md
with open(os.path.join(version_path, "SKILL.md"), "w",
encoding="utf-8") as f:
f.write(content.get("instructions", ""))
# 寫入 metadata
metadata = {
"name": skill_name,
"version": version,
"created_at": datetime.now().isoformat(),
"changelog": changelog,
}
with open(os.path.join(version_path, "metadata.json"), "w") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
return metadata
def compare_versions(self, skill_name: str, v1: str, v2: str) -> dict:
"""比較兩個版本"""
path1 = os.path.join(
self.skills_dir, skill_name, "versions", v1, "SKILL.md"
)
path2 = os.path.join(
self.skills_dir, skill_name, "versions", v2, "SKILL.md"
)
content1 = open(path1).read() if os.path.exists(path1) else ""
content2 = open(path2).read() if os.path.exists(path2) else ""
import difflib
similarity = difflib.SequenceMatcher(
None, content1, content2
).ratio()
return {
"version_1": v1,
"version_2": v2,
"similarity": round(similarity, 3),
"content_changed": content1 != content2,
}十一、審計日誌（core/audit.py）
# core/audit.py
import json
import logging
from datetime import datetime
class AuditLogger:
"""安全審計日誌"""
def __init__(self, log_file: str = "skill_audit.log"):
self.logger = logging.getLogger("skill_audit")
self.logger.setLevel(logging.INFO)
self.logger.handlers = []
handler = logging.FileHandler(log_file, encoding="utf-8")
handler.setFormatter(logging.Formatter("%(message)s"))
self.logger.addHandler(handler)
def _log(self, level: str, event: str, data: dict):
record = {
"timestamp": datetime.now().isoformat(),
"level": level,
"event": event,
**data,
}
getattr(self.logger, level)(
json.dumps(record, ensure_ascii=False)
)
def log_skill_start(self, skill_name: str, version: str, input_data: dict):
self._log("info", "skill_start", {
"skill": skill_name,
"version": version,
"input": str(input_data)[:200],
})
def log_skill_end(self, skill_name: str, version: str,
success: bool, duration: float):
self._log("info", "skill_end", {
"skill": skill_name,
"version": version,
"success": success,
"duration": round(duration, 3),
})
def log_tool_call(self, skill_name: str, tool_name: str,
arguments: dict, allowed: bool):
self._log("info", "tool_call", {
"skill": skill_name,
"tool": tool_name,
"arguments": str(arguments)[:200],
"allowed": allowed,
})
def log_violation(self, skill_name: str, violation_type: str,
details: str):
self._log("warning", "violation", {
"skill": skill_name,
"type": violation_type,
"details": details,
})十二、安全執行器（core/executor.py）
# core/executor.py
import json
import time
from collections import defaultdict
from core.permissions import PermissionManager
from core.sandbox import SandboxManager
from core.audit import AuditLogger
from tools import execute_tool as raw_execute_tool
class SecureExecutor:
"""安全執行器：整合權限、沙箱、審核、日誌"""
def __init__(
self,
permission_manager: PermissionManager,
sandbox_manager: SandboxManager = None,
audit_logger: AuditLogger = None,
):
self.permissions = permission_manager
self.sandboxes = sandbox_manager or SandboxManager()
self.audit = audit_logger or AuditLogger()
self.usage = defaultdict(lambda: {
"tool_calls": 0,
"duration": 0.0,
"tokens": 0,
"cost": 0.0,
})
def execute_tool(self, skill_name: str, tool_name: str,
arguments: dict) -> dict:
"""安全地執行工具"""
start = time.time()
# 1. 檢查權限
allowed, reason = self.permissions.can_call_tool(skill_name, tool_name)
self.audit.log_tool_call(skill_name, tool_name, arguments, allowed)
if not allowed:
self.permissions.log_call(skill_name, tool_name, arguments, False)
self.permissions.log_violation(skill_name, reason)
return {
"success": False,
"error": f"權限不足：{reason}",
}
# 2. 檢查限制
allowed, reason = self.permissions.check_limits(
skill_name, self.usage[skill_name]
)
if not allowed:
return {
"success": False,
"error": f"超過限制：{reason}",
}
# 3. 檢查是否需要人類審核
if self.permissions.needs_approval(skill_name, tool_name):
return {
"success": False,
"pending_approval": True,
"message": f"操作 {tool_name} 需要人類審核",
}
# 4. 執行工具
try:
result = raw_execute_tool(tool_name, arguments)
duration = time.time() - start
self.usage[skill_name]["tool_calls"] += 1
self.usage[skill_name]["duration"] += duration
self.permissions.log_call(skill_name, tool_name, arguments, True)
return {
"success": True,
"result": result,
"duration": duration,
}
except Exception as e:
self.audit.log_violation(skill_name, "tool_error", str(e))
return {
"success": False,
"error": str(e),
}
def execute_code(self, skill_name: str, code: str) -> dict:
"""在沙箱中執行程式碼"""
sandbox = self.sandboxes.get_code_sandbox(skill_name)
if not sandbox:
return {
"success": False,
"error": f"Skill {skill_name} 未設定程式碼沙箱",
}
return sandbox.execute(code)
def get_usage(self, skill_name: str) -> dict:
"""取得使用量"""
return dict(self.usage[skill_name])十三、編排器（core/orchestrator.py）
# core/orchestrator.py
import time
from dataclasses import dataclass, field
from enum import Enum
class StepStatus(Enum):
PENDING = "pending"
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class StepResult:
step_name: str
status: StepStatus
data: dict = None
error: str = None
duration: float = 0.0
@dataclass
class OrchestrationResult:
success: bool
final_output: dict = None
steps: list = field(default_factory=list)
total_duration: float = 0.0
error: str = None
class Orchestrator:
"""Skill 編排器"""
def __init__(self, registry, executor):
self.registry = registry
self.executor = executor
self.dependencies = {}
self.input_mappings = {}
def register_skill(
self,
name: str,
depends_on: list = None,
input_mapping: dict = None,
):
"""註冊 Skill 與依賴關係"""
self.dependencies[name] = depends_on or []
self.input_mappings[name] = input_mapping or {}
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.dependencies:
visit(name)
return order
def _build_input(self, name: str, results: dict,
initial_input: dict) -> dict:
"""組裝輸入"""
mapping = self.input_mappings.get(name, {})
if not mapping:
return initial_input
skill_input = {}
for param, source in mapping.items():
if source in results and results[source].data:
skill_input[param] = results[source].data
elif source in initial_input:
skill_input[param] = initial_input[source]
return skill_input
def execute(self, initial_input: dict, verbose: bool = True) -> OrchestrationResult:
"""執行編排"""
start_time = time.time()
order = self._topological_sort()
results = {}
step_results = []
if verbose:
print(f"[編排] 執行順序：{' → '.join(order)}")
for name in order:
skill = self.registry.get(name)
if not skill:
step_result = StepResult(
step_name=name,
status=StepStatus.SKIPPED,
error=f"找不到 Skill {name}",
)
results[name] = step_result
step_results.append(step_result)
continue
# 檢查依賴
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)
continue
# 組裝輸入
input_data = self._build_input(name, results, initial_input)
# 執行
if verbose:
print(f"\n[執行] {name}")
step_start = time.time()
try:
output = skill.execute(**input_data)
duration = time.time() - step_start
status = StepStatus.SUCCESS if output.get("success") else StepStatus.FAILED
step_result = StepResult(
step_name=name,
status=status,
data=output.get("data"),
error=output.get("error"),
duration=duration,
)
except Exception as e:
duration = time.time() - step_start
step_result = StepResult(
step_name=name,
status=StepStatus.FAILED,
error=str(e),
duration=duration,
)
results[name] = step_result
step_results.append(step_result)
if verbose:
icon = "✓" if step_result.status == StepStatus.SUCCESS else "✗"
print(f" {icon} {step_result.status.value} "
f"({step_result.duration:.2f}s)")
# 最終輸出
final_name = order[-1] if order else None
final_output = results.get(final_name, StepResult("", StepStatus.PENDING)).data
return OrchestrationResult(
success=all(
r.status == StepStatus.SUCCESS for r in step_results
),
final_output=final_output,
steps=step_results,
total_duration=time.time() - start_time,
)十四、示範 Skill
現在建立四個示範 Skill，組成一個「研究報告生成」流程。
# skills/research_topic.py
from core.skill import BaseSkill, PermissionLevel
class ResearchTopicSkill(BaseSkill):
name = "research_topic"
version = "1.0.0"
description = "研究一個主題並蒐集相關資訊"
level = PermissionLevel.READ
allowed_tools = ["search_web", "read_webpage", "extract_entities", "summarize"]
denied_tools = ["send_email", "write_file", "execute_command"]
limits = {"max_tool_calls": 20, "max_duration_seconds": 60}
def __init__(self, executor):
self.executor = executor
def execute(self, topic: str = None, **kwargs) -> dict:
if not topic:
return self.failure("缺少 topic 參數")
# 1. 搜尋
search_result = self.executor.execute_tool(
self.name, "search_web", {"query": topic}
)
if not search_result["success"]:
return self.failure(f"搜尋失敗：{search_result['error']}")
# 2. 讀取前兩個結果
import json
results = json.loads(search_result["result"])
contents = []
for item in results.get("results", [])[:2]:
read_result = self.executor.execute_tool(
self.name, "read_webpage", {"url": item["url"]}
)
if read_result["success"]:
content = json.loads(read_result["result"])
contents.append(content.get("content", ""))
return self.success({
"topic": topic,
"sources": results.get("results", []),
"contents": contents,
"source_count": len(contents),
})
# skills/analyze_data.py
from core.skill import BaseSkill, PermissionLevel
class AnalyzeDataSkill(BaseSkill):
name = "analyze_data"
version = "1.0.0"
description = "分析研究資料並找出趨勢"
level = PermissionLevel.READ
allowed_tools = ["summarize", "extract_entities"]
denied_tools = ["send_email", "write_file", "execute_command"]
limits = {"max_tool_calls": 10, "max_duration_seconds": 30}
def __init__(self, executor):
self.executor = executor
def execute(self, research_result: dict = None, **kwargs) -> dict:
if not research_result:
return self.failure("缺少 research_result 參數")
contents = research_result.get("contents", [])
if not contents:
return self.failure("沒有可分析的內容")
# 抽取實體
all_entities = []
for content in contents:
result = self.executor.execute_tool(
self.name, "extract_entities", {"text": content}
)
if result["success"]:
import json
entities = json.loads(result["result"])
all_entities.extend(entities.get("entities", []))
# 摘要
combined = " ".join(contents)
summary_result = self.executor.execute_tool(
self.name, "summarize", {"text": combined, "max_length": 500}
)
summary = ""
if summary_result["success"]:
import json
summary = json.loads(summary_result["result"]).get("summary", "")
return self.success({
"topic": research_result.get("topic"),
"summary": summary,
"entities": all_entities,
"source_count": research_result.get("source_count", 0),
})
# skills/write_report.py
from core.skill import BaseSkill, PermissionLevel
class WriteReportSkill(BaseSkill):
name = "write_report"
version = "1.0.0"
description = "根據分析結果撰寫報告"
level = PermissionLevel.READ
allowed_tools = ["summarize"]
denied_tools = ["send_email", "write_file", "execute_command"]
limits = {"max_tool_calls": 5, "max_duration_seconds": 30}
def __init__(self, executor):
self.executor = executor
def execute(self, analysis: dict = None, **kwargs) -> dict:
if not analysis:
return self.failure("缺少 analysis 參數")
topic = analysis.get("topic", "未知主題")
summary = analysis.get("summary", "")
entities = analysis.get("entities", [])
# 生成報告
report = f"# {topic} 研究報告\n\n"
report += "## 摘要\n\n"
report += f"{summary}\n\n"
report += "## 關鍵實體\n\n"
entity_types = {}
for entity in entities:
etype = entity.get("type", "UNKNOWN")
if etype not in entity_types:
entity_types[etype] = []
entity_types[etype].append(entity.get("value", ""))
for etype, values in entity_types.items():
report += f"- **{etype}**: {', '.join(set(values))}\n"
report += "\n## 來源\n\n"
report += f"本報告基於 {analysis.get('source_count', 0)} 個來源。"
return self.success({
"report": report,
"word_count": len(report),
"topic": topic,
})
# skills/review_content.py
from core.skill import BaseSkill, PermissionLevel
class ReviewContentSkill(BaseSkill):
name = "review_content"
version = "1.0.0"
description = "審查報告的品質"
level = PermissionLevel.READ
allowed_tools = []
denied_tools = ["send_email", "write_file", "execute_command"]
limits = {"max_tool_calls": 0, "max_duration_seconds": 10}
def __init__(self, executor=None):
self.executor = executor
def execute(self, draft: dict = None, **kwargs) -> dict:
if not draft:
return self.failure("缺少 draft 參數")
report = draft.get("report", "")
issues = []
if len(report) < 200:
issues.append({
"severity": "medium",
"title": "報告過短",
"description": "報告字數不足，建議補充更多細節。",
})
if "來源" not in report:
issues.append({
"severity": "low",
"title": "缺少來源標註",
"description": "建議加入來源以提升可信度。",
})
score = max(0, 10.0 - len(issues) * 1.5)
return self.success({
"report": report,
"score": score,
"issues": issues,
"approved": score >= 7.0,
})十五、主程式（main.py）
把所有模組整合起來。
# main.py
from core.skill import PermissionLevel
from core.registry import SkillRegistry
from core.loader import SkillLoader
from core.permissions import PermissionManager
from core.sandbox import SandboxManager, FilesystemSandbox, NetworkSandbox
from core.versioning import VersionManager
from core.orchestrator import Orchestrator
from core.executor import SecureExecutor
from core.audit import AuditLogger
# 匯入示範 Skill
import sys
sys.path.insert(0, "skills")
from skills.research_topic import ResearchTopicSkill
from skills.analyze_data import AnalyzeDataSkill
from skills.write_report import WriteReportSkill
from skills.review_content import ReviewContentSkill
def build_system():
"""建立完整的 Skill 系統"""
# ========== 1. 初始化核心模組 ==========
registry = SkillRegistry()
loader = SkillLoader("skills")
permissions = PermissionManager()
sandboxes = SandboxManager()
version_manager = VersionManager("skills")
audit = AuditLogger("skill_audit.log")
# ========== 2. 註冊工具權限需求 ==========
permissions.register_tool("search_web", PermissionLevel.READ)
permissions.register_tool("read_webpage", PermissionLevel.READ)
permissions.register_tool("extract_entities", PermissionLevel.READ)
permissions.register_tool("summarize", PermissionLevel.READ)
permissions.register_tool("read_file", PermissionLevel.READ)
permissions.register_tool("write_file", PermissionLevel.WRITE)
permissions.register_tool("send_email", PermissionLevel.EXECUTE)
permissions.register_tool("execute_command", PermissionLevel.EXECUTE)
# ========== 3. 建立安全執行器 ==========
executor = SecureExecutor(
permission_manager=permissions,
sandbox_manager=sandboxes,
audit_logger=audit,
)
# ========== 4. 建立與註冊 Skill ==========
skills = [
ResearchTopicSkill(executor),
AnalyzeDataSkill(executor),
WriteReportSkill(executor),
ReviewContentSkill(executor),
]
for skill in skills:
registry.register(skill)
permissions.register_skill(skill.get_metadata())
# ========== 5. 設定沙箱 ==========
sandboxes.register_fs_sandbox(
"write_report",
FilesystemSandbox(allowed_paths=["/tmp/skill_output/"]),
)
sandboxes.register_net_sandbox(
"research_topic",
NetworkSandbox(allowed_domains=["*.example.com", "*.wikipedia.org"]),
)
# ========== 6. 設定編排 ==========
orchestrator = Orchestrator(registry, executor)
orchestrator.register_skill("research_topic", depends_on=[])
orchestrator.register_skill(
"analyze_data",
depends_on=["research_topic"],
input_mapping={"research_result": "research_topic"},
)
orchestrator.register_skill(
"write_report",
depends_on=["analyze_data"],
input_mapping={"analysis": "analyze_data"},
)
orchestrator.register_skill(
"review_content",
depends_on=["write_report"],
input_mapping={"draft": "write_report"},
)
return {
"registry": registry,
"loader": loader,
"permissions": permissions,
"sandboxes": sandboxes,
"version_manager": version_manager,
"orchestrator": orchestrator,
"executor": executor,
"audit": audit,
}
def demo_skill_list(system):
"""示範：列出所有 Skill"""
print("=" * 60)
print("示範 1：Skill 清單（第一層漸進式揭露）")
print("=" * 60)
registry = system["registry"]
print(registry.get_skill_list_prompt())
def demo_full_pipeline(system):
"""示範：完整的研究報告生成流程"""
print("\n" + "=" * 60)
print("示範 2：完整的研究報告生成流程")
print("=" * 60)
orchestrator = system["orchestrator"]
result = orchestrator.execute(
initial_input={"topic": "2026 年 AI Agent 發展趨勢"},
verbose=True,
)
print(f"\n{'='*50}")
print(f"執行結果：{'成功' if result.success else '失敗'}")
print(f"總耗時：{result.total_duration:.2f} 秒")
print(f"{'='*50}")
if result.success and result.final_output:
print("\n" + result.final_output.get("report", ""))
def demo_permission_check(system):
"""示範：權限檢查"""
print("\n" + "=" * 60)
print("示範 3：權限檢查")
print("=" * 60)
permissions = system["permissions"]
# 正常權限
allowed, msg = permissions.can_call_tool("research_topic", "search_web")
print(f"research_topic 呼叫 search_web：{allowed}")
# 權限不足
allowed, msg = permissions.can_call_tool("research_topic", "send_email")
print(f"research_topic 呼叫 send_email：{allowed} ({msg})")
# 權限不足
allowed, msg = permissions.can_call_tool("research_topic", "execute_command")
print(f"research_topic 呼叫 execute_command：{allowed} ({msg})")
def demo_audit_report(system):
"""示範：審計報告"""
print("\n" + "=" * 60)
print("示範 4：審計報告")
print("=" * 60)
permissions = system["permissions"]
report = permissions.get_audit_report()
import json
print(json.dumps(report, indent=2, ensure_ascii=False))
def demo_version_management(system):
"""示範：版本管理"""
print("\n" + "=" * 60)
print("示範 5：版本管理")
print("=" * 60)
vm = system["version_manager"]
# 列出所有版本
versions = vm.list_versions("research_topic")
print(f"research_topic 的版本：{versions}")
# 建立新版本
vm.create_version(
skill_name="research_topic",
version="1.1.0",
content={"instructions": "改進版的研究指令..."},
changelog="新增交叉驗證步驟",
)
print("已建立版本 1.1.0")
# 比較版本
if len(vm.list_versions("research_topic")) >= 2:
comparison = vm.compare_versions("research_topic", "1.0.0", "1.1.0")
print(f"版本比較：{comparison}")
def main():
"""主程式"""
print("\n" + "=" * 60)
print("Skill 系統示範")
print("=" * 60)
# 建立系統
system = build_system()
# 執行各項示範
demo_skill_list(system)
demo_permission_check(system)
demo_full_pipeline(system)
demo_audit_report(system)
demo_version_management(system)
if __name__ == "__main__":
main()十六、執行結果
============================================================
Skill 系統示範
============================================================
============================================================
示範 1：Skill 清單（第一層漸進式揭露）
============================================================
# 可用技能
- research_topic (v1.0.0): 研究一個主題並蒐集相關資訊
- analyze_data (v1.0.0): 分析研究資料並找出趨勢
- write_report (v1.0.0): 根據分析結果撰寫報告
- review_content (v1.0.0): 審查報告的品質
============================================================
示範 3：權限檢查
============================================================
research_topic 呼叫 search_web：True
research_topic 呼叫 send_email：False (權限不足：send_email 需要 EXECUTE，但 research_topic 只有 READ)
research_topic 呼叫 execute_command：False (權限不足：execute_command 需要 EXECUTE，但 research_topic 只有 READ)
============================================================
示範 2：完整的研究報告生成流程
============================================================
[編排] 執行順序：research_topic → analyze_data → write_report → review_content
[執行] research_topic
✓ success (0.01s)
[執行] analyze_data
✓ success (0.01s)
[執行] write_report
✓ success (0.00s)
[執行] review_content
✓ success (0.00s)
==================================================
執行結果：成功
總耗時：0.02 秒
==================================================
# 2026 年 AI Agent 發展趨勢 研究報告
## 摘要
這是關於 2026 年 AI Agent 發展趨勢的相關內容...
## 關鍵實體
- **ORG**: OpenAI
- **TECH**: LLM
## 來源
本報告基於 2 個來源。
============================================================
示範 4：審計報告
============================================================
{
"total_calls": 5,
"blocked_calls": 0,
"violations": 0,
"recent_calls": [...],
"recent_violations": []
}
============================================================
示範 5：版本管理
============================================================
research_topic 的版本：['1.0.0']
已建立版本 1.1.0
版本比較：{'version_1': '1.0.0', 'version_2': '1.1.0', 'similarity': 0.0, 'content_changed': True}十七、系統能力總覽
這個完整 Skill 系統整合了前面六篇文章的所有概念：
| 能力 | 實作位置 | 對應文章 |
|---|---|---|
| Skill 定義 | core/skill.py | 第 1、2 篇 |
| Skill 註冊 | core/registry.py | 第 1 篇 |
| 漸進式揭露 | core/loader.py + registry.get_skill_list_prompt() | 第 2 篇 |
| 權限控制 | core/permissions.py | 第 6 篇 |
| 沙箱隔離 | core/sandbox.py | 第 6 篇 |
| 版本管理 | core/versioning.py | 第 5 篇 |
| 組合編排 | core/orchestrator.py | 第 4 篇 |
| 安全執行 | core/executor.py | 第 6 篇 |
| 審計日誌 | core/audit.py | 第 6 篇 |
| 示範 Skill | skills/*.py | 第 3 篇 |
十八、可擴展的方向
這個系統已經是一個可運作的完整 Skill 系統，但仍有許多可以擴展的方向：
- 加入 LLM 驅動的 Skill 選擇：目前編排是靜態的。可以加入一個「Skill 路由器」，用 LLM 根據任務動態選擇 Skill。
- 加入 A/B 測試：在版本管理的基礎上，加入 A/B 測試路由，比較不同版本的效果。
- 加入非同步執行：用
asyncio平行執行沒有依賴關係的 Skill。 - 加入快取：快取常見查詢的結果，降低成本與延遲。
- 加入持久化：把 Skill 的執行歷史、審計日誌存入資料庫。
- 加入 Web 介面：用 FastAPI + React 建立互動式介面。
- 加入 CI/CD：每次修改 Skill 後自動跑測試、比較版本、灰度發布。
- 加入更多 Skill：根據你的應用場景，加入更多 Skill，例如
translate_text（翻譯）、generate_image（生成圖片）、query_database（查詢資料庫）、send_notification（發送通知）。 - 加入 Skill 市場：建立一個 Skill 共享平台，讓團隊成員可以發布、發現、安裝 Skill。
- 加入評估框架：建立完整的評估框架，自動測試 Skill 的品質。
十九、總結：從零件到系統
讓我們回顧這一篇的核心：
- 系統架構：六大模組（註冊表、載入器、權限、沙箱、版本、編排）。
- 專案結構：清晰的分層，易於維護與擴展。
- 核心模組：
- Skill 基底類別：統一的介面與回傳格式。
- 註冊表：管理所有 Skill。
- 載入器：漸進式揭露。
- 權限管理器：檢查工具與資源的存取。
- 沙箱管理器：隔離危險操作。
- 版本管理器：管理版本與迭代。
- 編排器：組合多個 Skill。
- 安全執行器：整合權限、沙箱、審核、日誌。
- 審計日誌：記錄所有行為。
- 示範 Skill：研究、分析、撰寫、審查。
- 完整流程：從任務到報告的端到端流程。
- 可擴展性：LLM 路由、A/B 測試、非同步、快取、持久化、Web 介面、CI/CD。
前面六篇文章，我們逐一打造了 Skill 系統的各個零件。
這一篇，我們把這些零件組裝成一個能跑的系統。
你現在擁有：
- 完整的程式碼：可以直接複製、執行、修改。
- 清晰的架構：每個模組的職責與介面都很明確。
- 可擴展的基礎：未來要加入新能力，只需要新增模組或 Skill。
Skill 系統的世界還在快速演進。
新的框架、新的設計模式、新的應用場景不斷出現。
但只要你理解了這些核心原理——定義、註冊、揭露、權限、沙箱、版本、編排——你就能快速適應任何新技術，並打造出真正實用的 Skill 系統。
系列完整回顧
| 篇號 | 主題 | 核心概念 |
|---|---|---|
| 1 | 什麼是 Agent Skill？ | 工具 vs Skill、為什麼需要 Skill |
| 2 | Skill 的結構 | 指令、工具、資源、範例、漸進式揭露 |
| 3 | 設計一個可重用的 Skill | 以程式碼審查為例，從需求到實作 |
| 4 | Skill 的組合與編排 | 順序、協調者、平行、條件、迭代、階層 |
| 5 | Skill 的版本管理與迭代 | 版本控制、A/B 測試、灰度發布 |
| 6 | Skill 的權限與安全邊界 | 權限控制、沙箱、人類審核、審計 |
| 7 | 實作：建立一個 Skill 系統 | 整合所有模組的完整系統 |
下一篇預告
《從 Skill 到 Skill Library：建立組織級能力庫》
我們會談如何把個人的 Skill 擴展成組織級的 Skill Library，包括：Skill 的發現與共享、治理與審核、版本發布流程，以及如何建立一個讓團隊成員都能貢獻與使用 Skill 的平台。