Django實(shí)現(xiàn)聊天機(jī)器人

更新時(shí)間：2021年05月31日 15:51:58 作者：大江狗

本文基于channels + websocket結(jié)合Celery和Python爬蟲技術(shù)打造了一個(gè)會(huì)算術(shù)懂詩(shī)文的聊天機(jī)器人，是非常難得的一個(gè)Django綜合應(yīng)用項(xiàng)目哦，感興趣的朋友可以參考下

實(shí)現(xiàn)原理

用戶在聊天界面調(diào)用Celery異步任務(wù)，Celery異步任務(wù)執(zhí)行完畢后發(fā)送結(jié)果給channels，然后channels通過websocket將結(jié)果實(shí)時(shí)推送給用戶。對(duì)于簡(jiǎn)單的算術(shù)運(yùn)算，Celery一般自行計(jì)算就好了。對(duì)于網(wǎng)上查找詩(shī)人簡(jiǎn)介這樣的任務(wù)，Celery會(huì)調(diào)用Python爬蟲(requests+parsel)爬取古詩(shī)文網(wǎng)站上的詩(shī)人簡(jiǎn)介，把爬取結(jié)果實(shí)時(shí)返回給用戶。

接下來我們來看下具體的代碼實(shí)現(xiàn)吧。

第一步安裝環(huán)境依賴

首先在虛擬環(huán)境中安裝django和以下主要項(xiàng)目依賴。本項(xiàng)目使用了最新版本，為3.X版本。

 # 主要項(xiàng)目依賴
 pip install django
 pip install channels
 pip install channels_redis
 pip install celery
 pip install redis
 pip install eventlet # windows only

 # 爬蟲依賴
 pip install requests
 pip install parsel

新建一個(gè)名為myproject的項(xiàng)目，新建一個(gè)app名為bots。如果windows下安裝報(bào)錯(cuò)，如何解決自己網(wǎng)上去找吧，很容易解決。修改settings.py, 將channels和chat加入到INSTALLED_APPS里，并添加相應(yīng)配置，如下所示：

 INSTALLED_APPS = [
       'django.contrib.admin',
       'django.contrib.auth',
       'django.contrib.contenttypes',
       'django.contrib.sessions',
       'django.contrib.messages',
       'django.contrib.staticfiles',
       'channels', # channels應(yīng)用     
       'bots', # bots應(yīng)用
    ]

 # 設(shè)置ASGI應(yīng)用
 ASGI_APPLICATION = 'myproject.asgi.application'

# 生產(chǎn)環(huán)境中使用redis做后臺(tái)，安裝channels_redis
import os
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [os.environ.get('REDIS_URL', 'redis://127.0.0.1:6379/2')],
        },
    },
}

最后將bots應(yīng)用的urls.py加入到項(xiàng)目urls.py中去，這和常規(guī)Django項(xiàng)目無異。

 # myproject/urls.py
 from django.conf.urls import include
 from django.urls import path
 from django.contrib import admin
 
 urlpatterns = [
     path('bots/', include('bots.urls')),
     path('admin/', admin.site.urls),
 ]

第二步配置Celery

pip安裝好Celery和redis后，我們要對(duì)其進(jìn)行配置。分別修改myproject目錄下的__init__.py和celery.py（新建), 添加如下代碼：

# __init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)

# celery.py
import os
from celery import Celery

# 設(shè)置環(huán)境變量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
# 實(shí)例化
app = Celery('myproject')

# namespace='CELERY'作用是允許你在Django配置文件中對(duì)Celery進(jìn)行配置
# 但所有Celery配置項(xiàng)必須以CELERY開頭，防止沖突
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自動(dòng)從Django的已注冊(cè)app中發(fā)現(xiàn)任務(wù)
app.autodiscover_tasks()

# 一個(gè)測(cè)試任務(wù)
@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

接著修改settings.py, 增加如下Celery配置：

# Celery配置
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_TIMEZONE = TIME_ZONE

# celery內(nèi)容等消息的格式設(shè)置，默認(rèn)json
CELERY_ACCEPT_CONTENT = ['application/json', ]
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

完整Celery配置見：Django進(jìn)階：萬字長(zhǎng)文教你使用Celery執(zhí)行異步和周期性任務(wù)(多圖)

第三步編寫機(jī)器人聊天主頁面

本例我們只需要利用django普通視圖函數(shù)編寫1個(gè)頁面，用于展示首頁(index)與用戶交互的聊天頁面。這個(gè)頁面對(duì)應(yīng)的路由及視圖函數(shù)如下所示：

 # bots/urls.py
 from django.urls import path
 from . import views
 
 urlpatterns = [
     path('', views.index, name='index'),
 ]
 
 # bots/views.py
 from django.shortcuts import render
 
 def index(request):
     return render(request, 'bots/index.html', {})

接下來我們編寫模板文件index.html，它的路徑位置如下所示：

 bots/
     __init__.py
     templates/
         bots/
             index.html
     urls.py
     views.py

index.html內(nèi)容如下所示。

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8"/>
    <title>Django+Channels+Celery聊天機(jī)器人</title>
</head>
<body>

<textarea id="chat-log" cols="100" rows="20" readonly></textarea>
<br/>
<input id="chat-message-input" type="text" size="100" 
      placeholder="輸入`help`獲取幫助信息."/><br/><input id="chat-message-submit" type="button" value="Send"/>
   <script>
    var wss_protocol = (window.location.protocol == 'https:') ? 'wss://': 'ws://';
    var chatSocket = new WebSocket(
        wss_protocol + window.location.host + '/ws/bots/'
        );

    chatSocket.onopen = function(e) {
document.querySelector('#chat-log').value +=
('歡迎來到大江狗Django聊天機(jī)器人. 請(qǐng)輸入`help`獲取幫助信息.\n')}

    chatSocket.onmessage = function(e) {
        var data = JSON.parse(e.data);
        var message = data['message'];
        document.querySelector('#chat-log').value += (message + '\n');
    };

    chatSocket.onclose = function(e) {
document.querySelector('#chat-log').value +=
('Socket closed unexpectedly, please reload the page.\n')};

    document.querySelector('#chat-message-input').focus();
    document.querySelector('#chat-message-input').onkeyup = function(e) {
        if (e.keyCode === 13) {  // enter, return
            document.querySelector('#chat-message-submit').click();
        }
    };

    document.querySelector('#chat-message-submit').onclick = function(e) {
        var messageInputDom = document.querySelector('#chat-message-input');
        var message = messageInputDom.value;
        chatSocket.send(JSON.stringify({
            'message': message
        }));
     messageInputDom.value = '';
    };
</script>

</body>
</html>

第四步編寫后臺(tái)websocket路由及處理方法

當(dāng) channels 接受 WebSocket 連接時(shí), 它也會(huì)根據(jù)根路由配置去查找相應(yīng)的處理方法。只不過channels的websocket路由不在urls.py中配置，處理函數(shù)也不寫在views.py。在channels中，這兩個(gè)文件分別變成了routing.py和consumers.py。

在bots應(yīng)用下新建routing.py, 添加如下代碼。它的作用是將發(fā)送至ws/bots/的websocket請(qǐng)求轉(zhuǎn)由BotConsumer處理。

from django.urls import re_path

from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/bots/$', consumers.BotConsumer.as_asgi()),
]

注意：定義websocket路由時(shí)，推薦使用常見的路徑前綴 (如/ws) 來區(qū)分 WebSocket 連接與普通 HTTP 連接，因?yàn)樗鼘⑹股a(chǎn)環(huán)境中部署 Channels 更容易，比如nginx把所有/ws的請(qǐng)求轉(zhuǎn)給channels處理。

與Django類似，我們還需要把這個(gè)app的websocket路由加入到項(xiàng)目的根路由中去。編輯myproject/asgi.py, 添加如下代碼：

# myproject/asgi.py
import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
import chat.routing
import bots.routing

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    # websocket請(qǐng)求使用的路由
    "websocket": AuthMiddlewareStack(
        URLRouter(
            bots.routing.websocket_urlpatterns
        )
    )
})

接下來在bots應(yīng)用下新建consumers.py, 添加如下代碼:

import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer

from . import tasks

COMMANDS = {
    'help': {
        'help': '命令幫助信息.',
    },
    'add': {
        'args': 2,
        'help': '計(jì)算兩個(gè)數(shù)之和, 例子: `add 12 32`.',
        'task': 'add'
    },
    'search': {
        'args': 1,
        'help': '通過名字查找詩(shī)人介紹，例子: `search 李白`.',
        'task': 'search'
    },
}



class BotConsumer(WebsocketConsumer):
    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        response_message = '請(qǐng)輸入`help`獲取命令幫助信息。'
        message_parts = message.split()
        if message_parts:
            command = message_parts[0].lower()
            if command == 'help':
                response_message = '支持的命令有:\n' + '\n'.join(
                    [f'{command} - {params["help"]} ' for command, params in COMMANDS.items()])
            elif command in COMMANDS:
                if len(message_parts[1:]) != COMMANDS[command]['args']:
                    response_message = f'命令`{command}`參數(shù)錯(cuò)誤，請(qǐng)重新輸入.'
                else:
                    getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])
                    response_message = f'收到`{message}`任務(wù).'
                    
        async_to_sync(self.channel_layer.send)(
            self.channel_name,
            {
                'type': 'chat.message',
                'message': response_message
            }
        )

    def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        self.send(text_data=json.dumps({
            'message': f'[機(jī)器人]: {message}'
        }))

上面代碼中最重要的一行如下所示。BotConsumer在接收到路由轉(zhuǎn)發(fā)的前端消息后，對(duì)其解析，將當(dāng)前頻道名和解析后的參數(shù)一起交由Celery異步執(zhí)行。Celery執(zhí)行任務(wù)完成以后會(huì)將結(jié)果發(fā)到這個(gè)頻道，這樣就實(shí)現(xiàn)了channels和Celery的通信。

getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])

第五步編寫Celery異步任務(wù)

在bots目錄下新建`tasks.py`，添加如下代碼：

from asgiref.sync import async_to_sync
from celery import shared_task
from channels.layers import get_channel_layer
from parsel import Selector
import requests

channel_layer = get_channel_layer()

@shared_task
def add(channel_name, x, y):
    message = '{}+{}={}'.format(x, y, int(x) + int(y))
    async_to_sync(channel_layer.send)(channel_name, {"type": "chat.message", "message": message})
    print(message)

@shared_task
def search(channel_name, name):
    spider = PoemSpider(name)
    result = spider.parse_page()
    async_to_sync(channel_layer.send)(channel_name, {"type": "chat.message", "message": str(result)})
    print(result)

class PoemSpider(object):
    def __init__(self, keyword):
        self.keyword = keyword
        self.url = "https://so.gushiwen.cn/search.aspx"
        
    def parse_page(self):
        params = {'value': self.keyword}
        response = requests.get(self.url, params=params)
        if response.status_code == 200:
            # 創(chuàng)建Selector類實(shí)例
            selector = Selector(response.text)
            # 采用xpath選擇器提取詩(shī)人介紹
            intro = selector.xpath('//textarea[starts-with(@id,"txtareAuthor")]/text()').get()
            print("{}介紹:{}".format(self.keyword, intro))
            if intro:
                return intro

        print("請(qǐng)求失敗 status:{}".format(response.status_code))
        return "未找到詩(shī)人介紹。"

以上兩個(gè)任務(wù)都以channel_name為參數(shù)，任務(wù)執(zhí)行完畢后通過channel_layer的send方法將結(jié)果發(fā)送到指定頻道。

注意：

- 默認(rèn)獲取channel_layer的方式是調(diào)用接口：channels.layers.get_channel_layer()。如果是在consumer中調(diào)用接口的話可以直接使用self.channel_layer。

- 對(duì)于channel layer的方法（包括send()、group_send()，group_add()等）都屬于異步方法，這意味著在調(diào)用的時(shí)候都需要使用await，而如果想要在同步代碼中使用它們，就需要使用裝飾器asgiref.sync.async_to_sync

第六步運(yùn)行看效果

如果不出意外，你現(xiàn)在的項(xiàng)目布局應(yīng)該如下所示。說實(shí)話，整個(gè)項(xiàng)目一共沒幾個(gè)文件，Python的簡(jiǎn)潔和效率真是出了名的好啊。

連續(xù)運(yùn)行如下命令，就可以看到我們文初的效果啦。

 # 啟動(dòng)django測(cè)試服務(wù)器
 python manage.py makemigrations
 python manage.py migrate
 python manage.py runserver
 
 # windows下啟動(dòng)Celery需eventlet
 # 啟動(dòng)Celery前確定redis服務(wù)已開啟哦
 Celery -A myproject worker -l info -P eventlet

小結(jié)

本文我們使用Django + Channels + Celery + Redis打造了一個(gè)聊天機(jī)器人，既會(huì)算算術(shù)，還會(huì)查古詩(shī)文。借用這個(gè)實(shí)現(xiàn)原理，你可以打造非常有趣的實(shí)時(shí)聊天應(yīng)用哦，比如在線即時(shí)問答，在線客服，實(shí)時(shí)查詢訂單，Django版的siri美女等等。

Django Channels + Websocket + Celery聊天機(jī)器人項(xiàng)目源碼地址：https://github.com/shiyunbo/django-channels-chatbot

以上就是Django實(shí)現(xiàn)聊天機(jī)器人的詳細(xì)內(nèi)容，更多關(guān)于Django 聊天機(jī)器人的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: