免费天气技能 (专业版)

概述

专业版在免费版零配置查询能力之上,扩展多数据源冗余、企业集成、缓存加速、监控告警等生产能力。支持同时对接 wttr.in、Open-Meteo、OpenWeather 等多个数据源,自动故障切换,提供 99.9% 可用性 SLA。适合需要将天气数据深度集成到生产系统的企业、IoT 平台、商业应用等场景。

专业版与免费版命令行格式完全兼容,现有调用代码无需修改,可直接获得企业级能力。

核心能力

核心功能执行

用input_params参数进行配置。

输入: 用户提供核心功能执行所需的指令和必要参数。 处理: 按照skill规范执行核心功能执行操作,遵循单一意图原则。 输出: 返回核心功能执行的执行结果,包含操作状态和输出数据。

• 执行此能力时使用input_params参数,支持创建/查询/导出操作

参数配置与调用

用config_options参数进行配置。

输入: 用户提供参数配置与调用所需的指令和必要参数。 处理: 按照skill规范执行参数配置与调用操作,遵循单一意图原则。 输出: 返回参数配置与调用的执行结果,包含操作状态和输出数据。

• 执行此能力时使用config_options参数,支持修改/重置/导入操作

结果处理与输出

用output_format参数进行配置。

输入: 用户提供结果处理与输出所需的指令和必要参数。 处理: 按照skill规范执行结果处理与输出操作,遵循单一意图原则。 输出: 返回结果处理与输出的执行结果,包含操作状态和输出数据。

• 执行此能力时使用output_format参数,支持导出/保存/转换操作 能力覆盖范围：本skill的核心能力覆盖以下场景关键词：高可靠天气查询平、支持多源冗余、自定义数据源与监、面向企业、运维与生产环境的、核心能力、适用场景、生产系统集成、运维监控、商业应用、保障场景、差异化、专业版支持多数据、源冗余与企业级集、与免费版命令行格、式完全兼容、适用关键词、企业天气集成、多源冗余等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持。

使用场景

场景一: 生产系统天气集成

将天气数据深度集成到企业业务系统。

import os
import requests
import redis
import json
from datetime import datetime, timedelta
from typing import Optional

class ProWeatherService:
    """企业级天气服务"""

    SOURCES = [
        {"name": "wttr.in", "url": "https://wttr.in", "priority": 1},
        {"name": "open-meteo", "url": "https://api.open-meteo.com", "priority": 2},
        {"name": "openweather", "url": "https://api.openweathermap.org", "priority": 3},
        {"name": "weatherapi", "url": "https://api.weatherapi.com", "priority": 4},
        {"name": "private", "url": os.environ.get("PRIVATE_WEATHER_URL", ""), "priority": 5},
    ]

    def __init__(self):
        self.redis = redis.Redis.from_url(os.environ.get("REDIS_URL"))
        self.cache_ttl = 300  # 5 分钟
    def query(self, city, force_refresh=False):
        """查询天气 (带缓存与故障转移)"""
        cache_key = f"weather:{city}"
        if not force_refresh:
            cached = self.redis.get(cache_key)
            if cached:
                return json.loads(cached)

        # 多源故障转移
        for source in sorted(self.SOURCES, key=lambda x: x["priority"]):
            try:
                data = self._query_source(source, city)
                if data:
                    self.redis.setex(cache_key, self.cache_ttl, json.dumps(data, ensure_ascii=False))
                    self._log_success(source["name"], city)
                    return data
            except Exception as e:
                self._log_failure(source["name"], city, str(e))
                continue

        raise RuntimeError(f"所有数据源查询 {city} 失败")

    def _query_source(self, source, city):
        """从指定数据源查询"""
        if source["name"] == "wttr.in":
            resp = requests.get(
                f"{source['url']}/{city}",
                params={"format": "j1"},
                timeout=10,
            )
            return self._normalize_wttr(resp.json())
        elif source["name"] == "open-meteo":
            coords = self._geocode(city)
            resp = requests.get(
                f"{source['url']}/v1/forecast",
                params={"latitude": coords[0], "longitude": coords[1], "current_weather": True},
                timeout=10,
            )
            return self._normalize_open_meteo(resp.json(), city)

    def batch_query(self, cities):
        """批量查询多城市"""
        import concurrent.futures
        results = {}
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = {executor.submit(self.query, c): c for c in cities}
            for future in concurrent.futures.as_completed(futures):
                city = futures[future]
                try:
                    results[city] = future.result()
                except Exception as e:
                    results[city] = {"error": str(e)}
        return results

    def health_check(self):
        """数据源健康检查"""
        results = []
        for source in self.SOURCES:
            try:
                start = datetime.now()
                self._query_source(source, "Beijing")
                latency = (datetime.now() - start).total_seconds() * 1000
                results.append({
                    "source": source["name"],
                    "status": "healthy",
                    "latency_ms": round(latency, 2),
                })
            except Exception as e:
                results.append({
                    "source": source["name"],
                    "status": "unhealthy",
                    "error": str(e),
                })
        return results

场景二: IoT 设备天气联动

为 IoT 平台提供天气数据,联动智能设备。

def iot_weather_pipeline(devices):
    """IoT 设备天气联动"""
    service = ProWeatherService()
    for device in devices:
        city = device["location"]
        weather = service.query(city)
        # 根据天气控制设备
        if weather["current"]["temperature"] > 28:
            control_device(device["id"], "ac", "on", temp=24)
        elif weather["current"]["precip_prob"] > 70:
            control_device(device["id"], "window", "close")

场景三: 运维监控告警

监控天气数据源可用性,异常时告警。

def setup_weather_monitoring():
    """配置天气服务监控"""
    payload = {
        "metrics": [
            "source_availability",
            "query_latency",
            "cache_hit_rate",
            "failover_count",
        ],
        "alerts": [
            {
                "metric": "source_availability",
                "threshold": 0.8,
                "window": "5m",
                "channel": "slack",
            },
            {
                "metric": "failover_count",
                "threshold": 10,
                "window": "1h",
                "channel": "pagerduty",
            },
        ],
        "dashboard": True,
    }
    resp = requests.post(
        f"{API_BASE}/monitoring",
        headers={"X-API-Key": API_KEY},
        json=payload,
        timeout=30,
    )
    return resp.json()

不适用场景

以下场景免费天气技能专业版不适合处理：

• 物理硬件维修
• 网络物理布线
• 数据中心选址

触发条件

需要系统监控、日志分析、运维告警、部署管理时使用。不适用于非本工具能力范围的需求。

快速开始

Step 1: 部署缓存层

# 启动 Redis
docker run -d --name weather-redis -p 6379:6379 redis:alpine

# 配置环境变量
export REDIS_URL="redis://localhost:6379/0"
export WEATHER_EDITION="pro"
export WEATHER_SOURCES="wttr.in,open-meteo,openweather"

Step 2: 配置多数据源

# /etc/weather-pro/sources.yaml
sources:
  - name: wttr.in
    url: https://wttr.in
    priority: 1
    timeout: 10
    rate_limit: 10  # per minute
  - name: open-meteo
    url: https://api.open-meteo.com
    priority: 2
    timeout: 10
    rate_limit: 100
  - name: openweather
    url: https://api.openweathermap.org
    priority: 3
    timeout: 10
    api_key: ${OPENWEATHER_API_KEY}
  - name: weatherapi
    url: https://api.weatherapi.com
    priority: 4
    timeout: 10
    api_key: ${WEATHERAPI_API_KEY}
  - name: private
    url: https://internal-weather.local
    priority: 5
    timeout: 5

failover:
  strategy: priority
  retry: 2
  backoff: exponential

cache:
  enabled: true
  backend: redis
  ttl_seconds: 300
  key_prefix: "weather:"

monitoring:
  enabled: true
  metrics: [availability, latency, cache_hit_rate, failover_count]
  alerting:
    slack: ${SLACK_WEBHOOK_URL}
    pagerduty: ${PAGERDUTY_KEY}

Step 3: 启动服务

# 启动天气服务守护进程
weather-service --config /etc/weather-pro/sources.yaml --daemon

# 验证健康状态
curl http://localhost:8080/health

Step 4: 接入业务系统

# 业务系统调用本地代理
curl "http://localhost:8080/weather/Beijing"
curl "http://localhost:8080/batch/weather" -d '{"cities":["北京","上海"]}'

命令参数说明

• -API-Key: 命令参数,用于指定操作选项

示例

数据归档配置

def archive_weather(cities, interval_minutes=30):
    """定期归档天气数据"""
    service = ProWeatherService()
    while True:
        for city in cities:
            data = service.query(city)
            archive_to_db(city, data)
        time.sleep(interval_minutes * 60)

def archive_to_db(city, data):
    """归档到数据库"""
    payload = {
        "city": city,
        "data": data,
        "timestamp": datetime.now().isoformat(),
    }
    requests.post(
        f"{API_BASE}/archive",
        headers={"X-API-Key": API_KEY},
        json=payload,
        timeout=30,
    )

自定义数据源接入

class CustomWeatherSource:
    """自定义数据源适配器"""

    def __init__(self, config):
        self.url = config["url"]
        self.auth = config.get("auth")

    def query(self, city):
        """实现自定义查询逻辑"""
        headers = {}
        if self.auth:
            headers["Authorization"] = f"Bearer {self.auth}"

        resp = requests.get(
            f"{self.url}/weather",
            headers=headers,
            params={"city": city},
            timeout=10,
        )
        return self.normalize(resp.json())

    def normalize(self, raw):
        """转换为统一格式"""
        return {
            "location": raw.get("city"),
            "current": {
                "temperature": str(raw.get("temp")) + "°C",
                "condition": raw.get("weather"),
                "humidity": str(raw.get("humidity")) + "%",
                "wind": f"{raw.get('wind_dir')} {raw.get('wind_speed')}km/h",
            },
        }

监控仪表盘

def render_dashboard():
    """渲染监控仪表盘"""
    service = ProWeatherService()
    health = service.health_check()
    cache_stats = service.redis.info("stats")

    dashboard = f"""
    天气服务监控仪表盘
    ==================
    数据源健康状态:
    {chr(10).join([f"  - {s['source']}: {s['status']} ({s.get('latency_ms', 'N/A')}ms)" for s in health])}

    缓存统计:
    - 命中次数: {cache_stats['keyspace_hits']}
    - 未命中: {cache_stats['keyspace_misses']}
    - 命中率: {int(cache_stats['keyspace_hits']) / max(1, int(cache_stats['keyspace_hits']) + int(cache_stats['keyspace_misses'])) * 100:.1f}%
    """
    return dashboard

最佳实践

1. 多源策略

# 按优先级与延迟动态选择数据源
def select_best_source(sources):
    """根据健康状态与延迟选择最佳数据源"""
    healthy = [s for s in sources if s["status"] == "healthy"]
    return min(healthy, key=lambda x: x["latency_ms"])

2. 智能缓存

def adaptive_cache_ttl(weather):
    """根据天气状况动态调整缓存时间"""
    # 极端天气短缓存
    if weather["current"]["precip_prob"] > 80:
        return 60  # 1 分钟
    # 普通天气长缓存
    return 600  # 10 分钟

3. 限流保护

from functools import wraps
import time

def rate_limit(calls, period):
    """限流装饰器"""
    def decorator(func):
        last_reset = [time.time()]
        call_count = [0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            if now - last_reset[0] > period:
                last_reset[0] = now
                call_count[0] = 0
            if call_count[0] >= calls:
                raise RuntimeError("Rate limit exceeded")
            call_count[0] += 1
            return func(*args, **kwargs)
        return wrapper
    return decorator

常见问题

Q1: 专业版与免费版调用方式是否兼容?

完全兼容。专业版在内部提供多源冗余、缓存、监控,对外接口与免费版一致,现有代码无需修改。

Q2: 多数据源如何选择?

按优先级配置,主源故障自动切换到次源。可基于延迟动态调整优先级。

Q3: 缓存策略如何配置?

默认 5 分钟 TTL,可根据天气状况动态调整。极端天气缩短至 1 分钟,稳定天气延长至 10 分钟。

Q4: SLA 如何保障?

99.9% 月度可用性。多源冗余 + 缓存使得单源故障不影响整体可用性。

Q5: 支持私有化部署吗?

支持。可将全部组件部署在企业内网,数据不出域。

依赖说明

运行环境

• Agent 平台: 支持 SKILL.md 规范的任意 AI Agent (Claude Code、Cursor、Codex、Gemini CLI 等)
• 操作系统: Windows / macOS / Linux (生产环境推荐 Linux)
• 网络: 需访问多个天气数据源
• Python: 3.9+ (用于服务端)

依赖详情

API Key 配置

# 专业版基础配置
export WEATHER_EDITION="pro"
export REDIS_URL="redis://localhost:6379/0"

# 多数据源凭证 (按需配置)
export OPENWEATHER_API_KEY="..."
export WEATHERAPI_API_KEY="..."
export PRIVATE_WEATHER_URL="https://internal-weather.local"

# 监控告警
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/xxx"
export PAGERDUTY_KEY="..."

# 归档数据库
export ARCHIVE_DB_URL="db://user:pass@host:5432/weather_archive"

可用性分类

• 分类: MD+EXEC (Markdown 指令 + 命令行执行)
• 说明: 本 Skill 面向企业与生产环境,通过自然语言指令驱动 Agent 调用多数据源天气服务,提供高可用、可监控的企业级天气能力
• 专业版特性: 多源冗余、批量查询、Redis 缓存、监控告警、自定义数据源、SLA 保障
• 兼容性: 与免费版命令行格式完全兼容,现有调用代码无需修改

错误处理

已知限制

• 需要API Key，无Key环境无法使用