本篇旨在將前一篇的聊天機器人改成串流版.
前一篇測試中我們已利用 JSON 格式化回覆字串模板與採用對話歷史成功地製做了一個有記憶且能自動判斷是否需要搜尋網路的 OpenAI 聊天機器人, 當提出的問題超出模型的認知範圍時, 應用程式會去搜尋網路取得參考資料提供給模型去生成回應, 這種情況需要呼叫 GPT 模型兩次, 前一篇測試為了簡單起見都使用非串流回應, 事實上第二次呼叫 GPT 時可以使用串流回應 (但第一次因為要完整取得 JSON 格式的回應所以不能用串流方式), 作法參考 :
完整程式碼如下 :
# cli_chatbot_search_7.py
from openai import OpenAI, APIError
import requests
import json
def ask_gpt(messages, model='gpt-3.5-turbo'):
try:
reply=client.chat.completions.create(
model=model,
messages=messages
)
return reply.choices[0].message.content
except APIError as e:
return str(e)
def ask_gpt_s(messages, model='gpt-3.5-turbo'):
try:
# 啟用 stream=True 讓 GPT 逐步回應
chunks=client.chat.completions.create(
model=model,
messages=messages,
stream=True # 啟用串流
)
reply='' # 儲存完整回應
print('GPT :', end=' ', flush=True) # 立即輸出回應字元
for chunk in chunks:
if chunk.choices and chunk.choices[0].delta.content:
text=chunk.choices[0].delta.content or '' # 用 or 處理 None 問題
reply += text
print(text, end='', flush=True) # 即時輸出
print() # 換行
return reply # 回傳完整回應
except APIError as e:
return str(e)
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
}
url='https://www.googleapis.com/customsearch/v1'
r=requests.get(url, params=params)
data=r.json()
return data.get('items', [])
# 設定金鑰變數
openai_key='你的 OpenAI API Key'
custom_search_key='你的 Custom Search JSON API Key'
cx='你的 Custom Search Engine ID' # Custom Search Engine ID
# 建立 OpenAI 物件
client=OpenAI(api_key=openai_key)
# 設定 GPT 系統角色
sys_role=input('請設定 GPT 角色 : ')
if sys_role.strip() == '':
sys_role='你是繁體中文AI助理'
print(sys_role)
# 建立 JSON 格式回覆字串模板
template='''
請確認你是否知道下面這件事:
{}
如果知道, 請務必只用下列 JSON 格式回答:
{{
"need_search": "N",
"keyword": "",
"reply":"你的答案"
}}
如果不知道, 請務必只用下列 JSON 格式回答:
{{
"need_search": "Y",
"keyword": "你建議的搜尋關鍵字",
"reply": ""
}}
'''
# 初始化聊天訊息 (用來記憶聊天歷史)
messages=[{'role': 'system', 'content': sys_role}]
# 聊天機器人 (無窮迴圈)
while True:
prompt=input('You : ') # 輸入提示詞
if prompt.strip() == '': # 按 Enter 跳出迴圈結束聊天
print('GPT : 再見。')
break
# 記錄使用者輸入到歷史訊息
messages.append({'role': 'user', 'content': prompt})
# 初次查詢 GPT : 確定是否需要搜尋網路
json_query_str=template.format(prompt)
messages.append({'role': 'user', 'content': json_query_str})
reply=ask_gpt(messages) # 呼叫 OpenAI API
messages.pop() # 移除這個判斷訊息避免影響聊天歷史
result=json.loads(reply) # 將 JSON 字串轉成字典
if result['need_search']=='Y': # 需要搜尋網路
keyword=result['keyword'] # 剔除 '#s' 取得使用者之提示詞
web_msg='以下是 Google 搜尋所得資料:\n' # 初始化搜尋結果字串
for item in search_google(keyword, cx, custom_search_key):
web_msg += f'標題: {item["title"]}\n'
web_msg += f'描述: {item["snippet"]}\n\n'
web_msg += '請依據上述資料回答下列問題(**直接給出答案**):\n' + prompt
# 記錄 Google 搜尋結果
messages.append({'role': 'system', 'content': web_msg})
# 再次查詢 GPT : 根據所提供的網路搜尋結果回答
gpt_reply=ask_gpt_s(messages)
else: # 不需搜尋網路, 加入使用者提示詞
gpt_reply=result['reply']
print(f'GPT : {gpt_reply}')
# 記錄 GPT 回應到歷史訊息
messages.append({'role': 'assistant', 'content': gpt_reply})
主要的修改之處如下 :
- 新增了一個串流版呼叫 GPT 模型的函式 ask_gpt_s() 給搜尋完網路後第二次呼叫 GPT 模型時使用, 它會將收到的一個個 token 即時輸出, 但同時也會串接為完整的回應字串傳回給主程式來儲存對話歷史.
- 由於第二次呼叫 GPT 的回應已在 ask_gpt_s() 裡面以串流方式輸出, 因此在不需要搜尋網路的 else 區塊要用 print() 輸出回應.
執行結果如下 :
>>> %Run cli_chatbot_search_7.py
請設定 GPT 角色 :
你是繁體中文AI助理
You : 2024台灣總統大選是誰當選?
GPT : 賴清德當選為2024年台灣總統。
You : 那 2018 年呢?
GPT : 蔡英文當選為2018年台灣總統。
You : 那 2010 年呢?
GPT : 馬英九當選為2010年台灣總統。
You : 2024 台灣金曲歌王是誰?
GPT : 2024年台灣金曲歌王是MC HotDog。
You : 愛情是甚麼?
GPT : 愛情是強烈的依附和愛的感覺的發展，通常是對另一個人的情感結合，是受社會因素影響的生理、心理和主觀情感的複雜現象。
You : 請詳細說明
GPT : 愛情是生理、心理和主觀情感結合的複雜現象，受到社會因素影響。不同時代、文化、學科和學者對愛情有著不同的理解和定義，愛情通常包含亲密、激情、承諾這三個要素。有學者將愛情分為不同類型，包括基於對方外表的情慾之愛、不重承諾只欲征服的遊戲之愛、重情重義的友誼之愛、雙方皆可得益的現實之愛、妒忌與爭執的依附之愛以及完全無私的愛等。
You : 甚麼是量子疊加態?
GPT : 量子叠加态是指一个量子系统的几个量子态经过归一化的线性组合所得到的状态，在量子力学中，系统可以同时处于多个可能的状态之一，这种状态被称为量子叠加态。
You : 請進一步說明.
GPT : 量子叠加态是量子力学中的一个重要概念，表示量子系统可以同时处于多个可能的状态之一。这种状态的性质可以用数学上的线性组合来描述，是量子态的一个基本特征之一。量子叠加态的具体物理意义和应用在量子力学和量子信息科学等领域有着重要的应用和研究价值。
You :
GPT : 再見。
可見此程式的回應有兩種, 如果是需要搜尋網路的, 回應會以串流方式像打字機那樣一個字一個字輸出模型的回應; 反之則會一次輸出全部 (取得 JSON 回應字串). 另外, 從後面關於量子態的回應為殘體中文可知, 即使已經設定 GPT 角色為 "繁體中文 AI 助理", 但還是有些時候輸出殘體中文 (隨機性緣故, 雖然機率較低).
沒有留言 :
張貼留言