干货 | 大语言模型插件功能在携程的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的使用步骤，我们可以回答这个问题：

1）我们可以事先在应用服务端定义一个查询天气的函数，函数描述为“查询某时某地的天气情况”，函数传参为“日期”和“地点”，在函数内部编写具体查询天气的代码，如从气象局网站获取对应的天气信息。

2）除此之外，我们还可以定义一些其他自定义的函数，如base65加解密函数、ip信息查询函数等。

3）在调用大语言模型的API时，将这些定义的函数按照api规范连同问题“今天上海天气如何”一起传给模型。

4）通常情况下模型会返回和问题相匹配的函数，即查询天气的函数，同时返回函数传参“今天”和“上海”。

5）根据这个返回内容，我们再实际调用查询天气的函数，获取到天气信息。

6）最后将天气信息返回给用户。

由此可以看出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步：

1）客户端发起握手请求(GET)，服务端返回本次连接的前置基础信息

{      "sid": "FSDjX-WRwSA4zTZMALqx",  // 会话的ID，它必须包含在后续所有HTTP请求的查询参数中      "upgrades": ["websocket"],  // 数组包含服务器支持的所有“更好”传输的列表      "pingInterval": 25000,  // 心跳检测时间，25秒      "pingTimeout": 20000  // # 心跳超时时间，20秒  }

2）客户端带上sid(POST)，长轮询，发送连接请求

3）客户端带上sid(GET)，长轮询，获取连接确认

4）升级建立WebSocket连接，响应码为101，且一直处于连接状态

5）客户端接收数据 (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 torchfrom transformers import AutoTokenizer, AutoModeldef 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 Flaskfrom flask_socketio import SocketIO# flask原始 appweb_app = Flask(__name__, static_folder=Config.STATIC_PATH)# socketIO包装appsocketio = SocketIO(web_app, cors_allowed_origins="*", logger=True)# 可监听连接和断开@socketio.on('connect')def handle_connect():    print("connect")@socketio.on('disconnect')def handle_disconnect():    print("disconnect")# 本地启动appif __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