Agent Skills 和工具使用：构建专业化 AI 能力

引言

Agent Skills 是 Anthropic 推出的革命性功能，它允许开发者构建专业化、模块化、可重用的 Agent 能力。本文将详细介绍如何使用 Agent Skills 和设计高效的工具系统。

什么是 Agent Skills

Agent Skills 是一种组织 Agent 能力的新方式：

专业化能力包：将特定领域的知识和工具组织在一起
动态发现：Agent 可以自动发现可用的 Skills
按需加载：只在需要时加载相关 Skills
可重用：在不同 Agent 和项目间共享

Skills 的结构

skills/
├── code-review/
│   ├── instructions.md        # Skill 说明
│   ├── checklist.json         # 检查清单
│   ├── examples/              # 示例
│   │   ├── good-example.md
│   │   └── bad-example.md
│   └── tools/                 # 相关工具
│       └── linter-config.json
├── data-analysis/
│   ├── instructions.md
│   ├── templates/
│   │   └── analysis-report.md
│   └── scripts/
│       ├── clean_data.py
│       └── visualize.py
└── api-testing/
    ├── instructions.md
    ├── test-cases/
    └── tools/
        └── test-runner.ts

创建 Agent Skill

1. 基础 Skill 模板

1# Code Review Skill
2
3## 目的
4提供专业的代码审查能力，帮助识别代码质量问题、安全漏洞和性能瓶颈。
5
6## 能力范围
7- 代码质量评估
8- 安全漏洞检测
9- 性能分析
10- 最佳实践检查
11
12## 使用场景
131. Pull Request 审查
142. 代码重构前的评估
153. 新人代码指导
164. 生产问题分析
17
18## 审查标准
19
20### 代码质量
21- [ ] 命名是否清晰
22- [ ] 函数是否单一职责
23- [ ] 是否有重复代码
24- [ ] 注释是否充分
25
26### 性能
27- [ ] 时间复杂度是否合理
28- [ ] 是否有不必要的计算
29- [ ] 数据库查询是否优化
30- [ ] 是否有内存泄漏风险
31
32### 安全
33- [ ] 输入是否验证
34- [ ] 是否有 SQL 注入风险
35- [ ] 敏感数据是否加密
36- [ ] 权限检查是否完整
37
38## 输出格式
39
40### 问题报告
41{
42  "severity": "high | medium | low",
43  "category": "quality | performance | security",
44  "location": "file:line",
45  "description": "问题描述",
46  "suggestion": "改进建议",
47  "example": "代码示例"
48}
49
50## 相关资源
51- OWASP Top 10: https://owasp.org/www-project-top-ten/
52- Clean Code: https://github.com/ryanmcdermott/clean-code-javascript

2. 实现 Skill 加载器

1import os
2import json
3from pathlib import Path
4
5class SkillLoader:
6    def __init__(self, skills_dir="./skills"):
7        self.skills_dir = Path(skills_dir)
8        self.loaded_skills = {}
9
10    def discover_skills(self):
11        """发现所有可用的 Skills"""
12        skills = []
13
14        for skill_dir in self.skills_dir.iterdir():
15            if skill_dir.is_dir():
16                instructions_file = skill_dir / "instructions.md"
17                if instructions_file.exists():
18                    skill_info = {
19                        "name": skill_dir.name,
20                        "path": str(skill_dir),
21                        "instructions": instructions_file.read_text()
22                    }
23
24                    # 读取元数据（如果存在）
25                    metadata_file = skill_dir / "metadata.json"
26                    if metadata_file.exists():
27                        skill_info["metadata"] = json.loads(
28                            metadata_file.read_text()
29                        )
30
31                    skills.append(skill_info)
32
33        return skills
34
35    def load_skill(self, skill_name):
36        """加载特定 Skill"""
37        if skill_name in self.loaded_skills:
38            return self.loaded_skills[skill_name]
39
40        skill_path = self.skills_dir / skill_name
41        if not skill_path.exists():
42            raise ValueError(f"Skill '{skill_name}' 不存在")
43
44        # 读取指令
45        instructions = (skill_path / "instructions.md").read_text()
46
47        # 读取工具
48        tools = self._load_tools(skill_path / "tools")
49
50        # 读取示例
51        examples = self._load_examples(skill_path / "examples")
52
53        skill = {
54            "name": skill_name,
55            "instructions": instructions,
56            "tools": tools,
57            "examples": examples
58        }
59
60        self.loaded_skills[skill_name] = skill
61        return skill
62
63    def _load_tools(self, tools_dir):
64        """加载 Skill 相关的工具"""
65        if not tools_dir.exists():
66            return []
67
68        tools = []
69        for tool_file in tools_dir.glob("*.json"):
70            tool_config = json.loads(tool_file.read_text())
71            tools.append(tool_config)
72
73        return tools
74
75    def _load_examples(self, examples_dir):
76        """加载示例"""
77        if not examples_dir.exists():
78            return []
79
80        examples = []
81        for example_file in examples_dir.glob("*.md"):
82            examples.append({
83                "name": example_file.stem,
84                "content": example_file.read_text()
85            })
86
87        return examples
88
89# 使用
90loader = SkillLoader()
91
92# 发现所有 Skills
93available_skills = loader.discover_skills()
94print(f"发现 {len(available_skills)} 个 Skills:")
95for skill in available_skills:
96    print(f"- {skill['name']}")
97
98# 加载特定 Skill
99code_review_skill = loader.load_skill("code-review")

3. Agent 中使用 Skills

1class SkillfulAgent:
2    def __init__(self):
3        self.skill_loader = SkillLoader()
4        self.active_skills = []
5
6    async def activate_skill(self, skill_name):
7        """激活一个 Skill"""
8        skill = self.skill_loader.load_skill(skill_name)
9        self.active_skills.append(skill)
10
11        # 将 Skill 指令添加到 Agent 上下文
12        await self.add_to_context(skill["instructions"])
13
14        # 注册 Skill 工具
15        for tool in skill["tools"]:
16            await self.register_tool(tool)
17
18        return f"已激活 Skill: {skill_name}"
19
20    async def execute_with_skill(self, task, skill_name):
21        """使用特定 Skill 执行任务"""
22        # 激活 Skill
23        await self.activate_skill(skill_name)
24
25        # 构建提示词
26        prompt = self._build_prompt_with_skill(task, skill_name)
27
28        # 执行任务
29        response = await self.call_claude(prompt)
30
31        return response
32
33    def _build_prompt_with_skill(self, task, skill_name):
34        skill = next(s for s in self.active_skills if s["name"] == skill_name)
35
36        prompt = f"""
37        <skill>
38        {skill["instructions"]}
39        </skill>
40
41        <examples>
42        {self._format_examples(skill["examples"])}
43        </examples>
44
45        <task>
46        {task}
47        </task>
48
49        请使用上述 Skill 的指导完成任务。
50        """
51
52        return prompt
53
54# 使用
55agent = SkillfulAgent()
56
57# 审查代码
58review_result = await agent.execute_with_skill(
59    task="审查这个 Pull Request",
60    skill_name="code-review"
61)

工具设计最佳实践

1. 工具定义规范

1def create_tool_definition(
2    name: str,
3    description: str,
4    parameters: dict,
5    examples: list = None
6):
7    """创建标准化的工具定义"""
8    return {
9        "name": name,
10        "description": f"""
11        {description}
12
13        使用场景：
14        {parameters.get('use_cases', '通用场景')}
15
16        最佳实践：
17        {parameters.get('best_practices', '按照文档使用')}
18
19        注意事项：
20        {parameters.get('warnings', '无')}
21
22        示例：
23        {format_examples(examples) if examples else '参见文档'}
24        """,
25        "input_schema": {
26            "type": "object",
27            "properties": parameters.get("properties", {}),
28            "required": parameters.get("required", [])
29        }
30    }
31
32# 示例：创建搜索工具
33search_tool = create_tool_definition(
34    name="web_search",
35    description="搜索互联网获取实时信息",
36    parameters={
37        "use_cases": """
38        - 查找最新资讯
39        - 研究技术文档
40        - 验证事实数据
41        """,
42        "best_practices": """
43        - 使用具体的搜索词
44        - 包含时间范围
45        - 避免过于宽泛
46        """,
47        "warnings": """
48        - 搜索结果可能不准确
49        - 需要人工验证重要信息
50        """,
51        "properties": {
52            "query": {
53                "type": "string",
54                "description": "搜索查询"
55            },
56            "num_results": {
57                "type": "integer",
58                "default": 5
59            }
60        },
61        "required": ["query"]
62    },
63    examples=[
64        {
65            "query": "Claude 4 发布时间",
66            "num_results": 3
67        }
68    ]
69)

2. 工具错误处理

1class RobustTool:
2    def __init__(self, tool_func, max_retries=3):
3        self.tool_func = tool_func
4        self.max_retries = max_retries
5        self.call_history = []
6
7    async def execute(self, **kwargs):
8        """执行工具，包含错误处理"""
9        for attempt in range(self.max_retries):
10            try:
11                # 验证输入
12                validated_input = self.validate_input(kwargs)
13
14                # 记录调用
15                self.call_history.append({
16                    "attempt": attempt + 1,
17                    "input": validated_input,
18                    "timestamp": time.time()
19                })
20
21                # 执行工具
22                result = await self.tool_func(**validated_input)
23
24                # 验证输出
25                validated_output = self.validate_output(result)
26
27                return {
28                    "success": True,
29                    "result": validated_output,
30                    "attempt": attempt + 1
31                }
32
33            except ValidationError as e:
34                # 输入验证失败，不重试
35                return {
36                    "success": False,
37                    "error": f"输入验证失败: {str(e)}",
38                    "error_type": "validation"
39                }
40
41            except Exception as e:
42                # 执行失败，可能重试
43                if attempt < self.max_retries - 1:
44                    # 等待后重试
45                    await asyncio.sleep(2 ** attempt)
46                    continue
47                else:
48                    return {
49                        "success": False,
50                        "error": f"执行失败: {str(e)}",
51                        "error_type": "execution",
52                        "attempts": self.max_retries
53                    }
54
55    def validate_input(self, kwargs):
56        """验证输入参数"""
57        # 实现输入验证逻辑
58        return kwargs
59
60    def validate_output(self, result):
61        """验证输出结果"""
62        # 实现输出验证逻辑
63        return result

3. 工具组合

1class ToolChain:
2    """将多个工具链接成流水线"""
3
4    def __init__(self, tools):
5        self.tools = tools
6
7    async def execute(self, initial_input):
8        """顺序执行所有工具"""
9        current_input = initial_input
10
11        for i, tool in enumerate(self.tools):
12            print(f"执行工具 {i+1}/{len(self.tools)}: {tool.name}")
13
14            # 执行工具
15            result = await tool.execute(**current_input)
16
17            # 检查是否成功
18            if not result.get("success"):
19                return {
20                    "success": False,
21                    "failed_at": i,
22                    "error": result.get("error")
23                }
24
25            # 将输出作为下一个工具的输入
26            current_input = result["result"]
27
28        return {
29            "success": True,
30            "final_result": current_input
31        }
32
33# 示例：创建工具链
34tool_chain = ToolChain([
35    FetchWebpageTool(),
36    ExtractContentTool(),
37    SummarizeTool(),
38    TranslateTool()
39])
40
41# 执行
42result = await tool_chain.execute({
43    "url": "https://example.com/article"
44})

高级工具模式

1. 条件工具使用

1class ConditionalToolRouter:
2    """根据条件选择工具"""
3
4    def __init__(self, tools_map):
5        self.tools_map = tools_map
6
7    async def route_and_execute(self, context, task):
8        """根据上下文路由到合适的工具"""
9
10        # 分析任务类型
11        task_type = await self.analyze_task_type(task)
12
13        # 选择工具
14        if task_type not in self.tools_map:
15            return {
16                "success": False,
17                "error": f"没有找到处理 '{task_type}' 的工具"
18            }
19
20        tool = self.tools_map[task_type]
21
22        # 执行工具
23        return await tool.execute(context=context, task=task)
24
25    async def analyze_task_type(self, task):
26        """分析任务类型"""
27        prompt = f"""
28        分析以下任务的类型：
29
30        {task}
31
32        可能的类型：
33        - web_search: 需要搜索网络
34        - data_analysis: 需要分析数据
35        - code_generation: 需要生成代码
36        - file_operation: 需要操作文件
37
38        只返回类型名称。
39        """
40
41        response = await call_claude(prompt)
42        return response.strip()
43
44# 使用
45router = ConditionalToolRouter({
46    "web_search": WebSearchTool(),
47    "data_analysis": DataAnalysisTool(),
48    "code_generation": CodeGenerationTool(),
49    "file_operation": FileOperationTool()
50})
51
52result = await router.route_and_execute(
53    context=current_context,
54    task="搜索 Claude 4 的最新功能"
55)

2. 并行工具执行

1class ParallelToolExecutor:
2    """并行执行多个工具"""
3
4    def __init__(self, tools):
5        self.tools = tools
6
7    async def execute_all(self, inputs):
8        """并行执行所有工具"""
9        tasks = [
10            tool.execute(**input_data)
11            for tool, input_data in zip(self.tools, inputs)
12        ]
13
14        results = await asyncio.gather(*tasks, return_exceptions=True)
15
16        # 处理结果
17        processed_results = []
18        for i, result in enumerate(results):
19            if isinstance(result, Exception):
20                processed_results.append({
21                    "tool_index": i,
22                    "success": False,
23                    "error": str(result)
24                })
25            else:
26                processed_results.append({
27                    "tool_index": i,
28                    "success": True,
29                    "result": result
30                })
31
32        return processed_results
33
34# 示例：并行分析
35executor = ParallelToolExecutor([
36    SentimentAnalysisTool(),
37    KeywordExtractionTool(),
38    LanguageDetectionTool(),
39    TopicModelingTool()
40])
41
42results = await executor.execute_all([
43    {"text": article_text},
44    {"text": article_text},
45    {"text": article_text},
46    {"text": article_text}
47])

3. 自适应工具选择

1class AdaptiveToolSelector:
2    """根据历史表现自适应选择工具"""
3
4    def __init__(self, tools):
5        self.tools = tools
6        self.performance_history = {
7            tool.name: {"success": 0, "failure": 0, "avg_time": 0}
8            for tool in tools
9        }
10
11    async def select_and_execute(self, task):
12        """选择最佳工具并执行"""
13
14        # 让 Agent 评估每个工具
15        evaluations = await self.evaluate_tools(task)
16
17        # 结合历史表现评分
18        scored_tools = []
19        for tool_name, ai_score in evaluations.items():
20            history = self.performance_history[tool_name]
21
22            # 计算综合评分
23            success_rate = (
24                history["success"] /
25                (history["success"] + history["failure"])
26                if (history["success"] + history["failure"]) > 0
27                else 0.5
28            )
29
30            combined_score = 0.6 * ai_score + 0.4 * success_rate
31
32            scored_tools.append((tool_name, combined_score))
33
34        # 选择得分最高的工具
35        best_tool_name = max(scored_tools, key=lambda x: x[1])[0]
36        best_tool = next(t for t in self.tools if t.name == best_tool_name)
37
38        # 执行并记录性能
39        start_time = time.time()
40        try:
41            result = await best_tool.execute(task=task)
42            success = True
43        except Exception as e:
44            result = {"error": str(e)}
45            success = False
46
47        execution_time = time.time() - start_time
48
49        # 更新历史
50        self.update_performance(best_tool_name, success, execution_time)
51
52        return result
53
54    async def evaluate_tools(self, task):
55        """让 AI 评估工具适用性"""
56        tool_descriptions = "\n".join([
57            f"- {tool.name}: {tool.description}"
58            for tool in self.tools
59        ])
60
61        prompt = f"""
62        任务：{task}
63
64        可用工具：
65        {tool_descriptions}
66
67        请为每个工具评分（0-1），表示它处理此任务的适合程度。
68
69        输出 JSON 格式：
70        {{
71            "tool_name": score,
72            ...
73        }}
74        """
75
76        response = await call_claude(prompt)
77        return json.loads(response)
78
79    def update_performance(self, tool_name, success, execution_time):
80        """更新工具性能记录"""
81        history = self.performance_history[tool_name]
82
83        if success:
84            history["success"] += 1
85        else:
86            history["failure"] += 1
87
88        # 更新平均执行时间
89        total_executions = history["success"] + history["failure"]
90        history["avg_time"] = (
91            (history["avg_time"] * (total_executions - 1) + execution_time)
92            / total_executions
93        )

实战案例：数据分析 Skill

1# skills/data-analysis/instructions.md
2"""
3# Data Analysis Skill
4
5## 目的
6提供全面的数据分析能力，从数据清洗到可视化。
7
8## 工作流程
91. 数据加载和探索
102. 数据清洗和预处理
113. 统计分析
124. 可视化
135. 生成报告
14
15## 可用工具
16- load_data: 加载各种格式的数据
17- clean_data: 清洗和预处理
18- analyze_stats: 统计分析
19- create_visualization: 创建图表
20- generate_report: 生成分析报告
21"""
22
23class DataAnalysisSkill:
24    def __init__(self):
25        self.tools = {
26            "load_data": LoadDataTool(),
27            "clean_data": CleanDataTool(),
28            "analyze_stats": StatisticalAnalysisTool(),
29            "create_visualization": VisualizationTool(),
30            "generate_report": ReportGenerationTool()
31        }
32
33    async def analyze_dataset(self, dataset_path):
34        """完整的数据分析流程"""
35
36        # 1. 加载数据
37        print("加载数据...")
38        data = await self.tools["load_data"].execute(path=dataset_path)
39
40        # 2. 数据探索
41        print("探索数据...")
42        exploration_report = await self.explore_data(data)
43
44        # 3. 数据清洗
45        print("清洗数据...")
46        cleaned_data = await self.tools["clean_data"].execute(data=data)
47
48        # 4. 统计分析
49        print("统计分析...")
50        stats = await self.tools["analyze_stats"].execute(data=cleaned_data)
51
52        # 5. 创建可视化
53        print("创建可视化...")
54        visualizations = await self.create_visualizations(cleaned_data, stats)
55
56        # 6. 生成报告
57        print("生成报告...")
58        report = await self.tools["generate_report"].execute(
59            data=cleaned_data,
60            stats=stats,
61            visualizations=visualizations
62        )
63
64        return report
65
66    async def explore_data(self, data):
67        """探索数据特征"""
68        prompt = f"""
69        请探索这个数据集并提供概述：
70
71        数据形状：{data.shape}
72        列名：{list(data.columns)}
73        数据类型：{data.dtypes.to_dict()}
74        缺失值：{data.isnull().sum().to_dict()}
75
76        请提供：
77        1. 数据集概述
78        2. 潜在的数据质量问题
79        3. 建议的分析方向
80        """
81
82        response = await call_claude(prompt)
83        return response
84
85    async def create_visualizations(self, data, stats):
86        """创建多个可视化图表"""
87
88        # 决定需要哪些图表
89        viz_plan = await self.plan_visualizations(data, stats)
90
91        # 并行创建所有图表
92        viz_tasks = [
93            self.tools["create_visualization"].execute(
94                data=data,
95                type=viz["type"],
96                config=viz["config"]
97            )
98            for viz in viz_plan
99        ]
100
101        visualizations = await asyncio.gather(*viz_tasks)
102
103        return visualizations

工具生态系统

常用工具类别

数据获取工具
- Web搜索
- API 调用
- 数据库查询
- 文件读取
数据处理工具
- 数据清洗
- 格式转换
- 数据验证
- 数据聚合
分析工具
- 统计分析
- 机器学习
- 文本分析
- 图像处理
输出工具
- 报告生成
- 可视化
- 文件写入
- 通知发送

构建工具库

1class ToolRegistry:
2    """工具注册中心"""
3
4    def __init__(self):
5        self.tools = {}
6        self.categories = {}
7
8    def register(self, tool, category=None):
9        """注册工具"""
10        self.tools[tool.name] = tool
11
12        if category:
13            if category not in self.categories:
14                self.categories[category] = []
15            self.categories[category].append(tool.name)
16
17    def get_tool(self, name):
18        """获取工具"""
19        if name not in self.tools:
20            raise ValueError(f"工具 '{name}' 未注册")
21        return self.tools[name]
22
23    def list_tools(self, category=None):
24        """列出工具"""
25        if category:
26            if category not in self.categories:
27                return []
28            return [
29                self.tools[name]
30                for name in self.categories[category]
31            ]
32        return list(self.tools.values())
33
34    def search_tools(self, query):
35        """搜索工具"""
36        results = []
37        for tool in self.tools.values():
38            if query.lower() in tool.name.lower() or \
39               query.lower() in tool.description.lower():
40                results.append(tool)
41        return results
42
43# 使用
44registry = ToolRegistry()
45
46# 注册工具
47registry.register(WebSearchTool(), category="data_acquisition")
48registry.register(DatabaseQueryTool(), category="data_acquisition")
49registry.register(DataCleaningTool(), category="data_processing")
50registry.register(VisualizationTool(), category="output")
51
52# 查找工具
53search_tools = registry.search_tools("search")
54acquisition_tools = registry.list_tools(category="data_acquisition")

最佳实践总结

Skills 设计

单一职责：每个 Skill 聚焦一个专业领域
完整文档：提供清晰的说明和示例
模块化：Skill 之间低耦合
可测试：提供测试用例
版本管理：追踪 Skill 版本变化

工具设计

清晰描述：工具功能和使用场景
健壮性：错误处理和重试机制
性能优化：缓存和批处理
安全性：输入验证和权限检查
可监控：记录调用和性能指标

Agent 集成

按需加载：只加载必要的 Skills
上下文感知：根据任务选择 Skills
工具链接：组合多个工具完成复杂任务
持续学习：从使用中优化 Skills 和工具

总结

Agent Skills 和工具系统是构建强大 AI Agent 的基础。通过模块化、专业化的设计，我们可以创建可重用、可维护的 Agent 能力库，大幅提升 Agent 开发效率和质量。