Django实现聊天机器人
演示效果如下所示：
实现原理
用户在聊天界面调用Celery异步任务，Celery异步任务执行完毕后发送结果给channels，然后channels通过websocket将结果实时推送给用户。对于简单的算术运算，Celery一般自行计算就好了。对于网上查找诗人简介这样的任务，Celery会调用Python爬虫(requests+parsel)爬取古诗文网站上的诗人简介，把爬取结果实时返回给用户。
接下来我们来看下具体的代码实现吧。
第一步 安装环境依赖
首先在虚拟环境中安装django和以下主要项目依赖。本项目使用了最新版本，为3.X版本。
# 主要项目依赖 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
新建一个名为myproject的项目，新建一个app名为bots。如果windows下安装报错，如何解决自己网上去找吧，很容易解决。修改settings.py, 将channels和chat加入到INSTALLED_APPS里，并添加相应配置，如下所示：
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'channels', # channels应用
'bots', # bots应用
]
# 设置ASGI应用
ASGI_APPLICATION = 'myproject.asgi.application'
# 生产环境中使用redis做后台，安装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应用的urls.py加入到项目urls.py中去，这和常规Django项目无异。
# 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后，我们要对其进行配置。分别修改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
# 设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
# 实例化
app = Celery('myproject')
# namespace='CELERY'作用是允许你在Django配置文件中对Celery进行配置
# 但所有Celery配置项必须以CELERY开头，防止冲突
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动从Django的已注册app中发现任务
app.autodiscover_tasks()
# 一个测试任务
@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内容等消息的格式设置，默认json CELERY_ACCEPT_CONTENT = ['application/json', ] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json'
完整Celery配置见：Django进阶：万字长文教你使用Celery执行异步和周期性任务(多图)
第三步 编写机器人聊天主页面
本例我们只需要利用django普通视图函数编写1个页面，用于展示首页(index)与用户交互的聊天页面。这个页面对应的路由及视图函数如下所示：
# 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内容如下所示。
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8"/>
<title>Django+Channels+Celery聊天机器人</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聊天机器人. 请输入`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>
第四步 编写后台websocket路由及处理方法
当 channels 接受 WebSocket 连接时, 它也会根据根路由配置去查找相应的处理方法。只不过channels的websocket路由不在urls.py中配置，处理函数也不写在views.py。在channels中，这两个文件分别变成了routing.py和consumers.py。
在bots应用下新建routing.py, 添加如下代码。它的作用是将发送至ws/bots/的websocket请求转由BotConsumer处理。
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/bots/$', consumers.BotConsumer.as_asgi()),
]
注意：定义websocket路由时，推荐使用常见的路径前缀 (如/ws) 来区分 WebSocket 连接与普通 HTTP 连接， 因为它将使生产环境中部署 Channels 更容易，比如nginx把所有/ws的请求转给channels处理。
与Django类似，我们还需要把这个app的websocket路由加入到项目的根路由中去。编辑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请求使用的路由
"websocket": AuthMiddlewareStack(
URLRouter(
bots.routing.websocket_urlpatterns
)
)
})
接下来在bots应用下新建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': '计算两个数之和, 例子: `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}`参数错误，请重新输入.'
else:
getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])
response_message = f'收到`{message}`任务.'
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'[机器人]: {message}'
}))
上面代码中最重要的一行如下所示。BotConsumer在接收到路由转发的前端消息后，对其解析，将当前频道名和解析后的参数一起交由Celery异步执行。Celery执行任务完成以后会将结果发到这个频道，这样就实现了channels和Celery的通信。
getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])
第五步 编写Celery异步任务
在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:
# 创建Selector类实例
selector = Selector(response.text)
# 采用xpath选择器提取诗人介绍
intro = selector.xpath('//textarea[starts-with(@id,"txtareAuthor")]/text()').get()
print("{}介绍:{}".format(self.keyword, intro))
if intro:
return intro
print("请求失败 status:{}".format(response.status_code))
return "未找到诗人介绍。"
以上两个任务都以channel_name为参数，任务执行完毕后通过channel_layer的send方法将结果发送到指定频道。
注意：
- 默认获取channel_layer的方式是调用接口：channels.layers.get_channel_layer()。如果是在consumer中调用接口的话可以直接使用self.channel_layer。
- 对于channel layer的方法（包括send()、group_send()，group_add()等）都属于异步方法，这意味着在调用的时候都需要使用await，而如果想要在同步代码中使用它们，就需要使用装饰器asgiref.sync.async_to_sync
第六步 运行看效果
如果不出意外，你现在的项目布局应该如下所示。说实话，整个项目一共没几个文件，Python的简洁和效率真是出了名的好啊。
连续运行如下命令，就可以看到我们文初的效果啦。
# 启动django测试服务器 python manage.py makemigrations python manage.py migrate python manage.py runserver # windows下启动Celery需eventlet # 启动Celery前确定redis服务已开启哦 Celery -A myproject worker -l info -P eventlet
小结
本文我们使用Django + Channels + Celery + Redis打造了一个聊天机器人，既会算算术，还会查古诗文。借用这个实现原理，你可以打造非常有趣的实时聊天应用哦，比如在线即时问答，在线客服，实时查询订单，Django版的siri美女等等。
Django Channels + Websocket + Celery聊天机器人项目源码地址：https://github.com/shiyunbo/django-channels-chatbot
以上就是Django实现聊天机器人的详细内容，更多关于Django 聊天机器人的资料请关注我们其它相关文章！
相关推荐
-
nodejs实现聊天机器人功能
技术栈 服务端: koa.koa-route.koa-websocket.request. 客户端: html.css.js.websocket. 远程聊天API: http://api.qingyunke.com/api.php?key=free&appid=0&msg=msg. 客户端展示 开发步骤 1.在桌面创建bbs文件夹,然后在文件夹内打开cmd,输入: $ npm init 初始化箱项目,生成package.json包管理文件 2.cmd输入: $ npm install ko
-
Python如何实现机器人聊天
今天午休的时候,无意之中看了一篇博客,名字叫Python实现机器人,感觉挺有的意思的. 于是用其写了一个简单的Python聊天,源码如下所示: # -*- coding: utf-8 -*- import aiml import sys import os def get_module_dir(name): print("module", sys.modules[name]) path = getattr(sys.modules[name], '__file__', None) pri
-
python操作微信自动发消息的实现(微信聊天机器人)
前言 最近在学习python,发现一个微信自动发消息的小demo感觉很有意思,试了一下,不成功,因为demo中用的是itchat这个库来操作微信,而这个库是通过微信网页版来操作微信的,现在微信网页版已经不能登录了所以失败,我又试了第二种方法,我试图找到微信界面上的搜索框控件,使用搜索框控件找到想法消息的人,然后发送消息,结果就是又失败了,为啥呢?经过我翻翻翻,找找找,发现微信的界面是使用duilib实现的,界面都是画上去的,控件只是逻辑上存在,而实际没有,我们根本获取不到,然后我只能通过手动移动
-
Python QQBot库的QQ聊天机器人
本文实例为大家分享了Python QQBot库的QQ聊天机器人的具体代码,供大家参考,具体内容如下 项目地址:https://github.com/pandolia/qqbot 1.安装 pip install qqbot 2.主动发出消息 from qqbot import _bot as bot # 登录QQ bot.Login(['-q', '2816626661']) ''' buddy 获取指定名称/备注的好友 group 获取群 ''' buddy = bot.List('buddy
-
vue.js实现h5机器人聊天（测试版）
本文实例为大家分享了vue.js实现h5机器人聊天的具体代码,供大家参考,具体内容如下 <!DOCTYPE html> <html> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width,initial-scale=1,minimum-scale=1,maximum-scale=1,user-sca
-
基于python的itchat库实现微信聊天机器人(推荐)
一.开始之前必须安装itchat库 pip install itchat(使用pip必须在电脑的环境变量中添加Python的路径) 或 conda install request 二.开始编程前,我们需要在图灵机器人官网注册自己的图灵机器人,来实现我们程序的智能聊天功能 1.图灵机器人官网(http://www.turingapi.com/) 2.注册登录后点击创建机器人 3.创建成功后,可以获得机器人提供的API接口(apikey) 三.代码实现 import itchat import re
-
Python使用20行代码实现微信聊天机器人
近来,打开微信群发消息,就会秒收到一些活跃分子的回复,有的时候感觉对方回答很在理,但是有的时候发现对方的回答其实是驴唇不对马嘴,仔细深究发现,原来对方是机器人.今天,小编就带大家用20行代码,带你一起打造一个微信聊天机器人,让你的微信群一直嗨不停~~ 首先我们需要安装一个微信相关的第三方库,itchat,在Windows上通过命令:pip install itchat,就可以将其安装. 其二,我们需要去图灵机器人官网:http://www.tuling123.com,注册一下,即可获得一个机器人
-
Python实战整活之聊天机器人
一.前言 刚刚学了一些python文件读写的内容,先跑过来整活了.顺便复习一下之前学的东西. import time doc_local='D:\learning_folder\interaction.txt' def iRead(): fr = open(doc_local, 'r') message=fr.read() return message def iWrite(message): fw = open(doc_local, 'w') fw.write(message) fw.clos
-
使用python接入微信聊天机器人
本文实例为大家分享了python接入微信聊天机器人的具体代码,供大家参考,具体内容如下 1.安装库wxpy: pip install -U wxpy or pip install -U wxpy -i https://pypi.doubanio.com/simple/ 2.简单操作上手: from wxpy import * bot = Bot(cache_path=True) #扫码登录验证 friends_stat = bot.friends().stats() friend_loc = [
-
jquery实现聊天机器人
本文实例为大家分享了jquery实现聊天机器人的具体代码,供大家参考,具体内容如下 <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <meta
-
python微信聊天机器人改进版(定时或触发抓取天气预报、励志语录等,向好友推送)
最近想着做一个微信机器人,主要想要实现能够每天定时推送天气预报或励志语录,励志语录要每天有自动更新,定时或当有好友回复时,能够随机推送不同的内容.于是开始了分析思路.博主是采用了多线程群发,因为微信对频繁发送消息过快还会出现发送失败的问题,因此还要加入time.sleep(1),当然时间根据自身情况自己定咯.本想把接入写诗机器人,想想自己的渣电脑于是便放弃了,感兴趣的可以尝试一下.做完会有不少收获希望对你有帮助. (1)我们要找个每天定时更新天气预报的网站,和一个更新励志语录的网站.当然如果你想