AI大模型探索之路-实战篇15： Agent智能数据分析平台之整合封装Tools和Memory功能代码

系列篇章💥

AI大模型探索之路-实战篇4：深入DB-GPT数据应用开发框架调研
AI大模型探索之路-实战篇5：探索Open Interpreter开放代码解释器调研
AI大模型探索之路-实战篇6：掌握Function Calling的详细流程
AI大模型探索之路-实战篇7：Function Calling技术实战自动生成函数
AI大模型探索之路-实战篇8：多轮对话与Function Calling技术应用
AI大模型探索之路-实战篇9：探究Agent智能数据分析平台的架构与功能
AI大模型探索之路-实战篇10：数据预处理的艺术：构建Agent智能数据分析平台的基础
AI大模型探索之路-实战篇11： Function Calling技术整合：强化Agent智能数据分析平台功能
AI大模型探索之路-实战篇12： 构建互动式Agent智能数据分析平台：实现多轮对话控制
AI大模型探索之路-实战篇13： 从对话到报告：打造能记录和分析的Agent智能数据分析平台
AI大模型探索之路-实战篇14： 集成本地Python代码解释器：强化Agent智能数据分析平台

---

---

一、前言

在前面篇章中我们实现了Agent智能数据分析平台中的Tools和Memory相关代码落地实践，本文中我们将对这两大块功能代码进行整合封装。

二、Memory功能实现之在线云盘类封装

1、创建OpenAi客户端

## 导入依赖
import openai
import os
import numpy as np
import pandas as pd
import json
import io
from openai import OpenAI
import inspect
import pymysql
import tiktoken
from docx import Document
import matplotlib.pyplot as plt
import seaborn as sns
import tempfile
import ast
from IPython.display import display, Markdown, Code
import shutil
import copy
from openai import  APIConnectionError,AuthenticationErroropenai.api_key = os.getenv("OPENAI_API_KEY")client = OpenAI(api_key=openai.api_key)

2、定义本地云盘文件目录创建方法

import osdef create_or_get_folder(folder_name):"""根据项目创建云盘目录"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"full_path = os.path.join(base_path, folder_name)# 如果目录不存在，则创建它if not os.path.exists(full_path):os.makedirs(full_path)print(f"目录 {folder_name} 创建成功")else:print(f"目录 {folder_name} 已存在")return full_path

print(create_or_get_folder(folder_name = "测试函数"))

3、定义doc文档创建方法

def create_or_get_doc(folder_name, doc_name):"""创建或获取文件路径"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"full_path_folder=os.path.join(base_path,folder_name)file_path_doc = os.path.join(base_path+"/"+folder_name, f'{doc_name}.doc')# 检查目录是否存在，如果不存在则创建if not os.path.exists(full_path_folder):os.makedirs(full_path_folder)# 检查文件是否存在if os.path.exists(file_path_doc):# 文件存在，打开并追加内容document = Document(file_path_doc)else:# 文件不存在，创建一个新的文档对象document = Document()# 保存文档document.save(file_path_doc)return file_path_doc

create_or_get_doc(folder_name="测试函数",doc_name="数据分析")

4、定义文件内容追加方法

def append_content_in_doc(folder_name, doc_name, qa_string):""""往文件里追加内容@param folder_name=目录名，doc_name=文件名，qa_string=追加的内容"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"## 目录地址full_path_folder=base_path+"/"+folder_name## 文件地址full_path_doc = os.path.join(full_path_folder, doc_name)+".doc"  # 检查目录是否存在，如果不存在则创建if not os.path.exists(full_path_folder):os.makedirs(full_path_folder)# 检查文件是否存在if os.path.exists(full_path_doc):# 文件存在，打开并追加内容document = Document(full_path_doc)else:# 文件不存在，创建一个新的文档对象document = Document()# 追加内容document.add_paragraph(qa_string)# 保存文档document.save(full_path_doc)print(f"内容已追加到 {doc_name}")

测试

my_dict = "天青色等烟雨，而我在等你"
append_content_in_doc(folder_name="测试函数",doc_name="数据分析",qa_string=my_dict)

5、定义获取文件内容的方法

## 实现根据项目和文件获取文件内容的方法from docx import Document
import osdef get_file_content(folder_name, doc_name):"""实现根据项目名和文件名获取文件内容的方法@param project_name:项目名，file_name：文件名@return 文件内容"""# 构建文件的完整路径base_path = "/root/autodl-tmp/iquery项目/iquery云盘"file_path = os.path.join(folder_name, doc_name)full_path = os.path.join(base_path, file_path)+".doc"# 确保文件存在if not os.path.exists(full_path):return "文件不存在"try:# 加载文档doc = Document(full_path)content = []# 遍历文档中的每个段落，并收集文本for para in doc.paragraphs:content.append(para.text)# 将所有段落文本合并成一个字符串返回return '\n'.join(content)except Exception as e:return f"读取文件时发生错误: {e}"

测试

get_file_content(folder_name="测试函数",doc_name="数据分析")

6、定义清理文件内容的方法

from docx import Documentdef clear_content_in_doc(folder_name, doc_name):# 打开文档base_path = "/root/autodl-tmp/iquery项目/iquery云盘"file_path = os.path.join(base_path+"/"+folder_name, f'{doc_name}.doc')doc = Document(file_path)# 遍历每一个段落，设置其文本为空字符串for p in doc.paragraphs:for run in p.runs:run.text = ''# 保存修改后的文档doc.save(file_path)print("文档内容清除完毕")

测试

clear_content_in_doc(folder_name="测试函数",doc_name="数据分析")

7、定义获取文件列表的方法

def list_files_in_folder(folder_name):"""列举当前文件夹的全部文件"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"full_path = os.path.join(base_path,folder_name )file_names = [f for f in os.listdir(full_path) if os.path.isfile(os.path.join(full_path, f))]return file_names

测试

list_files_in_folder(folder_name="测试函数")

8、定义文件重命名的方法

def rename_doc(folder_name, doc_name, new_name):"""修改指定的文档名称"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"file_path = os.path.join(base_path+"/"+folder_name, f'{doc_name}.doc')new_file_path = os.path.join(base_path+"/"+folder_name, f'{new_name}.doc')# 重命名文件os.rename(file_path, new_file_path)return new_name

测试

rename_doc(folder_name="测试函数",doc_name="数据分析",new_name="数据可视化分析报告")

9、定义删除文件的方法

def delete_all_files_in_folder(folder_name):"""删除某文件夹内全部文件"""# 定义要删除的目录路径base_path = "/root/autodl-tmp/iquery项目/iquery云盘"full_path = os.path.join(base_path,folder_name)# 遍历整个目录for filename in os.listdir(full_path):# 构造文件或者文件夹的绝对路径file_path = os.path.join(full_path, filename)try:# 如果是文件，则删除文件if os.path.isfile(file_path) or os.path.islink(file_path):os.unlink(file_path)# 如果是文件夹，则删除文件夹elif os.path.isdir(file_path):shutil.rmtree(file_path)print("文件已清除完毕")except Exception as e:print('Failed to delete %s. Reason: %s' % (file_path, e))

测试

delete_all_files_in_folder(folder_name = "测试函数")

10、定义追加图片的方法

	from docx import Document
import matplotlib.pyplot as plt
import os
import tempfiledef append_img_in_doc(folder_name, doc_name, img):""""往文件里追加图片@param folder_name=目录名，doc_name=文件名，img=图片对象，数据类型为matplotlib.figure.Figure对象"""base_path = "/root/autodl-tmp/iquery项目/iquery云盘"## 目录地址full_path_folder=base_path+"/"+folder_name## 文件地址full_path_doc = os.path.join(full_path_folder, doc_name)+".doc"# 检查目录是否存在，如果不存在则创建if not os.path.exists(full_path_folder):os.makedirs(full_path_folder)# 检查文件是否存在if os.path.exists(full_path_doc):print(full_path_doc)# 文件存在，打开并追加内容document = Document(full_path_doc)else:# 文件不存在，创建一个新的文档对象document = Document()# 追加图片# 将matplotlib的Figure对象保存为临时图片文件with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as tmpfile:img.savefig(tmpfile.name, format='png')# 将图片插入到.docx文档中document.add_picture(tmpfile.name)# 保存文档document.save(full_path_doc)print(f"图片已追加到 {doc_name}")

import matplotlib.pyplot as plt# 创建一个图形
fig, ax = plt.subplots()
ax.plot([1, 2, 3, 4, 5])

append_img_in_doc(folder_name="测试函数",doc_name="数据分析",img=fig)

11、定义一个云盘文件操作类

class CloudFile():"""用于操作云盘文件"""def __init__(self, project_name, part_name, doc_content = None):# 项目名称，即项目文件夹名称self.project_name = project_name# 项目某部分名称，即项目文件名称self.part_name = part_name# 项目文件夹ID# 若项目文件夹ID为空，则获取项目文件夹IDfolder_path=create_or_get_folder(folder_name=project_name)# 创建时获取当前项目中其他文件名称列表self.doc_list = list_files_in_folder(folder_name=project_name)file_path = create_or_get_doc(folder_name=project_name, doc_name=part_name)# 项目文件具体内容，相当于多轮对话内容self.doc_content = doc_content# 若初始content不为空，则将其追加入文档内if doc_content != None:append_content_in_doc(folder_name=project_name, doc_name=part_name, dict_list=doc_content)def get_doc_content(self):"""根据项目某文件的文件ID，获取对应的文件内容"""     self.doc_content = get_file_content(folder_name=self.project_name, doc_name=self.part_name)return self.doc_contentdef append_doc_content(self, content):"""根据项目某文件的文件ID，追加文件内容"""  append_content_in_doc(folder_name=self.project_name,doc_name=self.part_name, dict_list=content)def clear_content(self):"""清空某文件内的全部内容"""  clear_content_in_doc(folder_name=self.project_name, doc_name=self.part_name)def delete_all_files(self):"""删除当前项目文件夹内的全部文件"""  delete_all_files_in_folder(folder_name=self.project_name)def update_doc_list(self):"""更新当前项目文件夹内的全部文件名称"""self.doc_list = list_files_in_folder(folder_name=self.project_name)def rename_doc(self, new_name):"""修改当前文件名称"""self.part_name = rename_doc_in_drive(folder_name=self.project_name, doc_name=self.part_name, new_name=new_name)

本地存储测试

c1 = CloudFile(project_name='测试项目', part_name='测试文档1')

三、Memory功能实现之消息工具类封装

1、定义消息管理类

class MessageManager():"""MessageManager，用于创建Chat模型能够接收和解读的messages对象。该对象是原始Chat模型接收的\messages对象的更高级表现形式，MessageManager类对象将字典类型的list作为其属性之一，同时还能\能区分系统消息和历史对话消息，并且能够自行计算当前对话的token量，并执能够在append的同时删\减最早对话消息，从而能够更加顺畅的输入大模型并完成多轮对话需求。"""def __init__(self, system_content_list=[], question='你好。',tokens_thr=None, project=None):self.system_content_list = system_content_list# 系统消息文档列表，相当于外部输入文档列表system_messages = []# 除系统消息外历史对话消息history_messages = []# 用于保存全部消息的listmessages_all = []# 系统消息字符串system_content = ''# 历史消息字符串，此时为用户输入信息history_content = question# 系统消息+历史消息字符串content_all = ''# 输入到messages中系统消息个数，初始情况为0num_of_system_messages = 0# 全部信息的token数量all_tokens_count = 0encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")# 将外部输入文档列表依次保存为系统消息if system_content_list != []:      for content in system_content_list:system_messages.append({"role": "system", "content": content})# 同时进行全文档拼接system_content += content# 计算系统消息tokensystem_tokens_count = len(encoding.encode(system_content))# 拼接系统消息messages_all += system_messages# 计算系统消息个数num_of_system_messages = len(system_content_list)# 若存在最大token数量限制if tokens_thr != None:# 若系统消息超出限制if system_tokens_count >= tokens_thr:print("system_messages的tokens数量超出限制，当前系统消息将不会被输入模型")            # 删除系统消息system_messages = []messages_all = []# 系统消息个数清零num_of_system_messages = 0# 系统消息token数清零system_tokens_count = 0all_tokens_count += system_tokens_count# 创建首次对话消息history_messages = [{"role": "user", "content": question}]# 创建全部消息列表messages_all += history_messages# 计算用户问题tokenuser_tokens_count = len(encoding.encode(question))# 计算总token数all_tokens_count += user_tokens_count# 若存在最大token限制if tokens_thr != None:# 若超出最大token限制if all_tokens_count >= tokens_thr:print("当前用户问题的tokens数量超出限制，该消息无法被输入到模型中")  # 同时清空系统消息和用户消息history_messages = []system_messages = []messages_all = []num_of_system_messages = 0all_tokens_count = 0# 全部messages信息self.messages = messages_all# system_messages信息self.system_messages = system_messages# user_messages信息self.history_messages = history_messages# messages信息中全部content的token数量self.tokens_count = all_tokens_count# 系统信息数量self.num_of_system_messages = num_of_system_messages# 最大token数量阈值self.tokens_thr = tokens_thr# token数计算编码方式self.encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")# message挂靠的项目self.project = project# 删除部分对话信息def messages_pop(self, manual=False, index=None):def reduce_tokens(index):drop_message = self.history_messages.pop(index)self.tokens_count -= len(self.encoding.encode(str(drop_message)))if self.tokens_thr is not None:while self.tokens_count >= self.tokens_thr:reduce_tokens(-1)if manual:if index is None:reduce_tokens(-1)elif 0 <= index < len(self.history_messages) or index == -1:reduce_tokens(index)else:raise ValueError("Invalid index value: {}".format(index))# 更新messagesself.messages = self.system_messages + self.history_messages# 增加部分对话信息def messages_append(self, new_messages):# 若是单独一个字典，或JSON格式字典if type(new_messages) is dict or type(new_messages) is openai.types.chat.chat_completion_message.ChatCompletionMessage:self.messages.append(new_messages)self.tokens_count += len(self.encoding.encode(str(new_messages)))# 若新消息也是MessageManager对象elif isinstance(new_messages, MessageManager):self.messages += new_messages.messagesself.tokens_count += new_messages.tokens_count# 重新更新history_messagesself.history_messages = self.messages[self.num_of_system_messages: ]# 再执行pop，若有需要，则会删除部分历史消息self.messages_pop()# 复制信息def copy(self):# 创建一个新的 MessageManager 对象，复制所有重要的属性system_content_str_list = [message["content"] for message in self.system_messages]new_obj = MessageManager(system_content_list=copy.deepcopy(system_content_str_list),  # 使用深复制来复制系统消息question=self.history_messages[0]["content"] if self.history_messages else '',tokens_thr=self.tokens_thr)# 复制任何其他需要复制的属性new_obj.history_messages = copy.deepcopy(self.history_messages)  # 使用深复制来复制历史消息new_obj.messages = copy.deepcopy(self.messages)  # 使用深复制来复制所有消息new_obj.tokens_count = self.tokens_countnew_obj.num_of_system_messages = self.num_of_system_messagesreturn new_obj# 增加系统消息def add_system_messages(self, new_system_content):system_content_list = self.system_content_listsystem_messages = []# 若是字符串，则将其转化为listif type(new_system_content) == str:new_system_content = [new_system_content]system_content_list.extend(new_system_content)new_system_content_str = ''for content in new_system_content:new_system_content_str += contentnew_token_count = len(self.encoding.encode(str(new_system_content_str)))self.tokens_count += new_token_countself.system_content_list = system_content_listfor message in system_content_list:system_messages.append({"role": "system", "content": message})self.system_messages = system_messagesself.num_of_system_messages = len(system_content_list)self.messages = system_messages + self.history_messages# 再执行pop，若有需要，则会删除部分历史消息self.messages_pop()# 删除系统消息def delete_system_messages(self):system_content_list = self.system_content_listif system_content_list != []:system_content_str = ''for content in system_content_list:system_content_str += contentdelete_token_count = len(self.encoding.encode(str(system_content_str)))self.tokens_count -= delete_token_countself.num_of_system_messages = 0self.system_content_list = []self.system_messages = []self.messages = self.history_messages# 清除对话消息中的function消息def delete_function_messages(self):# 用于删除外部函数消息history_messages = self.history_messages# 从后向前迭代列表for index in range(len(history_messages) - 1, -1, -1):message = history_messages[index]## 这儿估计有问题if message.get("function_call") or message.get("role") == "function":self.messages_pop(manual=True, index=index)

2、测试-查看消息管理器

msg1 = MessageManager()
msg1.system_messages

3、测试-追加消息

msg1.history_messages

msg1.messages_append({"role": "user", "content": "你好，有什么可以帮你？"})
msg1.history_messages

4、测试-删除消息

msg1.messages_pop(manual=True, index=-1)
msg1.history_messages

5、添加背景知识

# 数据字典文件
with open('/root/autodl-tmp/iquery项目/data/数据字典/iquery数据字典.md', 'r', encoding='utf-8') as f:data_dictionary = f.read()

# 数据分析报告编写专家文档
with open('/root/autodl-tmp/iquery项目/data/业务知识/本公司数据分析师业务介绍.md', 'r', encoding='utf-8') as f:da_instruct = f.read()

msg2 = MessageManager(system_content_list=[data_dictionary, da_instruct])
msg2.system_messages

输出

四、Tools功能之函数封装

1、获取表结构基本信息（工具函数）

定义一个SQL工具函数，用于获取表结构信息，作为大模型生成SQL的背景知识

## mysql hive sparksql
def sql_inter(sql_query, g='globals()'):"""用于获取iquery数据库中各张表的有关相关信息，\核心功能是将输入的SQL代码传输至iquery数据库所在的MySQL环境中进行运行，\并最终返回SQL代码运行结果。需要注意的是，本函数是借助pymysql来连接MySQL数据库。:param sql_query: 字符串形式的SQL查询语句，用于执行对MySQL中iquery数据库中各张表进行查询，并获得各表中的各类相关信息:param g: g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可:return：sql_query在MySQL中的运行结果。"""mysql_pw = "iquery_agent"connection = pymysql.connect(host='localhost',  # 数据库地址user='iquery_agent',  # 数据库用户名passwd=mysql_pw,  # 数据库密码db='iquery',  # 数据库名charset='utf8'  # 字符集选择utf8)try:with connection.cursor() as cursor:# SQL查询语句sql = sql_querycursor.execute(sql)# 获取查询结果results = cursor.fetchall()finally:connection.close()return json.dumps(results)

sql_inter(sql_query='SELECT COUNT(*) FROM user_demographics;', g=globals())

2、提取SQL数据到python变量（辅助函数）

定义一个辅助函数，用于将查询到的记录提取保存到本地python变量中

def extract_data(sql_query,df_name,g='globals()'):"""用于借助pymysql，将MySQL中的iquery数据库中的表读取并保存到本地Python环境中。:param sql_query: 字符串形式的SQL查询语句，用于提取MySQL中iquery数据库中的某张表。:param df_name: 将MySQL数据库中提取的表格进行本地保存时的变量名，以字符串形式表示。:param g: g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可:return：表格读取和保存结果"""mysql_pw = "iquery_agent"connection = pymysql.connect(host='localhost',  # 数据库地址user='iquery_agent',  # 数据库用户名passwd=mysql_pw,  # 数据库密码db='iquery',  # 数据库名charset='utf8'  # 字符集选择utf8)globals()[df_name] = pd.read_sql(sql_query, connection)return "已成功完成%s变量创建" % df_name

extract_data(sql_query = 'SELECT * FROM user_demographics;', df_name = 'user_demographics_df', g = globals())

从python变量中取出数据查看

user_demographics_df.head()

3、python代码解释器（工具函数）

def python_inter(py_code,g='globals()'):"""用于对iquery数据库中各张数据表进行查询和处理，并获取最终查询或处理结果。:param py_code: 字符串形式的Python代码，用于执行对iquery数据库中各张数据表进行操作:param g: g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可:return：代码运行的最终结果"""    # 添加图片对象，如果存在绘图代码，则创建fig对象py_code = insert_fig_object(py_code)global_vars_before = set(globals().keys())try:exec(py_code, globals())except Exception as e:return str(e)global_vars_after = set(globals().keys())new_vars = global_vars_after - global_vars_beforeif new_vars:result = {var: globals()[var] for var in new_vars}return str(result)else:try:return str(eval(py_code, globals()))except Exception as e:return "已经顺利执行代码"

检查图形对象，赋值给fig（方便通过全局变量fig进行统一的绘图对象操作）

def insert_fig_object(code_str,g='globals()'):"""为图片创建fig对象:param g: g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可"""#print("开始画图了")global fig# 检查是否已存在 fig 对象的创建if 'fig = plt.figure' in code_str or 'fig, ax = plt.subplots()' in code_str:return code_str  # 如果存在，则返回原始代码字符串# 定义可能的库别名和全名plot_aliases = ['plt.', 'matplotlib.pyplot.','plot']sns_aliases = ['sns.', 'seaborn.']# 寻找第一次出现绘图相关代码的位置first_plot_occurrence = min((code_str.find(alias) for alias in plot_aliases + sns_aliases if code_str.find(alias) >= 0), default=-1)# 如果找到绘图代码，则在该位置之前插入 fig 对象的创建if first_plot_occurrence != -1:plt_figure_index = code_str.find('plt.figure')if plt_figure_index != -1:# 寻找 plt.figure 后的括号位置，以确定是否有参数closing_bracket_index = code_str.find(')', plt_figure_index)# 如果找到了 plt.figure()，则替换为 fig = plt.figure()modified_str = code_str[:plt_figure_index] + 'fig = ' + code_str[plt_figure_index:closing_bracket_index + 1] + code_str[closing_bracket_index + 1:]else:modified_str = code_str[:first_plot_occurrence] + 'fig = plt.figure()\n' + code_str[first_plot_occurrence:]return modified_strelse:return code_str  # 如果没有找到绘图代码，则返回原始代码字符串

图形绘制代码测试

import matplotlib.pyplot as plt
# 数据
categories = ['Category 1', 'Category 2', 'Category 3', 'Category 4']
values = [4, 7, 1, 8]# 创建 Figure 对象
fig, ax = plt.subplots()# 在 Axes 对象 ax 上创建条形图
ax.bar(categories, values)# 添加标题和标签
ax.set_title('Bar Chart Example')
ax.set_xlabel('Categories')
ax.set_ylabel('Values')# 显示图表
plt.show()

insert_fig_object方法测试

code_string = """
import matplotlib.pyplot as plt
# 数据
categories = ['Category 1', 'Category 2', 'Category 3', 'Category 4']
values = [4, 7, 1, 8]# 创建 Figure 对象
fig, ax = plt.subplots()# 在 Axes 对象 ax 上创建条形图
ax.bar(categories, values)# 添加标题和标签
ax.set_title('Bar Chart Example')
ax.set_xlabel('Categories')
ax.set_ylabel('Values')# 显示图表
plt.show()
"""

print(insert_fig_object(code_str = code_string, g=globals()))

python_inter(py_code = code_string, g=globals())

查看fig对象

4、function函数信息生成器（辅助函数）

定义一个用于生成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_descriptionuser_prompt = '根据这个函数的函数说明，请帮我创建一个JSON格式的字典，这个字典有如下5点要求：\1.字典总共有三个键值对；\2.第一个键值对的Key是字符串name，value是该函数的名字：%s，也是字符串；\3.第二个键值对的Key是字符串description，value是该函数的函数的功能说明，也是字符串；\4.第三个键值对的Key是字符串parameters，value是一个JSON Schema对象，用于说明该函数的参数输入规范。\5.输出结果必须是一个JSON格式的字典，只输出这个字典即可，前后不需要任何前后修饰或说明的语句' % function_nameresponse = 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 functionsmax_attempts = 4attempts = 0while 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

5、function函数调用辅助类

负责承接外部函数调用时相关功能支持。
类属性包括外部函数列表、外部函数参数说明列表、以及调用方式说明三项。

class AvailableFunctions():"""外部函数类，主要负责承接外部函数调用时相关功能支持。类属性包括外部函数列表、外部函数参数说明列表、以及调用方式说明三项。"""def __init__(self, functions_list=[], functions=[], function_call="auto"):self.functions_list = functions_listself.functions = functionsself.functions_dic = Noneself.function_call = None# 当外部函数列表不为空、且外部函数参数解释为空时，调用auto_functions创建外部函数解释列表if functions_list != []:self.functions_dic = {func.__name__: func for func in functions_list}self.function_call = function_callif functions == []:self.functions = auto_functions(functions_list)# 增加外部函数方法，并且同时可以更换外部函数调用规则def add_function(self, new_function, function_description=None, function_call_update=None):self.functions_list.append(new_function)self.functions_dic[new_function.__name__] = new_functionif function_description == None:new_function_description = auto_functions([new_function])self.functions.append(new_function_description)else:self.functions.append(function_description)if function_call_update != None:self.function_call = function_call_update

af = AvailableFunctions(functions_list=[sql_inter, extract_data, python_inter])
af.functions_list

af.functions_dic

af.function_call

af.functions

输出：

[{'type': 'function','function': {'name': 'sql_inter','description': '用于获取iquery数据库中各张表的有关相关信息，核心功能是将输入的SQL代码传输至iquery数据库所在的MySQL环境中进行运行，并最终返回SQL代码运行结果。','parameters': {'type': 'object','properties': {'sql_query': {'type': 'string','description': '字符串形式的SQL查询语句，用于执行对MySQL中iquery数据库中各张表进行查询，并获得各表中的各类相关信息。'},'g': {'type': 'string','description': 'g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可。'}},'required': ['sql_query', 'g']}}},{'type': 'function','function': {'name': 'extract_data','description': '用于借助pymysql，将MySQL中的iquery数据库中的表读取并保存到本地Python环境中。','parameters': {'type': 'object','properties': {'sql_query': {'type': 'string','description': '字符串形式的SQL查询语句，用于提取MySQL中iquery数据库中的某张表。'},'df_name': {'type': 'string','description': '将MySQL数据库中提取的表格进行本地保存时的变量名，以字符串形式表示。'},'g': {'type': 'string', 'description': 'g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可'}},'required': ['sql_query', 'df_name']}}},{'type': 'function','function': {'name': 'python_inter','description': '用于对iquery数据库中各张数据表进行查询和处理，并获取最终查询或处理结果。','parameters': {'$schema': 'http://-schema.org/draft-07/schema#','type': 'object','properties': {'py_code': {'type': 'string','description': '字符串形式的Python代码，用于执行对iquery数据库中各张数据表进行操作'},'g': {'type': 'string', 'description': 'g，字符串形式变量，表示环境变量，无需设置，保持默认参数即可'}},'required': ['py_code', 'g']}}}]

三、结语

本文中我们封装落地了Agent智能数据分析平台中的Tools和Memory相关代码，下一篇章中我将落地实践Agent智能数据分析平台的核心模块Plan，探索发掘人类意图，优化整个决策流程。

🎯🔖更多专栏系列文章：AIGC-AI大模型探索之路

😎 作者介绍：我是寻道AI小兵，资深程序老猿，从业10年+、互联网系统架构师，目前专注于AIGC的探索。
📖 技术交流：建立有技术交流群，可以扫码👇 加入社群，500本各类编程书籍、AI教程、AI工具等你领取！
如果文章内容对您有所触动，别忘了点赞、⭐关注，收藏！加入我，让我们携手同行AI的探索之旅，一起开启智能时代的大门！