保姆级教程:如何用Python自制聊天机器人?

原创
01/15 15:58
阅读数 38
导语 | 近日, ChatGPT 作为 2023 年最火的技术之一,它将人工智能的应用和发展推向了一个新的高度。各种大模型也雨后春笋般涌现,基于大模型的聊天机器人层出不穷,腾讯的混元大模型便是其中一大翘楚。如果你连不上混元大模型,本文也给出一种解决方案让你能跑起来代码,看得到效果。今天,我们特邀了小樱桃 CEO 杜金房老师,他将为我们介绍如何使用 Python 连接腾讯混元大模型制作一个聊天机器人,并介绍底层通信协议和实现原理。

 

作者简介

 

 

杜金房,小樱桃 CEO,北京信悦通科技创始人,烟台小樱桃网络科技创始人,FreeSWITCH 中文社区创始人,RTS 社区和 RTSCon 创始人,《FreeSWITCH 权威指南》、《Kamailio 实战》、《深入理解 FFmpeg》作者,FreeSWITCH 开源项目核心 Committer。致力打造 XSwitch 通信云平台,基于腾讯云建设。现在,他正在写一本新书《大道至简,给所有人看的编程书》,旨在带大家洞悉编程本质。写作过程中,以编程课的形式在微信小程序上滚动发布,便于读者追更。

 

引言

 

腾讯混元大模型(Tencent Hunyuan)是由腾讯研发的大语言模型,具备强大的中文创作能力,复杂语境下的逻辑推理能力,以及可靠的任务执行能力。可应用于聊天、文档、会议、广告、营销等各种场景。

如果你没有用过混元大模型,可以先在微信中搜索微信小程序“腾讯混元助手”体验一下。腾讯混元大模型也支持 API 调用,可以在https://cloud.tencent.com/product/hunyuan 购买。下面的例子需要使用 API。

 

准备

 

下面步骤主要基于你已经熟悉 Python 程序,且已经有了一个腾讯云账号,并购买了混元大模型服务。

在第一次使用腾讯云 API 之前,我们首先需要在腾讯云控制台(https://console.cloud.tencent.com/cam/capi)上申请安全凭证,安全凭证包括 SecretID 和 SecretKey, SecretID 是用于标识 API 调用者的身份,SecretKey 是用于加密签名字符串和服务器端验证签名字符串的密钥。SecretKey 必须严格保管,避免泄露。

https://console.cloud.tencent.com/hunyuan进入控制台,点击左侧菜单栏的“ API 密钥”,然后点击“新建密钥”按钮,生成一个 API 密钥。在后面的代码中我们将用到这个 API 密钥。

确保已经安装了 Python 开发环境,腾讯云 SDK 支持 Python 2 和 Python 3,但建议使用 Python 3。

首先,我们安装腾讯云 Python SDK:

 

pip install --upgrade tencentcloud-sdk-python

其次,设置环境变量,在终端中输入以下命令,将 SecretID 和 SecretKey 替换成你的 API 密钥:

 

export TENCENTCLOUD_SECRET_ID=你的SecretID
export TENCENTCLOUD_SECRET_KEY=你的SecretKey

到此,环境就已准备好了。

 

快速体验

 

腾讯云提供了一个 Demo,可以从以下地址获取:

https://github.com/TencentCloud/tencentcloud-sdk-python/blob/master/examples/hunyuan/v20230901/chat_std.py

下载后,运行:pythonchat_std.py 随后,你将看到如下输出:

 

当然可以!这是一个爆笑笑话:

老师让小明造一句有“哥哥”二字的话。小明:“哥哥昨天晚上又玩游戏了。”

老师:“那你知道哥哥昨天玩什么游戏?”

小明:“不知道,我只知道哥哥又输了。”

这个笑话的笑点在于,它使用了双关语和幽默元素。第一个双关语是“哥哥”,既表示兄弟之间的称呼,也可以表示游戏角色的名称。第二个双关语是“又输了”,既表示游戏输了,也可以表示哥哥输了。这种双关语和幽默元素的结合,使得这个笑话非常搞笑。

如果看到类似上面的内容,就表示成功了。当然,每次输出的内容都不一样。

 

输出调试信息

 

当然,上述示例也会打印一些调试信息,下面是其中的一条 JSON 示例。可以看出,大模型返回了一个 Note,应该是规避一些法律风险。Choices 是一个数组,里面包含了多个候选项;Delta 是增量数据;Role 是角色,这里是 assistant,表示大语言模型助手;而实际的内容在 Content 中。

 

{
  "Note": "以上内容为AI生成,不代表开发者立场,请勿删除或修改本标记",
  "Choices": [
    {
      "FinishReason": "",
      "Delta": {
        "Role": "assistant",
        "Content": "当"
      }
    }
  ],
  "Created": 1702001460,
  "Id": "bd2df0ed-4766-4478-bf9f-db43e70e7a63",
  "Usage": {
    "PromptTokens": 7,
    "CompletionTokens": 1,
    "TotalTokens": 8
  }
}

GPT 的全称是 Generative Pre-trained Transformer,即生成式预训练大语言模型。这里的生成代表生成内容。一般来说,生成内容需要很高的算力,比较慢,需要一个过程,因此,现阶段几乎所有的大语言模型在成内容时都是渐近生成的。当你在使用 ChatGPT 等聊天的时候,也是看到字或词是一个一个“蹦”出来的。下面是几个 JSON 示例(简单起见,只放了 Delta 部分):

 

"Delta":{"Role":"assistant","Content":"当"}
"Delta":{"Role":"assistant","Content":"然可"}
"Delta":{"Role":"assistant","Content":"以"}
"Delta":{"Role":"assistant","Content":"!这是一"}
"Delta":{"Role":"assistant","Content":"个爆"}
"Delta":{"Role":"assistant","Content":"笑笑"}
"Delta":{"Role":"assistant","Content":"话"}

代码解析

 

下面是完整的示例代码,我在其中添加了详细的注释:

 

# -*- coding: utf-8 -*-
import json
import os
# 导入腾讯云 SDK 相应的包
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models
from tencentcloud.common.profile.client_profile import ClientProfile

try:
    # 实例化一个认证对象,入参需要传入腾讯云账户 secretId,secretKey ,从环境变量中读取
    cred = credential.Credential(
        os.environ.get("TENCENTCLOUD_SECRET_ID"),
        os.environ.get("TENCENTCLOUD_SECRET_KEY"))
    # 初始化一个客户端,入参需要传入腾讯云服务器地域,这里传入广州地区
    client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")
    # 为了方便学习,输出一些调试信息
    client.set_stream_logger(level="DEBUG")
    # 创建一个标准请求对象
    req = models.ChatStdRequest()
    # 创建一个消息对象,这里的消息对象可以有多个,这里我们只用一个
    msg = models.Message()
    # 指定角色,这里是用户,相当于用户输入的提问信息
    msg.Role = "user"
    # 具体的提问内容
    msg.Content = "你好,可以讲个笑话吗"
    # 将上面的消息放到一个消息列表(数组)中
    req.Messages = [msg]
    # 从客户端发送请求,得到一个响应(resp)
    resp = client.ChatStd(req)
    full_content = ""
    # 循环从响应中获取内容
    for event in resp:
        print(event) # 输出调试信息
        data = json.loads(event['data'])
        for choice in data['Choices']:
            # 将收到的内容拼接起来,成为一个大的字符串
            full_content += choice['Delta']['Content']
    # 打印完整输出内容
    print(full_content)

except TencentCloudSDKException as err:
    print(err)

聊天

 

为了能愉快地聊天,我们还需要一个循环,这样可以不断地从控制台输入问题,然后得到回答。不过,控制台输入不是我们的重点,篇幅关系,在此就忽略了。

下面,我们讲一下聊天的逻辑。理论上,每次运行时,只需要更换 msg.Content = "你好,可以讲个笑话吗"部分的代码,大模型就会输出不同的内容。这部分就是大家常说的“提示词”。

但是,大模型是非常健忘的。也就是说,它不记得它说过什么。如果你想跟大模型优雅地对话,你需要替它记住聊过的内容。我们将代码改成如下的样子:

 

msg = models.Message()
msg.Role = "user"
msg.Content = "你好,可以讲个笑话吗"

msg2 = models.Message()
msg2.Role = "assistant"
msg2.Content = '''
当然可以!这是一个爆笑笑话:

老师让小明造一句有“哥哥”二字的话。小明:“哥哥昨天晚上又玩游戏了。”

老师:“那你知道哥哥昨天玩什么游戏?”

小明:“不知道,我只知道哥哥又输了。”

这个笑话的笑点在于,它使用了双关语和幽默元素。第一个双关语是“哥哥”,既表示兄弟之间的称呼,也可以表示游戏角色的名称。第二个双关语是“又输了”,既表示游戏输了,也可以表示哥哥输了。这种双关语和幽默元素的结合,使得这个笑话非常搞笑。
'''

msg3 = models.Message()
msg3.Role = "user"
msg3.Content = "不好笑,再讲一个"

req.Messages = [msg, msg2, msg3]

从上述代码中,我们给 req.Messages 提供了三条信息,前两条是第一次的用户提问和应答,第三条是用户的第二次追问。这样,大模型就能根据对话的上下文生成新的内容。

在实际应用中,可以将问答交互的流程保存到内存列表对象中,也可以保存到数据库中,在下次提问时附上前面的内容就可以了。关于这一点,我们就不特别举例了。

值得一提的是,大模型对上下文的支持是有限制的,因此,列表不能无限的长。下面部分来自官方文档说明,供参考:

● 长度最多为 40, 按对话时间从旧到新在数组中排列。

● Message 的 Role 当前可选值:user、assistant,其中,user 和 assistant 需要交替出现 (一问一答),最后一个为 user 提问,且 Content 不能为空。

● Messages 中 Content 总长度不超过 16000 token,超过则会截断最前面的内容,只保留尾部内容。建议不超过 4000 token。

 

Token

 

大模型一般都使用 Token 做为输入和输出的计量单位,它代表了大模型计算能力。混元大模型中,1 token 约等于 1.8 个中文汉字或 3 个英文字母。在腾讯云平台上有一个 Token 计算器,可以帮助你计算 Token 数量。

 

 

进阶

 

上面我们讲了使用 Python 跟大模型聊天的关键点,并给出了一个完整的示例代码。考虑到篇幅关系,我们并没有写一个完整的客户端程序。下面的几种建议读者可以自行练习:

● 把聊天的历史记录记在内存中,每次请求提供上下文。

● 让 Python 脚本接受命令行输入,通过命令行版聊天。

● 使用 PyGUI 之类的客户端库做一个 GUI 程序,让用户可以更方便地使用。

● 在 Python 程序中增加一个 HTTP Server,然后再写一个前端的网页输入框,在浏览器中聊天。

 

自制大模型

 

考虑到现在腾讯混元大模型还需要申请才能使用,而且也并不是所有读者都能立即申请到大模型体验。下面,我们自己动手做一个简单的大模型,用于测试我们的程序,同时,也学习一下相应的底层交互原理。

 

实现HTTP Server

 

我们先实现一个简单的 HTTP Server,用于接收用户的输入,然后模拟大模型返回输出。这里我们使用一个轻量级的 Web 服务器。下面是完整的代码及注释:

 

# ! -*- coding: utf-8 -*-
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

# 监听地址和端口
host = ('0.0.0.0', 8080)
# 当收到HTTP请求时执行如下内容
class Resquest(BaseHTTPRequestHandler):
    def handler(self):
        print("data:", self.rfile.readline().decode())
        self.wfile.write(self.rfile.readline())
    # 当收到GET请求时返回 Hello World,主要是为了测试服务器的正确性,可有可无
    def do_GET(self):
        data = {
            'Hello': 'World',
        }
        self.send_response(200)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())
    # 当收到POST请求时,返回结果
    def do_POST(self):
        # 返回 200 OK
        self.send_response(200)
        # 返回 event-stream 模式的数据,这是SSE协议要求的,后面还有解释
        self.send_header('Content-type', 'text/event-stream')
        self.end_headers()
        # 将一句话拆成多个词返回,模拟大模型的输出
        texts = ['这', '是', '一个', '笑话', '。']
        # 输出的JSON大致的结构
        data = {
            'Choices': [
                {
                    'Delta': {
                        'Role': 'assistant',
                        'Content': '',
                    },
                },
            ],
        }
        # 循环输出每一个词
        for i in range(texts.__len__()):
            # 将本次要输出的字符放到 Content 中
            data.get('Choices')[0].get('Delta')['Content'] = texts[i]
            if i == texts.__len__() - 1:
                # 是后一个输出,加上个 stop 标志,表示说话完成
                data.get('Choices')[0]['FinishReason'] = 'stop'
            self.wfile.write('data: '.encode('utf-8'))
            self.wfile.write(json.dumps(data).encode('utf-8'))
            # 每次输出完毕,加上两个换行符
            self.wfile.write(b'\n\n')

if __name__ == '__main__':
    # 启动服务器,监听相应的HTTP端口
    server = HTTPServer(host, Resquest)
    print("Starting server, listen at: %s:%s" % host)
    server.serve_forever()

写好后,存成 mock_server.py,执行pythonmock_server.py可以启动一个 Web 服务器,监听8080端口。

接下来,我们可以打开另一个终端窗口,使用 curl 命令发一个请求,就可以看到返回的结果:

 

curl localhost:8080

{"hello": "world"}

如果相看更详细的 HTTP 信息(包括请求和返回的 HTTP 头域),可以使用-v 参数:

 

curl -v localhost:8080
*   Trying 127.0.0.1:8080...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET / HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.85.0
> Accept: */*
>
* Mark bundle as not supporting multiuse
* HTTP 1.0, assume close after body
< HTTP/1.0 200 OK
< Server: BaseHTTP/0.6 Python/3.10.9
< Date: Fri, 08 Dec 2023 07:00:04 GMT
< Content-type: application/json
<
* Closing connection 0
{"hello": "world"}

发一个 POST 请求试试,随便传一个参数:

 

curl -XPOST -d a=1 localhost:8080

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": "这"}}]}

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": "是"}}]}

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": "一个"}}]}

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": "笑话"}}]}

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": "。"}, "FinishReason": "stop"}]}

注意,在实际操作中,你看到的中文可能是 JSON 编码器转义过的,如\u8fd9就是 “这” 这个字。

 

让聊天程序连接我们自己的大模型

 

上面我们已经准备好了一个大模型的模拟器,现在我们来修改一下聊天程序,让它连接我们自己的大模型。

首先,打开 chat_std.py,找到如下一行:

 

    client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")

改成如下的样子,详见代码内注释。

 

  # 创建一个客户端配置对象
    client_profile = ClientProfile()
    # 将服务器地址改成我们自己的服务器地址
    client_profile.httpProfile.endpoint = "localhost:8080"
    # 默认是 https,改成 http 协议
    client_profile.httpProfile.protocol = "http"
    client_profile.httpProfile.scheme = "http"
    # 增加一个 client_profile 参数,将上述设置传入底层的 HTTP 客户端
    client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou", client_profile)

保存后,运行 pythonchat_std.py,你将看到如下输出:

 

python chat_std.py
2023-12-08 15:09:04,374 10230 request.py L109 DEBUG GetResponse <Response [200]>
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": "这"}}]}'}
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": "是"}}]}'}
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": "一个"}}]}'}
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": "笑话"}}]}'}
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": "。"}}]}'}
{'data': ' {"Choices": [{"Delta": {"Role": "assistant", "Content": ""}, "FinishReason": "stop"}]}'}
这是一个笑话。

这就表示,我们的聊天机器人程序成功连接到了我们自制的大模型。这种通过 Mock 方式模拟的 API 服务器在开发中经常用到,它可以简化开发过程,人为制造一些流程和场景,甚至在做压力测试时也可以大大节省 Token 的费用。正所谓“磨刀不误砍柴工”。

 

SSE

 

最后,我们来说一下 SSE 协议。SSE 的全称是 Server-Sent Event,即服务端推送事件,是一个流式协议,用于 HTTP 长连接中的服务端事件推送。大模型类的聊天程序广泛使用了该协议,包括 ChatGPT 和腾讯混元大模型。

SSE 协议很简单,在 HTTP 输出头中增加Content-Type: text/event-stream,然后在每次输出的数据前面加data:(注意这里有一个空格),并在每次输出的数据后面加上两个换行符 \n\n。这样,客户端就可以通过 data: 来判断每次输出的数据了。上面例子中的数据输出就是 SSE,下面再写一遍加深印象:

 

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": 
"这"}}]}

data: {"Choices": [{"Delta": {"Role": "assistant", "Content": 
"是"}}]}

如果一行数据比较长,也可以分成多行输出,但每行都要以 data: 开头,以 \n 结尾,最后一行以 \n\n 结尾。如(注意,下面是一个句子):

 

data: 这
data: 是一个
data: 很长的句子。

前面可以加一个id:,表示数据的序号,这样客户端就可以判断数据的顺序了。如:

 

id: 1
data: Hello

也可以给这个句子(事件)起个名字,如:

 

event: 笑话101
data: 这是一个很好笑的笑话。

以冒号开头的行是一个注释,如:

 

: 这是一个注释

HTTP 请求只能发送一次,服务端在 HTTP 连接不断开的情况下,可以一直推送数据,这也就是你在使用 ChatGPT 时经常看到的慢慢往外“蹦”字的效果。同样,用户的每次询问都需要重新发起一个新的 HTTP 请求。

理解SSE协议有助于理解程序跟大模型间的通信原理,从而更好地理解大模型及其API。

 

小结

 

本文带你手把手使用 Python 语言写了一个简单的大模型聊天程序,并介绍了相应的调用流程和底层通信协议。如果你连不上混元大模型,也可以使用文中的模拟程序(Mock)来测试你的聊天机器人程序。

在「腾讯云TVP」公众号后台回复「聊天机器人」,即可下载本文完整的源代码,并订阅老师新书《大道至简,给所有人看的编程书》。

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部