OpenAI插件实战：用Python Flask快速搭建一个天气查询插件（含完整API代码）

用Python Flask构建OpenAI天气插件：从API设计到生产部署全指南

天气数据作为高频查询需求，正成为企业服务智能化升级的入口级功能。本文将手把手教你如何基于Flask框架开发符合OpenAI插件规范的天气服务，重点解决三个核心问题：如何设计符合AI交互特点的API？如何确保服务在真实业务场景下的稳定性？以及如何让语言模型准确理解你的服务能力？

1. 插件架构设计原则

开发AI插件与普通API服务的本质区别在于：你的调用者不再是人类开发者，而是具备自然语言理解能力的大模型。这要求我们在设计时遵循三个特殊原则：

语义化参数设计：传统API可能要求严格的参数格式（如城市ID或经纬度坐标），但AI插件应该接受"北京朝阳区"、"埃菲尔铁塔附近"这类自然语言描述。我们的天气服务需要内置地理位置解析模块：

from geopy.geocoders import Nominatim def resolve_location(text): geolocator = Nominatim(user_agent="weather_plugin") location = geolocator.geocode(text) return { 'address': location.address, 'latitude': location.latitude, 'longitude': location.longitude }

错误处理友好性：当用户询问"上海明天会下雨吗"而你的服务只支持当天数据时，错误响应应该包含模型可理解的说明：

{ "error": "unsupported_time_range", "message": "本服务仅提供实时天气数据，如需预报请使用专业气象服务", "resolution": "请查询当前天气或联系管理员开通预报功能" }

多粒度响应设计：考虑不同场景下的信息密度需求，响应结构应该支持可选的详情模式：

2. Flask服务核心实现

我们采用模块化设计，将服务拆分为三个主要组件：

2.1 路由控制器

from flask import Flask, jsonify, request from weather_service import get_current_weather app = Flask(__name__) @app.route('/weather', methods=['GET']) def weather_endpoint(): location = request.args.get('location') detail_mode = request.args.get('detail', 'false').lower() == 'true' try: weather_data = get_current_weather(location, detail_mode) return jsonify(weather_data) except LocationResolutionError as e: return jsonify({ "error": "location_resolution_failed", "message": str(e) }), 400

2.2 认证中间件

虽然示例使用无认证模式，但生产环境建议至少启用API Key验证：

from functools import wraps def require_api_key(view_func): @wraps(view_func) def decorated(*args, **kwargs): api_key = request.headers.get('X-API-KEY') if not validate_api_key(api_key): return jsonify({"error": "invalid_api_key"}), 403 return view_func(*args, **kwargs) return decorated

2.3 缓存层集成

使用Redis缓存高频查询结果，显著降低上游天气API的调用成本：

import redis from datetime import timedelta redis_client = redis.StrictRedis(host='localhost', port=6379, db=0) def get_cached_weather(location): cache_key = f"weather:{location}" cached_data = redis_client.get(cache_key) if cached_data: return json.loads(cached_data) fresh_data = fetch_from_upstream(location) redis_client.setex(cache_key, timedelta(minutes=30), json.dumps(fresh_data)) return fresh_data

3. OpenAPI规范编写技巧

规范的API描述是模型正确使用插件的关键。以下是天气插件的OpenAPI定义要点：

openapi: 3.0.2 info: title: Weather Plugin API description: 提供实时天气查询服务，支持全球主要城市 version: 1.0.0 servers: - url: https://api.yourdomain.com/v1 paths: /weather: get: summary: 获取指定地点天气 parameters: - name: location in: query required: true schema: type: string example: "北京市海淀区" description: 城市名称或具体地址 - name: detail in: query schema: type: boolean description: 是否返回详细气象数据 responses: '200': description: 成功获取天气数据 content: application/json: schema: $ref: '#/components/schemas/WeatherResponse' components: schemas: WeatherResponse: type: object properties: location: type: string description: 解析后的标准地址 temperature: type: number format: float description: 摄氏温度 condition: type: string enum: [sunny, cloudy, rainy, snowy] description: 天气状况分类 humidity: type: integer description: 湿度百分比 minimum: 0 maximum: 100

特别注意：

• 每个参数必须包含example和description
• 枚举值需要明确定义可能取值
• 错误响应需要单独定义schema

4. 生产环境部署方案

4.1 基础设施配置

推荐使用容器化部署，以下Dockerfile包含最佳实践：

FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 设置健康检查 HEALTHCHECK --interval=30s --timeout=3s \ CMD curl -f http://localhost:5000/health || exit 1 EXPOSE 5000 CMD ["gunicorn", "-w 4", "-b :5000", "app:app"]

配套的docker-compose.yml集成Redis和监控：

version: '3' services: web: build: . ports: - "5000:5000" environment: - REDIS_HOST=redis depends_on: - redis redis: image: redis:6-alpine volumes: - redis_data:/data prometheus: image: prom/prometheus ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml volumes: redis_data:

4.2 性能优化策略

通过负载测试确定最佳配置参数：

启用Gzip压缩和静态资源缓存：

from flask_compress import Compress Compress(app)

4.3 监控告警设置

Prometheus监控指标示例：

scrape_configs: - job_name: 'weather_api' metrics_path: '/metrics' static_configs: - targets: ['web:5000']

关键告警规则：

• API错误率5分钟内>1%
• 平均响应时间>1s持续10分钟
• 内存使用率>80%持续5分钟

5. 插件调试与优化

5.1 测试用例设计

使用pytest编写全面的集成测试：

def test_weather_endpoint(client): # 测试正常查询 response = client.get('/weather?location=北京') assert response.status_code == 200 assert 'temperature' in response.json # 测试详细模式 response = client.get('/weather?location=上海&detail=true') assert 'humidity' in response.json # 测试错误处理 response = client.get('/weather?location=不存在的城市') assert response.status_code == 400

5.2 模型交互优化

在ai-plugin.json中添加模型提示：

{ "description_for_model": "调用本插件时，请遵循以下规则：1. 当用户询问天气时自动触发 2. 地点参数支持中文城市名或地标 3. 时间相关查询会自动转为当前时间" }

5.3 用户体验改进

添加响应模板示例帮助模型生成友好回复：

{ "template": "当前{location}天气：{condition}，温度{temperature}℃{humidity?，湿度{humidity}%}{wind_speed?，风速{wind_speed}km/h}", "examples": [ "当前北京天气：晴天，温度23℃", "当前上海天气：小雨，温度18℃，湿度85%，风速12km/h" ] }

实际部署时发现，当服务响应时间超过2秒时，模型会放弃等待。因此我们在所有耗时操作中添加了进度提示：

@app.route('/weather') def weather(): if estimate_processing_time() > 1.5: return jsonify({ "progress": "fetching_data", "message": "正在获取最新气象数据，请稍候..." })