在构建现代AI Agent的过程中，一个至关重要的组成部分就是其“技能”（Skills）系统。技能，本质上是AI Agent能够调用的外部函数或工具，它赋予了Agent超越其固有语言模型能力的无限可能性。无论是查询天气、发送邮件、操作数据库，还是与其他软件进行交互，都离不开一个强大而灵活的技能框架。本文将深入探讨如何使用Python和流行的AI开发框架，构建一个可扩展的、健壮的技能系统。
什么是AI Agent的技能（Skills）？
想象一下，你有一个非常聪明但无法离开房间的AI朋友。它知道世界上所有的事情，能写诗作赋，也能进行复杂的逻辑推理，但它无法帮你打开窗户、倒杯水或者查看今天的天气预报。技能（Skills）就是为这个AI朋友提供的“手臂”和“腿”，让它们能够与外部世界进行交互。
在技术层面，这通常通过Function Calling（函数调用）机制来实现。开发者定义一系列函数，描述它们的功能、参数和返回值。当AI在思考过程中认为需要调用某个技能时，它会生成一个包含函数名和所需参数的JSON请求，交由系统执行该函数，然后将执行结果返回给AI，供其进一步处理。
设计原则：构建一个模块化的技能系统
一个优秀的技能系统应遵循以下原则：
- 可插拔（Pluggable）: 新的技能应该能轻松添加进来，而不影响现有系统。
- 可描述（Descriptive）: 每个技能都需要有清晰的、AI能理解的描述，说明它的用途和参数。
- 安全（Secure）: 系统需要对技能的执行进行严格的安全检查，防止恶意调用。
- 可追踪（Traceable）: 需要记录技能的调用过程，便于调试和审计。
实践：从零构建一个基础技能系统
让我们从一个简单的例子开始，逐步构建一个功能完备的技能系统。
第一步：定义技能接口
首先，我们定义一个基类来规范所有技能的行为。
from abc import ABC, abstractmethod
import inspect
import json
class Skill(ABC):
"""技能基类，所有自定义技能都应继承此类"""
@property
@abstractmethod
def name(self) -> str:
"""技能名称，必须唯一"""
pass
@property
@abstractmethod
def description(self) -> str:
"""技能描述，用于AI理解"""
pass
@property
@abstractmethod
def parameters(self) -> dict:
"""技能参数定义，遵循JSON Schema规范"""
pass
@abstractmethod
def execute(self, **kwargs) -> any:
"""执行技能的核心方法"""
pass
def get_function_definition(self):
"""获取该技能的函数定义，用于传递给AI模型"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}第二步：实现具体技能
现在，我们利用这个基类来创建一些实用的技能。
import requests
import datetime
class GetCurrentWeatherSkill(Skill):
"""获取当前天气的技能"""
@property
def name(self) -> str:
return "get_current_weather"
@property
def description(self) -> str:
return "获取指定城市的当前天气情况，包括温度、湿度和天气状况。"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {
"city_name": {
"type": "string",
"description": "城市名称，例如 '北京', '上海'"
},
"country_code": {
"type": "string",
"description": "国家代码 (可选，默认为CN)",
"default": "CN"
}
},
"required": ["city_name"],
}
def execute(self, city_name: str, country_code: str = "CN") -> dict:
# 这里使用OpenWeatherMap API作为示例 (需要注册并获取API KEY)
api_key = "YOUR_OPENWEATHERMAP_API_KEY"
url = f"http://api.openweathermap.org/data/2.5/weather?q={city_name},{country_code}&appid={api_key}&units=metric"
try:
response = requests.get(url)
response.raise_for_status()
data = response.json()
weather_info = {
"city": data['name'],
"temperature": round(data['main']['temp'], 1),
"humidity": data['main']['humidity'],
"description": data['weather'][0]['description']
}
return weather_info
except requests.exceptions.RequestException as e:
return {"error": f"无法获取天气信息: {str(e)}"}
except KeyError:
return {"error": "天气信息格式异常"}
class GetTimeSkill(Skill):
"""获取当前时间的技能"""
@property
def name(self) -> str:
return "get_time"
@property
def description(self) -> str:
return "获取当前的日期和时间。"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {},
"required": [],
}
def execute(self) -> str:
now = datetime.datetime.now()
return now.strftime("%Y-%m-%d %H:%M:%S")
class CalculatorSkill(Skill):
"""一个简单的计算器技能"""
@property
def name(self) -> str:
return "calculate"
@property
def description(self) -> str:
return "执行基本的数学运算，如加法、减法、乘法、除法。"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["add", "subtract", "multiply", "divide"],
"description": "运算类型"
},
"a": {
"type": "number",
"description": "第一个操作数"
},
"b": {
"type": "number",
"description": "第二个操作数"
}
},
"required": ["operation", "a", "b"]
}
def execute(self, operation: str, a: float, b: float) -> float:
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
elif operation == "multiply":
return a * b
elif operation == "divide":
if b != 0:
return a / b
else:
raise ValueError("除数不能为零")
else:
raise ValueError(f"不支持的运算类型: {operation}")第三步：创建技能注册与调度中心
我们需要一个中心来管理所有可用的技能，并根据AI的请求来调度它们。
from typing import Dict, List, Any
import json
class SkillRegistry:
"""技能注册与调度中心"""
def __init__(self):
self.skills: Dict[str, Skill] = {}
def register_skill(self, skill: Skill):
"""注册一个技能"""
if skill.name in self.skills:
print(f"警告: 技能 '{skill.name}' 已存在，将被覆盖。")
self.skills[skill.name] = skill
print(f"技能 '{skill.name}' 已成功注册。")
def get_all_function_definitions(self) -> List[dict]:
"""获取所有已注册技能的函数定义列表"""
return [skill.get_function_definition() for skill in self.skills.values()]
def execute_skill(self, tool_call: dict) -> dict:
"""
执行一个技能调用
:param tool_call: 包含 'name' 和 'arguments' 的字典
:return: 执行结果
"""
name = tool_call['name']
arguments_str = tool_call['arguments']
if name not in self.skills:
return {"error": f"未知技能: {name}"}
skill = self.skills[name]
try:
# 将字符串参数解析为字典
arguments = json.loads(arguments_str)
# 安全性检查：确保参数只包含定义过的字段
allowed_params = set(skill.parameters.get('properties', {}).keys())
provided_params = set(arguments.keys())
if not provided_params.issubset(allowed_params):
return {"error": f"技能 '{name}' 收到了未定义的参数: {provided_params - allowed_params}"}
result = skill.execute(**arguments)
return {"result": result}
except json.JSONDecodeError:
return {"error": f"技能 '{name}' 的参数格式错误，无法解析 JSON: {arguments_str}"}
except Exception as e:
return {"error": f"技能 '{name}' 执行失败: {str(e)}"}
# --- 示例：如何使用 ---
registry = SkillRegistry()
# 注册我们刚才定义的技能
registry.register_skill(GetCurrentWeatherSkill())
registry.register_skill(GetTimeSkill())
registry.register_skill(CalculatorSkill())
# 获取所有技能定义，这通常是传递给AI模型的工具列表
available_functions = registry.get_all_function_definitions()
print("已注册的技能:")
for func_def in available_functions:
print(json.dumps(func_def, indent=2))
# 模拟AI返回的调用请求
ai_response_with_tool_call = {
"name": "get_current_weather",
"arguments": '{"city_name": "北京", "country_code": "CN"}'
}
# 调度中心执行该调用
execution_result = registry.execute_skill(ai_response_with_tool_call)
print("\n技能执行结果:", execution_result)第四步：模拟与AI模型的集成
最后，我们模拟一个简单的交互循环，展示技能系统如何与AI模型（这里用一个模拟函数代替）协同工作。
import random
def mock_ai_model(user_input: str, available_tools: List[dict]) -> dict:
"""
模拟AI模型的响应。
在实际应用中，这里会是调用真实AI模型API的代码。
"""
print(f"\n--- AI 思考: 用户输入 '{user_input}' ---")
# 简单的规则引擎来模拟AI判断是否需要调用工具
if "天气" in user_input:
return {
"type": "tool_call",
"name": "get_current_weather",
"arguments": json.dumps({"city_name": "上海"}) # 模拟AI提取了城市名
}
elif "时间" in user_input:
return {
"type": "tool_call",
"name": "get_time",
"arguments": "{}"
}
elif "计算" in user_input or ("+" in user_input or "-" in user_input):
# 简单模拟数学表达式识别
if "+" in user_input:
nums = [float(x) for x in user_input.split("+")]
op = "add"
else: # For simplicity, assume it's add
nums = [random.uniform(1, 10), random.uniform(1, 10)]
op = "add"
return {
"type": "tool_call",
"name": "calculate",
"arguments": json.dumps({"operation": op, "a": nums[0], "b": nums[1]})
}
else:
return {
"type": "message",
"content": "我理解了您的问题，但我认为不需要调用任何工具。"
}
# 主交互循环
def main_interaction_loop():
print("--- 欢迎使用你的AI Agent! (输入 'quit' 退出) ---")
while True:
user_input = input("\n你: ")
if user_input.lower() == 'quit':
print("再见!")
break
# AI模型分析输入并决定是否调用工具
ai_decision = mock_ai_model(user_input, registry.get_all_function_definitions())
if ai_decision['type'] == 'tool_call':
print(f"AI 决定调用技能: {ai_decision['name']}")
result = registry.execute_skill(ai_decision)
# 将工具执行结果反馈给AI，进行下一步处理
# (在真实场景中，这结果会再次传回给AI模型)
print(f"AI (已获得工具结果): 工具 '{ai_decision['name']}' 返回了: {result.get('result', result.get('error'))}")
# 模拟AI基于结果生成最终回复
final_response = f"好的，我已经帮您查询了。{result.get('result', result.get('error', '操作完成'))}"
print(f"AI: {final_response}")
elif ai_decision['type'] == 'message':
print(f"AI: {ai_decision['content']}")
# 运行交互
if __name__ == "__main__":
main_interaction_loop()结语：技能驱动的未来
通过以上代码示例，我们构建了一个从基础到实用的AI Agent技能系统。这个系统不仅清晰地分离了技能定义、注册管理和调度执行，还通过SkillRegistry实现了良好的可扩展性。
在真实的项目中，这个系统还会面临更多挑战，例如：
- 异步执行: 对于耗时较长的技能（如文件上传、复杂计算），需要支持异步调用。
- 认证与授权: 技能调用可能需要API密钥或其他身份凭证。
- 缓存机制: 对于不常变动的数据（如静态配置），可以引入缓存减少重复调用。
- 错误处理与重试: 更完善的错误处理和重试逻辑是生产环境的必备品。
掌握并实践这套技能系统的设计思想，是构建强大、灵活AI Agent的关键一步。它让Agent不再局限于文本对话，而是真正成为了能够与数字世界无缝交互的智能体。