使用 Python、WebSocket、ChatterBot 和 Bocadillo 构建实时聊天机器人服务器

本文改编自Bocadillo 官方教程。

大家好！今天的帖子有点特别。你们中有些人可能还记得我几个月前写的这篇文章：《我是如何构建一个Python Web框架并成为一名开源维护者的》。

从那以后，我一直在继续开发Bocadillo，这段时间真是太棒了！事实上，就在上周，我得知我将在五月底飞往慕尼黑，在PyConWeb 2019上发表演讲！这将是我第一次参加会议并发表演讲，所以不用说，我超级兴奋！🙌🤩

另一个好消息是，Bocadillo v0.13 版本刚刚发布：

在这样良好的氛围下，我终于决定发布一篇详尽的教程。

废话不多说，剧情是这样的：我们要尝试构建一个聊天机器人服务器！

Bocadillo 内置了许多功能，因此这是一个了解使用 Bocadillo 构建 Web 服务的一些方面的好机会。

在本教程中，你不仅可以体验聊天机器人，还能学习如何：

使用WebSocket实时处理多个连接。
创建 REST 端点。
使用提供程序将可重用资源注入到视图中。
使用pytest测试Bocadillo 应用程序。

还在疑惑我们该如何构建像聊天机器人这样看似复杂的东西吗？其实，你可能知道 Python 拥有庞大的数据科学生态系统。我敢打赌，肯定已经存在现成的聊天机器人框架了。

原来真的有！经过一番研究，我偶然发现了ChatterBot。它看起来相当可靠且流行，所以我们将用它来构建Diego，一个友好的对话式智能助手。别担心，这不需要任何数据科学或聊天机器人技术方面的背景知识！

听起来很刺激？好，让我们开始吧！🙌

项目设置

首先，让我们来设置一下项目：

打开终端，在电脑上的某个位置创建一个空目录，然后cd进入该目录：

mkdir ~/dev/bocadillo-chatbot
cd ~/dev/bocadillo-chatbot

安装 Bocadillo 和 ChatterBot。这里我们使用pipenv来安装依赖项，但您也可以使用普通的pip+virtualenv符号。

# Note: pytz is required by chatterbot.
pipenv install bocadillo chatterbot pytz

创建一个空app.py脚本。稍后我们将在这里创建应用程序：

touch app.py

现在我们应该有如下目录结构：

$ tree
.
├── Pipfile
├── Pipfile.lock
└── app.py

引导应用程序

现在，让我们来编写应用程序框架app.py。请稍等——第一段像样的代码即将出现：

# app.py
from bocadillo import App

app = App()

if __name__ == "__main__":
    app.run()

如果你曾经使用过 Flask，或者几乎任何 Python Web 框架，那么这个看起来应该会非常眼熟。甚至有点乏味。不过谁在乎呢？它能用！自己看看：

python app.py

如果你访问http://localhost:8000并得到404 Not Found响应，那就一切正常！Ctrl+C在终端中输入命令停止服务器。

编写 WebSocket 端点

现在我们终于可以进入正题了！首先我们要构建的是WebSocket端点。

如果你还不熟悉 WebSocket，别担心——这里有一个十个字的简要概括：它允许服务器和客户端以双向方式交换消息。它是为 Web 重新设计的经典套接字技术。

由于它们具有双向性，因此非常适合我们正在构建的这种应用程序——客户端和服务器之间的某种对话（即我们的聊天机器人）。

如果您有兴趣了解更多关于 Python 中的 WebSocket 的知识，我强烈推荐这个演讲：WebSocket 入门指南。

好的，我们暂时先不接入聊天机器人。相反，我们先让服务器将收到的任何消息都发送回去——这种行为也称为“回显”端点。

在app对象声明和代码块之间添加以下内容：app.run()app.py

@app.websocket_route("/conversation")
async def converse(ws):
    async for message in ws:
        await ws.send(message)

这里简单解释几句，供好奇者参考：

这将定义一个 WebSocket 端点，该端点可在该ws://localhost:8000/conversation位置访问。
该async for message in ws:行代码遍历通过 WebSocket 接收的消息。
最后，将接收到的数据原样await ws.send(message)发送回客户端。message

尝试使用 WebSocket 端点

我们不妨尝试创建一个 WebSocket 客户端？别担心，我们不需要编写任何 JavaScript 代码。我们将使用 Python 和Bocadillo 自带的websockets库。

创建一个client.py文件并将以下代码粘贴到其中。这段代码的作用是连接到 WebSocket 端点并运行一个简单的 REPL：

# client.py
import asyncio
from contextlib import suppress
import websockets

async def client(url: str):
    async with websockets.connect(url) as websocket:
        while True:
            message = input("> ")
            await websocket.send(message)
            response = await websocket.recv()
            print(response)

with suppress(KeyboardInterrupt):
    # See asyncio docs for the Python 3.6 equivalent to .run().
    asyncio.run(client("ws://localhost:8000/conversation"))

运行服务器端应用程序，python app.py并在另一个终端中启动client.py脚本。你应该会看到>提示符。如果看到了，就可以开始聊天了！

$ python client.py
> Hi!
Hi!
> Is there anyone here?
Is there anyone here?
>

很酷吧？🤓

输入Ctrl+C此命令退出会话并关闭 WebSocket 连接。

你好，迭戈！

既然我们已经能够让服务器和客户端进行通信，那么我们何不将回声实现替换为一个真正智能且友好的聊天机器人呢？

这时ChatterBot就派上用场了！我们将创建一个名为Diego 的聊天机器人——一个会说异步萨尔萨舞的聊天机器人。🕺

请新建一个chatbot.py文件，并将 Diego 的名字添加到该文件中：

# chatbot.py
from chatterbot import ChatBot
from chatterbot.trainers import ChatterBotCorpusTrainer

diego = ChatBot("Diego")

trainer = ChatterBotCorpusTrainer(diego)
trainer.train(
    "chatterbot.corpus.english.greetings",
    "chatterbot.corpus.english.conversations"
)

（ChatterBot 的聊天机器人默认功能比较弱，所以上面的代码使用英语语料库训练 Diego，使其更聪明一些。）

此时，您可以在 Python 解释器中尝试运行聊天机器人：

$ python
>>> from chatbot import diego  # Be patient — this may take a few seconds to load!
>>> diego.get_response("Hi, there!")
<Statement text:There should be one-- and preferably only one --obvious way to do it.>

（嗯，有趣的回答！🐍）

现在让我们把 Diego 连接到 WebSocket 端点：每次我们收到一个新的message，我们都会把它交给 Diego，然后把他的响应发送回去。

# app.py
from chatbot import diego

...

@app.websocket_route("/conversation")
async def converse(ws):
    async for message in ws:
        response = diego.get_response(message)
        await ws.send(str(response))

如果您运行之前设置的服务器/客户端程序，现在可以看到 Diego 通过 WebSocket 与我们进行通信！

$ python client.py
> Hi there!
I am a chat bot. I am the original chat bot. Did you know that I am incapable of error?
> Where are you?
I am on the Internet.
>

看来迭戈是个爱开玩笑的人。😉

将聊天机器人重构为提供者

现在客户可以通过 WebSocket 连接与 Diego 聊天了。太棒了！

然而，我们目前的设置存在一些功能性问题：

加载 Diego 相当耗时：在普通笔记本电脑上大约需要十秒钟。
由于import脚本顶部存在加载语句，每次导入app模块时都会加载 Diego。这可不好！
Diego 作为全局依赖项注入到 WebSocket 端点中：我们无法将其替换为其他实现（这在测试期间尤其有用），而且乍一看并不清楚端点是否依赖于它。

仔细想想，Diego 是一种资源——理想情况下，它应该只在处理连接请求时才提供给 WebSocket 端点。

所以，一定有更好的办法……而且确实有：那就是服务提供商。✨

Providers 是 Bocadillo 的一个独特功能。它们的灵感来源于pytest fixtures，并提供了一种优雅、模块化和灵活的方式来管理资源并将其注入到 Web 视图中。

我们用它们来修复代码吧？

首先，我们把迭戈放到providerconf.py剧本里：

# providerconf.py
from chatterbot import ChatBot
from chatterbot.trainers import ChatterBotCorpusTrainer
from bocadillo import provider

@provider(scope="app")
def diego():
    diego = ChatBot("Diego")

    trainer = ChatterBotCorpusTrainer(diego)
    trainer.train(
        "chatterbot.corpus.english.greetings",
        "chatterbot.corpus.english.conversations",
    )

    return diego

上面的代码声明了一个diego提供程序，现在我们可以将其注入到 WebSocket 视图中。我们只需将其声明为视图的一个参数即可。

让我们通过更新app.py脚本来实现这一点。以下是完整的脚本：

from bocadillo import App

app = App()

@app.websocket_route("/conversation")
async def converse(ws, diego):  # <-- 👋, Diego!
    async for message in ws:
        response = diego.get_response(message)
        await ws.send(str(response))

if __name__ == "__main__":
    app.run()

无需导入——Diego 会在处理 WebSocket 连接请求时自动注入到 WebSocket 视图中。✨

好了，准备好试一试了吗？

运行app.py脚本。您应该会看到与 Bocadillo 在启动时设置 Diego 相关的其他日志：

$ python app.py
INFO: Started server process [29843]
INFO: Waiting for application startup.
[nltk_data] Downloading package averaged_perceptron_tagger to
[nltk_data]     /Users/Florimond/nltk_data...
[nltk_data]   Package averaged_perceptron_tagger is already up-to-
[nltk_data]       date!
[nltk_data] Downloading package punkt to /Users/Florimond/nltk_data...
[nltk_data]   Package punkt is already up-to-date!
[nltk_data] Downloading package stopwords to
[nltk_data]     /Users/Florimond/nltk_data...
[nltk_data]   Package stopwords is already up-to-date!
Training greetings.yml: [####################] 100%
Training conversations.yml: [####################] 100%
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

运行client.py脚本，开始聊天吧！你应该感觉和以前没什么区别。特别是迭戈的回复速度，依然很快。

$ python client.py
> Hello!
Hi
> I would like to order a sandwich
Yes it is.
>

瞧！Bocadillo 提供程序实现了美观、模块化和灵活的依赖注入。

跟踪客户

让我们更进一步。没错，我们已经相当巧妙地实现了通过 WebSocket 与聊天机器人进行对话。现在，我们该如何跟踪当前有多少客户端正在与聊天机器人对话呢？

如果您想知道的话——是的，我们也可以与供应商合作实现这一点！

让我们添加一个clients提供商providerconf.py：

# providerconf.py
from bocadillo import provider

...

@provider(scope="app")
def clients():
    return set()

现在，我们添加另一个提供程序，它返回一个上下文管理器，负责将连接注册ws到客户端集合中。顺便一提，这是一个工厂提供程序的示例，但你现在不需要理解全部代码。

# providerconf.py
from contextlib import contextmanager
from bocadillo import provider
...

@provider
def save_client(clients):
    @contextmanager
    def _register(ws):
        clients.add(ws)
        try:
            yield ws
        finally:
            clients.remove(ws)

    return _register

在 WebSocket 视图中，使用新的save_client提供程序注册 WebSocket 客户端：

# app.py

...

@app.websocket_route("/conversation")
async def converse(ws, diego, save_client):
    with save_client(ws):
        async for message in ws:
            response = diego.get_response(message)
            await ws.send(str(response))

就是这样！当客户与迭戈聊天时，它将出现在场景中clients。

我们何不利用这些信息做点什么呢？

通过 REST 端点公开客户端数量

最后，我们暂时放下 WebSocket，回到经典的 HTTP 协议。我们将创建一个简单的 REST 端点来查看当前连接的客户端数量。

返回app.py并添加以下代码：

# app.py

...

@app.route("/client-count")
async def client_count(req, res, clients):
    res.media = {"count": len(clients)}

clients如果你曾经使用过 Flask 或 Falcon，那么这段代码应该不会让你感到陌生。我们在这里所做的只是将（从提供商处获取的）数量clients以 JSON 响应的形式发送出去。

来吧！python app.py运行几个实例。在浏览器中打开http://localhost:8000/client-count，python client.py查看有多少客户端连接。点击其中一个客户端，看看客户端数量是否减少！Ctrl+C

成功了吗？恭喜！✨

测试

我们想一起探讨的功能基本都完成了。当然，我还有一些想法可以作为练习供你们探索，但在那之前，我们先来编写一些测试。

Bocadillo 的设计原则之一是让编写高质量应用程序变得轻松便捷。因此，Bocadillo 内置了所有必要的工具，方便用户为这款聊天机器人服务器编写测试。

你可以使用你喜欢的测试框架来编写这些测试。在本教程中，我们将选择pytest 。首先让我们安装它：

pipenv install --dev pytest

现在，我们来搭建测试环境。我们将编写一个pytest fixture来设置测试客户端。该测试客户端会暴露一个类似 Requests 的 API，以及一些用于测试 WebSocket 端点的辅助函数。此外，我们实际上并不需要在这里测试聊天机器人，所以我们会diego用一个“echo”模拟对象覆盖提供程序——这样做的一个好处是可以大大加快测试速度。

所以，请创建一个conftest.py脚本，并将以下内容放入其中：

# conftest.py
import pytest
from bocadillo import provider
from bocadillo.testing import create_client

from app import app

@provider
def diego():
    class EchoDiego:
        def get_response(self, query):
            return query

    return EchoDiego()

@pytest.fixture
def client():
    return create_client(app)

现在是时候编写一些测试了！test_app.py在项目根目录下创建一个文件：

touch test_app.py

首先，我们来测试一下是否可以连接到 WebSocket 端点，以及发送消息后是否能收到 Diego 的响应：

# test_app.py

def test_connect_and_converse(client):
    with client.websocket_connect("/conversation") as ws:
        ws.send_text("Hello!")
        assert ws.receive_text() == "Hello!"

现在，让我们测试一下客户端连接到 WebSocket 端点时客户端计数器的递增情况：

# test_app.py
...

def test_client_count(client):
    assert client.get("/client-count").json() == {"count": 0}

    with client.websocket_connect("/conversation"):
        assert client.get("/client-count").json() == {"count": 1}

    assert client.get("/client-count").json() == {"count": 0}

使用以下命令运行这些测试：

pytest

你猜怎么着？

==================== test session starts =====================
platform darwin -- Python 3.7.2, pytest-4.3.1, py-1.8.0, pluggy-0.9.0
rootdir: ..., inifile: pytest.ini
collected 2 items

test_app.py ..                                         [100%]

================== 2 passed in 0.08 seconds ==================

考试通过！✅🎉

总结

如果你已经做到这一步——恭喜！你刚刚构建了一个由 WebSocket、ChatterBot和 Bocadillo 提供支持的聊天机器人服务器。

在本文中，我们已经了解了如何：

创建一个博卡迪略项目。
编写一个 WebSocket 端点。
编写一个HTTP端点。
利用提供者将资源与其消费者解耦。
测试 WebSocket 和 HTTP 端点。

本教程的完整代码可在 GitHub 上的 Bocadillo 代码库中找到：获取代码！总而言之，服务器端和程序providerconf.py端加起来只有大约 60 行代码——性价比很高！

显然，我们仅仅触及了Bocadillo功能的冰山一角。本教程的目标是引导您逐步构建一个最小有意义应用程序（MEN）。

你可以很轻松地对我们共同搭建的这个聊天机器人服务器进行迭代开发。我很想看看你的成果！

想挑战一下自己吗？这里有一些建议：

添加一个使用模板渲染的主页。网页浏览器应通过 JavaScript 程序连接到聊天机器人服务器。您可能还需要提供静态文件才能实现这一点。
训练迭戈回答诸如“你目前正在和多少人交谈？”之类的问题。
目前，所有客户端都与同一个 Diego 实例通信。然而，如果每个客户端都拥有自己的 Diego，以确保对话的个性化，那就更好了。您可以考虑使用基于 cookie 的会话和工厂提供程序来实现这一目标。

希望您喜欢这篇教程！如果您想支持本项目，请务必给仓库点个星标。如果您不想错过任何新版本发布和公告，欢迎在 Twitter 上关注@bocadillopy！

文章来源：https://dev.to/bocadillo/building-a-real-time-chatbot-server-in-python-with-websocket-chatterbot-and-bocadillo-482g

菜单

分享

使用 Python、WebSocket、ChatterBot 和 Bocadillo 构建实时聊天机器人服务器

使用 Python、WebSocket、ChatterBot 和 Bocadillo 构建实时聊天机器人服务器

项目设置

引导应用程序

编写 WebSocket 端点

尝试使用 WebSocket 端点

你好，迭戈！

将聊天机器人重构为提供者

跟踪客户

通过 REST 端点公开客户端数量

测试

总结

系统设计面试中的 19 种微服务模式

使用 React 和 AWS Amplify 实现无服务器架构第三部分：跟踪应用使用情况

模型-视图-控制器（MVC）模式到底是什么？DEV 全球项目展示挑战赛，由 Mux 主办：快来展示你的项目吧！

我在两年内从 PHP 开发人员晋升为高级 C#/.NET 开发人员。

了解 Docker：第 12 部分 – 传递构建参数

Yarn 和第三方 NPM 客户端的黑暗未来 DEV 的全球展示与讲述挑战赛，由 Mux 呈现：展示你的项目！

CSS DEV 的全球展示挑战赛“响应式字体”由 Mux 呈现：展示你的项目！

我是如何以学生开发者的身份免费获得 Tabnine Pro 的，你也可以！

五大顶级JS框架

从 Rector PHP 开始：利用自动化改进您的 PHP 代码