用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_agentweather_plugin) location geolocator.geocode(text) return { address: location.address, latitude: location.latitude, longitude: location.longitude }错误处理友好性当用户询问上海明天会下雨吗而你的服务只支持当天数据时错误响应应该包含模型可理解的说明{ error: unsupported_time_range, message: 本服务仅提供实时天气数据如需预报请使用专业气象服务, resolution: 请查询当前天气或联系管理员开通预报功能 }多粒度响应设计考虑不同场景下的信息密度需求响应结构应该支持可选的详情模式字段基础模式详情模式说明temperature✓✓当前温度condition✓✓天气状况humidity✗✓湿度百分比wind_speed✗✓风速(km/h)forecast✗✓未来3小时预测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) }), 4002.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 decorated2.3 缓存层集成使用Redis缓存高频查询结果显著降低上游天气API的调用成本import redis from datetime import timedelta redis_client redis.StrictRedis(hostlocalhost, port6379, db0) def get_cached_weather(location): cache_key fweather:{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(minutes30), json.dumps(fresh_data)) return fresh_data3. 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枚举值需要明确定义可能取值错误响应需要单独定义schema4. 生产环境部署方案4.1 基础设施配置推荐使用容器化部署以下Dockerfile包含最佳实践FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 设置健康检查 HEALTHCHECK --interval30s --timeout3s \ 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_HOSTredis 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 性能优化策略通过负载测试确定最佳配置参数并发用户数平均响应时间所需Worker数推荐配置100200ms21CPU 1GB100-500300ms42CPU 2GB500-2000500ms84CPU 4GB启用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上海detailtrue) assert humidity in response.json # 测试错误处理 response client.get(/weather?location不存在的城市) assert response.status_code 4005.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: 正在获取最新气象数据请稍候... })
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_agentweather_plugin) location geolocator.geocode(text) return { address: location.address, latitude: location.latitude, longitude: location.longitude }错误处理友好性当用户询问上海明天会下雨吗而你的服务只支持当天数据时错误响应应该包含模型可理解的说明{ error: unsupported_time_range, message: 本服务仅提供实时天气数据如需预报请使用专业气象服务, resolution: 请查询当前天气或联系管理员开通预报功能 }多粒度响应设计考虑不同场景下的信息密度需求响应结构应该支持可选的详情模式字段基础模式详情模式说明temperature✓✓当前温度condition✓✓天气状况humidity✗✓湿度百分比wind_speed✗✓风速(km/h)forecast✗✓未来3小时预测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) }), 4002.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 decorated2.3 缓存层集成使用Redis缓存高频查询结果显著降低上游天气API的调用成本import redis from datetime import timedelta redis_client redis.StrictRedis(hostlocalhost, port6379, db0) def get_cached_weather(location): cache_key fweather:{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(minutes30), json.dumps(fresh_data)) return fresh_data3. 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枚举值需要明确定义可能取值错误响应需要单独定义schema4. 生产环境部署方案4.1 基础设施配置推荐使用容器化部署以下Dockerfile包含最佳实践FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 设置健康检查 HEALTHCHECK --interval30s --timeout3s \ 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_HOSTredis 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 性能优化策略通过负载测试确定最佳配置参数并发用户数平均响应时间所需Worker数推荐配置100200ms21CPU 1GB100-500300ms42CPU 2GB500-2000500ms84CPU 4GB启用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上海detailtrue) assert humidity in response.json # 测试错误处理 response client.get(/weather?location不存在的城市) assert response.status_code 4005.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: 正在获取最新气象数据请稍候... })
