【AI Agent开发实战】Day3——自定义Skill开发：从概念到实战

# 天气查询Skill

## 使用方法

用户说"北京天气怎么样"时，调用此Skill。

## 工具

使用心知天气API获取实时天气数据。

## 工作流程

1. 提取用户提到的城市名称
2. 调用天气API获取数据
3. 格式化结果返回给用户

## 输出格式

返回格式：
– 城市：xxx
– 天气：xxx
– 温度：xxx°C
– 湿度：xxx%
– 建议：xxx

def get_weather(city):
"""获取天气数据"""
# 这里使用心知天气API作为示例
api_key = "YOUR_API_KEY" # 需要替换为实际API Key
url = f"https://api.seniverse.com/v3/weather/now.json?key={api_key}&location={city}&language=zh-Hans&unit=c"

try:
response = requests.get(url, timeout=10)
data = response.json()

if 'results' in data:
result = data['results'][0]
now = result['now']
location = result['location']

return f"{location['name']}：{now['text']}，温度{now['temperature']}°C，湿度{now['humidity']}%"
else:
return f"未找到{city}的天气数据"
except Exception as e:
return f"查询失败：{str(e)}"

if __name__ == "__main__":
if len(sys.argv) > 1:
city = sys.argv[1]
print(get_weather(city))
else:
print("请提供城市名称")

# 旅行助手Skill

## 功能

1. 查询天气
2. 搜索航班
3. 搜索酒店
4. 生成行程

## 使用场景

– "帮我查一下下周去上海的航班"
– "北京有哪些五星级酒店"
– "上海三天行程怎么安排"

## 工作流程

根据用户需求选择对应工具：
1. 天气查询 → weather.py
2. 航班搜索 → flights.py
3. 酒店搜索 → hotels.py
4. 行程生成 → itinerary.py

## 参数

– origin：出发城市
– destination：目的地
– date：日期
– budget：预算

参考 `references/best-practices.md` 获取详细的代码规范。
参考 `references/common-issues.md` 了解常见问题清单。

template_path = os.path.join(os.path.dirname(__file__), '../assets/template.xlsx')
output_path = 'report.xlsx'
shutil.copy(template_path, output_path)
# 然后打开文件进行填充

logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)
logger = logging.getLogger(__name__)

def query_weather(city):
logger.info(f"查询城市：{city}")
# 执行查询
logger.info(f"查询成功：{result}")

# 智能提醒Skill

## 功能

1. 设置提醒（一次性或周期性）
2. 查看所有活跃提醒
3. 取消指定提醒

## 使用示例

– "明天上午9点提醒我开会"
– "每周五下午3点提醒我周报"
– "我有哪些提醒"
– "取消明天9点的提醒"

## 参数

– time：时间（支持自然语言，如"明天9点"、"下周一"）
– content：提醒内容
– repeat：重复周期（daily/weekly/monthly或不设置）

## 数据存储

提醒数据存储在 `~/.openclaw/reminders.json`

REMINDERS_FILE = os.path.expanduser("~/.openclaw/reminders.json")

def load_reminders():
if os.path.exists(REMINDERS_FILE):
with open(REMINDERS_FILE, 'r') as f:
return json.load(f)
return []

def save_reminders(reminders):
os.makedirs(os.path.dirname(REMINDERS_FILE), exist_ok=True)
with open(REMINDERS_FILE, 'w') as f:
json.dump(reminders, f, ensure_ascii=False, indent=2)

def parse_time(time_str):
"""解析时间字符串"""
now = datetime.now()

if "明天" in time_str:
return (now + timedelta(days=1)).replace(hour=9, minute=0, second=0)
elif "下周" in time_str:
return now + timedelta(days=7)
elif "下个月" in time_str:
return now + timedelta(days=30)
else:
# 简单解析小时
try:
hour = int(time_str.split("点")[0][-2:])
return now.replace(hour=hour, minute=0, second=0)
except:
return None

def main():
if len(sys.argv) < 3:
print("用法：set_reminder.py <时间> <内容>")
sys.exit(1)

time_str = sys.argv[1]
content = sys.argv[2]

remind_time = parse_time(time_str)
if not remind_time:
print("无法解析时间，请使用自然语言如'明天9点'")
sys.exit(1)

reminder = {
"id": len(load_reminders()) + 1,
"time": remind_time.isoformat(),
"content": content,
"done": False
}

reminders = load_reminders()
reminders.append(reminder)
save_reminders(reminders)

print(f"提醒已设置：{remind_time.strftime('%Y-%m-%d %H:%M')} – {content}")

if __name__ == "__main__":
main()

class TestWeather(unittest.TestCase):
def test_beijing(self):
result = get_weather("北京")
self.assertIn("北京", result)

if __name__ == "__main__":
unittest.main()

CACHE_DIR = os.path.expanduser("~/.openclaw/cache")

def get_cached(key, ttl_minutes=60):
"""获取缓存数据"""
cache_file = os.path.join(CACHE_DIR, hashlib.md5(key.encode()).hexdigest() + ".json")

if os.path.exists(cache_file):
with open(cache_file, 'r') as f:
cache = json.load(f)

if datetime.now().timestamp() – cache['timestamp'] < ttl_minutes * 60:
return cache['data']

return None

def set_cache(key, data):
"""设置缓存数据"""
os.makedirs(CACHE_DIR, exist_ok=True)
cache_file = os.path.join(CACHE_DIR, hashlib.md5(key.encode()).hexdigest() + ".json")

with open(cache_file, 'w') as f:
json.dump({
'timestamp': datetime.now().timestamp(),
'data': data
}, f)

executor = ThreadPoolExecutor(max_workers=4)

async def fetch_weather_async(cities):
loop = asyncio.get_event_loop()
tasks = [loop.run_in_executor(executor, get_weather, city) for city in cities]
return await asyncio.gather(*tasks)

def validate_city(city):
"""验证城市名称"""
if not city or len(city) > 50:
return False
# 只允许中文、英文和常见符号
if not re.match(r'^[\u4e00-\u9fa5a-zA-Z\s]+$', city):
return False
return True

def safe_execute(command, args):
"""安全执行命令"""
# 使用shlex.quote转义参数
safe_args = [shlex.quote(str(arg)) for arg in args]
full_command = f"{command} {' '.join(safe_args)}"
# 执行…

def send_webhook(url, data):
"""发送Webhook通知"""
try:
response = requests.post(
url,
json=data,
headers={'Content-Type': 'application/json'},
timeout=10
)
return response.status_code == 200
except Exception as e:
print(f"Webhook发送失败：{e}")
return False

def query_database(sql, params=None):
"""执行数据库查询"""
db_path = os.path.expanduser("~/.openclaw/skills/my-skill/data.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)

results = cursor.fetchall()
conn.close()
return results

def recognize_intent(text):
"""识别用户意图"""
for intent, patterns in INTENT_PATTERNS.items():
for pattern in patterns:
if pattern in text:
return intent
return "unknown"

def extract_time(text):
"""提取时间表达式"""
patterns = [
(r'(\d+)年(\d+)月(\d+)日', 'date'),
(r'明天', 'tomorrow'),
(r'今天', 'today'),
(r'下周', 'next_week'),
(r'(\d+)点', 'hour')
]

for pattern, entity_type in patterns:
match = re.search(pattern, text)
if match:
return {"type": entity_type, "match": match.group(0)}

return None

def t(key, lang="zh"):
"""翻译函数"""
return TRANSLATIONS.get(lang, {}).get(key, key)

class TestWeatherSkill(unittest.TestCase):
def test_city_validation(self):
self.assertTrue(validate_city("北京"))
self.assertFalse(validate_city("北京<script>"))

def test_time_parsing(self):
result = parse_time("明天9点")
self.assertIsNotNone(result)

if __name__ == "__main__":
unittest.main()

# 验证Agent是否调用了my-weather Skill
# 检查日志或调用记录
assert was_skill_called("my-weather")

if os.path.exists(old_file) and not os.path.exists(new_file):
with open(old_file, 'r') as f:
old_data = json.load(f)

new_data = transform_format(old_data)

with open(new_file, 'w') as f:
json.dump(new_data, f)

class NewsAggregator:
def __init__(self, interests):
self.interests = interests
self.sources = [
{"name": "科技", "url": "https://tech.example.com/rss"},
{"name": "财经", "url": "https://finance.example.com/rss"},
{"name": "体育", "url": "https://sports.example.com/rss"}
]

def fetch_articles(self, source):
"""从指定来源获取文章"""
try:
response = requests.get(source["url"], timeout=10)
articles = self.parse_rss(response.text)
return articles
except Exception as e:
print(f"获取失败：{source['name']} – {e}")
return []

def filter_by_interest(self, articles):
"""根据兴趣过滤文章"""
filtered = []
for article in articles:
for interest in self.interests:
if interest.lower() in article.get("title", "").lower():
filtered.append(article)
break
return filtered

def generate_summary(self, articles):
"""生成新闻摘要"""
if not articles:
return "今日暂无符合条件的新闻"

summary = f"今日为您筛选了{len(articles)}篇精彩文章：\n\n"
for i, article in enumerate(articles[:5], 1):
summary += f"{i}. {article['title']}\n"
summary += f" 来源：{article.get('source', '未知')}\n"
summary += f" 摘要：{article.get('description', '')[:50]}…\n\n"

return summary

def main():
interests = ["AI", "科技", "人工智能"]
aggregator = NewsAggregator(interests)

all_articles = []
for source in aggregator.sources:
articles = aggregator.fetch_articles(source)
all_articles.extend(articles)

filtered = aggregator.filter_by_interest(all_articles)
summary = aggregator.generate_summary(filtered)

print(summary)

if __name__ == "__main__":
main()

def analyze_logs(log_content):
"""分析Skill调用日志"""
success = 0
failure = 0
errors = defaultdict(int)

for line in log_content.split('\n'):
if '"level":"INFO"' in line and '调用成功' in line:
success += 1
elif '"level":"ERROR"' in line:
failure += 1
# 提取错误类型
match = re.search(r'"error":"([^"]+)"', line)
if match:
errors[match.group(1)] += 1

total = success + failure
success_rate = success / total if total > 0 else 0

return {
"total": total,
"success": success,
"failure": failure,
"success_rate": success_rate,
"error_types": errors
}

def benchmark_skill(func, runs=100):
"""基准测试"""
durations = []

for _ in range(runs):
start = time.time()
try:
func()
duration = (time.time() – start) * 1000 # 毫秒
durations.append(duration)
except Exception as e:
print(f"执行失败：{e}")

return {
"mean_ms": mean(durations),
"stdev_ms": stdev(durations) if len(durations) > 1 else 0,
"min_ms": min(durations),
"max_ms": max(durations)
}