OpenAI API 學習筆記 : 用 Function calling 讓模型開外掛 (三)

在前一面的測試中, 我們已利用 Function calling 功能實作了一個可呼叫外部函式 (搜尋 Google)  的命令列聊天機器人, 並且加上了記憶歷史對話的功能, 參考 :

# OpenAI API 學習筆記 : 用 Function calling 讓模型開外掛 (一)

# OpenAI API 學習筆記 : 用 Function calling 讓模型開外掛 (二)

本篇則是要用 Gradio 的 Chatbot 元件將命令列介面改寫為一個網頁應用程式 (Web app) 介面, 做法上最主要的差別是, 在 Chatbot 元件上顯示對話泡泡時, 紀錄對話歷史的字典串列中的每個對話字典必須只能包含 user 與 assistant 這兩種鍵, 否則會出現例外錯誤. 

本系列全部文章索引參考 : 

# OpenAI API 學習筆記索引 

1. 使用串列紀錄聊天歷史 : 

Gradio 提供 Chatbot 與 Blocks 來實作聊天機器人, 用法參考 :

# Gradio 學習筆記索引

此外還提供 State 元件可用來記憶聊天歷史, 此處先來實作不使用 State 元件, 而是使用一般串列來管理對話歷史, 程式碼如下 : 

# gradio_chatbot_function_calling_1.py

import gradio as gr

from openai import OpenAI, APIError

from dotenv import load_dotenv

import os

import json

import requests

# 載入環境變數並檢查

load_dotenv()

openai_api_key=os.environ.get('OPENAI_API')

custom_search_key=os.environ.get('GOOGLE_CUSTOM_SEARCH_API')

cx=os.environ.get('SEARCH_ENGINE_ID')

if not all([openai_api_key, custom_search_key, cx]):

    raise ValueError("缺少必要的環境變數")

# 初始化 OpenAI 客戶端

client=OpenAI(api_key=openai_api_key)

# 定義外部工具函式串列

tools=[{

    "type": "function",

    "function": {

        "name": "search_google",

        "description": "用 Google Custom Search API 取得搜尋結果",

        "parameters": {

            "type": "object",

            "properties": {

                "query": {"type": "string", "description": "搜尋關鍵字"},

                "cx": {"type": "string", "description": "搜尋引擎 ID"},

                "api_key": {"type": "string", "description": "API 金鑰"},

                "num": {"type": "integer", "description": "結果數量", "default": 3},

                "gl": {"type": "string", "description": "地理區域", "default": "tw"},

                "lr": {"type": "string", "description": "語言", "default": "lang_zh_TW"}

                },

            "required": ["query", "cx", "api_key"]

            }

        }

    }]

# 定義外部工具函式 : 搜尋 Google Custom Search 函式

def search_google(query, cx, api_key, num=3, gl='tw', lr='lang_zh_TW'):

    params={'q': query, 'key': api_key, 'cx': cx, 'gl': gl, 'lr': lr, 'num': num}

    try:

        r=requests.get('https://www.googleapis.com/customsearch/v1', params=params)

        r.raise_for_status()

        data=r.json()

        if 'error' in data:

            return f"Google API 錯誤：{data['error']['message']}"

        return data.get('items', [])

    except requests.RequestException as e:

        return f'Google 搜尋失敗：{str(e)}'

# OpenAI API 調用

def ask_gpt(messages, tools=[], model='gpt-3.5-turbo'):

    try:

        reply=client.chat.completions.create(model=model, messages=messages, tools=tools)

        if not reply.choices or not reply.choices[0].message:

            return 'API 回應格式錯誤'

        return reply

    except APIError as e:

        return f'OpenAI API 錯誤：{str(e)}'

# 聊天函數

def chat(prompt, model='gpt-3.5-turbo'):

    global messages  # 取用全域變數 (對話歷史)

    messages.append({'role': 'user', 'content': prompt}) # 儲存提示詞

    # 第一次調用：檢查是否需要工具

    reply=ask_gpt(messages, tools=tools, model=model)

    if isinstance(reply, str): # 字串=直接回應 (答案或錯誤)

        messages.append({'role': 'assistant', 'content': reply})

        return messages, ''

    # 依據 tool_cools 判斷是否需呼叫外部工具函式

    reply_message=reply.choices[0].message # ChatCompletionMessage 物件

    if reply_message.tool_calls: # tool_calls 不是 None : 需呼叫函式

        messages.append({  # 將要呼叫的函式資訊存入對話歷史

            'role': 'assistant',

            'tool_calls': [

                {'id': tc.id,

                 'type': 'function',

                 'function': {

                     'name': tc.function.name,

                     'arguments': tc.function.arguments

                     }

                 } for tc in reply_message.tool_calls]

            })  

        for tool_call in reply_message.tool_calls: # 逐一呼叫串列中的函式

            function_name=tool_call.function.name  # 取得函式名稱

            args=json.loads(tool_call.function.arguments)  # 取得函式參數

            args['api_key']=custom_search_key  # 更改為自己的金鑰

            args['cx']=cx  # 更改為自己的搜尋引擎ID

            func=globals().get(function_name)

            if not func:  # 函式不存在 

                messages.append({  # 將無法呼叫之函式資訊存入對話歷史

                    'role': 'tool',

                    'tool_call_id': tool_call.id,

                    'content': f'函式 {function_name} 未找到'

                    })  

                continue

            results=func(**args)  # 函式存在 : 呼叫外部函式

            if isinstance(results, str):  # 傳回值為字串 (出現錯誤)

                content=results  # 儲存錯誤訊息

            else:  # 傳回值為串列 (搜尋結果) : 將串列內容串成字串

                r=[f'標題: {r["title"]}\n描述: {r["snippet"]}' for r in results]

                content="\n".join(r)

            messages.append({  # 將搜尋結果字串與呼叫之函式資訊存入對話歷史

                'role': 'tool',

                'tool_call_id': tool_call.id,

                'name': function_name,

                'content': content

                })

        # 第二次呼叫模型：根據搜尋結果回答

        reply2=ask_gpt(messages, tools=tools, model=model)

        if isinstance(reply2, str): # 傳回值為字串: 直接回應

            messages.append({'role': 'assistant', 'content': reply2})

        else:

            messages.append({'role': 'assistant', 'content': reply2.choices[0].message.content})

    else:

        messages.append({'role': 'assistant', 'content': reply_message.content})

    # 限制對話歷史長度

    if len(messages) > MAX_HISTORY_LENGTH:

        messages=[messages[0]] + messages[-(MAX_HISTORY_LENGTH - 1):]

    # 過濾 : Chatbot 元件僅能顯示 role 為 user 和 assistant 之訊息

    display_messages=[

        {"role": m["role"], "content": m["content"]}

        for m in messages

        if m.get("role") in ("user", "assistant") and m.get("content")

        ]

    return display_messages, ''  # 傳回顯示訊息並清除輸入框

# 初始化

messages=[{'role': 'system', 'content': '你是繁體中文AI助理'}]

MAX_HISTORY_LENGTH=10  # 設定對話歷史長度

# Gradio 介面

with gr.Blocks(title="OpenAI 聊天機器人") as blocks:

    chatbot=gr.Chatbot(type='messages', height=400, placeholder='我們的對話', show_copy_button=True)

    with gr.Column():

        prompt=gr.Textbox(label="您的詢問:")

        model_select=gr.Dropdown(choices=['gpt-3.5-turbo', 'gpt-4'], label="選擇模型", value='gpt-3.5-turbo')

        send_btn=gr.Button("送出")

    send_btn.click(chat, inputs=[prompt, model_select], outputs=[chatbot, prompt])

    prompt.submit(chat, inputs=[prompt, model_select], outputs=[chatbot, prompt])

blocks.launch()

請注意上面程式碼中的藍色字部分, 在前兩篇測試中, 第一次查詢 GPT 時攜帶 tools 參數上去, 當收到需要呼叫外部函式的回應時, 我們直接將回應中的 ChatCompletionMessage 物件放進對話歷史串列 messages 中; 事實上根據最新 OpenAI API SDK (v1.x) 的要求, 應該將其內容打包為包含 role 鍵的字典後再放進 messages 串列中為宜, 此處是將外部函式 id, 函式名稱, 以及參數打幫放在 tool_calls 鍵裡面後以 assistant 角色放進 messages 串列中. 

其次是在最後要將 messages 串列傳回給 Chatbot 輸出包含 role 鍵 (a)的內容必須與傳送給模型的訊息格式一樣, 遵照最新 OpenAI API SDK (v1.x) 的要求包含 role 與 content 等鍵

執行結果如下 : 

2. 使用 State 物件紀錄聊天歷史 : 

上面範例使用一個串列 messages 來記錄對話歷史, 事實上 Gradio 有提供一個 State 物件用來記憶聊天紀錄, 用法參考 :

# Python 學習筆記 : 簡單好用的 Web app 套件 gradio (九)

# OpenAI API 學習筆記 : 結合搜尋引擎的聊天機器人 (五)

將上面開外掛的聊天機器人程式改寫為如下使用 State 物件的版本 : 

# gradio_chatbot_function_calling_2.py

import gradio as gr

from openai import OpenAI, APIError

from dotenv import load_dotenv

import os

import json

import requests

# 載入環境變數並檢查

load_dotenv()

openai_api_key=os.environ.get('OPENAI_API')

custom_search_key=os.environ.get('GOOGLE_CUSTOM_SEARCH_API')

cx=os.environ.get('SEARCH_ENGINE_ID')

if not all([openai_api_key, custom_search_key, cx]):

    raise ValueError("缺少必要的環境變數")

# 初始化 OpenAI 客戶端

client=OpenAI(api_key=openai_api_key)

# 定義外部工具函式串列

tools=[{

    "type": "function",

    "function": {

        "name": "search_google",

        "description": "用 Google Custom Search API 取得搜尋結果",

        "parameters": {

            "type": "object",

            "properties": {

                "query": {"type": "string", "description": "搜尋關鍵字"},

                "cx": {"type": "string", "description": "搜尋引擎 ID"},

                "api_key": {"type": "string", "description": "API 金鑰"},

                "num": {"type": "integer", "description": "結果數量", "default": 3},

                "gl": {"type": "string", "description": "地理區域", "default": "tw"},

                "lr": {"type": "string", "description": "語言", "default": "lang_zh_TW"}

                },

            "required": ["query", "cx", "api_key"]

            }

        }

    }]

# 定義外部工具函式 : 搜尋 Google Custom Search 函式

def search_google(query, cx, api_key, num=3, gl='tw', lr='lang_zh_TW'):

    params={'q': query, 'key': api_key, 'cx': cx, 'gl': gl, 'lr': lr, 'num': num}

    try:

        r=requests.get('https://www.googleapis.com/customsearch/v1', params=params)

        r.raise_for_status()

        data=r.json()

        if 'error' in data:

            return f"Google API 錯誤：{data['error']['message']}"

        return data.get('items', [])

    except requests.RequestException as e:

        return f'Google 搜尋失敗：{str(e)}'

# OpenAI API 調用

def ask_gpt(messages, tools=[], model='gpt-3.5-turbo'):

    try:

        reply=client.chat.completions.create(model=model, messages=messages, tools=tools)

        if not reply.choices or not reply.choices[0].message:

            return 'API 回應格式錯誤'

        return reply

    except APIError as e:

        return f'OpenAI API 錯誤：{str(e)}'

# 聊天函數

def chat(messages, prompt, model='gpt-3.5-turbo'):

    messages.append({'role': 'user', 'content': prompt}) # 儲存提示詞

    # 第一次調用：檢查是否需要工具

    reply=ask_gpt(messages, tools=tools, model=model)

    if isinstance(reply, str): # 字串=直接回應 (答案或錯誤)

        messages.append({'role': 'assistant', 'content': reply})

        return messages, messages, ''  # 更新 state, chatbot, prompt

    # 依據 tool_cools 判斷是否需呼叫外部工具函式

    reply_message=reply.choices[0].message # ChatCompletionMessage 物件

    if reply_message.tool_calls: # tool_calls 不是 None : 需呼叫函式

        messages.append({  # 將要呼叫的函式資訊存入對話歷史

            'role': 'assistant',

            'tool_calls': [

                {'id': tc.id,

                 'type': 'function',

                 'function': {

                     'name': tc.function.name,

                     'arguments': tc.function.arguments

                     }

                 } for tc in reply_message.tool_calls]

            })  

        for tool_call in reply_message.tool_calls: # 逐一呼叫串列中的函式

            function_name=tool_call.function.name  # 取得函式名稱

            args=json.loads(tool_call.function.arguments)  # 取得函式參數

            args['api_key']=custom_search_key  # 更改為自己的金鑰

            args['cx']=cx  # 更改為自己的搜尋引擎ID

            func=globals().get(function_name)

            if not func:  # 函式不存在 

                messages.append({  # 將無法呼叫之函式資訊存入對話歷史

                    'role': 'tool',

                    'tool_call_id': tool_call.id,

                    'content': f'函式 {function_name} 未找到'

                    })  

                continue

            results=func(**args)  # 函式存在 : 呼叫外部函式

            if isinstance(results, str):  # 傳回值為字串 (出現錯誤)

                content=results  # 儲存錯誤訊息

            else:  # 傳回值為串列 (搜尋結果) : 將串列內容串成字串

                r=[f'標題: {r["title"]}\n描述: {r["snippet"]}' for r in results]

                content="\n".join(r)

            messages.append({  # 將搜尋結果字串與呼叫之函式資訊存入對話歷史

                'role': 'tool',

                'tool_call_id': tool_call.id,

                'name': function_name,

                'content': content

                })

        # 第二次呼叫模型：根據搜尋結果回答

        reply2=ask_gpt(messages, tools=tools, model=model)

        if isinstance(reply2, str): # 傳回值為字串: 直接回應

            messages.append({'role': 'assistant', 'content': reply2})

        else:

            messages.append({'role': 'assistant', 'content': reply2.choices[0].message.content})

    else:

        messages.append({'role': 'assistant', 'content': reply_message.content})

    # 限制對話歷史長度

    if len(messages) > MAX_HISTORY_LENGTH:

        messages=[messages[0]] + messages[-(MAX_HISTORY_LENGTH - 1):]

    # 過濾 : Chatbot 元件僅能顯示 role 為 user 和 assistant 之訊息

    display_messages=[

        {"role": m["role"], "content": m["content"]}

        for m in messages

        if m.get("role") in ("user", "assistant") and m.get("content")

        ]

    return messages, display_messages, ''  # 傳回 state, chatbot, prompt

MAX_HISTORY_LENGTH=10  # 設定對話歷史長度

# Gradio 介面

with gr.Blocks(title="OpenAI 聊天機器人") as blocks:

    chatbot=gr.Chatbot(type='messages',

                       height=400,

                       placeholder='我們的對話',

                       show_copy_button=True)

    # 使用 State 儲存對話歷史

    state=gr.State([{'role': 'system', 'content': '你是繁體中文AI助理'}])  

    with gr.Column():

        prompt=gr.Textbox(label="您的詢問:")

        model_select=gr.Dropdown(choices=['gpt-3.5-turbo', 'gpt-4'],

                                 label="選擇模型",

                                 value='gpt-3.5-turbo')

        send_btn=gr.Button("送出")

    send_btn.click(chat, inputs=[state, prompt, model_select], outputs=[state, chatbot, prompt])

    prompt.submit(chat, inputs=[state, prompt, model_select], outputs=[state, chatbot, prompt])

blocks.launch()

與上面範例不同的是, 此處將儲存對話歷史的 State 物件做為輸入與輸出元件的第一元素, [state, prompt, model_select] 對應到按鈕事件 (click) 與提交 (submit) 事件的處理函式 chat() 的傳入參數 (messages, prompt, model); 同樣地, chat() 傳回的 messages, display_messages, '' 元組則對應到 outputs 的三元素 [state, chatbot, prompt]. 

其實 State 物件就是包裝過的串列, 用法與上面的 messages 串列完全相同, 所以此例同樣可用 MAX_HISTORY_LENGTH 來限制對話歷史的記憶長度.