系列篇章💥

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的探索之旅，一起開啟智能時代的大門！