侧边栏壁纸
博主头像
码森林博主等级

一起走进码森林,享受编程的乐趣,发现科技的魅力,创造智能的未来!

  • 累计撰写 145 篇文章
  • 累计创建 73 个标签
  • 累计收到 4 条评论

目 录CONTENT

文章目录

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

码森林
2023-06-20 / 0 评论 / 0 点赞 / 665 阅读 / 3,098 字 / 正在检测是否收录...
温馨提示:
本文最后更新于 2023-06-20,若内容或图片失效,请留言反馈。部分素材来自网络,若不小心影响到您的利益,请联系我们删除。

大家好,我是 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-0613gpt-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 需要查询的城市时,它会提示我;当我告知了城市,基于上下文它回答了我的问题。

image-20230620224248718

更多详细信息

image-20230620224500064

未来几天的天气预报

image-20230620224637090

结合天气推理

image-20230620224746510

根据天气制定旅游计划

image-20230620225058764

其他相关问题

image-20230620225244561

结尾

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

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

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

扫码_搜索联合传播样式-标准色版

有想讨论副业或 AI 的也可以关注我,或私聊我加微信一起探讨~

以下是个人搞得副业,长按或扫描二维码支持一下~

image-20230617123908688
0

评论区