前言
本項目基于機器學習和語義識別技術(shù)，讓機器人理解文本并進行合適的答復(fù)?；锇閭兛梢酝ㄟ^該工程源碼，進行個人二次開發(fā)，比如使用語音與機器人交流，實現(xiàn)智能問答、智能音箱及智能機器寵物等等。
當然針對現(xiàn)在最火爆的ChatGPT等通用大語言模型，伙伴們可以直接將其應(yīng)用在模塊實現(xiàn)第6部分，其它詳細的接口使用操作，大家可以關(guān)注我博客的其它關(guān)于ChatGPT接口使用的說明。
總體設(shè)計
本部分包括系統(tǒng)整體結(jié)構(gòu)圖和系統(tǒng)流程圖。
系統(tǒng)整體結(jié)構(gòu)圖
系統(tǒng)整體結(jié)構(gòu)如圖所示。
系統(tǒng)流程圖
系統(tǒng)流程如圖所示。
運行環(huán)境
本部分包括 Python 環(huán)境、Pycharm 環(huán)境和 ChatterBot 環(huán)境。
Python 環(huán)境
需要 Python 3.6 及以上配置，進入 python 官方網(wǎng)站：www.python.org，選擇自己所需版本號，單擊 DownLoad，添加環(huán)境變量。
Pycharm 環(huán)境
PyCharm 是 一 款 功 能 強 大 的 Python 編輯器，具有跨平臺性 ， 下 載 地 址 ：
http://www.jetbrains.com/pycharm/download/#section=windowsPyCharm ，把 pycharm和 python解釋器進行連接。
ChatterBot 環(huán)境
基于 chatterbot 0.8.7 開發(fā)，打開 cmd 進入 python 所在的磁盤，輸入：
pip install –ignore-installed –upgrade chatterbot0.8.7
等待安裝即可。
模塊實現(xiàn)
本項目包括 6 個模塊：模型構(gòu)建、服務(wù)器端、客戶端、語音錄入、接口調(diào)用、模型訓練
及保存，下面分別給出各模塊的功能介紹及相關(guān)代碼。
1. 模型構(gòu)建
進入百度云官網(wǎng)：https://ai.baidu.com/，進入我的控制臺，打開百度語音進入語音應(yīng)用
管理界面，創(chuàng)建一個新的應(yīng)用，如圖所示。
記錄 APPID、API Key 和 Secret Key 三個值。
請求方式和參數(shù)網(wǎng)址：https://cloud.baidu.com/doc/SPEECH/s/Qk38y8lrl
通過調(diào)用百度語音 API 實現(xiàn)相應(yīng)的轉(zhuǎn)換功能，在百度云應(yīng)用管理中心可以看到記錄，即為調(diào)用成功。
2. 服務(wù)器端
此模塊主要處理用戶的登錄校驗，房間的人員消息處理，通過 config.py 中配置的列表PORT 生成幾個不同房間，相關(guān)代碼如下：
#導(dǎo)入 port 列表
import asynchat
import asyncore
from config import PORT
class CommandHandler:
"""
命令處理類
"""
def unknown(self, session, cmd):
# 響應(yīng)未知命令
# 通過 aynchat.async_chat.push 方法發(fā)送消息
session.push(('Unknown command {} \n'.format(cmd)).encode("utf-8"))
def handle(self, session, line):
line = line.decode()
# 命令處理
if not line.strip():
return
parts = line.split(' ', 1)
cmd = parts[0]
try:
line = parts[1].strip()
except IndexError:
line = ''
# 通過協(xié)議代碼執(zhí)行相應(yīng)的方法
method = getattr(self, 'do_' + cmd, None)
try:
method(session, line)
except TypeError:
self.unknown(session, cmd)
if __name__ == '__main__':
for i in range(len(PORT)):
ChatServer(PORT[i])
print("Chat server run at '127.0.0.1:{0}'".format(PORT[i]))
try:
asyncore.loop()
except KeyboardInterrupt:
print("Chat server exit")
在開啟客戶端前，運行服務(wù)器端，CommandHandler 類拆解 client 客戶端發(fā)送信息中的命令，并綁定函數(shù)。通過 config.py 中配置列表 PORT = range(1, 3)生成兩個房間，地址分別是 127.0.0.1:1 和 127.0.0.1:2，如圖所示。
3. 客戶端
該模塊提供登錄窗口、聊天窗口以及各種響應(yīng)事件，相關(guān)代碼如下：
1）登錄窗口設(shè)計
設(shè)計登錄窗口的 GUI 界面，添加包括標題、地址欄、用戶信息欄和登錄退出按鈕的幾個控件并綁定登錄、退出等各種事件。
class LoginFrame(wx.Frame):
"""
登錄窗口
"""
def __init__(self, parent, id, title, size):
# 初始化，添加控件并綁定事件
wx.Frame.__init__(self, parent, id, title)
self.SetSize(size)
self.Center()
self.serverAddressLabel = wx.StaticText(self, label="服務(wù)地址", pos=(45, 40), size=(120, 25))
self.userNameLabel = wx.StaticText(self, label="用戶名", pos=(45, 90), size=(120, 25))
self.serverAddress = wx.TextCtrl(self, value=default_server,
pos=(120, 37), size=(150, 25), style=wx.TE_PROCESS_ENTER)
self.userName = wx.TextCtrl(self, pos=(120, 87), size=(150, 25), style=wx.TE_PROCESS_ENTER)
self.loginButton = wx.Button(self, label='登錄', pos=(50, 145), size=(90, 30))
self.exitButton = wx.Button(self, label='退出', pos=(180, 145), size=(90, 30))
# 綁定登錄方法
self.loginButton.Bind(wx.EVT_BUTTON, self.login)
# 綁定退出方法
self.exitButton.Bind(wx.EVT_BUTTON, self.exit)
# 服務(wù)器輸入框Tab事件
self.serverAddress.SetFocus()
self.Bind(wx.EVT_TEXT_ENTER, self.usn_focus, self.serverAddress)
# 用戶名回車登錄
self.Bind(wx.EVT_TEXT_ENTER, self.login, self.userName)
self.Show()
# 回車調(diào)到用戶名輸入欄
def usn_focus(self, event):
self.userName.SetFocus()
def login(self, event):
# 登錄處理
try:
serverAddress = self.serverAddress.GetLineText(0).split(':')
con.open(serverAddress[0], port=int(serverAddress[1]), timeout=10)
response = con.read_some()
if response != b'Connect Success':
self.showDialog('Error', 'Connect Fail!', (200, 100))
return
con.write(('login ' + str(self.userName.GetLineText(0)) + '\n').encode("utf-8"))
response = con.read_some()
if response == b'UserName Empty':
self.showDialog('Error', 'UserName Empty!', (200, 100))
elif response == b'UserName Exist':
self.showDialog('Error', 'UserName Exist!', (200, 100))
else:
self.Close()
ChatFrame(None, 2, title='當前用戶：'+str(self.userName.GetLineText(0)), size=(515, 400))
except Exception:
self.showDialog('Error', 'Connect Fail!', (95, 20))
def exit(self, event):
self.Close()
# 顯示錯誤信息對話框
def showDialog(self, title, content, size):
dialog = wx.Dialog(self, title=title, size=size)
dialog.Center()
wx.StaticText(dialog, label=content)
dialog.ShowModal()
2）聊天窗口設(shè)計
設(shè)計聊天窗口的 GUI 界面，添加包括當前用戶顯示、信息輸入框、語音輸入按鈕、發(fā)送和關(guān)閉按鈕等各種控件并綁定發(fā)送消息、輸入消息等事件。
class ChatFrame(wx.Frame):
"""
聊天窗口
"""
def __init__(self, parent, id, title, size):
# 初始化，添加控件并綁定事件
wx.Frame.__init__(self, parent, id, title, style=wx.SYSTEM_MENU | wx.CAPTION | wx.CLOSE_BOX |
wx.DEFAULT_FRAME_STYLE)
self.SetSize(size)
self.Center()
self.chatFrame = wx.TextCtrl(self, pos=(5, 5), size=(490, 310), style=wx.TE_MULTILINE | wx.TE_READONLY)
self.sayButton = wx.Button(self, label="語音", pos=(5, 320), size=(58, 25))
self.message = wx.TextCtrl(self, pos=(65, 320), size=(280, 25), style=wx.TE_PROCESS_ENTER)
self.sendButton = wx.Button(self, label="發(fā)送", pos=(360, 320), size=(58, 25))
# self.usersButton = wx.Button(self, label="Users", pos=(373, 320), size=(58, 25))
self.closeButton = wx.Button(self, label="關(guān)閉", pos=(436, 320), size=(58, 25))
self.sendButton.Bind(wx.EVT_BUTTON, self.send) # 發(fā)送按鈕綁定發(fā)送消息方法
self.message.SetFocus() # 輸入框回車焦點
self.sayButton.Bind(wx.EVT_LEFT_DOWN, self.sayDown) # SAY按鈕按下
self.sayButton.Bind(wx.EVT_LEFT_UP, self.sayUp) # Say按鈕彈起
self.Bind(wx.EVT_TEXT_ENTER, self.send, self.message) # 回車發(fā)送消息
# self.usersButton.Bind(wx.EVT_BUTTON, self.lookUsers) # Users按鈕綁定獲取在線用戶數(shù)量方法
self.closeButton.Bind(wx.EVT_BUTTON, self.close) # 關(guān)閉按鈕綁定關(guān)閉方法
treceive = threading.Thread(target=self.receive) # 接收信息線程
treceive.start()
# self.ShowFullScreen(True) # 全屏
self.Show()
def sayDown(self, event):
trecording = threading.Thread(target=recording)
trecording.start()
def sayUp(self, event):
sayText = getText(r"E:\python_pycharm\ChatBot\voice\voice.wav")
self.message.AppendText(str(sayText))
self.send(self)
def send(self, event):
# 發(fā)送消息
message = str(self.message.GetLineText(0)).strip()
global bot_use
if message != '':
if message == "機器人":
bot_use = "ChatBot"
self.message.Clear()
con.write(('noone_say 成功呼叫語音助手！' + '\n').encode("utf-8"))
return
elif message == "用戶聊天":
bot_use = "User"
self.message.Clear()
con.write(('noone_say 語音助手已離開' + '\n').encode("utf-8"))
return
con.write(('say ' + message + '\n').encode("utf-8"))
self.message.Clear()
# 機器人回復(fù)
if bot_use == "ChatBot":
answer = chatbot(message)
con.write(('chatbot_say ' + answer + '\n').encode("utf-8"))
elif bot_use == "User":
return
if VOICE_SWITCH:
# 寫本地音樂文件
baidu_api(answer)
# 新建線程播放音樂
tplay_mp3 = threading.Thread(target=play_mp3)
tplay_mp3.start()
# thread.start_new_thread(play_mp3, ())
return
# def lookUsers(self, event):
# # 查看當前在線用戶
# con.write(b'look\n')
def close(self, event):
# 關(guān)閉窗口
tremove_voice = threading.Thread(target=remove_voice)
tremove_voice.start()
# thread.start_new_thread(remove_voice, ())
con.write(b'logout\n')
con.close()
self.Close()
def receive(self):
# 接受服務(wù)器的消息
while True:
sleep(1)
result = con.read_very_eager()
if result != '':
self.chatFrame.AppendText(result)
def saytime(self):
i = 0
while True:
self.chatFrame.AppendText('正在錄音...' + str(i) + '秒\n')
sleep(1)
i = i + 1
4. 語音錄入
該模塊提供錄音功能并將文件保存在本地， Recorder 的采樣率與百度語音轉(zhuǎn)文字的采樣率相同，相關(guān)代碼如下：
from pyaudio import PyAudio, paInt16
import numpy as np
import wave
class Recoder:
NUM_SAMPLES = 2000
# py audio內(nèi)置緩沖大小
SAMPLING_RATE = 16000 # 取樣頻率
LEVEL = 500 # 聲音保存的閾值
COUNT_NUM = 20 # NUM_SAMPLES個取樣之內(nèi)出現(xiàn)COUNT_NUM個大于LEVEL的取樣則記錄聲音
SAVE_LENGTH = 8 # 聲音記錄的最小長度：SAVE_LENGTH * NUM_SAMPLES 個取樣
TIME_COUNT = 20 # 錄音時間，單位s
Voice_String = []
def savewav(self, filename):
wf = wave.open(filename, 'wb')
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(self.SAMPLING_RATE)
wf.writeframes(np.array(self.Voice_String).tostring())
# wf.writeframes(self.Voice_String.decode())
wf.close()
def recoder(self):
pa = PyAudio()
stream = pa.open(format=paInt16, channels=1, rate=self.SAMPLING_RATE, input=True,
frames_per_buffer=self.NUM_SAMPLES)
save_count = 0
save_buffer = []
time_count = self.TIME_COUNT
while True:
time_count -= 1
# print time_count
# 讀入NUM_SAMPLES個取樣
string_audio_data = stream.read(self.NUM_SAMPLES)
# 將讀入的數(shù)據(jù)轉(zhuǎn)換為數(shù)組
audio_data = np.fromstring(string_audio_data, dtype=np.short)
# 計算大于LEVEL的取樣的個數(shù)
large_sample_count = np.sum(audio_data > self.LEVEL)
print(np.max(audio_data))
# 如果個數(shù)大于COUNT_NUM，則至少保存SAVE_LENGTH個塊
if large_sample_count > self.COUNT_NUM:
save_count = self.SAVE_LENGTH
else:
save_count -= 1
if save_count < 0:
save_count = 0
if save_count > 0:
# 將要保存的數(shù)據(jù)存放到save_buffer中
# print save_count > 0 and time_count >0
save_buffer.append(string_audio_data)
else:
# print save_buffer
# 將save_buffer中的數(shù)據(jù)寫入WAV文件，WAV文件的文件名是保存的時刻
# print "debug"
if len(save_buffer) > 0:
self.Voice_String = save_buffer
save_buffer = []
print("Recode a piece of voice successfully!")
return True
if time_count == 0:
if len(save_buffer) > 0:
self.Voice_String = save_buffer
save_buffer = []
print("Recode a piece of voice successfully!")
return True
else:
return False
def recording():
r = Recoder()
r.recoder()
r.savewav(r"E:\python_pycharm\ChatBot\voice\voice.wav")
# if tsayTime != '':
# stop_thread(tsayTime)
語音錄入時，控制面板輸出如圖所示。
此時在相應(yīng)路徑中可以看到錄制成功的語音文件，格式為.Wav，如圖所示。
5. 接口調(diào)用
該模塊調(diào)用 chatbot 返回文本信息、百度 API 語音識別、百度 API 轉(zhuǎn)文本為語音，相關(guān)代碼如下：
import pygame
from chatterbot import ChatBot
import requests
import json
from config import *
import time
import os
import random
import urllib.request
import base64
# 初始化百度返回的音頻文件地址，后面會變?yōu)槿肿兞浚S需改變
mp3_url = r'E:\python_pycharm\ChatBot\voice\\voice_du\\voice_ss.mp3'
''
# 播放Mp3文件
def play_mp3():
# 接受服務(wù)器的消息
pygame.mixer.init()
pygame.mixer.music.load(mp3_url)
pygame.mixer.music.play()
while pygame.mixer.music.get_busy():
time.sleep(1)
pygame.mixer.music.stop()
pygame.mixer.quit()
# 刪除聲音文件
def remove_voice():
path = r"E:\python_pycharm\ChatBot\voice\voice_du"
for i in os.listdir(path):
path_file = os.path.join(path, i)
try:
os.remove(path_file)
except:
continue
# 聊天機器人回復(fù)
def chatbot(info):
my_bot = ChatBot("", read_only=True,
database="./db.sqlite3")
res = my_bot.get_response(info)
return str(res)
# 百度講文本轉(zhuǎn)為聲音文件保存在本地 tts地址，無需token實時認證
def baidu_api(answer):
api_url = '{11}?idx={0}&tex={1}&cuid={2}&cod={3}&lan={4}&ctp={5}&pdt={6}&spd={7}&per={8}&vol={9}&pit={10}'\
.format(baidu_api_set["idx"], answer, baidu_api_set["cuid"], baidu_api_set["cod"], baidu_api_set["lan"],
baidu_api_set["ctp"], baidu_api_set["pdt"], baidu_api_set["spd"], baidu_api_set["per"],
baidu_api_set["vol"], baidu_api_set["pit"], baidu_api_url)
res = requests.get(api_url, headers=headers2)
# 本地Mp3語音文件保存位置
iname = random.randrange(1, 99999)
global mp3_url
mp3_url = r'E:\python_pycharm\ChatBot\voice\voice_du\voice_tts' + str(iname) + '.mp3'
with open(mp3_url, 'wb') as f:
f.write(res.content)
# 百度講文本轉(zhuǎn)為聲音文件保存在本地 方法2 tsn地址
def baidu_api2(answer):
# 獲取access_token
token = getToken()
get_url = baidu_api_url2 % (urllib.parse.quote(answer), "test", token)
voice_data = urllib.request.urlopen(get_url).read()
# 本地Mp3語音文件保存位置
name = random.randrange(1, 99999)
global mp3_url
mp3_url = r'E:\python_pycharm\ChatBot\voice\voice_du\voice_tsn' + str(name) + '.mp3'
voice_fp = open(mp3_url, 'wb+')
voice_fp.write(voice_data)
voice_fp.close()
return
# 百度語音轉(zhuǎn)文本
def getText(filename):
# 獲取access_token
token = getToken()
data = {}
data['format'] = 'wav'
data['rate'] = 16000
data['channel'] = 1
data['cuid'] = str(random.randrange(123456, 999999))
data['token'] = token
wav_fp = open(filename, 'rb')
voice_data = wav_fp.read()
data['len'] = len(voice_data)
data['speech'] = base64.b64encode(voice_data).decode('utf-8')
post_data = json.dumps(data)
# 語音識別的api url
upvoice_url = 'http://vop.baidu.com/server_api'
r_data = urllib.request.urlopen(upvoice_url, data=bytes(post_data, encoding="utf-8")).read()
print(json.loads(r_data))
err = json.loads(r_data)['err_no']
if err == 0:
return json.loads(r_data)['result'][0]
else:
return json.loads(r_data)['err_msg']
# 獲取百度API調(diào)用的認證，實時生成，因為有時間限制
def getToken():
# token認證的url
api_url = "https://openapi.baidu.com/oauth/2.0/token?" \
"grant_type=client_credentials&client_id=%s&client_secret=%s"
token_url = api_url % (BaiDu_API_Key_GetVoi, BaiDu_Secret_Key_GetVoi)
r_str = urllib.request.urlopen(token_url).read()
token_data = json.loads(r_str)
token_str = token_data['access_token']
return token_str
6.模型訓練及保存
該模塊主要訓練 chatbot 機器人，保存在本地 sqlite 數(shù)據(jù)庫，提供兩種語料訓練方法。如果使用ChatGPT的話，可以將此步省略，直接使用其API接口
1) 語句訓練
直接寫訓練語句，也可開啟通過聊天時的語句自動，相關(guān)代碼如下：
my_bot.train(["你今年幾歲了呀？", "我今年三歲半了！", ])
my_bot.train([
"你吃飯了嗎",
"我吃飽了",
"你叫什么名字呀",
"我叫小郵",
])
2) 語料庫訓練
通過自定義語料庫訓練。
安裝chatbot后默認提供的中文語料格式E:\Python36\Lib\sitepackages\chatterbot_corpus\data\Chinese。參照格式，寫入自己的語料文件。
#使用自定義語句訓練
my_bot.set_trainer(ChatterBotCorpusTrainer)
my_bot.train("chatterbot.corpus.chinese")
系統(tǒng)測試
本部分包括模型效果和模型應(yīng)用。
1. 模型效果
將所需語料庫數(shù)據(jù)帶入模型進行訓練，如圖所示。
訓練結(jié)果如圖所示。
2. 模型應(yīng)用
本部分包括程序運行、應(yīng)用使用說明和程序功能實現(xiàn)。
1) 程序運行
- rainChat.py 訓練本地 chatbot 機器人（每次更新訓練內(nèi)容，運行一次即可）；如果使用ChatGPT的話，可以將此步省略，直接使用其API接口
- server.py 開啟服務(wù)器；
- client.py 運行客戶端，每次運行都可登陸一個用戶。
2) 應(yīng)用使用說明
運行程序后，用戶初始登錄界面如圖所示。界面從上至下，兩個文本框：一個輸入服務(wù)地址，一個輸入用戶名；兩個按鈕：一個用于登錄，一個用于項目退出。
用戶輸入服務(wù)地址和登錄后進入聊天界面。主體為聊天展示框，底部從左到右分別有語音輸入按鈕、文字輸入文本框、發(fā)送按鈕和聊天室關(guān)閉按鈕，如圖所示。
3) 程序功能實現(xiàn)
（1）在聊天框輸入“機器人”可以呼叫語音助手，輸入“用戶聊天”退出語音機器人模式，如圖所示。
（2）長按“語音”按鈕錄入語音與機器人進行對話，如圖所示。
（3）退出機器人聊天模式還可實現(xiàn)多人聊天，如圖所示。
源代碼下載地址
基于Python+百度語音的智能語音ChatGPT聊天機器人（機器學習+深度學習+語義識別）含全部工程源碼文章來源：http://www.zghlxwxcb.cn/news/detail-468462.html
其它資料下載
如果大家想繼續(xù)了解人工智能相關(guān)學習路線和知識體系，歡迎大家翻閱我的另外一篇博客《重磅 | 完備的人工智能AI 學習——基礎(chǔ)知識學習路線，所有資料免關(guān)注免套路直接網(wǎng)盤下載》
這篇博客參考了Github知名開源平臺，AI技術(shù)平臺以及相關(guān)領(lǐng)域?qū)＜遥篋atawhale，ApacheCN，AI有道和黃海廣博士等約有近100G相關(guān)資料，希望能幫助到所有小伙伴們。文章來源地址http://www.zghlxwxcb.cn/news/detail-468462.html
到了這里，關(guān)于基于Python+百度語音的智能語音ChatGPT聊天機器人（機器學習+深度學習+語義識別）含全部工程源碼 適合個人二次開發(fā)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
