成学，携程高级安全研发工程师，关注Python/ target=_blank class=infotextkey>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的使用步骤，我们可以回答这个问题：
由此可以看出Function Calling（函数调用）的本质就是利用大语言模型的文字分析能力，在提供的一系列函数中，找出能够回答问题的最合适的函数，函数内部的具体逻辑则交给开发者自己实现，而不是大语言模型实现。
再进一步分析可以发现，如果没有Function Calling，其实通过prompt提示我们也可以实现Function Calling的功能，例如prompt类似于“我有一些函数，定义为xxxx，我想知道“今天上海天气如何”，请告诉我用哪个函数可以解答这个问题，并告诉我函数的传参“。当然这个prompt效果并不一定很好，每个人的prompt也不尽相同，那么Function Calling这个功能就应运而生了，针对这种场景进行调优并规范了函数的定义和返回格式，方便了开发者的使用。
2.2 如何实现异步
在定义插件时，有一些插件如Ping插件、IP扫描插件等，由于网络耗时或执行本身比较慢，提问后无法立马返回结果，所以需要使用异步的方式，等后台服务执行完成后，再把结果返回给前端。对于这种场景，我们需要主动向前端推送消息，常用的方法就是使用WebSocket。
WebSocket是从html5开始提供的一种浏览器与服务器进行全双工通讯的网络技术，它的目的是在浏览器和服务器之间建立一个不受限的双向通信的通道。比如说，服务器可以在任意时刻发送消息给浏览器。它不是一种全新的协议，而是利用了HTTP协议来建立连接，属于应用层协议。
它具有如下优点：
除了WebSocket，我们还可以选择使用socketIO。Socket.IO也可以实现客户端和服务段之间双向通信。但与WebSocket不同的是，socketIO是一个第三方库，他具有WebSocket的基本功能，同时也增强了一些的功能。比如：
但需要强调的是：Socket.IO与WebSocket并不能兼容，尽管 Socket.IO 确实在可能的情况下使用 WebSocket 进行传输，但它为每个数据包添加了额外的元数据。因此WebSocket客户端将无法成功连接到Socket.IO服务器，而Socket.IO客户端也将无法连接到普通WebSocket服务器。
socketIO服务连接时，可以在f12中看到连接的过程：
总共分为5步：
{
"sid": "FSDjX-WRwSA4zTZMALqx", // 会话的ID，它必须包含在后续所有HTTP请求的查询参数中
"upgrades": ["websocket"], // 数组包含服务器支持的所有“更好”传输的列表
"pingInterval": 25000, // 心跳检测时间，25秒
"pingTimeout": 20000 // # 心跳超时时间，20秒
}
以下实现案例基于国内开源大语言模型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 res
3.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 每个用户的自定义插件
目前的插件功能可以支持我们这些项目的开发者实现自定义插件，这些插件也必须提前写入项目中，并不能支持终端用户直接自定义自己的插件。后续我们会调研可行性方案，让终端的用户自己编写对应的插件代码，实现每个用户都能定义自己的插件。
京公网安备 11010802035086号