智能体技能开发实战:从天气查询到多模态输出
1. 什么是Agent SkillsAgent Skills智能体技能本质上是一组可复用的功能模块让AI智能体具备完成特定任务的能力。就像给机器人安装不同的工具包——拧螺丝需要电动起子切割金属需要激光刀每个技能都对应着解决某类问题的专业工具。在实际开发中一个完整的Skill通常包含三个核心组件意图识别理解用户想要什么比如查天气对应天气查询技能逻辑处理执行具体操作调用API、计算数据、操作设备等结果格式化将输出调整为人类可读的格式语音/文字/图表2. 开发环境准备2.1 基础工具链推荐使用Python 3.8作为开发语言配合以下工具包pip install semantic-kernel0.9.6b1 # 微软开源的SK框架 pip install python-dotenv # 环境变量管理注意建议使用虚拟环境隔离依赖避免版本冲突。可通过python -m venv myenv创建。2.2 项目结构规划典型技能项目的目录结构应保持模块化skills/ ├── weather/ # 技能名称 │ ├── __init__.py # 空文件标识Python包 │ ├── config.json # 技能参数配置 │ └── skill.py # 核心逻辑实现 └── calendar/ # 另一个技能 kernel.py # 技能调度中枢3. 技能开发实战天气查询案例3.1 定义技能元数据在weather/config.json中声明技能基础信息{ skill_name: WeatherQuery, description: 提供全球城市天气查询服务, endpoint: /weather, parameters: [ { name: location, type: string, required: true, description: 城市名称如北京 }, { name: unit, type: enum, options: [celsius, fahrenheit], default: celsius } ] }3.2 核心逻辑实现在skill.py中编写业务代码import requests from datetime import datetime class WeatherSkill: def __init__(self, api_key): self.base_url https://api.weatherapi.com/v1 self.api_key api_key async def get_current_weather(self, location: str, unit: str celsius) - str: 获取实时天气数据 params { key: self.api_key, q: location, aqi: no } try: resp requests.get(f{self.base_url}/current.json, paramsparams) data resp.json() temp data[current][temp_ unit[:1]] condition data[current][condition][text] update_time datetime.strptime( data[current][last_updated], %Y-%m-%d %H:%M ).strftime(%m月%d日 %H:%M) return f{location}当前天气{condition}温度{temp}度{unit}数据更新时间{update_time} except Exception as e: return f天气查询失败{str(e)}3.3 技能注册与测试在kernel.py中集成技能from semantic_kernel import Kernel from skills.weather.skill import WeatherSkill import os kernel Kernel() weather_skill WeatherSkill(os.getenv(WEATHER_API_KEY)) # 注册技能到内核 kernel.register_skill( skill_nameweather, skill_instanceweather_skill, description提供天气查询功能 ) # 测试调用 async def test_skill(): result await kernel.run_skill_function( skill_nameweather, function_nameget_current_weather, input_params{location: 上海, unit: celsius} ) print(result) if __name__ __main__: import asyncio asyncio.run(test_skill())4. 高级功能实现技巧4.1 多模态输出支持扩展技能使其支持富文本响应from typing import Dict, Any import matplotlib.pyplot as plt import io import base64 def generate_weather_card(data: Dict[str, Any]) - Dict[str, str]: 生成带图表的天气卡片 fig, ax plt.subplots(figsize(6, 3)) ax.bar([温度, 湿度, 风速], [data[temp], data[humidity], data[wind_kph]]) plt.tight_layout() buf io.BytesIO() plt.savefig(buf, formatpng) buf.seek(0) img_base64 base64.b64encode(buf.read()).decode(utf-8) return { text: f{data[location]}天气简报, image: fdata:image/png;base64,{img_base64}, quick_replies: [三天预报, 空气质量, 灾害预警] }4.2 技能组合调用实现多技能协作的旅行规划场景async def plan_trip(kernel: Kernel, destination: str): 组合调用天气、地图、翻译等技能 weather await kernel.run_skill_function( weather, get_current_weather, {location: destination} ) attractions await kernel.run_skill_function( map, get_top_attractions, {city: destination, limit: 5} ) return { weather: weather, attractions: attractions, advice: await generate_tips(kernel, destination) }5. 性能优化与生产部署5.1 缓存策略实现使用Redis缓存高频查询结果import redis from functools import wraps redis_client redis.StrictRedis(hostlocalhost, port6379, db0) def cache_weather(ttl: int 3600): 天气数据缓存装饰器 def decorator(func): wraps(func) async def wrapper(location: str, *args, **kwargs): cache_key fweather:{location.lower()} cached redis_client.get(cache_key) if cached: return cached.decode() result await func(location, *args, **kwargs) redis_client.setex(cache_key, ttl, result) return result return wrapper return decorator # 使用示例 cache_weather(ttl1800) async def get_current_weather(self, location: str, unit: str): # 原有实现...5.2 容器化部署Dockerfile配置示例FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . ENV WEATHER_API_KEY${API_KEY} EXPOSE 8000 CMD [gunicorn, -w 4, -k uvicorn.workers.UvicornWorker, kernel:app]6. 避坑指南与经验总结6.1 常见错误排查API限流问题现象突然返回429错误解决方案实现令牌桶算法限流from ratelimit import limits, sleep_and_retry sleep_and_retry limits(calls30, period60) def call_api(): # API调用代码编码问题中文字符乱码时确保所有文件头部添加# -*- coding: utf-8 -*-6.2 性能优化技巧批量处理对城市列表查询改造为批量接口预处理对静态数据如城市编码表启动时加载到内存异步化使用aiohttp替代requests实现并发请求import aiohttp async def batch_query(locations: list): async with aiohttp.ClientSession() as session: tasks [fetch_one(session, loc) for loc in locations] return await asyncio.gather(*tasks)6.3 技能商店发布制作技能包的标准流程压缩项目目录zip -r weather_skill.zip skills/weather/生成校验文件sha256sum weather_skill.zip checksum.txt编写使用说明README.md包含技能功能描述参数说明示例调用代码依赖项列表