不再受天气干扰！ChatGPT打破不可知的神秘，为你实时查询天气！

大家好，我是 AI 研习者轻寒。在前一篇文章《解锁无限可能：ChatGPT + 钉钉，打造属于你的 AI 助手未来！》 中，从 0 到 1 完成了把 ChatGPT 接入到钉钉中，我命名版本为 v1.0.0。本文介绍如何使用 OpenAI 的 function calling 实现实时天气查询接入，命名版本为v1.0.3（中间还接了其他的接口）。

Function Calling（官方描述）

Function Calling 即函数调用，在 API 调用中，您可以向gpt-3.5-turbo-0613和描述函数gpt-4-0613，并让模型智能地选择输出包含参数的 JSON 对象以调用这些函数。Chat Completions API 不调用该函数；相反，该模型会生成 JSON，您可以使用它来调用代码中的函数。

最新的模型 (gpt-3.5-turbo-0613和gpt-4-0613) 已经过微调，可以检测何时应该调用函数（取决于输入）并使用符合函数签名的 JSON 进行响应。这种能力也带来了潜在的风险。我们强烈建议在代表用户采取影响世界的行动（发送电子邮件、在线发布内容、进行购买等）之前构建用户确认流程。

在幕后，函数以模型训练过的语法注入到系统消息中。这意味着函数根据模型的上下文限制进行计数，并作为输入令牌计费。如果遇到上下文限制，我们建议限制函数的数量或您为函数参数提供的文档的长度。

函数调用使您可以更可靠地从模型中获取结构化数据。例如，您可以：

• 创建通过调用外部 API（例如 ChatGPT 插件）回答问题的聊天机器人
  • 例如定义函数，如send_email(to: string, body: string), 或get_current_weather(location: string, unit: 'celsius' | 'fahrenheit')
• 将自然语言转换为 API 调用
  • 例如转换“谁是我的顶级客户？” 并get_customers(min_revenue: int, created_before: string, limit: int)调用您的内部 API
• 从文本中提取结构化数据
  • 例如，定义一个名为extract_data(name: string, birthday: string), 或sql_query(query: string)
• …

函数调用的基本步骤顺序如下：

1. 使用用户查询和functions 参数中定义的一组函数调用模型。
2. 模型可以选择调用一个函数；如果是这样，内容将是一个符合您的自定义模式的字符串化 JSON 对象（注意：该模型可能会生成无效的 JSON 或幻觉参数）。
3. 在您的代码中将字符串解析为 JSON，并使用提供的参数（如果存在）调用您的函数。
4. 通过将函数响应附加为新消息再次调用模型，并让模型汇总结果返回给用户。

示例详见：https://platform.openai.com/docs/guides/gpt/function-calling

功能分析

1. 根据用户提问关于天气问题时，自动调用天气查询接口函数；
2. 需要实现一个天气查询的函数，这里采用高德天气 API；
3. 根据高德天气API请求参数分析，核心需要提供城市编码，其他几个参数都固定了；
4. 如何获取城市编码，高德提供了一个 excel 文档，下载地址；
5. 读取 excel 文档数据我们可以进行相关检索匹配获取到城市编码 adcode；
6. 根据城市编码 adcode 调用高德天气 API，并返回结果，用于通过 ChatGPT 回答用户；
7. 在函数中我们还需要做一些异常处理，用于通过 ChatGPT 提示用户。

核心代码（版本v1.0.3）

以下代码可能存在不足之处，仅作参考。

requirements.txt

Flask==2.2.3
requests==2.28.2
urllib3==1.25.11
python-dotenv==1.0.0
aiohttp==3.8.3
openai~=0.27.4
pandas~=1.5.3
openpyxl~=3.1.2

安装依赖

pip install -r requirements.txt

自定义天气查询函数

import datetime
import json

import requests
import pandas as pd

functions = [
    {
        "name": "get_weather_info",
        "description": "根据中国城市名称和中国城市编码获取城市的中国区域编码，再根据中国城市的区域编码获取天气预报",
        "parameters": {
            "type": "object",
            "properties": {
                "city_name": {
                    "type": "string",
                    "description": "中国城市名称",
                },
                "district_name": {
                    "type": "string",
                    "description": "中国区县名称",
                },

            },
            "required": ["city_name"],
        },
    }
]


def get_weather_info(city_name: str, district_name: str):
    """根据城市名称和区县名称获取城市的区域编码，再根据城市的区域编码获取天气预报"""
    if city_name is None and district_name is None:
        return "{errorMsg: '输入的城市信息可能有误或未提供城市信息'}"
    # 读取Excel文件
    df = pd.read_excel("AMap_adcode_citycode.xlsx", sheet_name="Sheet1",
                       dtype={'district_name': str, 'adcode': str, 'city_name': str})
    # 将所有NaN值转换成0
    df = df.dropna()

    if district_name is not None and district_name != '':
        # 根据'district_name'列检索数据
        result = df[df['district_name'].str.contains(district_name)]
        json_data = result.to_json(orient='records', force_ascii=False)
        # 解析 JSON 数据
        json_array = json.loads(json_data)

    # 如果区域名称为空，用城市名称去查
    if district_name is None and city_name != '':
        # 根据'district_name'列检索数据
        result = df[df['district_name'].str.contains(city_name)]
        json_data = result.to_json(orient='records', force_ascii=False)
        # 解析 JSON 数据
        json_array = json.loads(json_data)

    # 如果没数据直接返回空
    if len(json_array) == 0:
        # 根据'district_name'列检索数据
        result = df[df['district_name'].str.contains(city_name)]
        json_data = result.to_json(orient='records', force_ascii=False)
        # 解析 JSON 数据
        json_array = json.loads(json_data)

    print(json_array)

    # 如果只有一条直接返回
    if len(json_array) == 1:
        return request_weather_api(json_array[0]['adcode'])

    # 如果有多条再根据district_name进行检索
    if len(json_array) > 1 and district_name != '':
        for obj in json_array:
            print(obj)
            if district_name is not None and district_name in obj['district_name']:
                return request_weather_api(obj['adcode'])
            if city_name in obj['district_name']:
                return request_weather_api(obj['adcode'])
    return "{errorMsg: '输入的城市信息可能有误或未提供城市信息'}"


def request_weather_api(adcode: str):
    """根据城市的区域编码获取天气预报"""
    response = requests.get("https://restapi.amap.com/v3/weather/weatherInfo?", {
        "key": "your key",
        "city": adcode,
        "extensions": "all",
        "output": "JSON"
    })
    # 解析并转换为 Python 字典
    data_dict = json.loads(response.content)
    print(data_dict)
    if data_dict["status"] == "0":
        return "{errorMsg: '输入的城市信息可能有误或未提供城市信息'}"
    if data_dict["forecasts"] is None or len(data_dict["forecasts"]) == 0:
        return "{errorMsg: '输入的城市信息可能有误或未提供城市信息'}"
    # 让 ChatGPT 知道当前时间  
    data_dict["currentTime"] = datetime.datetime.now()
    return json.dumps(data_dict["forecasts"])

Webhook.py

import openai
import json
import requests
import urllib3
from flask import Flask, request

from tools.life_tools import functions, get_weather_info

urllib3.disable_warnings()
requests.adapters.DEFAULT_RETRIES = 3
s = requests.Session()
# 关闭多余连接
s.keep_alive = False
# 取消验证证书
s.verify = False
# 关闭在设置了verify=False后的错误提示
urllib3.disable_warnings()
app = Flask(__name__)

messages = []  # AI对话形成上下文连贯

# Bearer加上空格然后替换openai的apikey 从openai官网获取
openai.api_key = "your openai apikey"
openai.api_base = "your openai proxy url"
DINGTALK_SEND_URL = "your dingtalk send url"


def call_functions(response_message):
    # 步骤三：调用函数。
    # 注意: JSON响应可能不总是有效; 请确保处理错误
    global function_response
    available_functions = {
        "get_weather_info": get_weather_info,
    }  # 在这个示例中只有一个函数，但你可以有多个。
    function_name = response_message["function_call"]["name"]
    function_to_call = available_functions[function_name]
    function_args = json.loads(response_message["function_call"]["arguments"])

    function_response = function_to_call(
      city_name=function_args.get("city_name"),
      district_name=function_args.get("district_name"),
    )

    # 步骤 4：将有关函数调用和函数响应的信息发送至GPT。
    messages.append(response_message)  # 与助手的回复延长对话
    messages.append(
        {
            "role": "function",
            "name": function_name,
            "content": function_response,
        }
    )  # 使用功能响应来延长对话
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-0613",
        messages=messages,
    )  # 从GPT获得一个新的响应，在那里它可以看到函数的响应。
    return response


@app.route("/webhook/event", methods=['POST'])
def event():  # AI聊天
    global messages
    # 接口请求参数
    json_data = request.get_json()

    messages.append({"role": "user", "content": json_data['text']['content']})

    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-0613",
        messages=messages,
        functions=functions,
        function_call="auto",  # auto是默认值，但我们会明确说明
    )
    response_message = response["choices"][0]["message"]

    # 第二步：检查GPT是否要调用一个函数。
    if response_message.get("function_call"):
        response = call_functions(response_message)

    messages.append(response['choices'][0]['message'])
    answer = response['choices'][0]['message']['content']
    print("----" * 20)
    print(answer)
    print("----" * 20)
    json_send_message = {"msgtype": "text", "text": {"content": answer}}
    response = requests.post(DINGTALK_SEND_URL, headers={'Content-Type': 'application/json'}, json=json_send_message)
    print(response.text)
    if len(messages) > 16:  # 对会话数进行限制
        del messages[0]
    return 'success'


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=18888, debug=True)  # 运行在有公网IP的服务器，同时开发18888端口

部署方式

我这里采用 Docker 部署。先创建一个简单 Dockerfile 用于构建镜像，Dockerfile 与上面的 webhook.py 和 requirements.txt 同一目录。

FROM python:3.9.17-slim-bullseye

WORKDIR /app

COPY requirements.txt requirements.txt
RUN pip3 install -r requirements.txt

COPY . .

CMD [ "python3", "webhook.py" ]

在 Terminal 中执行如下命令，完成镜像构建。

docker build -t ding-chatbot:1.0.0
docker images # 获取镜像ID cc07f3641130

我这里采用阿里云进行镜像管理。

docker tag cc07f3641130 registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.3
docker push registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.3

编写 docker-compose.yaml 部署脚本。

version: '3.9'
services:
  chatbot:
    image: registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.3
    ports:
      - 18888:18888
    networks:
      - xunlu
networks:
  xunlu:
    external: true

在服务器上 docker-compose.yaml 目录下，通过以下命令完成部署。

docker-compose up -d

部署完成后，把公网可访问的 URL 填写到钉钉开放平台的消息接收地址里。

遇到问题

1. 用户提问可能不会告诉你具体的地理位置，这里就要通过返回错误信息让 ChatGPT 提示用户，如 return "{errorMsg: '输入的城市信息可能有误或未提供城市信息'}"。
2. 大家知道 ChatGPT 3.5 模型数据是基于 2021 年的数据，所以它不知道当前时间，所以在响应给 ChatGPT 时需要添加一个当前时间，如 data_dict["currentTime"] = datetime.datetime.now()。
3. 从高德城市编码表中读取数据，进行检索，如何让拿到的城市编码 adcode 更准确，参照以上逻辑，同时我对编码表做了调整。

演示成功

异常情况

当我未告知 ChatGPT 需要查询的城市时，它会提示我；当我告知了城市，基于上下文它回答了我的问题。

更多详细信息

未来几天的天气预报

结合天气推理

根据天气制定旅游计划

其他相关问题

结尾

本次教程就到这里，后续会继续迭代优化，比如基于 LangChain 去实现一些功能，大家有什么建议可以留言。

理解新范式，拥抱新时代，把握新机会。

想要了解更多 AI 内容，记得关注我哦！觉得对你有用的话，记得点赞，转发给你的朋友！如果有什么问题可以私信我~