大模型开发(十三):Function calling调用外部工具API,实现实时天气查询

全文共1.2w余字,预计阅读时间约34~50分钟 | 满满干货(附代码案例),建议收藏!

本文目标:完整构建一个借助Function calling功能调用外部工具API的开发流程,实现天气信息的实时查询

本文代码切换使用gpt3.5和gpt4接口,请注意甄别

image-20230726102748156

代码下载点这里

一、介绍

Function calling功能从名称上来看是进行外部函数的调用,但实际上“外部函数”的功能绝不仅仅是进行类似上一篇文章中大模型开发(十二):Function calling 流程优化并实现多轮对话任务中的自定义函数这种数值运算,更常用的是通过在函数内添加外部工具的API来增加函数功能

例如可以在函数中封装调用外部实时查询天气工具API,从而实现一键实时查询天气的功能;封装一些外部数据库API,从而实现一键查询数据库数据功能;甚至可以把谷歌搜索“封装”在一个函数内,通过调用谷歌搜索API来对某些信息进行实时搜索。这些函数一旦作为外部函数“接入”Chat模型,就相当于是给Chat模型增加了一些额外能力,例如实时查询天气数据的能力、直接对数据库操作的能力、借助谷歌搜索实时获取最新信息的能力等。

从另一个角度来说,能够实时接入天气数据的大模型(LLMs)本质上就是一个天气查询的智能对话机器人,能够直接操作数据库的大模型(LLMs)本质上就是一个定制化的SQL代码解释器,而一个接入谷歌搜索、能够获取实时信息的大模型(LLMs),就是一个简易版的New bing聊天机器人,这才是Function calling的核武级应用场景——围绕大语言模型实现AI应用的敏捷开发。

本文以实时查询天气的API工具为例,介绍让Chat模型接入天气查询API的方法。

二、OpenWeather注册及API key获取方法

2.1 什么是外部工具API

API的全称是应用程序接口,例如在Python环境中使用的ChatCompletion.create函数,实际上就是OpenAI公司提供的GPT系列模型应用程序的一个接口,通过这个API(create函数或者openai库),能够顺利的调用GPT模型,并根据输入信息实时输出模型结果。而其他的很多应用程序,也都有相应的API。例如谷歌搜索,也可以通过调用API的形式来实现搜索功能

2.2 OpenWeatherAPI介绍

OpenWeather是一家提供全球范围内的气象数据服务的公司,该公司的服务包括实时天气信息、天气预报、历史天气数据以及各种气象相关的报告等,并且OpenWeather开放了一定使用限度内完全免费的API,可以在代码环境中通过调用OpenWeather API来进行实时天气查询、天气预报等功能,开发者可以将OpenWeather的天气预报功能加入到自己的应用或网站中。

一个开放的天气查询API,意味着可以将其封装为一个本地的函数,即通过调用OpenWeather API来获取实时天气信息,然后再用Function calling功能调用该函数,这样的操作就相当于给Chat模型增加了实时获取天气信息的能力,而后可通过Chat模型实时查询天气或根据天气情况询问穿衣建议等各种需求,但这个实现起来并不复杂。

2.3 OpenWeather注册并获取API key

为了能够调用OpenWeather服务,和OpenAI的API使用过程类似,首先需要先注册OpenWeather账号,并获取OpenWeather API Key。

OpenWeather API key获取流程如下:(官网地址:https://openweathermap.org/,无需魔法上网):

  • Step 1:注册

登录OpenWeather官网并点击Sign—>create account,按照提示完成填写即可。

image-20230725144332355

image-20230725144425459

  • Step 2:获取API-key

注册等登陆完成后,可在API keys页面查看当前账户的API key

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3guZfJu8-1690338934424)(https://snowball100.oss-cn-beijing.aliyuncs.com/images/202307261019025.png)]

image-20230725144906692

  • Step 3:设置为环境变量

和OpenAI API key类似,为方便后续调用,也可以直接将OpenWeather API key设置为环境变量,变量名为OPENWEATHER_API_KEY。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-a392u3Vc-1690338934425)(https://snowball100.oss-cn-beijing.aliyuncs.com/images/202307261019027.png)]

如果不清楚如何配置本地环境变量的,参考这篇:大模型开发(四):OpenAI API调用方法

一般来说首次注册用户,首个API key需要等待2-5小时才会被激活,在此期间使用该API key会返回401错误。

三、OpenWeather API使用方法

3.1 API功能说明及计费规则

关于OpenWeather API的功能、调用方法和免费额度,可以在OpenWeather API介绍页面进行查看:https://openweathermap.org/api

image-20230726083515355

最新版的3.0API能够进行实时天气查询、48小时预报、8天天气预报等,并能够查询历史40年的天气信息同时对于全部注册用户,每个用户每天有1000次免费查询的额度,超过1000次则按照0.0015美元/次进行计费。

3.2 实时天气查询API调用方法

在官网先查看实时天气信息的API Doc,该页面包含了借助API进行实时天气信息查询方法:https://openweathermap.org/current

image-20230726084136085

不同于OpenAI提供了Python库作为API接入方式,OpenWeather API是通过https协议进行通信,它是一种RESTful风格API,通过SSL或TLS协议对通信进行加密,并使用HTTP协议的原生语义进行操作,通过使用不同的HTTP方法(GET、POST、PUT、DELETE等)对资源进行增、删、改、查等操作。

使用Python调用RESTful风格API基本步骤为:

  • Step 1 构建请求

即输入某个特定的端点,也是后续请求发送的目标端点,对于OpenWeather实时查询天气的API来说,这个端点就是https://api.openweathermap.org/data/2.5/weather

  • Step 2 设置查询参数

即创建请求相关参数对象,例如如果是调用OpenWeather API进行天气查询,则需要按照官网说明设置以下参数:

image-20230726084811916

这些参数分别是城市名称或者位置坐标、API key、返回结果的格式、天气相关单位标准以及输出天气信息的语言,在实际调用OpenWeather API时,可以通过灵活调整这些参数的取值,来获得期望的结果;

  • Step 3 发送请求

借助python中的request库来发出请求

request库是一个专门为发送HTTP请求提供支持的库,借助request库可以非常便捷的进行get或者post请求,并且能够非常便捷的进行返回结果的解析;

  • Step 4 解析请求

等待API服务返回结果之后,即可围绕该结果进行解析,将其转化为当前环境可读的形式。具体返回的结果的格式可以通过mode参数进行设置,默认情况下是JOSN格式。

按照上述流程,进行通过OpenWeather API获取实时天气信息连通性测试,代码如下:

import requests

open_weather_key = os.getenv("OPENWEATHER_API_KEY")

# Step 1.构建请求
url = "https://api.openweathermap.org/data/2.5/weather"

# Step 2.设置查询参数
params = {
    
    
    "q": "Beijing",               # 查询北京实时天气
    "appid": open_weather_key,    # 输入API key
    "units": "metric",            # 使用摄氏度而不是华氏度
    "lang":"zh_cn"                # 输出语言为简体中文
}

# Step 3.发送GET请求
response = requests.get(url, params=params)

# Step 4.解析响应
data = response.json()

# 打印response对象
print("Response: ", response)

# 打印response对象的类型
print("Type of response: ", type(response))

# Step 4.解析响应
data = response.json()

# 打印解析后的json数据
print("Data: ", data)

看下输出结果:

image-20230726085715526

若API key未激活,则返回的response将显示401,表示本次请求Unauthorized(未经授权)。此外,若response出现404,则表示URL路径无效,需要根据官网提供的信息重新核对URL路径。

这样更直观:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3SQySoBn-1690338934426)(https://snowball100.oss-cn-beijing.aliyuncs.com/images/202307261019033.png)]

返回结果中包含了天气条件、温度、湿度、风速、天气描述等信息。

四、使用Function calling实现Chat模型实时天气查询

  • Step 1:编写实时天气信息获取函数

为了确保能与大语言模型之间的顺畅通信,需要函数的输入和输出都是字符串格式。具体函数编写如下:

def get_weather(loc):
    """
    查询即时天气函数
    :param loc: 必要参数,字符串类型,用于表示查询天气的具体城市名称,\
    注意,中国的城市需要用对应城市的英文名称代替,例如如果需要查询北京市天气,则loc参数需要输入'Beijing';
    :return:OpenWeather API查询即时天气的结果,具体URL请求地址为:https://api.openweathermap.org/data/2.5/weather\
    返回结果对象类型为解析之后的JSON格式对象,并用字符串形式进行表示,其中包含了全部重要的天气信息
    """
    # Step 1.构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2.设置查询参数
    params = {
    
    
        "q": loc,               
        "appid": open_weather_key,    # 输入API key
        "units": "metric",            # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                # 输出语言为简体中文
    }

    # Step 3.发送GET请求
    response = requests.get(url, params=params)
    
    # Step 4.解析响应
    data = response.json()
    return json.dumps(data)
  • Step 2:测试实时天气信息获取函数是否能正常运行
get_weather("shanghai")

输出如下:
image-20230726090949516

  • Step 3:测试Chat模型是否理解OpenWeather相关内容

需要验证Chat模型本身是否知道OpenWeather,代码如下:

response = openai.ChatCompletion.create(
  model="gpt-4-0613",
  messages=[
    {
    
    "role": "user", "content": "你知道OpenWeather是什么吗?"}
  ]
)
response.choices[0].message['content']

看下结果:

image-20230726091402672

需要验证Chat模型能否针对函数结果进行解读,即能否在基于data信息理解基础上,给出一些天气查询问题准确的回答,代码如下:

hangzhou_data = get_weather("hangzhou")
response = openai.ChatCompletion.create(
  model="gpt-4-0613",
  messages=[
    {
    
    "role": "system", "content": "天气信息来源于OpenWeather API:https://api.openweathermap.org/data/2.5/weather"},
    {
    
    "role": "system", "content": "这是今日杭州市的天气:%s" % hangzhou_data},
    {
    
    "role": "user", "content": "请问今日杭州天气如何?请用简体中文回答"}
  ]
)
response.choices[0].message['content']

看下结果:

image-20230726092023679

最关键的一个验证:需要检查模型本身是否知道中文输入的这些城市名称对应的英文名称。

因为在进行提问时问题里面的城市名称是中文,但OpenAIWeather API要求输入英文名称,尽管在函数参数说明中注明了loc函数需要转化成英文,但对于外部函数的参数整理是Chat模型自发进行的整理,外部无法干预,模型能否按照要求、在对话过程中提取参数并将其整理为指定格式,完全依赖模型自身的推理能力,验证代码如下:

response = openai.ChatCompletion.create(
  model="gpt-4-0613",
  messages=[
    {
    
    "role": "user", "content": "请问,杭州市的英文名称是?"}
  ]
)
response.choices[0].message['content']

看下结果:

image-20230726092347583

  • Step 4:创建get_weather函数的functions参数

导入Chat模型的functions参数编写函数

如果不清楚这个函数的,看这篇文章:大模型开发(十二):Function calling 流程优化并实现多轮对话任务

def auto_functions(functions_list):
    """
    Chat模型的functions参数编写函数
    :param functions_list: 包含一个或者多个函数对象的列表;
    :return:满足Chat模型functions参数要求的functions对象
    """
    def functions_generate(functions_list):
        # 创建空列表,用于保存每个函数的描述字典
        functions = []
        # 对每个外部函数进行循环
        for function in functions_list:
            # 读取函数对象的函数说明
            function_description = inspect.getdoc(function)
            # 读取函数的函数名字符串
            function_name = function.__name__

            system_prompt = '以下是某函数说明:%s' % function_description
            user_prompt = '根据这个函数的函数说明,请帮我创建一个JSON格式的字典,这个字典有如下5点要求:\
                           1.字典总共有三个键值对;\
                           2.第一个键值对的Key是字符串name,value是该函数的名字:%s,也是字符串;\
                           3.第二个键值对的Key是字符串description,value是该函数的函数的功能说明,也是字符串;\
                           4.第三个键值对的Key是字符串parameters,value是一个JSON Schema对象,用于说明该函数的参数输入规范。\
                           5.输出结果必须是一个JSON格式的字典,只输出这个字典即可,前后不需要任何前后修饰或说明的语句' % function_name

            response = openai.ChatCompletion.create(
                              model="gpt-4-0613",
                              messages=[
                                {
    
    "role": "system", "content": system_prompt},
                                {
    
    "role": "user", "content": user_prompt}
                              ]
                            )
            functions.append(json.loads(response.choices[0].message['content']))
        return functions
    
    max_attempts = 3
    attempts = 0

    while attempts < max_attempts:
        try:
            functions = functions_generate(functions_list)
            break  # 如果代码成功执行,跳出循环
        except Exception as e:
            attempts += 1  # 增加尝试次数
            print("发生错误:", e)
            if attempts == max_attempts:
                print("已达到最大尝试次数,程序终止。")
                raise  # 重新引发最后一个异常
            else:
                print("正在重新运行...")
    return functions

然后调用这个函数,得到其JSON Schema格式,代码如下:

functions_list = [get_weather]
functions = auto_functions(functions_list)
functions

看下结果:

image-20230726094104533

能发现,模型提取的函数关键信息非常到位。

  • Step 5:实现实时天气查询功能的Function calling

导入之前写的两个函数:自动执行外部函数调用的Chat对话模型run_conversation和多轮对话

如果不清楚这两个函数的,看这篇文章:大模型开发(十二):Function calling 流程优化并实现多轮对话任务

def run_conversation(messages, functions_list=None, model="gpt-4-0613"):
    """
    能够自动执行外部函数调用的Chat对话模型
    :param messages: 必要参数,字典类型,输入到Chat模型的messages参数对象
    :param functions_list: 可选参数,默认为None,可以设置为包含全部外部函数的列表对象
    :param model: Chat模型,可选参数,默认模型为gpt-4
    :return:Chat模型输出结果
    """
    # 如果没有外部函数库,则执行普通的对话任务
    if functions_list == None:
        response = openai.ChatCompletion.create(
                        model=model,
                        messages=messages,
                        )
        response_message = response["choices"][0]["message"]
        final_response = response_message["content"]
        
    # 若存在外部函数库,则需要灵活选取外部函数并进行回答
    else:
        # 创建functions对象
        functions = auto_functions(functions_list)
        # 创建外部函数库字典
        available_functions = {
    
    func.__name__: func for func in functions_list}

        # first response
        response = openai.ChatCompletion.create(
                        model=model,
                        messages=messages,
                        functions=functions,
                        function_call="auto")
        response_message = response["choices"][0]["message"]

        # 判断返回结果是否存在function_call,即判断是否需要调用外部函数来回答问题
        if response_message.get("function_call"):
            # 需要调用外部函数
            # 获取函数名
            function_name = response_message["function_call"]["name"]
            # 获取函数对象
            fuction_to_call = available_functions[function_name]
            # 获取函数参数
            function_args = json.loads(response_message["function_call"]["arguments"])
            # 将函数参数输入到函数中,获取函数计算结果
            function_response = fuction_to_call(**function_args)

            # messages中拼接first response消息
            messages.append(response_message)  
            # messages中拼接函数输出结果
            messages.append(
                {
    
    
                    "role": "function",
                    "name": function_name,
                    "content": function_response,
                }
            )  
            # 第二次调用模型
            second_response = openai.ChatCompletion.create(
                model=model,
                messages=messages,
            )  
            # 获取最终结果
            final_response = second_response["choices"][0]["message"]["content"]
        else:
            final_response = response_message["content"]
    
    return final_response
def chat_with_model(functions_list=None, 
                    prompt="你好呀", 
                    model="gpt-4-0613", 
                    system_message=[{
    
    "role": "system", "content": "你是以为乐于助人的助手。"}]):
    
    messages = system_message
    messages.append({
    
    "role": "user", "content": prompt})
    
    while True:           
        answer = run_conversation(messages=messages, 
                                    functions_list=functions_list, 
                                    model=model)
        
        
        print(f"模型回答: {
      
      answer}")

        # 询问用户是否还有其他问题
        user_input = input("您还有其他问题吗?(输入退出以结束对话): ")
        if user_input == "退出":
            break

        # 记录用户回答
        messages.append({
    
    "role": "user", "content": user_input})

直接启动外链函数库之后的多轮对话机器人chat_with_model,查看对话效果,代码如下:

  • 首先测试在不带入外部函数时,Chat模型能否准确回答当前天气情况;

image-20230726095007218

看下结果:

image-20230726095105728

在原始状态下,大模型并不具备实时查询天气信息的能力。

在对话中允许模型挂载外部函数库functions_list,并再次询问当前天气情况,得到结果如下:

image-20230726100237194

能够发现,此时在外部函数库的加持下,Chat模型就具备了实时查询天气的能力,能够对天气问题做出准确回答。

至此,就完整构建了一个借助Function calling功能来调用外部工具API的开发流程,借助Function calling,只要能够获取并灵活使用外部工具API,就能快速为大语言模型“赋能”,其背后的技术想象空间非常大,就OpenWeather API一项,就可以进一步探索为Chat模型增加未来天气预测、历史天气统计等功能,以满足不同场景的应用需求,而每个功能的加入,都相当于是完成了一个简单的AI应用开发,而这也是Function calling功能号称能够大幅加快AI应用开发的根本原因。

五、使用Few-sho提示优化函数

Chat模型的functions参数编写函数auto_functions运行过程中会出现这种情况:

image-20230726101227647

这是因为auto_functions会判断get_weather函数参数类型是array类型而不是object类型,这种错误的推理并不会导致auto_functions函数执行时报错,但会导致后续的run_conversation函数运行报错。

可以考虑通过Few-shot的方式来大幅提高auto_functions在执行外部函数参数格式推理时的准确性:即通过为模型编写一些JSON Schema编写示例,来提升推理准确率。具体auto_functions修改方法如下:

def auto_functions(functions_list):
    """
    Chat模型的functions参数编写函数
    :param functions_list: 包含一个或者多个函数对象的列表;
    :return:满足Chat模型functions参数要求的functions对象
    """
    def functions_generate(functions_list):
        # 创建空列表,用于保存每个函数的描述字典
        functions = []
        
        def chen_ming_algorithm(data):
            """
            陈明算法函数,该函数定义了一种特殊的数据集计算过程
            :param data: 必要参数,表示带入计算的数据表,用字符串进行表示
            :return:陈明函数计算后的结果,返回结果为表示为JSON格式的Dataframe类型对象
            """
            df_new = pd.read_json(data)
            res = np.sum(df_new, axis=1) - 1
            return res.to_json(orient='records')
        
        chen_ming_function_description = inspect.getdoc(chen_ming_algorithm)
        
        chen_ming_function_name = chen_ming_algorithm.__name__
        
        chen_ming_function = {
    
    "name": "chen_ming_algorithm",
                              "description": "用于执行陈明算法的函数,定义了一种特殊的数据集计算过程",
                              "parameters": {
    
    "type": "object",
                                             "properties": {
    
    "data": {
    
    "type": "string",
                                                                     "description": "执行陈明算法的数据集"},
                                                           },
                                             "required": ["data"],
                                            },
                             }

        
        # 对每个外部函数进行循环
        for function in functions_list:
            # 读取函数对象的函数说明
            function_description = inspect.getdoc(function)
            # 读取函数的函数名字符串
            function_name = function.__name__

            user_message1 = '以下是某函数说明:%s。' % chen_ming_function_description +\
                            '根据这个函数的函数说明,请帮我创建一个function对象,用于描述这个函数的基本情况。这个function对象是一个JSON格式的字典,\
                            这个字典有如下5点要求:\
                            1.字典总共有三个键值对;\
                            2.第一个键值对的Key是字符串name,value是该函数的名字:%s,也是字符串;\
                            3.第二个键值对的Key是字符串description,value是该函数的函数的功能说明,也是字符串;\
                            4.第三个键值对的Key是字符串parameters,value是一个JSON Schema对象,用于说明该函数的参数输入规范。\
                            5.输出结果必须是一个JSON格式的字典,只输出这个字典即可,前后不需要任何前后修饰或说明的语句' % chen_ming_function_name
            
            
            assistant_message1 = json.dumps(chen_ming_function)
            
            user_prompt = '现在有另一个函数,函数名为:%s;函数说明为:%s;\
                          请帮我仿造类似的格式为当前函数创建一个function对象。' % (function_name, function_description)

            response = openai.ChatCompletion.create(
                              model="gpt-4-0613",
                              messages=[
                                {
    
    "role": "user", "name":"example_user", "content": user_message1},
                                {
    
    "role": "assistant", "name":"example_assistant", "content": assistant_message1},
                                {
    
    "role": "user", "name":"example_user", "content": user_prompt}]
                            )
            functions.append(json.loads(response.choices[0].message['content']))
        return functions
    
    max_attempts = 3
    attempts = 0

    while attempts < max_attempts:
        try:
            functions = functions_generate(functions_list)
            break  # 如果代码成功执行,跳出循环
        except Exception as e:
            attempts += 1  # 增加尝试次数
            print("发生错误:", e)
            if attempts == max_attempts:
                print("已达到最大尝试次数,程序终止。")
                raise  # 重新引发最后一个异常
            else:
                print("正在重新运行...")
    return functions

Few-shot对模型推理能力的提升可以应用于提升AI应用开发稳定性中,大家可自行测试。

六、总结

本篇文章讲述了如何在OpenWeather注册并获取API key,同时解释了什么是外部工具API以及OpenWeatherAPI的基本介绍。之后,详解了OpenWeather API的使用方法,包括API功能说明、计费规则以及实时天气查询API的调用方法。在此基础上,展示了如何使用Function calling来实现Chat模型的实时天气查询,最后介绍了如何使用Few-shot提示来优化函数。此内容为实现Function calling调用外部工具API,具有参考和启发价值。

最后,感谢您阅读这篇文章!如果您觉得有所收获,别忘了点赞、收藏并关注我,这是我持续创作的动力。您有任何问题或建议,都可以在评论区留言,我会尽力回答并接受您的反馈。如果您希望了解某个特定主题,也欢迎告诉我,我会乐于创作与之相关的文章。谢谢您的支持,期待与您共同成长!

猜你喜欢

转载自blog.csdn.net/Lvbaby_/article/details/131933871