设为首页收藏本站
首页 > 论坛 > 行空板

[K10项目分享]项目实践案例征集 + AI + 大模型语音聊天机器人

御坂10032号 | 学徒 | 创造力： | 帖子： | 发消息 | 串个门 | 加好友 | 打招呼
2025-1-22 17:36:50 [显示全部楼层]
39浏览
0回复
0收藏
查看: 39|回复: 0

[K10项目分享] 项目实践案例征集 + AI + 大模型语音聊天机器人

[复制链接]
本帖最后由 御坂10032号 于 2025-1-23 04:01 编辑

项目背景



为满足新课标中八年级“物联网实践与探索”内容要求， 结合教材中人工智能教学相关的需求，因此设计了AI 大模型语音聊天机器人的项目。通过当前应用案例的学习，学生可以收获如何通过Mind + 结合对应的上位机Python 来来实现一个自定义的语音助手，以及上位机的相关应用。 比如说如何调用第三方的API接口来实现文字到语音或者语音到文字的互相转， base64编码，和反向代理服务器Nginx的基本使用。同时结合通义千问来实现AI助手的相关功能。


项目介绍


项目开发难度 ： 难（具体体现在上位机的逻辑处理上）
项目所需物料 : 行空板K10 + SD卡
项目所需第三方API Key
  • 讯飞语音转文字
  • 讯飞文字转语音
  • 阿里云通义千问

项目所需前置技术:
  • Arduino 环境和Mind + 环境的基础使用
  • 串口相关知识
  • C语言文件保存和Python文件保存相关知识
  • PIP包管理器的使用
  • WAV音频编码相关知识
  • Nginx反向代理服务器的使用
  • base64编码

项目原理


项目主要是采用了K10作为语音的采集和播放终端， 当K10上的按键A被按下时那么触发本地的语音采集。之后K10会将采集的WAV语音文件保存到内存卡中。 之后通过串口的方式将wav的语音文件发送到上位机Python。 当Python收到K10发送到的数据之后，首先将二进制的数据转换成Bin文件，之后再根据Wav格式的编码将bin文件转换成和内存卡内相同的wav文件并且保存在上位机（Python). 之后通过调用讯飞语音的语音转文字的API接口，上位机可以获取到语音数据中的文字信息。 同时当上位机拿到了文字信息之后。上位机会将文字信息通过调用通义千问的Turbo模型来实现AI的聊天功能。此时上位机继续将通义千问返回的消息发送给讯飞的文字转语音功能。当文字转换语音完毕之后。语音的数据将会编码为base64格式发送给上位机。上位机需要将base64编码转换成wav文件，接着将wav文件生成到由nginx代理的静态目录中。 当wav文件生成完毕，上位机会通过串口给K10发送一条命令，来通知K10语音文件已经准备就绪。K10随即发送HTTP GET请求来将上位机nginx代理的静态目录中的音频文件下载到SD卡中再由Mind + 提供的语音播放API进行播放从事实现整体的聊天功能。


简要时序图如下



项目实践案例征集 + AI + 大模型语音聊天机器人图1


项目实现步骤



1- SD卡准备阶段

由于需要使用到SD卡来保存音频的相关数据，因此在项目的开始之前最好确认一下SD卡的相关信息。如果SD卡的容量超过了32GB的话，请参考官方文档来对SD卡进行初始化操作。


2- 打开Mind + ， 并且选取开发板为K10
项目实践案例征集 + AI + 大模型语音聊天机器人图2

3- 选择网络模块的拓展
项目实践案例征集 + AI + 大模型语音聊天机器人图3
网络模块选择了WIFI，用于连接2.4GAP， 而HTTP模块的引入则可以发送对应的HTTP请求。


4- 返回编辑器打开手动编辑模式。并且引入必须的库文件。
  1. #include "asr.h"

  2. #include <DFRobot_Iot.h>

  3. #include "unihiker_k10.h"

  4. #include <DFRobot_HTTPClient.h>
复制代码


5- 定义程序需要使用的对象信息。
  1. // 创建对象

  2. DFRobot_Iot  myIot;

  3. UNIHIKER_K10 k10;

  4. ASR          asr;

  5. Music        music;

  6. DFRobot_HTTPClient http;
复制代码

6- 需要注意的一点是DFRobot_HTTPClient.h 这个库并不支持直接访问内部的httpclient对象，因此想要成功的从http响应的stream中获取数据的话，还需要做一点小小的修改。 我们可以使用everything这个软件来搜索DFRobot_HTTPClient.h
项目实践案例征集 + AI + 大模型语音聊天机器人图4


7- 修改DFRobot_HTTPClient.h 内部的httpclient的访问修饰符，从private 修改为public使其可以被外部访问直接调用。如下代码所示，为修改之后的内容。
  1. #ifndef DFROBOT_HTTPCLIENT_H

  2. #define DFROBOT_HTTPCLIENT_H

  3. 

  4. #include <Arduino.h>

  5. 

  6. #if defined(ARDUINO_ARCH_ESP8266)

  7. #include <ESP8266HTTPClient.h>

  8. #include <WiFiClientSecureBearSSL.h>

  9. #else

  10. #include <HTTPClient.h>

  11. #endif

  12. 

  13. 

  14. class DFRobot_HTTPClient

  15. {

  16. public:

  17.     void init();

  18.     void addParam(const String& name, const String& value);

  19.     void addParam(float name, const String& value) {this->addParam(String(name), value);}

  20.     void addParam(const String& name, float value) {this->addParam(name, String(value));}

  21.     void addParam(float name, float value) {this->addParam(String(name), String(value));}

  22. 

  23.     void addHeader(const String& name, const String& value);

  24.     void addHeader(float name, const String& value) {this->addHeader(String(name), value);}

  25.     void addHeader(const String& name, float value) {this->addHeader(name, String(value));}

  26.     void addHeader(float name, float value) {this->addHeader(String(name), String(value));}

  27. 

  28.     void addString(const String& text);

  29.     void addString(float text) {this->addString(String(text));}

  30. 

  31.     void GET(String url, float timeout = 10000);

  32.     void POST(String url, float timeout = 10000);

  33. 

  34.     String getLine();

  35.     String getString();

  36.     HTTPClient _httpclient;

  37.     String _params, _body; 

  38.     int _httpcode;

  39. };

  40. 

  41. #endif

复制代码


8 - 重启MIND + 并且重新载入工程使其修改的文件生效。

9- 在setup() 方法中初始化开发板信息、串口、语音合成播放速度、初始化SD卡和初始化wifi连接。
  1. // 主程序开始

  2. void setup() {

  3.     k10.begin();

  4.     Serial.begin(115200);

  5.     asr.setAsrSpeed(1);

  6.     k10.initSDFile();

  7.     myIot.wifiConnect("ImmortalWrt", "mazha1997");

  8.     while (!myIot.wifiStatus()) {}

  9. }
复制代码


10 - 现在初始化工作已经完成了，现在我们希望当按下K10上的A键的时候将会开始录制一段音频数据，并且保存到SD卡然后通过串口发送到Python上位机。
  1. void recordAndSendFile()
复制代码

代码解读： 上述代码中通过调用积木编程的API来实现了音频文件的录制和保存。 当录制完成之后， 通过串口将SD卡内的音频文件发送到了Python上位机。需要注意的是上述代码中的START_OF_FILE 和 END_OF_FILE 是用来区分音频的起始和结束标志位。 由于上位机也是无限循环等待，所以这里需要使用起始位的方式来进行处理从而来实现无限的对话功能。

11- 上位机文件接收和转码
如果想在python中使用串口功能的话则需要pyserial的这个库, 我们可以在pip中直接通过下述命令进行安装。
  1. pip install pyserial
复制代码
导入pyserial
  1. import serial
复制代码

之后以和开发板与之匹配的串口号和波特率打开串口。
  1. if __name__ == '__main__':

  2.     port = 'COM38'

  3.     baud_rate = 115200

  4.     ser = serial.Serial(port, baud_rate, timeout=1)
复制代码


我们希望程序在运行时一直监控K10 发送过来的消息， 一旦接收到开始标志即开始接收串口数据，然后保存为BIN文件方便后面的转码。所以我们需要在main入口中再加上一个while循环来读取串口的数据
  1. if __name__ == '__main__':

  2.     port = 'COM38'

  3.     baud_rate = 115200

  4.     ser = serial.Serial(port, baud_rate, timeout=1)

  5.     while True:

  6.     save_serial_data_to_bin()
复制代码
对于这个save_serial_data_to_bin() 方法则为下述定义
  1. def save_serial_data_to_bin():

  2.     try:

  3.         with open(OUTPUT_FILE_PATH, 'wb') as f:

  4.             in_block = False

  5. 

  6.             while True:

  7.                 if ser.in_waiting > 0:

  8.                     data = ser.readline()

  9. 

  10.                     if START_OF_FILE in data:

  11.                         in_block = True

  12.                         continue

  13. 

  14.                     if END_OF_FILE in data:

  15.                         in_block = False

  16.                         print("End of file marker detected. Stopping.")

  17.                         break

  18. 

  19.                     if in_block:

  20.                         f.write(data)

  21.                         # print(f"Data written: {data}")

  22. 

  23.     except serial.SerialException as e:

  24.         print(f"Error opening serial port: {e}")
复制代码
上述代码用于将K10发送的数据直接保存为BIN格式，方便我们之后的数据转码。修改后的代码如下所示
  1. if __name__ == '__main__':

  2.     port = 'COM38'

  3.     baud_rate = 115200

  4.     ser = serial.Serial(port, baud_rate, timeout=1)

  5.     while True:

  6.         save_serial_data_to_bin()

  7.         add_wav_header(OUTPUT_FILE_PATH, "sound.wav", sample_rate=32000, num_channels=1, bits_per_sample=16)
复制代码
对于add_wav_header方法的定义则是如下所示，用于将bin文件转换为wav格式（增加文件头信息等）。
  1. def add_wav_header(bin_file, wav_file, sample_rate=16000, num_channels=1, bits_per_sample=16):

  2.     with open(bin_file, "rb") as bin_f:

  3.         audio_data = bin_f.read()

  4. 

  5.     # 计算头部字段

  6.     subchunk2_size = len(audio_data)

  7.     chunk_size = 36 + subchunk2_size

  8.     byte_rate = sample_rate * num_channels * (bits_per_sample // 8)

  9.     block_align = num_channels * (bits_per_sample // 8)

  10. 

  11.     # 构建 WAV 文件头

  12.     wav_header = struct.pack(

  13.         '<4sI4s4sIHHIIHH4sI',

  14.         b'RIFF',  # ChunkID

  15.         chunk_size,  # ChunkSize

  16.         b'WAVE',  # Format

  17.         b'fmt ',  # Subchunk1ID

  18.         16,  # Subchunk1Size

  19.         1,  # AudioFormat (1 = PCM)

  20.         num_channels,  # NumChannels

  21.         sample_rate,  # SampleRate

  22.         byte_rate,  # ByteRate

  23.         block_align,  # BlockAlign

  24.         bits_per_sample,  # BitsPerSample

  25.         b'data',  # Subchunk2ID

  26.         subchunk2_size  # Subchunk2Size

  27.     )

  28. 

  29.     # 写入 WAV 文件

  30.     with open(wav_file, "wb") as wav_f:

  31.         wav_f.write(wav_header)  # 写入头部

  32.         wav_f.write(audio_data)  # 写入音频数据
复制代码

12 - 调用讯飞文字转换语音的API来将我们K10采集的人声转换成文字。首先在讯飞文档中心的语音识别的导航下的极速语音转写下找到WEBAPI的调用文档。点击接口demo下载来下载调用的demo程序。并且选择python进行下载。
项目实践案例征集 + AI + 大模型语音聊天机器人图5

之后我们需要注册讯飞开放平台，在讯飞开放平台中选择产品的试用
项目实践案例征集 + AI + 大模型语音聊天机器人图6。
在控制台中拷贝自己的Key和密钥信息。

项目实践案例征集 + AI + 大模型语音聊天机器人图7

之后我们来调用语音转换文字的功能。
  1. if __name__ == '__main__':

  2.     port = 'COM38'

  3.     baud_rate = 115200

  4.     ser = serial.Serial(port, baud_rate, timeout=1)

  5.     while True:

  6.         save_serial_data_to_bin()

  7.         add_wav_header(OUTPUT_FILE_PATH, "sound.wav", sample_rate=32000, num_channels=1, bits_per_sample=16)

  8.         api = RequestApi(appid="你的应用ID", secret_key="你的KEY", upload_file_path=r"sound.wav")

  9.         api.all_api_request()
复制代码


对应的请求和处理代码如下所示

  1.    def __init__(self, appid, secret_key, upload_file_path):

  2.         self.appid = appid

  3.         self.secret_key = secret_key

  4.         self.upload_file_path = upload_file_path

  5. 

  6.     # 根据不同的apiname生成不同的参数,本示例中未使用全部参数您可在官网(https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html)查看后选择适合业务场景的进行更换

  7.     def gene_params(self, apiname, taskid=None, slice_id=None):

  8.         appid = self.appid

  9.         secret_key = self.secret_key

  10.         upload_file_path = self.upload_file_path

  11.         ts = str(int(time.time()))

  12.         m2 = hashlib.md5()

  13.         m2.update((appid + ts).encode('utf-8'))

  14.         md5 = m2.hexdigest()

  15.         md5 = bytes(md5, encoding='utf-8')

  16.         # 以secret_key为key, 上面的md5为msg， 使用hashlib.sha1加密结果为signa

  17.         signa = hmac.new(secret_key.encode('utf-8'), md5, hashlib.sha1).digest()

  18.         signa = base64.b64encode(signa)

  19.         signa = str(signa, 'utf-8')

  20.         file_len = os.path.getsize(upload_file_path)

  21.         file_name = os.path.basename(upload_file_path)

  22.         param_dict = {}

  23. 

  24.         if apiname == api_prepare:

  25.             # slice_num是指分片数量，如果您使用的音频都是较短音频也可以不分片，直接将slice_num指定为1即可

  26.             slice_num = int(file_len / file_piece_sice) + (0 if (file_len % file_piece_sice == 0) else 1)

  27.             param_dict['app_id'] = appid

  28.             param_dict['signa'] = signa

  29.             param_dict['ts'] = ts

  30.             param_dict['file_len'] = str(file_len)

  31.             param_dict['file_name'] = file_name

  32.             param_dict['slice_num'] = str(slice_num)

  33.         elif apiname == api_upload:

  34.             param_dict['app_id'] = appid

  35.             param_dict['signa'] = signa

  36.             param_dict['ts'] = ts

  37.             param_dict['task_id'] = taskid

  38.             param_dict['slice_id'] = slice_id

  39.         elif apiname == api_merge:

  40.             param_dict['app_id'] = appid

  41.             param_dict['signa'] = signa

  42.             param_dict['ts'] = ts

  43.             param_dict['task_id'] = taskid

  44.             param_dict['file_name'] = file_name

  45.         elif apiname == api_get_progress or apiname == api_get_result:

  46.             param_dict['app_id'] = appid

  47.             param_dict['signa'] = signa

  48.             param_dict['ts'] = ts

  49.             param_dict['task_id'] = taskid

  50.         return param_dict

  51. 

  52.     # 请求和结果解析，结果中各个字段的含义可参考：https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html

  53.     def gene_request(self, apiname, data, files=None, headers=None):

  54.         response = requests.post(lfasr_host + apiname, data=data, files=files, headers=headers)

  55.         result = json.loads(response.text)

  56.         print(f"打印的数据:{result}")

  57.         print(f"打印的数据2:{str(result)}")

  58.         print(result["data"] if "data" in result else None)

  59.         if apiname == api_get_result:

  60.             try:

  61.                 global text

  62.                 # 从结果中提取数据

  63.                 res = json.loads(result["data"])

  64.                 data = res[0]["onebest"]

  65.                 text = data;

  66. 

  67.             except Exception as e:

  68.                 print(f"串口操作失败: {e}")

  69. 

  70.         if result["ok"] == 0:

  71.             print("{} success:".format(apiname) + str(result))

  72.             return result

  73.         else:

  74.             print("{} error:".format(apiname) + str(result))

  75.             exit(0)

  76.             return result

  77. 

  78.     # 预处理

  79.     def prepare_request(self):

  80.         return self.gene_request(apiname=api_prepare,

  81.                                  data=self.gene_params(api_prepare))

  82. 

  83.     # 上传

  84.     def upload_request(self, taskid, upload_file_path):

  85.         file_object = open(upload_file_path, 'rb')

  86.         try:

  87.             index = 1

  88.             sig = SliceIdGenerator()

  89.             while True:

  90.                 content = file_object.read(file_piece_sice)

  91.                 if not content or len(content) == 0:

  92.                     break

  93.                 files = {

  94.                     "filename": self.gene_params(api_upload).get("slice_id"),

  95.                     "content": content

  96.                 }

  97.                 response = self.gene_request(api_upload,

  98.                                              data=self.gene_params(api_upload, taskid=taskid,

  99.                                                                    slice_id=sig.getNextSliceId()),

  100.                                              files=files)

  101.                 if response.get('ok') != 0:

  102.                     # 上传分片失败

  103.                     print('upload slice fail, response: ' + str(response))

  104.                     return False

  105.                 print('upload slice ' + str(index) + ' success')

  106.                 index += 1

  107.         finally:

  108.             'file index:' + str(file_object.tell())

  109.             file_object.close()

  110.         return True

  111. 

  112.     # 合并

  113.     def merge_request(self, taskid):

  114.         return self.gene_request(api_merge, data=self.gene_params(api_merge, taskid=taskid))

  115. 

  116.     # 获取进度

  117.     def get_progress_request(self, taskid):

  118.         return self.gene_request(api_get_progress, data=self.gene_params(api_get_progress, taskid=taskid))

  119. 

  120.     # 获取结果

  121.     def get_result_request(self, taskid):

  122.         return self.gene_request(api_get_result, data=self.gene_params(api_get_result, taskid=taskid))

  123. 

  124.     def all_api_request(self):

  125.         # 1. 预处理

  126.         pre_result = self.prepare_request()

  127.         taskid = pre_result["data"]

  128.         # 2 . 分片上传

  129.         self.upload_request(taskid=taskid, upload_file_path=self.upload_file_path)

  130.         # 3 . 文件合并

  131.         self.merge_request(taskid=taskid)

  132.         # 4 . 获取任务进度

  133.         while True:

  134.             # 每隔20秒获取一次任务进度

  135.             progress = self.get_progress_request(taskid)

  136.             progress_dic = progress

  137.             if progress_dic['err_no'] != 0 and progress_dic['err_no'] != 26605:

  138.                 print('task error: ' + progress_dic['failed'])

  139.                 return

  140.             else:

  141.                 data = progress_dic['data']

  142.                 task_status = json.loads(data)

  143.                 if task_status['status'] == 9:

  144.                     print('task ' + taskid + ' finished')

  145.                     break

  146.                 print('The task ' + taskid + ' is in processing, task status: ' + str(data))

  147. 

  148.             # 每次获取进度间隔20S

  149.             time.sleep(20)

  150.         # 5 . 获取结果

  151.         self.get_result_request(taskid=taskid)
复制代码
上述的代码来自于讯飞官方，只需要替换你自己的key 和 密钥 即可完成语音转换文字的功能。
13 - 使用语音转换后的文本来发送到通义千问，获取大模型的回复。首先我们需要根据通义千问的官方文档来进行注册和试用操作。

项目实践案例征集 + AI + 大模型语音聊天机器人图8


根据官方的教程配置KEY到环境变量中
项目实践案例征集 + AI + 大模型语音聊天机器人图9

安装和调用大模型
项目实践案例征集 + AI + 大模型语音聊天机器人图10

  1. def Turbo(msg):

  2.     global ai_res

  3.     try:

  4.         client = OpenAI(

  5.             # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",

  6.             api_key=os.getenv("DASHSCOPE_API_KEY"),

  7.             base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",

  8.         )

  9. 

  10.         completion = client.chat.completions.create(

  11.             model="qwen-plus",  # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models

  12.             messages=[

  13.                 {'role': 'system', 'content': 'You are a helpful assistant.'},

  14.                 {'role': 'user', 'content': msg}

  15.             ]

  16.         )

  17.         ai_res = completion.choices[0].message.content

  18.         print(completion.choices[0].message.content)

  19.     except Exception as e:

  20.         print(f"错误信息：{e}")

  21.         print("请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code")
复制代码
之后我们便可以将上述语音转换文本的内容发送给通义千问模型（速度很快）然后等待通义千问返回结果， 同时将返回的结果定义并且保存到全局的变量中，用于下文的讯飞文字转换语音功能。


回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

为本项目制作心愿单
购买心愿单
心愿单 编辑
[[wsData.name]]

硬件清单

  • [[d.name]]
btnicon
我也要做!
点击进入购买页面
撰写新帖
关于楼主

楼主的其它帖子

background 创造力兑换 background 心愿单 background 创客导航
上海智位机器人股份有限公司 沪ICP备09038501号-4 备案 沪公网安备31011502402448

© 2013-2025 Comsenz Inc. Powered by Discuz! X3.4 Licensed

mail