1. 引言:为什么需要掌握 AI Agent Skill

随着大语言模型能力的持续提升,AI Agent 已经从简单的对话机器人演变为能够自主规划、调用工具、执行复杂任务的智能体。而 Skill(技能)正是赋予 Agent 领域能力的关键机制。本文将从基础概念出发,逐步深入到高级实战,帮助你系统掌握 AI Agent Skill 的设计、开发与调优方法。

无论你是刚接触 Agent 开发的初学者,还是希望提升 Agent 复杂任务处理能力的进阶开发者,本文都会提供可落地的代码示例和工程实践建议。

2. Skill 基础概念

2.1 什么是 Skill

Skill 是 Agent 可复用的能力单元,它将特定领域的知识、工具调用逻辑和提示词模板封装在一起。当 Agent 遇到匹配的任务时,会自动加载对应的 Skill 来完成任务。

一个完整的 Skill 通常包含以下组成部分:

  • 触发条件:定义何时启用该 Skill,通常基于任务描述或用户意图匹配。
  • 指令模板:指导模型如何执行任务的提示词,包含步骤、约束和输出格式。
  • 工具调用:Skill 内部可编排一个或多个外部工具(如搜索、代码执行、API 调用)。
  • 上下文管理:定义需要收集和传递的上下文信息。

2.2 Skill 与普通提示词的区别

普通提示词是一次性的指令文本,而 Skill 是结构化的、可复用的能力封装。Skill 具备以下优势:

  • 可复用性:同一 Skill 可在多个 Agent 或任务中复用。
  • 可组合性:多个 Skill 可以组合成更复杂的流程。
  • 可维护性:技能逻辑集中管理,便于迭代优化。
  • 可测试性:每个 Skill 可以独立测试和验证。

3. 环境准备与工具链

3.1 开发环境搭建

本文的实战示例基于 Python 3.10+ 和 LangChain 框架。首先安装必要的依赖:

pip install langchain langchain-openai langchain-community
pip install openai python-dotenv

创建项目目录结构:

agent-skill-project/
├── skills/
│   ├── web_search/
│   │   ├── SKILL.md
│   │   └── tools.py
│   ├── code_runner/
│   │   ├── SKILL.md
│   │   └── tools.py
│   └── data_analysis/
│       ├── SKILL.md
│       └── tools.py
├── agent.py
├── config.py
└── .env

3.2 配置环境变量

.env 文件中配置 API 密钥:

OPENAI_API_KEY=your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o

4. 第一个 Skill:从零开始

4.1 定义 Skill 元数据

每个 Skill 以目录形式组织,核心是 SKILL.md 文件。下面创建一个网页搜索 Skill:

---
name: web_search
description: 执行网络搜索并返回结构化结果,适用于查询最新信息、新闻、文档等场景。
version: 1.0.0
author: your-name
triggers:
  - 搜索
  - 查询
  - 查找资料
  - 最新信息
---
Web Search Skill
执行步骤
分析用户查询意图,提取关键词。
调用 search_web 工具执行搜索。
对结果进行去重和相关性排序。
返回前 5 条最相关的结果,包含标题、链接和摘要。
注意事项
搜索关键词应简洁,避免过长。
优先选择权威来源(官方文档、学术网站)。
如果结果不相关,尝试改写关键词重新搜索。

4.2 实现工具函数

tools.py 中实现搜索工具:

import requests
from typing import List, Dict
def search_web(query: str, num_results: int = 5) -> List[Dict]:
"""执行网络搜索,返回结构化结果列表。"""
# 这里以 DuckDuckGo 为例,实际可替换为其他搜索 API
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1
}
try:
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()
    data = response.json()
results = []
for topic in data.get("RelatedTopics", [])[:num_results]:
    if "Text" in topic:
        results.append({
            "title": topic.get("Text", "").split(" - ")[0],
            "url": topic.get("FirstURL", ""),
            "snippet": topic.get("Text", "")
        })
return results
except Exception as e:
return [{"error": f"搜索失败: {str(e)}"}]
def format_results(results: List[Dict]) -> str:
"""将搜索结果格式化为可读文本。"""
if not results:
return "未找到相关结果。"
lines = []
for i, r in enumerate(results, 1):
if "error" in r:
return r["error"]
lines.append(f"{i}. {r['title']}\n   {r['url']}\n   {r['snippet']}")
return "\n\n".join(lines)</code></pre>
4.3 将 Skill 接入 Agent
创建主 Agent 程序,加载并调用 Skill:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import Tool
from skills.web_search.tools import search_web, format_results
load_dotenv()
def create_agent():
"""创建带 Skill 能力的 Agent。"""
llm = ChatOpenAI(
model=os.getenv("MODEL_NAME", "gpt-4o"),
temperature=0.3
)
将 Skill 中的工具注册到 Agent
tools = [
Tool(
name="web_search",
func=lambda q: format_results(search_web(q)),
description="执行网络搜索,输入为查询关键词,返回结构化搜索结果。"
)
]
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个智能助手,可以调用工具完成任务。请根据用户需求选择合适的工具。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools, verbose=True)
if name == "main":
agent = create_agent()
result = agent.invoke({"input": "帮我搜索一下 2025 年 AI Agent 的最新发展趋势"})
print(result["output"])
5. Skill 高级设计模式
5.1 多工具编排
复杂任务往往需要多个工具协同。下面创建一个数据分析 Skill,它同时使用代码执行和文件读写工具:
skills/data_analysis/tools.py
import pandas as pd
import json
from typing import Any, Dict
def load_dataset(file_path: str) -> Dict[str, Any]:
"""加载 CSV 或 JSON 格式的数据集。"""
try:
if file_path.endswith(".csv"):
df = pd.read_csv(file_path)
elif file_path.endswith(".json"):
df = pd.read_json(file_path)
else:
return {"error": "不支持的文件格式"}
return {
"columns": list(df.columns),
"shape": df.shape,
"head": df.head(5).to_dict(orient="records"),
"dtypes": df.dtypes.astype(str).to_dict()
}
except Exception as e:
return {"error": f"加载失败: {str(e)}"}
def analyze_column(df_data: Dict, column: str) -> Dict[str, Any]:
"""对指定列进行统计分析。"""
try:
df = pd.DataFrame(df_data["head"])
series = pd.Series(df[column])
return {
"mean": float(series.mean()) if pd.api.types.is_numeric_dtype(series) else None,
"unique_values": series.nunique(),
"missing": int(series.isna().sum()),
"sample": series.head(3).tolist()
}
except Exception as e:
return {"error": f"分析失败: {str(e)}"}
5.2 条件分支与决策
高级 Skill 需要根据中间结果动态调整执行路径。在 SKILL.md 中定义分支逻辑:
name: smart_analysis
description: 智能数据分析,根据数据特征自动选择分析策略。
version: 2.0.0
Smart Analysis Skill
执行流程
加载数据集,获取基本结构信息。
判断数据类型:
如果包含数值列,执行统计分析和相关性分析。
如果包含文本列,执行关键词提取和情感分析。
如果包含时间列,执行趋势分析。
根据分析结果生成可视化建议。
输出结构化分析报告。
决策规则
数值列占比 > 60%:优先统计分析。
文本列占比 > 40%:优先文本分析。
时间列存在:增加趋势分析。
5.3 Skill 组合与链式调用
将多个 Skill 串联成工作流,实现复杂任务自动化:
from langchain.tools import Tool
from skills.web_search.tools import search_web, format_results
from skills.code_runner.tools import run_python_code
def research_and_summarize(topic: str) -> str:
"""组合 Skill:搜索 + 代码分析 + 总结。"""
第一步:搜索资料
search_results = format_results(search_web(topic, num_results=10))
第二步:用代码提取关键词
code = f"""
import re
from collections import Counter
text = """{search_results}"""
words = re.findall(r'\w+', text.lower())
stopwords = {{'the', 'a', 'an', 'and', 'or', 'for', 'with'}}
keywords = [w for w in words if w not in stopwords and len(w) > 3]
top_keywords = Counter(keywords).most_common(10)
print(top_keywords)
"""
analysis = run_python_code(code)
第三步:返回组合结果
return f"搜索到 {len(search_results)} 条结果,关键词分析:{analysis}"
注册为组合工具
combined_tool = Tool(
name="research_and_summarize",
func=research_and_summarize,
description="搜索资料并进行关键词分析,返回综合结果。"
)
6. 实战案例:构建智能客服 Agent
6.1 需求分析
本节构建一个完整的智能客服 Agent,它需要处理订单查询、退换货、产品咨询等常见问题。我们将设计三个 Skill:
order_query:查询订单状态和物流信息。
return_request:处理退换货申请。
product_info:提供产品参数和库存信息。
6.2 实现订单查询 Skill
skills/order_query/tools.py
import json
from datetime import datetime
from typing import Dict, Optional
模拟订单数据库
ORDERS_DB = {
"ORD2025001": {
"status": "已发货",
"items": ["无线鼠标", "机械键盘"],
"total": 599.00,
"shipping": "顺丰速运",
"tracking": "SF1234567890",
"estimated_delivery": "2025-03-20"
},
"ORD2025002": {
"status": "待付款",
"items": ["显示器支架"],
"total": 199.00,
"shipping": None,
"tracking": None,
"estimated_delivery": None
}
}
def query_order(order_id: str) -> Dict:
"""查询订单状态。"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return {"error": f"未找到订单 {order_id},请确认订单号是否正确。"}
result = {
"订单号": order_id.upper(),
"状态": order["status"],
"商品": ", ".join(order["items"]),
"金额": f"¥{order['total']:.2f}"
}
if order["tracking"]:
result["物流公司"] = order["shipping"]
result["运单号"] = order["tracking"]
result["预计送达"] = order["estimated_delivery"]
return result
def format_order_response(order_info: Dict) -> str:
"""格式化订单查询结果。"""
if "error" in order_info:
return order_info["error"]
lines = [f"您的订单信息如下:"]
for key, value in order_info.items():
lines.append(f"- {key}:{value}")
if order_info.get("状态") == "已发货":
lines.append("\n如需查询物流详情,请提供运单号。")
elif order_info.get("状态") == "待付款":
lines.append("\n请尽快完成付款,订单将在付款后 24 小时内发货。")
return "\n".join(lines)</code></pre>
6.3 实现退换货 Skill
skills/return_request/tools.py
from typing import Dict, List
RETURN_POLICY = {
"window_days": 7,
"conditions": [
"商品未经使用,包装完好",
"不影响二次销售",
"非定制类商品"
],
"process": [
"提交退换货申请",
"审核通过后寄回商品",
"仓库验收(1-3 个工作日)",
"退款原路返回(3-5 个工作日)"
]
}
def check_return_eligibility(order_id: str, item: str) -> Dict:
"""检查退换货资格。"""
模拟检查逻辑
eligible = True
reasons = []
if not order_id.startswith("ORD"):
eligible = False
reasons.append("订单号格式不正确")
if item in ["定制键盘", "已拆封耳机"]:
eligible = False
reasons.append("该商品不支持退换货")
return {
"eligible": eligible,
"reasons": reasons if reasons else ["符合退换货条件"],
"policy": RETURN_POLICY
}
def create_return_request(order_id: str, item: str, reason: str) -> Dict:
"""创建退换货申请。"""
eligibility = check_return_eligibility(order_id, item)
if not eligibility["eligible"]:
return {
"success": False,
"message": ";".join(eligibility["reasons"])
}
request_id = f"RET{order_id[-4:]}001"
return {
"success": True,
"request_id": request_id,
"message": f"退换货申请已提交,申请编号:{request_id}",
"next_steps": RETURN_POLICY["process"]
}</code></pre>
6.4 组装客服 Agent
customer_service_agent.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import Tool
from langchain.memory import ConversationBufferMemory
from skills.order_query.tools import query_order, format_order_response
from skills.return_request.tools import create_return_request, check_return_eligibility
from skills.product_info.tools import get_product_info
load_dotenv()
def create_customer_service_agent():
"""创建智能客服 Agent。"""
llm = ChatOpenAI(
model=os.getenv("MODEL_NAME", "gpt-4o"),
temperature=0.2
)
tools = [
Tool(
name="query_order",
func=lambda order_id: format_order_response(query_order(order_id)),
description="查询订单状态和物流信息。输入参数为订单号,格式如 ORD2025001。"
),
Tool(
name="create_return_request",
func=lambda order_id, item, reason: create_return_request(order_id, item, reason),
description="创建退换货申请。参数:订单号、商品名称、退换原因。"
),
Tool(
name="get_product_info",
func=lambda product_name: get_product_info(product_name),
description="查询产品参数、价格和库存信息。输入参数为产品名称。"
)
]
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的电商客服助手。请遵循以下规则:
用户询问订单状态时,使用 query_order 工具。
用户要求退换货时,先确认订单信息,再使用 create_return_request。
用户咨询产品时,使用 get_product_info。
回答要友好、专业,必要时提供额外帮助。
如果工具返回错误,向用户解释并引导正确操作。"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
agent = create_openai_tools_agent(llm, tools, prompt)
return AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=5
)
if name == "main":
agent = create_customer_service_agent()
测试对话
print("=== 测试 1:订单查询 ===")
response = agent.invoke({"input": "帮我查一下订单 ORD2025001 到哪了"})
print(response["output"])
print("\n=== 测试 2:退换货 ===")
response = agent.invoke({"input": "我想退掉 ORD2025002 里的显示器支架,还没付款"})
print(response["output"])
print("\n=== 测试 3:产品咨询 ===")
response = agent.invoke({"input": "你们有无线鼠标吗?多少钱?"})
print(response["output"])</code></pre>
7. Skill 性能优化
7.1 提示词优化策略
Skill 的执行效果高度依赖提示词质量。以下优化策略可以显著提升效果:
明确输出格式:在 SKILL.md 中定义结构化输出模板,减少模型自由发挥空间。
提供示例:每个 Skill 至少包含 2-3 个输入输出示例,帮助模型理解预期行为。
错误处理指引:明确工具调用失败时的降级策略和用户沟通方式。
上下文压缩:长对话中,使用摘要压缩历史消息,避免超出上下文窗口。
7.2 缓存与记忆机制
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
import hashlib
import json
启用 LLM 缓存,减少重复调用
set_llm_cache(InMemoryCache())
class SkillMemory:
"""Skill 级记忆管理,缓存工具调用结果。"""
def init(self, max_entries: int = 100):
self.cache = {}
self.max_entries = max_entries
def _key(self, tool_name: str, *args) -> str:
"""生成缓存键。"""
raw = f"{tool_name}:{json.dumps(args, ensure_ascii=False)}"
return hashlib.md5(raw.encode()).hexdigest()
def get(self, tool_name: str, *args):
"""获取缓存结果。"""
key = self._key(tool_name, *args)
return self.cache.get(key)
def set(self, tool_name: str, result, *args):
"""写入缓存,超出容量时淘汰最旧条目。"""
key = self._key(tool_name, *args)
if len(self.cache) >= self.max_entries:
oldest_key = next(iter(self.cache))
del self.cache[oldest_key]
self.cache[key] = result
def clear(self):
"""清空缓存。"""
self.cache.clear()
使用示例
memory = SkillMemory()
def cached_query_order(order_id: str):
"""带缓存的订单查询。"""
cached = memory.get("query_order", order_id)
if cached:
return cached
result = query_order(order_id)
memory.set("query_order", result, order_id)
return result
7.3 并行执行与异步优化
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict
async def run_skills_parallel(skill_calls: List[Dict]) -> List[Dict]:
"""并行执行多个 Skill 调用。"""
async def execute_one(call: Dict):
tool_name = call["tool"]
args = call.get("args", {})
# 这里根据工具名分发到对应函数
if tool_name == "web_search":
return await asyncio.to_thread(search_web, **args)
elif tool_name == "query_order":
return await asyncio.to_thread(query_order, **args)
elif tool_name == "get_product_info":
return await asyncio.to_thread(get_product_info, **args)
else:
return {"error": f"未知工具: {tool_name}"}
并发执行所有调用
results = await asyncio.gather(
*[execute_one(call) for call in skill_calls]
)
return results
使用示例
async def demo_parallel():
calls = [
{"tool": "web_search", "args": {"query": "AI Agent 最新进展"}},
{"tool": "query_order", "args": {"order_id": "ORD2025001"}},
{"tool": "get_product_info", "args": {"product_name": "无线鼠标"}}
]
results = await run_skills_parallel(calls)
for r in results:
print(r)
运行
asyncio.run(demo_parallel())
8. 测试与调试
8.1 单元测试 Skill
import unittest
from skills.order_query.tools import query_order, format_order_response
from skills.return_request.tools import check_return_eligibility
class TestOrderSkill(unittest.TestCase):
"""订单查询 Skill 单元测试。"""
def test_query_existing_order(self):
result = query_order("ORD2025001")
self.assertIn("状态", result)
self.assertEqual(result["状态"], "已发货")
def test_query_nonexistent_order(self):
result = query_order("ORD9999999")
self.assertIn("error", result)
def test_format_response(self):
result = query_order("ORD2025001")
formatted = format_order_response(result)
self.assertIn("订单号", formatted)
self.assertIn("ORD2025001", formatted)
class TestReturnSkill(unittest.TestCase):
"""退换货 Skill 单元测试。"""
def test_eligible_item(self):
result = check_return_eligibility("ORD2025001", "无线鼠标")
self.assertTrue(result["eligible"])
def test_ineligible_item(self):
result = check_return_eligibility("ORD2025001", "定制键盘")
self.assertFalse(result["eligible"])
if name == "main":
unittest.main()
8.2 调试技巧
调试 Skill 时,重点关注以下方面:
工具调用日志:开启 Agent 的 verbose 模式,观察每一步的工具调用和中间结果。
提示词追踪:记录发送给模型的完整提示词,检查 Skill 指令是否正确加载。
错误注入测试:模拟工具返回错误,验证 Agent 的降级处理逻辑。
边界条件测试:测试空输入、超长输入、特殊字符等边界情况。
9. 部署与监控
9.1 生产环境部署
deploy.py
import os
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from customer_service_agent import create_customer_service_agent
配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(name)
app = FastAPI(title="AI Agent Skill Service")
agent = create_customer_service_agent()
class ChatRequest(BaseModel):
message: str
session_id: str = "default"
class ChatResponse(BaseModel):
reply: str
session_id: str
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""处理用户消息并返回 Agent 回复。"""
try:
logger.info(f"收到消息: {request.message}")
response = agent.invoke({"input": request.message})
logger.info(f"Agent 回复: {response['output'][:100]}...")
return ChatResponse(
reply=response["output"],
session_id=request.session_id
)
except Exception as e:
logger.error(f"处理失败: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""健康检查接口。"""
return {"status": "healthy"}
if name == "main":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
9.2 监控与日志
monitoring.py
import time
import json
from datetime import datetime
from typing import Dict, Any
class SkillMonitor:
"""Skill 调用监控器。"""
def init(self):
self.metrics = {
"total_calls": 0,
"success_calls": 0,
"failed_calls": 0,
"avg_latency_ms": 0,
"tool_usage": {}
}
self._latencies = []
def record_call(self, tool_name: str, success: bool, latency_ms: float):
"""记录一次工具调用。"""
self.metrics["total_calls"] += 1
if success:
self.metrics["success_calls"] += 1
else:
self.metrics["failed_calls"] += 1
self._latencies.append(latency_ms)
self.metrics["avg_latency_ms"] = sum(self._latencies) / len(self._latencies)
if tool_name not in self.metrics["tool_usage"]:
self.metrics["tool_usage"][tool_name] = {"calls": 0, "failures": 0}
self.metrics["tool_usage"][tool_name]["calls"] += 1
if not success:
self.metrics["tool_usage"][tool_name]["failures"] += 1
def get_report(self) -> Dict[str, Any]:
"""生成监控报告。"""
return {
"timestamp": datetime.now().isoformat(),
"metrics": self.metrics,
"success_rate": (
self.metrics["success_calls"] / self.metrics["total_calls"]
if self.metrics["total_calls"] > 0 else 0
)
}
全局监控实例
monitor = SkillMonitor()
在工具调用处埋点
def monitored_call(tool_name: str, func, *args, **kwargs):
"""带监控的工具调用包装器。"""
start = time.time()
try:
result = func(*args, **kwargs)
monitor.record_call(tool_name, True, (time.time() - start) * 1000)
return result
except Exception as e:
monitor.record_call(tool_name, False, (time.time() - start) * 1000)
raise e
10. 最佳实践与常见陷阱
10.1 设计最佳实践
单一职责:每个 Skill 只做一件事,避免大而全的 Skill。
明确边界:清晰定义 Skill 的输入输出和触发条件,避免与其他 Skill 冲突。
版本管理:使用语义化版本号,记录变更日志,便于回滚。
渐进式复杂度:先实现最小可用版本,再逐步增加高级功能。
10.2 常见陷阱与解决方案
陷阱
表现
解决方案
提示词过长
模型忽略部分指令,输出不稳定
精简指令,将详细规则放入工具描述
工具调用循环
Agent 反复调用同一工具不退出
设置 max_iterations,增加退出条件
上下文溢出
长对话后报错或遗忘早期信息
使用记忆压缩、摘要或向量检索
错误处理缺失
工具异常导致整个流程失败
每个工具增加 try-except,返回友好错误
Skill 冲突
多个 Skill 同时匹配同一任务
设置优先级,细化触发条件
11. 总结与进阶方向
本文从基础概念到高级实战,系统介绍了 AI Agent Skill 的设计、开发、优化和部署方法。通过完整的代码示例,你可以快速上手构建自己的 Skill 体系。
后续进阶方向包括:
多 Agent 协作:设计多个专业 Agent 协同完成复杂任务。
Skill 自动生成:让 Agent 根据任务描述自动生成新 Skill。
强化学习优化:基于用户反馈自动调整 Skill 参数。
跨框架兼容:设计框架无关的 Skill 标准,便于迁移。
掌握 AI Agent Skill 的核心能力,将帮助你在智能化应用开发中占据先机。建议从本文的客服 Agent 案例入手,逐步扩展到你的业务场景中。

更多推荐