Django實(shí)現(xiàn)聊天機(jī)器人
演示效果如下所示：
用戶在聊天界面調(diào)用Celery異步任務(wù)，Celery異步任務(wù)執(zhí)行完畢后發(fā)送結(jié)果給channels，然后channels通過websocket將結(jié)果實(shí)時(shí)推送給用戶。對于簡單的算術(shù)運(yùn)算，Celery一般自行計(jì)算就好了。對于網(wǎng)上查找詩人簡介這樣的任務(wù)，Celery會(huì)調(diào)用Python爬蟲(requests+parsel)爬取古詩文網(wǎng)站上的詩人簡介，把爬取結(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_redisimport osCHANNEL_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后，我們要對其進(jìn)行配置。分別修改myproject目錄下的__init__.py和celery.py（新建), 添加如下代碼：
# __init__.pyfrom .celery import app as celery_app__all__ = (’celery_app’,)# celery.pyimport osfrom celery import Celery# 設(shè)置環(huán)境變量os.environ.setdefault(’DJANGO_SETTINGS_MODULE’, ’myproject.settings’)# 實(shí)例化app = Celery(’myproject’)# namespace=’CELERY’作用是允許你在Django配置文件中對Celery進(jìn)行配置# 但所有Celery配置項(xiàng)必須以CELERY開頭，防止沖突app.config_from_object(’django.conf:settings’, namespace=’CELERY’)# 自動(dòng)從Django的已注冊app中發(fā)現(xiàn)任務(wù)app.autodiscover_tasks()# 一個(gè)測試任務(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)jsonCELERY_ACCEPT_CONTENT = [’application/json’, ]CELERY_TASK_SERIALIZER = ’json’CELERY_RESULT_SERIALIZER = ’json’
完整Celery配置見：Django進(jìn)階：萬字長文教你使用Celery執(zhí)行異步和周期性任務(wù)(多圖)
第三步 編寫機(jī)器人聊天主頁面本例我們只需要利用django普通視圖函數(shù)編寫1個(gè)頁面，用于展示首頁(index)與用戶交互的聊天頁面。這個(gè)頁面對應(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 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ī)器人. 請輸入`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請求轉(zhuǎn)由BotConsumer處理。
from django.urls import re_pathfrom . import consumerswebsocket_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的請求轉(zhuǎn)給channels處理。
與Django類似，我們還需要把這個(gè)app的websocket路由加入到項(xiàng)目的根路由中去。編輯myproject/asgi.py, 添加如下代碼：
# myproject/asgi.pyimport osfrom channels.auth import AuthMiddlewareStackfrom channels.routing import ProtocolTypeRouter, URLRouterfrom django.core.asgi import get_asgi_applicationimport chat.routingimport bots.routingos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')application = ProtocolTypeRouter({ 'http': get_asgi_application(), # websocket請求使用的路由 'websocket': AuthMiddlewareStack(URLRouter( bots.routing.websocket_urlpatterns) )})
接下來在bots應(yīng)用下新建consumers.py, 添加如下代碼:
import jsonfrom asgiref.sync import async_to_syncfrom channels.generic.websocket import WebsocketConsumerfrom . import tasksCOMMANDS = { ’help’: {’help’: ’命令幫助信息.’, }, ’add’: {’args’: 2,’help’: ’計(jì)算兩個(gè)數(shù)之和, 例子: `add 12 32`.’,’task’: ’add’ }, ’search’: {’args’: 1,’help’: ’通過名字查找詩人介紹，例子: `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 = ’請輸入`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ò)誤，請重新輸入.’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 WebSocketself.send(text_data=json.dumps({ ’message’: f’[機(jī)器人]: {message}’}))
上面代碼中最重要的一行如下所示。BotConsumer在接收到路由轉(zhuǎn)發(fā)的前端消息后，對其解析，將當(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_syncfrom celery import shared_taskfrom channels.layers import get_channel_layerfrom parsel import Selectorimport requestschannel_layer = get_channel_layer()@shared_taskdef 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_taskdef 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 = keywordself.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選擇器提取詩人介紹 intro = selector.xpath(’//textarea[starts-with(@id,'txtareAuthor')]/text()’).get() print('{}介紹:{}'.format(self.keyword, intro)) if intro:return introprint('請求失敗 status:{}'.format(response.status_code))return '未找到詩人介紹。'
以上兩個(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。
- 對于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的簡潔和效率真是出了名的好啊。
連續(xù)運(yùn)行如下命令，就可以看到我們文初的效果啦。
# 啟動(dòng)django測試服務(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ì)查古詩文。借用這個(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ī)器人的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！
相關(guān)文章：
1. python 如何在 Matplotlib 中繪制垂直線2. bootstrap select2 動(dòng)態(tài)從后臺(tái)Ajax動(dòng)態(tài)獲取數(shù)據(jù)的代碼3. ASP常用日期格式化函數(shù) FormatDate()4. python中@contextmanager實(shí)例用法5. html中的form不提交（排除）某些input 原創(chuàng)6. CSS3中Transition屬性詳解以及示例分享7. js select支持手動(dòng)輸入功能實(shí)現(xiàn)代碼8. 如何通過python實(shí)現(xiàn)IOU計(jì)算代碼實(shí)例9. 開發(fā)效率翻倍的Web API使用技巧10. vue使用moment如何將時(shí)間戳轉(zhuǎn)為標(biāo)準(zhǔn)日期時(shí)間格式
網(wǎng)公網(wǎng)安備