Skills 组合与编排
单个 Skill 解决单一问题,而复杂的工作流往往需要多个 Skill 协同配合。
本篇介绍如何设计 Skill 之间的依赖关系,以及在 SKILL.md 中编排多 Skill 工作流。
组合模式:两种思路
| 模式 | 描述 | 适用场景 |
|---|---|---|
| 顺序编排 | Skill A 的输出作为 Skill B 的输入 | 数据处理流水线 |
| 并行调用 | 同时使用多个 Skill,结果合并 | 多源数据聚合 |
Claude 在单次对话中可以跨 Skill 调用,但一次只能"主动读取"一个 Skill 的正文。复杂的多 Skill 工作流,通常由一个"编排 Skill"来统筹调度。
顺序编排示例
以"上传 PDF → 提取文字 → 生成 Word 摘要报告"为例,三个 Skill 依次协作。
---
name: pdf-to-word-report
description: >
将 PDF 文档转换为 Word 格式的摘要报告,包含文字提取、
内容分析和格式化输出三个阶段。当用户需要从 PDF 生成报告、
提取 PDF 内容并整理为 Word 文档时触发。
---
# PDF 转 Word 摘要报告
## 工作流程
本 Skill 协调三个处理阶段,最终生成 .docx 报告。
### 阶段一:提取 PDF 文字
使用 pdf-reading Skill(位于 /mnt/skills/public/pdf-reading/)的处理逻辑:
```bash
python /mnt/skills/public/pdf-reading/scripts/extract.py \
--input <PDF路径> \
--output /home/claude/extracted_text.txt
```
完成后告知用户:已提取 {N} 页内容,共 {字数} 字。
### 阶段二:生成摘要内容
对提取的文字进行内容分析,生成:
- 文档主题(1句话)
- 关键章节摘要(每章 2-3 句)
- 重要数据与结论
将摘要内容写入 /home/claude/summary.json
### 阶段三:生成 Word 报告
使用 docx Skill 的规范,调用:
```bash
python /mnt/skills/public/docx/scripts/generate.py \
--template /mnt/skills/public/docx/assets/report_template.docx \
--data /home/claude/summary.json \
--output /mnt/user-data/outputs/report.docx
```
完成后调用 present_files 展示下载链接。
通过脚本实现 Skill 组合
当多个 Skill 共享脚本时,可以通过主脚本直接调用其他 Skill 的脚本,形成处理管道。
实例
# 文件路径:scripts/pipeline_orchestrator.py
# 编排多个 Skill 脚本的处理管道
import subprocess
import sys
import os
import json
SKILLS_BASE = "/mnt/skills/public"
def run_script(script_path: str, args: list) -> dict:
"""运行指定路径的脚本并返回 JSON 结果"""
cmd = [sys.executable, script_path] + args
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if result.returncode != 0:
return {"status": "error", "message": result.stderr}
try:
return json.loads(result.stdout)
except json.JSONDecodeError:
return {"status": "success", "raw_output": result.stdout}
def run_pipeline(pdf_path: str, output_dir: str) -> dict:
"""
执行完整的 PDF → Word 报告生成管道
阶段 1:PDF 文字提取
阶段 2:内容摘要生成(由 Claude 处理,此处跳过)
阶段 3:Word 文档生成
"""
os.makedirs(output_dir, exist_ok=True)
extracted_txt = os.path.join(output_dir, "extracted.txt")
final_docx = os.path.join(output_dir, "report.docx")
# 阶段 1:提取 PDF 文字
print("阶段 1:正在提取 PDF 文字...")
step1 = run_script(
f"{SKILLS_BASE}/pdf-reading/scripts/extract.py",
["--input", pdf_path, "--output", extracted_txt]
)
if step1.get("status") == "error":
return {"status": "error", "stage": 1, "message": step1["message"]}
print(f" 已提取 {step1.get('pages', '?')} 页")
# 阶段 3:生成 Word 报告(摘要由 Claude 在调用此脚本前生成并传入)
summary_json = os.path.join(output_dir, "summary.json")
if not os.path.exists(summary_json):
return {"status": "error", "stage": 3,
"message": "缺少 summary.json,请先完成阶段 2"}
print("阶段 3:正在生成 Word 报告...")
step3 = run_script(
f"{SKILLS_BASE}/docx/scripts/generate.py",
["--data", summary_json, "--output", final_docx]
)
if step3.get("status") == "error":
return {"status": "error", "stage": 3, "message": step3["message"]}
return {"status": "success", "output": final_docx}
if __name__ == "__main__":
pdf_path = sys.argv[1] if len(sys.argv) > 1 else ""
output_dir = sys.argv[2] if len(sys.argv) > 2 else "/home/claude/pipeline_output"
result = run_pipeline(pdf_path, output_dir)
print(json.dumps(result, ensure_ascii=False, indent=2))
# 编排多个 Skill 脚本的处理管道
import subprocess
import sys
import os
import json
SKILLS_BASE = "/mnt/skills/public"
def run_script(script_path: str, args: list) -> dict:
"""运行指定路径的脚本并返回 JSON 结果"""
cmd = [sys.executable, script_path] + args
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if result.returncode != 0:
return {"status": "error", "message": result.stderr}
try:
return json.loads(result.stdout)
except json.JSONDecodeError:
return {"status": "success", "raw_output": result.stdout}
def run_pipeline(pdf_path: str, output_dir: str) -> dict:
"""
执行完整的 PDF → Word 报告生成管道
阶段 1:PDF 文字提取
阶段 2:内容摘要生成(由 Claude 处理,此处跳过)
阶段 3:Word 文档生成
"""
os.makedirs(output_dir, exist_ok=True)
extracted_txt = os.path.join(output_dir, "extracted.txt")
final_docx = os.path.join(output_dir, "report.docx")
# 阶段 1:提取 PDF 文字
print("阶段 1:正在提取 PDF 文字...")
step1 = run_script(
f"{SKILLS_BASE}/pdf-reading/scripts/extract.py",
["--input", pdf_path, "--output", extracted_txt]
)
if step1.get("status") == "error":
return {"status": "error", "stage": 1, "message": step1["message"]}
print(f" 已提取 {step1.get('pages', '?')} 页")
# 阶段 3:生成 Word 报告(摘要由 Claude 在调用此脚本前生成并传入)
summary_json = os.path.join(output_dir, "summary.json")
if not os.path.exists(summary_json):
return {"status": "error", "stage": 3,
"message": "缺少 summary.json,请先完成阶段 2"}
print("阶段 3:正在生成 Word 报告...")
step3 = run_script(
f"{SKILLS_BASE}/docx/scripts/generate.py",
["--data", summary_json, "--output", final_docx]
)
if step3.get("status") == "error":
return {"status": "error", "stage": 3, "message": step3["message"]}
return {"status": "success", "output": final_docx}
if __name__ == "__main__":
pdf_path = sys.argv[1] if len(sys.argv) > 1 else ""
output_dir = sys.argv[2] if len(sys.argv) > 2 else "/home/claude/pipeline_output"
result = run_pipeline(pdf_path, output_dir)
print(json.dumps(result, ensure_ascii=False, indent=2))
避免循环依赖
当 Skill A 依赖 Skill B、Skill B 又依赖 Skill A 时,会形成循环依赖,导致无法执行。
设计 Skill 组合时,应保持单向依赖关系。
共享工具函数
多个 Skill 都需要的通用函数,可以提取为公共模块,避免重复实现。
实例
# 文件路径:/home/claude/shared/file_utils.py
# 多个 Skill 共享的文件工具函数
import os
UPLOAD_DIR = "/mnt/user-data/uploads"
OUTPUT_DIR = "/mnt/user-data/outputs"
def resolve_upload(filename: str) -> str:
"""将文件名解析为完整的上传路径"""
if os.path.isabs(filename):
return filename # 已经是绝对路径
return os.path.join(UPLOAD_DIR, filename)
def resolve_output(filename: str) -> str:
"""将文件名解析为完整的输出路径,并确保目录存在"""
if os.path.isabs(filename):
path = filename
else:
path = os.path.join(OUTPUT_DIR, filename)
os.makedirs(os.path.dirname(path), exist_ok=True)
return path
# 各 Skill 脚本中使用:
# import sys; sys.path.insert(0, "/home/claude/shared")
# from file_utils import resolve_upload, resolve_output
# 多个 Skill 共享的文件工具函数
import os
UPLOAD_DIR = "/mnt/user-data/uploads"
OUTPUT_DIR = "/mnt/user-data/outputs"
def resolve_upload(filename: str) -> str:
"""将文件名解析为完整的上传路径"""
if os.path.isabs(filename):
return filename # 已经是绝对路径
return os.path.join(UPLOAD_DIR, filename)
def resolve_output(filename: str) -> str:
"""将文件名解析为完整的输出路径,并确保目录存在"""
if os.path.isabs(filename):
path = filename
else:
path = os.path.join(OUTPUT_DIR, filename)
os.makedirs(os.path.dirname(path), exist_ok=True)
return path
# 各 Skill 脚本中使用:
# import sys; sys.path.insert(0, "/home/claude/shared")
# from file_utils import resolve_upload, resolve_output
