AI大模型探索之路-实战篇8：多轮对话与Function Calling技术应用

系列篇章💥

AI大模型探索之路-实战篇4：深入DB-GPT数据应用开发框架调研
AI大模型探索之路-实战篇5：探索Open Interpreter开放代码解释器调研
AI大模型探索之路-实战篇6：掌握Function Calling的详细流程
AI大模型探索之路-实战篇7：Function Calling技术实战自动生成函数

系列篇章💥
一、前言
二、封装调用2轮response的函数
- 1、输出函数列表
- 2、function函数调用封装
- 3、定义参数数据
- 4、测试函数tangseng_function
- 5、测试函数sunwukong_function
- 6、测试函数（描述中不指明信息）
三、增加多轮对话的效果
- 1、创建OpenAI客户端
- 2、定义工具函数
- 3、定义函数生成器
- 4、封装大模型2次API调用
- 5、多轮对话函数定义
- 6、函数列表定义
- 7、对话调用测试
四、结语

一、前言

继前文深入探讨了Function Calling的操作流程并成功实验自动生成function函数之后，本文将进一步深化我们的研究，具体考察OpenAI的Function Calling技术在现实应用中的表现。此次研究的重点在于封装一个能够调用两轮response的函数，并在此基础上执行多轮对话的测试，从而全面评估该技术在实际应用中的效率和效果。

通过精心设计的实验，我们将探索这项技术的潜力及其在智能数据分析平台上的应用前景。我们的目标是为平台的顺利实施构建一个稳固且高效的技术基础。这不仅涉及到技术层面的优化，也关系到如何在实际应用中提供更流畅的用户体验和更准确的数据处理能力。

在本文中，我们将详细阐述这一过程的每个步骤，从技术选型到功能实现，再到最终的效果评估。我们希望通过对OpenAI的Function Calling技术的深入分析和应用展示，为未来相关技术的发展和应用场景的拓展提供有益的参考和启示。

二、封装调用2轮response的函数

在实现function call的过程中，我们通常需要进行两次关键的调用。
1）第一次调用：目的是让强大的大型模型协助我们识别并选择恰当的工具函数。这就像在进行一场棋局，首先需要确定使用哪个棋子。一旦选定，接下来的步骤便是实际调用该工具函数，执行具体的任务。
2）第二次调用：这一次，我们再次借助大模型的强大能力，其任务是将工具函数处理后的结果进行整合和优化，以最清晰、最易于理解的形式呈现最终输出。这可以比喻为将散落的珠子串成一串光彩夺目的项链，使其更具有观赏价值和实用价值。

在本章节中，我们将重点介绍如何将这两阶段的调用过程进行有效的封装，使其成为一个独立、可复用的模块。这种封装不仅简化了操作流程，还提高代码的可维护性和扩展性，为未来的功能拓展或技术升级提供了便利。

1、输出函数列表

继上文中，我们定义了2个工具函数

def sunwukong_function(data):
    """
    孙悟空算法函数，该函数定义了数据集计算过程
    :param data: 必要参数，表示带入计算的数据表，用字符串进行表示
    :return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
    """
    data = io.StringIO(data)
    df_new = pd.read_csv(data, sep='\s+', index_col=0)
    res = df_new * 10
return json.dumps(res.to_string())

#在定义一个工具函数，一起测试
def tangseng_function(data):
    """
    唐僧算法函数，该函数定义了数据集计算过程
    :param data: 必要参数，表示带入计算的数据表，用字符串进行表示
    :return：tangseng_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
    """
    data = io.StringIO(data)
    df_new = pd.read_csv(data, sep='\s+', index_col=0)
    res = df_new * 1000000
    return json.dumps(res.to_string())
    
# 将2个函数放入函数列表
functions_list=[sunwukong_function,tangseng_function]

输出函数列表：

functions_list

输出：
在这里插入图片描述

2、function函数调用封装

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

        # 创建外部函数库字典
        available_functions = {func.__name__: func for func in functions_list}

        # 第一次调用大模型，寻找工具函数
        response = client.chat.completions.create(
                        model=model,
                        messages=messages,
                        tools=tools,
                        tool_choice="auto", )
        response_message = response.choices[0].message


        tool_calls = response_message.tool_calls

        if tool_calls:

            messages.append(response_message) 
            for tool_call in tool_calls:
                function_name = tool_call.function.name
                function_to_call = available_functions[function_name]
                function_args = json.loads(tool_call.function.arguments)
                ## 真正执行外部函数的就是这儿的代码
                function_response = function_to_call(**function_args)
                messages.append(
                    {
                        "tool_call_id": tool_call.id,
                        "role": "tool",
                        "name": function_name,
                        "content": function_response,
                    }
                ) 
            ## 第二次调用模型，将响应结果，给到大模型进行转为为更为可视化的数据格式，再返回给用户
            second_response = client.chat.completions.create(
                model=model,
                messages=messages,
            ) 
            # 获取最终结果
            final_response = second_response.choices[0].message.content
        else:
            final_response = response_message.content
                
    return final_response

3、定义参数数据

df_str = pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]}).to_string()
df_str

在这里插入图片描述

4、测试函数tangseng_function

# 定义消息列表
messages = [
        {"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
        {"role": "user", "content": '请在data上执行唐僧算法函数'}]


# API调用测试
run_conversation(messages = messages, functions_list = functions_list)

输出：
在这里插入图片描述

5、测试函数sunwukong_function

# 定义消息列表
messages = [
        {"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
        {"role": "user", "content": '请在data上执行孙悟空算法函数'}]

# API调用测试
run_conversation(messages = messages, functions_list = functions_list)

输出：
在这里插入图片描述

6、测试函数（描述中不指明信息）

# 定义消息列表
messages = [
        {"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
        {"role": "user", "content": '请解释一下data数据集'}]

# API调用测试
run_conversation(messages = messages, functions_list = functions_list)

输出：
在这里插入图片描述

根据结果可以看出，大模型并没有成功调用工具函数

三、增加多轮对话的效果

在实现function call技术的过程中，我们不仅需要关注单次调用的优化，还需要注重多轮对话的效果。本章节将介绍如何通过引入两个工具函数，来测试大型模型在面对涉及工具函数的问题时，是否能准确识别并调用相应的工具。

1、创建OpenAI客户端

import openai
import os
import numpy as np
import pandas as pd
import json
import io
from openai import OpenAI
import inspect
# 定义api key
openai.api_key = os.getenv("OPENAI_API_KEY")

#创建OpenAI客户端
client = OpenAI(api_key=openai.api_key )

2、定义工具函数

def sunwukong_function(data):
    """
    孙悟空算法函数，该函数定义了数据集计算过程
    :param data: 必要参数，表示带入计算的数据表，用字符串进行表示
    :return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
    """
    data = io.StringIO(data)
    df_new = pd.read_csv(data, sep='\s+', index_col=0)
    res = df_new * 10
    return json.dumps(res.to_string())
    
def tangseng_function(data):
    """
    唐僧算法函数，该函数定义了数据集计算过程
    :param data: 必要参数，表示带入计算的数据表，用字符串进行表示
    :return：tangseng_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
    """
    data = io.StringIO(data)
    df_new = pd.read_csv(data, sep='\s+', index_col=0)
    res = df_new * 1000000
    return json.dumps(res.to_string())

3、定义函数生成器

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 = client.chat.completions.create(
                              model="gpt-3.5-turbo",
                              messages=[
                                {"role": "system", "content": system_prompt},
                                {"role": "user", "content": user_prompt}
                              ]
                            )
            json_function_description=json.loads(response.choices[0].message.content.replace("```","").replace("json",""))
            json_str={"type": "function","function":json_function_description}
            functions.append(json_str)
        return functions
    ## 最大可以尝试4次
    max_attempts = 4
    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

4、封装大模型2次API调用

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

        # 创建外部函数库字典
        available_functions = {func.__name__: func for func in functions_list}

        # 第一次调用大模型
        response = client.chat.completions.create(
                        model=model,
                        messages=messages,
                        tools=tools,
                        tool_choice="auto", )
        response_message = response.choices[0].message


        tool_calls = response_message.tool_calls

        if tool_calls:

            messages.append(response_message) 
            for tool_call in tool_calls:
                function_name = tool_call.function.name
                function_to_call = available_functions[function_name]
                function_args = json.loads(tool_call.function.arguments)
                ## 真正执行外部函数的就是这儿的代码
                function_response = function_to_call(**function_args)
                messages.append(
                    {
                        "tool_call_id": tool_call.id,
                        "role": "tool",
                        "name": function_name,
                        "content": function_response,
                    }
                ) 
            ## 第二次调用模型
            second_response = client.chat.completions.create(
                model=model,
                messages=messages,
            ) 
            # 获取最终结果
            final_response = second_response.choices[0].message.content
        else:
            final_response = response_message.content
                
    return final_response

5、多轮对话函数定义

def chat_with_model(functions_list=None, 
                    prompt="你好", 
                    model="gpt-3.5-turbo", 
                    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})

6、函数列表定义

functions_list=[sunwukong_function,tangseng_function]

7、对话调用测试

chat_with_model(functions_list,prompt="你好")

gpt3.5生成json schema还是会不太稳定，经常会生成失败,如下：
在这里插入图片描述

调用成功效果如下：
在这里插入图片描述
以下是我测试的对话过程
问题1：你好
问题2：1+2
问题3：什么是孙悟空函数
问题4：数据集data以字符串形式呈现如下：’ x1 x2\n0 1 3\n1 2 4’，请在数据集data上执行孙悟空算法
智能助手回复结果如下：

四、结语

在本文中，我们深入探讨了function call技术中对大型模型API进行两次调用的进一步封装方法，并实践了这一方法以实现多轮对话的效果。通过精心设计的技术策略和实验，我们不仅优化了API调用流程，还成功地将这些技术应用于实际的对话场景中，使得交互更加流畅和自然。这一成果不仅展示了function call技术的高级应用，也为未来更复杂的应用场景提供了可能。

在这里插入图片描述