OpenClaw技能配置实战：如何高效实现Weather天气查询功能

1次阅读

共计 2241 个字符，预计需要花费 6 分钟才能阅读完成。

天气查询功能是现代智能助手的基础能力之一，广泛应用于出行规划、日程安排、灾害预警等场景。在 OpenClaw 平台实现该功能面临三大技术挑战：

数据实时性要求：天气数据需要保持较高更新频率（通常每小时更新）
地理位置适配：需支持多种位置输入格式（城市名、邮编、经纬度等）
服务稳定性：应对 API 服务商的限流策略和突发故障

主流天气 API 对比分析：

服务商	免费额度	数据精度	响应时间	特殊优势
WeatherAPI	100 万次 / 月	±1℃	300ms	历史天气数据完整
OpenWeather	5 万次 / 月	±2℃	500ms	多语言支持优秀
AccuWeather	500 次 / 天	±0.5℃	700ms	分钟级降水预报

推荐选用 WeatherAPI 作为基础方案，其优势在于：

充足的免费调用额度
99.9% 的 SLA 保障
简洁的 RESTful 接口设计

在 OpenClaw 控制台创建新技能，选择 ”Web Service” 类型
配置基础信息：
技能名称：WeatherQuery
调用频率：60 次 / 分钟（预留安全余量）
超时设置：2000ms

建立用户输入到 API 参数的转换规则：

parameter_mappings:
  - user_input: "{city}天气"
    api_param: "q={city}&lang=zh"
  - user_input: "{location}的天气"
    api_param: "q={location}&days=3"

三层容错设计：

初级重试：对 5xx 错误立即重试 1 次
备用切换：主 API 不可用时自动切换备用源
优雅降级：返回缓存数据并提示 ” 非实时数据 ”

# openclaw_weather_config.py
import requests
from cachetools import TTLCache

# 基础配置
API_KEY = "your_api_key"
BASE_URL = "https://api.weatherapi.com/v1/current.json"
CACHE = TTLCache(maxsize=1000, ttl=1800)  # 30 分钟缓存

def get_weather(location):
    """
    获取天气核心逻辑
    :param location: 城市名称 / 经纬度 / 邮编
    :return: 格式化天气数据
    """cache_key = f"weather_{location}"
    if cache_key in CACHE:
        return CACHE[cache_key]

    try:
        params = {
            'key': API_KEY,
            'q': location,
            'lang': 'zh'
        }
        response = requests.get(BASE_URL, params=params, timeout=1.5)
        response.raise_for_status()

        data = response.json()
        result = {"temp": data["current"]["temp_c"],
            "condition": data["current"]["condition"]["text"]
        }

        CACHE[cache_key] = result
        return result
    except Exception as e:
        return {"error": str(e)}

策略	命中率	响应时间	内存消耗
无缓存	0%	320ms	0MB
本地内存缓存	65%	15ms	50MB
Redis 集群缓存	78%	25ms	200MB

推荐中小规模部署使用本地内存缓存，当 QPS>500 时升级为 Redis 集群。

对同时到达的相同位置请求：

from threading import Lock

request_lock = Lock()
pending_requests = {}

def batch_get_weather(location):
    with request_lock:
        if location in pending_requests:
            return pending_requests[location]

        future = Future()
        pending_requests[location] = future

    result = get_weather(location)

    with request_lock:
        future.set_result(result)
        del pending_requests[location]

    return result