700字范文 > 树莓派简易聊天机器人

树莓派简易聊天机器人

时间：2022-05-11 11:08:08

相关推荐

树莓派简易聊天机器人

前言

这篇文章属于此系列：

一个用树莓派做的会聊天，能人脸识别（支持云台追踪）和发送邮件的小玩具

/yonglisikao/article/details/82804318

原理简析

简易的聊天机器人包括以下几个部分：输入声音（录音），把录音中有声音的片段提取出来（VAD），将声音转换为文字（语音识别），对文字进行回复，将文字转换为语音（语音识别），输出语音（播放）。

1.硬件要求

树莓派及基本配件，声音输出设备（音箱），USB声卡以及声音输入设备（麦克风），USB声卡与麦克风可由内置麦克风的USB摄像头代替

2.系统要求

Raspbain系统（其它Linux系统应该也可以）

3.软件要求

Python及一些库——baidu-aip（接入百度语音），pyaudio（输入和输出声音），requests（接入图灵机器人），webrtcvad（语音活动检测即VAD）等

4.其他要求

因为语音识别和语音合成使用的是百度云语音，所以需要在百度的AI开放平台注册自己的应用，获取相应口令。同样地，负责聊天的机器人使用的是图灵机器人，也需要去官网注册，获取相应口令。以上三个接口都是在线使用的。

环境搭建

关于声卡的配置请参考下面这篇文章：

/p/2e8eed5070ed

1.安装pyaudio

在命令行输入：

sudo apt-get install portaudio19-devsudo apt-get install python-all-dev

为了与系统的安装包隔离，建议使用虚拟环境（virtualenv），虚拟环境安装完成后，进入虚拟环境，下文所有操作均在虚拟环境下进行。

在命令行输入：

pip install pyaudio

如果你没有安装pip，请前往官网

2.安装baidu-aip，requests，webrtcvad

在命令行输入：

pip install baidu-aippip install requestspip install webrtcvad

至此，环境搭建完毕

代码实现

1.语音识别及语音合成

直接参考下面代码及注释。更多花样玩法，以及一些参数设置请参考官方文档

# coding=utf-8from aip import AipSpeech# 使用时请将下面的内容替换为你自己的APP_ID = '11549141'API_KEY = 'dNEEwltpoaUM086aHBAkxdjy'SECRET_KEY = 'jRjaMTPz9QegCcxmZ10XBg9OMj94nU5Y'# 初始化client = AipSpeech(APP_ID, API_KEY, SECRET_KEY)def text2sound(words='你好'):# 语音合成函数，传入欲合成的内容，返回成功与否，若成功默认将文件保存为'test.wav'result = client.synthesis(words, 'zh', 1, {'vol': 5, 'aue': 6, 'per': 4}) # 具体的参数设置请参考官方文档if not isinstance(result, dict):with open('test.wav', 'wb') as f:f.write(result)return Trueelse:return Falsedef sound2text(file_path='test.wav'):# 语音识别函数，传入文件名（默认为'test.wav'），返回识别结果或错误代码with open(file_path, 'rb') as fp:recog = client.asr(fp.read(), 'wav', 16000, {'dev_pid': 1536}) # 参数设置见文档if recog['err_no'] == 0:return recog['err_no']return recog['result'][0]# 调用示例text2sound('你吃饭了吗?')sound2text('test.wav')

2.语音输入，输出及VAD

以下代码改编自（非中文注释为原作者所加）：

/wangshub/python-vad

这里的录音直接融合了VAD。更改特定参数可改变灵敏度，参数的位置见注释。运行时会有报错，忽略即可，如果实在想解决，下面这篇文章有一些解决方案，但不一定能解决所有问题：

/u013372900/article/details/72784823

# coding=utf-8import webrtcvadimport collectionsimport sysimport signalimport pyaudiofrom array import arrayfrom struct import packimport waveimport timeimport osFORMAT = pyaudio.paInt16CHANNELS = 1RATE = 16000CHUNK_DURATION_MS = 30 # supports 10, 20 and 30 (ms)PADDING_DURATION_MS = 1500 # 1 sec jugementCHUNK_SIZE = int(RATE * CHUNK_DURATION_MS / 1000) # chunk to readCHUNK_BYTES = CHUNK_SIZE * 2 # 16bit = 2 bytes, PCMNUM_PADDING_CHUNKS = int(PADDING_DURATION_MS / CHUNK_DURATION_MS)NUM_WINDOW_CHUNKS = int(240 / CHUNK_DURATION_MS)# NUM_WINDOW_CHUNKS = int(400 / CHUNK_DURATION_MS) # 400 ms/ 30ms geNUM_WINDOW_CHUNKS_END = NUM_WINDOW_CHUNKS * 2def handle_int(sig, chunk):global leave, got_a_sentenceleave = Truegot_a_sentence = Truedef record_to_file(path, data, sample_width):"Records from the microphone and outputs the resulting data to 'path'"# sample_width, data = record()data = pack('<' + ('h' * len(data)), *data)wf = wave.open(path, 'wb')wf.setnchannels(1)wf.setsampwidth(sample_width)wf.setframerate(RATE)wf.writeframes(data)wf.close()def normalize(snd_data):"Average the volume out"MAXIMUM = 32767 # 16384times = float(MAXIMUM) / max(abs(i) for i in snd_data)r = array('h')for i in snd_data:r.append(int(i * times))return rsignal.signal(signal.SIGINT, handle_int)def record_sound(file_path='test.wav'):# 录音，有声音自动写入文件，默认为'test.wav'，声音结束后录音也停止，调用一次，录制一个片段vad = webrtcvad.Vad(3) # 这个参数可为1，2，3，可改变灵敏度，越大越粗犷pa = pyaudio.PyAudio()stream = pa.open(format=FORMAT,channels=CHANNELS,rate=RATE,input=True,start=False,# input_device_index=2,frames_per_buffer=CHUNK_SIZE)got_a_sentence = Falseleave = Falseno_time = 0while not leave:ring_buffer = collections.deque(maxlen=NUM_PADDING_CHUNKS)triggered = Falsevoiced_frames = []ring_buffer_flags = [0] * NUM_WINDOW_CHUNKSring_buffer_index = 0ring_buffer_flags_end = [0] * NUM_WINDOW_CHUNKS_ENDring_buffer_index_end = 0buffer_in = ''# WangS（原作者的名字，就喜欢这种造轮子的人）raw_data = array('h')index = 0start_point = 0StartTime = time.time()print("* recording: ")stream.start_stream()while not got_a_sentence and not leave:chunk = stream.read(CHUNK_SIZE)# add WangSraw_data.extend(array('h', chunk))index += CHUNK_SIZETimeUse = time.time() - StartTimeactive = vad.is_speech(chunk, RATE)sys.stdout.write('1' if active else '_')ring_buffer_flags[ring_buffer_index] = 1 if active else 0ring_buffer_index += 1ring_buffer_index %= NUM_WINDOW_CHUNKSring_buffer_flags_end[ring_buffer_index_end] = 1 if active else 0ring_buffer_index_end += 1ring_buffer_index_end %= NUM_WINDOW_CHUNKS_END# start point detectionif not triggered:ring_buffer.append(chunk)num_voiced = sum(ring_buffer_flags)if num_voiced > 0.8 * NUM_WINDOW_CHUNKS:sys.stdout.write(' Open ')triggered = Truestart_point = index - CHUNK_SIZE * 20 # start point# voiced_frames.extend(ring_buffer)ring_buffer.clear()# end point detectionelse:# voiced_frames.append(chunk)ring_buffer.append(chunk)num_unvoiced = NUM_WINDOW_CHUNKS_END - sum(ring_buffer_flags_end)if num_unvoiced > 0.90 * NUM_WINDOW_CHUNKS_END or TimeUse > 10:sys.stdout.write(' Close ')triggered = Falsegot_a_sentence = Truesys.stdout.flush()sys.stdout.write('\n')# data = b''.join(voiced_frames)stream.stop_stream()print("* done recording")got_a_sentence = False# write to fileraw_data.reverse()for index in range(start_point):raw_data.pop()raw_data.reverse()raw_data = normalize(raw_data)record_to_file(file_path, raw_data, 2)leave = Truestream.close()return TrueCHUNK = 512 # 512是树莓派能使用的最大的CHUNKdef play_sound(file_path='test.wav'):# 播放声音文件，默认为'test.wav'wf = wave.open(file_path, 'rb')p = pyaudio.PyAudio()stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),channels=wf.getnchannels(),rate=wf.getframerate(),output=True)data = wf.readframes(CHUNK)while data != b'':stream.write(data)data = wf.readframes(CHUNK)stream.stop_stream()stream.close()p.terminate()return# 调用示例record_sound('test.wav')play_sound('test.wav')

如果你对声音文件不了解，不明白chunk，rate等术语是什么意思，相信下面这篇文章会帮到你：

/lzxwalex/p/6922099.html

3.图灵机器人对话

代码如下：

# coding=utf-8import requestsimport jsondef tuling(text='I said nothing'):# 与图灵机器人对话tuling_url = '/openapi/api'tuling_date = {'key': '4527c40a4ee64b2bb4ca92cf020847be','info': text} # 使用时请将key更换为你自己的r = requests.post(tuling_url, data=tuling_date)return json.loads(r.text)['text']# 调用示例reply = tuling('你吃饭了吗？')print(reply)