微网站怎么做的好名字吗,58企业网站怎么做,网站建设价格套餐,工程中标查询网站系列篇章#x1f4a5;
AI大模型探索之路-实战篇4#xff1a;深入DB-GPT数据应用开发框架调研 AI大模型探索之路-实战篇5#xff1a;探索Open Interpreter开放代码解释器调研 AI大模型探索之路-实战篇6#xff1a;掌握Function Calling的详细流程 AI大模型探索之路-实战篇7…系列篇章
AI大模型探索之路-实战篇4深入DB-GPT数据应用开发框架调研 AI大模型探索之路-实战篇5探索Open Interpreter开放代码解释器调研 AI大模型探索之路-实战篇6掌握Function Calling的详细流程 AI大模型探索之路-实战篇7Function 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智能数据分析平台 目录 系列篇章一、前言二、Memory功能实现之在线云盘类封装1、创建OpenAi客户端2、定义本地云盘文件目录创建方法3、定义doc文档创建方法4、定义文件内容追加方法5、定义获取文件内容的方法6、定义清理文件内容的方法7、定义获取文件列表的方法8、定义文件重命名的方法9、定义删除文件的方法10、定义追加图片的方法11、定义一个云盘文件操作类 三、Memory功能实现之消息工具类封装1、定义消息管理类2、测试-查看消息管理器3、测试-追加消息4、测试-删除消息5、添加背景知识 四、Tools功能之函数封装1、获取表结构基本信息工具函数2、提取SQL数据到python变量辅助函数3、python代码解释器工具函数4、function函数信息生成器辅助函数5、function函数调用辅助类 三、结语 一、前言
在前面篇章中我们实现了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_keyopenai.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_pathprint(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_folderos.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_doccreate_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_folderbase_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_stringmy_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_folderbase_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(deleteFalse, suffix.png) as tmpfile:img.savefig(tmpfile.name, formatpng)# 将图片插入到.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数据分析,imgfig)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_pathcreate_or_get_folder(folder_nameproject_name)# 创建时获取当前项目中其他文件名称列表self.doc_list list_files_in_folder(folder_nameproject_name)file_path create_or_get_doc(folder_nameproject_name, doc_namepart_name)# 项目文件具体内容相当于多轮对话内容self.doc_content doc_content# 若初始content不为空则将其追加入文档内if doc_content ! None:append_content_in_doc(folder_nameproject_name, doc_namepart_name, dict_listdoc_content)def get_doc_content(self):根据项目某文件的文件ID获取对应的文件内容 self.doc_content get_file_content(folder_nameself.project_name, doc_nameself.part_name)return self.doc_contentdef append_doc_content(self, content):根据项目某文件的文件ID追加文件内容 append_content_in_doc(folder_nameself.project_name,doc_nameself.part_name, dict_listcontent)def clear_content(self):清空某文件内的全部内容 clear_content_in_doc(folder_nameself.project_name, doc_nameself.part_name)def delete_all_files(self):删除当前项目文件夹内的全部文件 delete_all_files_in_folder(folder_nameself.project_name)def update_doc_list(self):更新当前项目文件夹内的全部文件名称self.doc_list list_files_in_folder(folder_nameself.project_name)def rename_doc(self, new_name):修改当前文件名称self.part_name rename_doc_in_drive(folder_nameself.project_name, doc_nameself.part_name, new_namenew_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_thrNone, projectNone):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, manualFalse, indexNone):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_listcopy.deepcopy(system_content_str_list), # 使用深复制来复制系统消息questionself.history_messages[0][content] if self.history_messages else ,tokens_thrself.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(manualTrue, indexindex)2、测试-查看消息管理器
msg1 MessageManager()
msg1.system_messages3、测试-追加消息
msg1.history_messagesmsg1.messages_append({role: user, content: 你好有什么可以帮你})
msg1.history_messages4、测试-删除消息
msg1.messages_pop(manualTrue, index-1)
msg1.history_messages5、添加背景知识
# 数据字典文件
with open(/root/autodl-tmp/iquery项目/data/数据字典/iquery数据字典.md, r, encodingutf-8) as f:data_dictionary f.read()# 数据分析报告编写专家文档
with open(/root/autodl-tmp/iquery项目/data/业务知识/本公司数据分析师业务介绍.md, r, encodingutf-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, gglobals()):用于获取iquery数据库中各张表的有关相关信息\核心功能是将输入的SQL代码传输至iquery数据库所在的MySQL环境中进行运行\并最终返回SQL代码运行结果。需要注意的是本函数是借助pymysql来连接MySQL数据库。:param sql_query: 字符串形式的SQL查询语句用于执行对MySQL中iquery数据库中各张表进行查询并获得各表中的各类相关信息:param g: g字符串形式变量表示环境变量无需设置保持默认参数即可:returnsql_query在MySQL中的运行结果。mysql_pw iquery_agentconnection pymysql.connect(hostlocalhost, # 数据库地址useriquery_agent, # 数据库用户名passwdmysql_pw, # 数据库密码dbiquery, # 数据库名charsetutf8 # 字符集选择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_querySELECT COUNT(*) FROM user_demographics;, gglobals())2、提取SQL数据到python变量辅助函数
定义一个辅助函数用于将查询到的记录提取保存到本地python变量中
def extract_data(sql_query,df_name,gglobals()):用于借助pymysql将MySQL中的iquery数据库中的表读取并保存到本地Python环境中。:param sql_query: 字符串形式的SQL查询语句用于提取MySQL中iquery数据库中的某张表。:param df_name: 将MySQL数据库中提取的表格进行本地保存时的变量名以字符串形式表示。:param g: g字符串形式变量表示环境变量无需设置保持默认参数即可:return表格读取和保存结果mysql_pw iquery_agentconnection pymysql.connect(hostlocalhost, # 数据库地址useriquery_agent, # 数据库用户名passwdmysql_pw, # 数据库密码dbiquery, # 数据库名charsetutf8 # 字符集选择utf8)globals()[df_name] pd.read_sql(sql_query, connection)return 已成功完成%s变量创建 % df_nameextract_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,gglobals()):用于对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,gglobals()):为图片创建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, gglobals()))python_inter(py_code code_string, gglobals())查看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是字符串namevalue是该函数的名字%s也是字符串\3.第二个键值对的Key是字符串descriptionvalue是该函数的函数的功能说明也是字符串\4.第三个键值对的Key是字符串parametersvalue是一个JSON Schema对象用于说明该函数的参数输入规范。\5.输出结果必须是一个JSON格式的字典只输出这个字典即可前后不需要任何前后修饰或说明的语句 % function_nameresponse client.chat.completions.create(modelgpt-3.5-turbo,messages[{role: system, content: system_prompt},{role: user, content: user_prompt}])json_function_descriptionjson.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 functions5、function函数调用辅助类
负责承接外部函数调用时相关功能支持。 类属性包括外部函数列表、外部函数参数说明列表、以及调用方式说明三项。
class AvailableFunctions():外部函数类主要负责承接外部函数调用时相关功能支持。类属性包括外部函数列表、外部函数参数说明列表、以及调用方式说明三项。def __init__(self, functions_list[], functions[], function_callauto):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_descriptionNone, function_call_updateNone):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_updateaf AvailableFunctions(functions_list[sql_inter, extract_data, python_inter])
af.functions_listaf.functions_dicaf.function_callaf.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的探索之旅一起开启智能时代的大门
