大語言模型插件功能在攜程的Python實踐
作者簡介
成學，攜程高級安全研發工程師，關注Python/Golang后端開發、大語言模型等領域。
一、背景
2023年初，科技圈最火爆的話題莫過于大語言模型了，它是一種全新的聊天機器人模型，除了能應對基本的日常聊天外，還能勝任如文案編寫、旅游規劃等各項工作。
攜程信息安全部也緊跟時代步伐，在攜程內部推出基于大語言模型的智能聊天機器人，并發布網頁版1.0，讓所有“程里人”可以便捷無門檻地享受大語言模型帶來的便利。在運營半年多后，我們廣泛收集了用戶的使用感受以及建議，結合當下大語言模型的發展進程，新推出了網頁版2.0，相較于1.0版本，簡化了頁面展示，增加了歷史會話保存、自定義對話設置、支持插件、AI繪圖等功能。
其中插件功能可以擴展大語言模型的個性化能力，讓其如虎添翼，開發者可以根據特定需求定義特定插件供用戶使用。目前網頁版2.0版本支持10多種插件，如Google搜索插件可以對輸入的問題進行聯網查詢；Base64插件可以對Base64內容進行解密等。那么這些插件是如何實現的呢？讓我們一探究竟吧。
二、需求調研
2.1 Function Calling
目前多個大語言模型都推出了Function Calling（函數調用）能力，用于幫助開發者通過 API 方式實現類似于插件的能力。通過Function Calling，我們可以將多個自定義的函數描述連同提出的問題一起傳給大語言模型，它會分析這些函數描述與提問內容的相關性，并將最相關的函數及對應的函數傳參一起返回，我們再執行函數對應的業務邏輯，即可得到問題的答案。
Function Calling的大體使用流程如下：
舉個例子，我們的問題是“今天上海天氣如何”，大語言模型本身不能聯網，不知道當前上海的天氣信息，但是按照Function Calling的使用步驟，我們可以回答這個問題：
- 我們可以事先在應用服務端定義一個查詢天氣的函數，函數描述為“查詢某時某地的天氣情況”，函數傳參為“日期”和“地點”，在函數內部編寫具體查詢天氣的代碼，如從氣象局網站獲取對應的天氣信息。
- 除此之外，我們還可以定義一些其他自定義的函數，如base65加解密函數、ip信息查詢函數等。
- 在調用大語言模型的API時，將這些定義的函數按照api規范連同問題“今天上海天氣如何”一起傳給模型。
- 通常情況下模型會返回和問題相匹配的函數，即查詢天氣的函數，同時返回函數傳參“今天”和“上海”。
- 根據這個返回內容，我們再實際調用查詢天氣的函數，獲取到天氣信息。
- 最后將天氣信息返回給用戶。
由此可以看出Function Calling（函數調用）的本質就是利用大語言模型的文字分析能力，在提供的一系列函數中，找出能夠回答問題的最合適的函數，函數內部的具體邏輯則交給開發者自己實現，而不是大語言模型實現。
再進一步分析可以發現，如果沒有Function Calling，其實通過prompt提示我們也可以實現Function Calling的功能，例如prompt類似于“我有一些函數，定義為xxxx，我想知道“今天上海天氣如何”，請告訴我用哪個函數可以解答這個問題，并告訴我函數的傳參“。當然這個prompt效果并不一定很好，每個人的prompt也不盡相同，那么Function Calling這個功能就應運而生了，針對這種場景進行調優并規范了函數的定義和返回格式，方便了開發者的使用。
2.2 如何實現異步
在定義插件時，有一些插件如Ping插件、IP掃描插件等，由于網絡耗時或執行本身比較慢，提問后無法立馬返回結果，所以需要使用異步的方式，等后臺服務執行完成后，再把結果返回給前端。對于這種場景，我們需要主動向前端推送消息，常用的方法就是使用WebSocket。
WebSocket是從HTML5開始提供的一種瀏覽器與服務器進行全雙工通訊的網絡技術，它的目的是在瀏覽器和服務器之間建立一個不受限的雙向通信的通道。比如說，服務器可以在任意時刻發送消息給瀏覽器。它不是一種全新的協議，而是利用了HTTP協議來建立連接，屬于應用層協議。
它具有如下優點：
- 支持雙向通信，實時性更強
- 更好的二進制支持
- 較少的控制開銷。連接創建后，ws客戶端、服務端進行數據交換時，協議控制的數據包頭部較小。在不包含頭部的情況下，服務端到客戶端的包頭只有2~10字節（取決于數據包長度），客戶端到服務端的的話，需要加上額外的4字節的掩碼。而HTTP協議每次通信都需要攜帶完整的頭部
- 支持擴展。ws協議定義了擴展，用戶可以擴展協議，或者實現自定義的子協議（比如支持自定義壓縮算法等）
除了WebSocket，我們還可以選擇使用socketIO。Socket.IO也可以實現客戶端和服務段之間雙向通信。但與WebSocket不同的是，socketIO是一個第三方庫，他具有WebSocket的基本功能，同時也增強了一些的功能。比如：
- 兼容性：WebSocket是HTML5標準中的一部分，需要瀏覽器支持HTML5才能使用，而Socket.IO是基于WebSocket協議的封裝，可以在不支持WebSocket的瀏覽器上使用
- API：WebSocket只提供了底層的API，需要開發者自己實現消息的編解碼、心跳等功能，而Socket.IO提供了更高層次的API，封裝了消息的編解碼、心跳等功能，使用更加方便
- 處理異常：WebSocket在連接異常時會直接斷開連接，而Socket.IO的心跳機制會嘗試重新連接，提高了連接的穩定性
- 支持的協議：WebSocket只支持單一的協議，而Socket.IO支持多種協議，包括WebSocket、Flash Socket、AJAX長輪詢等
- HTTP 長輪詢回退：如果無法建立 WebSocket 連接，連接將回退到 HTTP 長輪詢
但需要強調的是：Socket.IO與WebSocket并不能兼容，盡管 Socket.IO 確實在可能的情況下使用 WebSocket 進行傳輸，但它為每個數據包添加了額外的元數據。因此WebSocket客戶端將無法成功連接到Socket.IO服務器，而Socket.IO客戶端也將無法連接到普通WebSocket服務器。
socketIO服務連接時，可以在f12中看到連接的過程：
總共分為5步：
- 客戶端發起握手請求(GET)，服務端返回本次連接的前置基礎信息
{
"sid": "FSDjX-WRwSA4zTZMALqx", // 會話的ID，它必須包含在后續所有HTTP請求的查詢參數中
"upgrades": ["websocket"], // 數組包含服務器支持的所有“更好”傳輸的列表
"pingInterval": 25000, // 心跳檢測時間，25秒
"pingTimeout": 20000 // # 心跳超時時間，20秒
}- 客戶端帶上sid(POST)，長輪詢，發送連接請求
- 客戶端帶上sid(GET)，長輪詢，獲取連接確認
- 升級建立WebSocket連接，響應碼為101，且一直處于連接狀態
- 客戶端接收數據 (GET)，長輪詢，WebSocket連接建立成功后關閉
三、 基本實現
以下實現案例基于國內開源大語言模型ChatGLM3，ChatGLM3 是智譜AI和清華大學 KEG 實驗室聯合發布的對話預訓練模型。
3.1 定義各種插件
根據ChatGLM3模型的插件規范，定義插件的相關信息，這里舉2個例子，Google搜索（同步插件）、Ping（異步插件）。
all_plugins = {
"google": {
"name_cn": "谷歌搜索", # 中文名稱
"sync": True, # 是否同步執行
"message": "{result}", # 返回給用戶的消息
# info內容為符合ChatGLM3 function call規范的函數定義
"info": {
"name": "google", # 函數名
"description": "當問題需要進行實時搜索(如今天的日期或者今天的天氣等)時, 或者無法回答時, 使用 google 搜索", # 函數描述
"parameters": {
"type": "object",
"properties": {
"keyword": { # 傳參參數名
"type": "string", # 參數數據類型
"description": "搜索的關鍵詞" # 參數描述
}
},
"required": ["keyword"] # 必填參數
}
}
},
"ping": {
"name_cn": "ping",
"sync": False,
"message": "使用ping插件，由于該任務執行時間比較長，完成后我會主動將結果發送給您。請耐心等待。如果您有其他問題，可以繼續提問。",
"info": {
"name": "ping",
"description": "使用ping工具對IP地址進行ping測試",
"parameters": {
"type": "object",
"properties": {
"addr": {
"type": "string",
"description": "被ping的ip或者域名"
}
},
"required": ["addr"]
}
}
}
}定義插件對應的函數實現：
class Functions:
@classmethod
def ping(cls, **kwargs):
"""ping實現"""
# 省略ping的代碼實現
pass
@classmethod
def google(cls, **kwargs):
"""google搜索實現"""
# 查詢關鍵字
keyword = kwargs['keyword']
# 搜索結果
search_context = []
# 使用google api搜索
res = server['service'].cse().list(q=keyword, cx=server['cx'], ).execute()
# 遍歷搜索結果
for row in res.get('items', []):
# 提取每條搜索結果的簡要信息
search_context.append(row['snippet'])
# 匯總搜索結果和問題組成prompt
prompt = [{"role": "user", "content": f"請結合以下內容，回答問題：{keyword}\n" + "\n".join(search_context)}]
# 調用大語言模型生成答案
return reply_text(prompt)3.2 使用Function Calling實現插件功能
大體邏輯為：將插件信息和用戶提問一起發送給大語言模型的API，得到與之匹配的插件，再調用插件對應的函數，得到結果返回給用戶。以下代碼為簡化的ChatGLM3示例代碼：
import torch
from transformers import AutoTokenizer, AutoModel
def main():
"""使用插件時回復文字"""
DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'
tokenizer = AutoTokenizer.from_pretrained('/home/chatglm3-6b', trust_remote_code=True)
model = AutoModel.from_pretrained('/home/chatglm3-6b', trust_remote_code=True).to(DEVICE).eval()
# 匯總所有的插件信息
tools = [plugin['info'] for plugin in all_plugins.values()]
# 將插件信息設置在對話歷史中
history = [{"role": "system", "content": "Answer the following questions as best as you can. You have access to the following tools:", "tools": tools}]
# 調用function calling
response, _ = model.chat(tokenizer, query, history=history)
# 獲取匹配的插件名稱
plugin_name = response.get("name", "")
# 獲取匹配的插件參數
arguments = response.get("parameters", {})
# 沒有匹配到插件則退出
if not plugin_name:
return None
# 獲取插件完整信息
plugin = all_plugins[plugin_name]
# 使用反射機制獲取插件對應的函數對象
func = getattr(Functions, plugin_name)
# 執行函數并返回結果
res = func(**arguments)
return res3.3 異步插件的實現
本項目Web后端使用的框架為flask，使用socketIO實現異步交互，需要安裝對應的庫：flask_socketio，啟動時，在flask的app上使用SocketIO包裝一下即可，這樣在同一個端口上同時開啟了http服務和socketIO服務，下面只展示基本關鍵代碼：
from flask import Flask
from flask_socketio import SocketIO
# flask原始 app
web_app = Flask(__name__, static_folder=Config.STATIC_PATH)
# socketIO包裝app
socketio = SocketIO(web_app, cors_allowed_origins="*", logger=True)
# 可監聽連接和斷開
@socketio.on('connect')
def handle_connect():
print("connect")
@socketio.on('disconnect')
def handle_disconnect():
print("disconnect")
# 本地啟動app
if __name__ == '__main__':
socketio.run(web_app, address, port, allow_unsafe_werkzeug=True)在socketIO中調用emit(event, *args, **kwargs)方法即可給指定目標(event，本項目對應為user)發送消息。
我們通過function calling獲取到對應插件時，如果是同步插件，則立即執行對應函數，如果是異步插件，應該異步開啟執行對應函數，并立馬結束當前會話，等異步函數執行完成后主動發送消息給前端用戶，因此我們需要修改一下上面的插件代碼：
def main(user, question):
"""使用插件時回復文字"""
...代碼同上
# 使用反射機制獲取插件對應的函數對象
func = getattr(Functions, plugin_name)
# 判斷插件是否同步
if plugin['sync']:
# 同步的插件，直接調用函數
res = func(**arguments)
else:
# 異步的插件，這里使用線程池示例執行異步任務
thread_pool = ThreadPool(3)
# 定義回調函數, 接收到結果后推送給前端
def callback(result):
# 推送給前端
socketio.emit(user, f"任務結果為: {result}")
# 異步調用
res = thread_pool.apply_async(func, kwds=arguments, callback=callback)
return res四、 未來規劃
4.1 更多的插件
上述插件案例只是插件功能的冰山一角，通過該功能我們可以定義各種實用的插件，目前攜程信息安全部的大語言模型智能聊天機器人只是支持一些基本的插件，也歡迎大家給我們提出寶貴的建議，集思廣益，一起開發出更多實用好用的插件。
4.2 每個用戶的自定義插件
目前的插件功能可以支持我們這些項目的開發者實現自定義插件，這些插件也必須提前寫入項目中，并不能支持終端用戶直接自定義自己的插件。后續我們會調研可行性方案，讓終端的用戶自己編寫對應的插件代碼，實現每個用戶都能定義自己的插件。
