如何將智能音箱接入ChatGPT _ChatGPT

本文介紹一下某米音箱如何接入目前大火的ChatGPT ,本文的主要內容是整理Github上yihong0618作者的代碼而來，經過歸納整理，并對代碼做了少量精簡和修改。這里對yihong0618表示感謝，希望大家都能成功的改造自己的音箱。
某米音箱是一個開發的平臺，支持個人和企業開發者定制音頻產品和定制化操作，在官網可以查看具體的教程。
下面開始詳細介紹如何接入chatgpt
系統架構

文章插圖
你需要一個主機來編寫程序，連接米家服務器，主機可以是你的個人電腦，樹莓派等Arm電腦，或者云端主機，我的測試中使用的是樹莓派4B，編程語言是Python，推薦使用云服務器這樣可以讓你的程序保持持續在線。
主機負責從米家服務上查詢你的問答，將問答傳輸給ChatGPT后，將GPT的內容通過智能音箱來進行播放。
查看你的賬戶ID
登陸米家官網，登陸你的賬號后在菜單欄會顯示你的賬號ID ，這個ID也可以在APP的個人信息里找到，記錄下你的賬戶名和密碼。

文章插圖
獲取你的賬戶ID
獲取你的機器型號
在你的智能音箱的底部的標簽上找到你的機器型號，我的機器是Pro 2代，在小米的底部標簽上可以找到如下內容
產品名稱：智能音箱型號: L15A記錄下你的型號我這里是L15A
安裝Python庫miservice
建議不要直接使用pip3 install miservice這種形式安裝，這種安裝方式的版本低一些，少幾個查詢智能音箱當前狀態的函數。
在Github上搜索MiService庫，作者是yihong0608，將代碼clone下來，使用如下方式安裝
git clone https://github.com/yihong0618/MiService.gitcdMiServicesudo pip3 install .庫中介紹了一些查詢你的機器信息和使用智能音箱文字轉音頻的命令，需要的同學，也可以體驗一下，取決你的智能音箱型號，很多命令可能是無效的。
同時重要的是你需要獲取你的智能音箱的tts文本轉語音命令，這里有一個列表，如果你的型號恰好在里面，那么你直接記錄一下命令即可，不需要使用后續的命令獲得

"LX06": "5-1","L05B": "5-3","S12A": "5-1","LX01": "5-1","L06A": "5-1","LX04": "5-1","L05C": "5-3","L17A": "7-3","X08E": "7-3","LX05A": "5-1","LX5A": "5-1","L15A": "7-3",

我的型號是L15A，所以我的tts命令是7-3
如果你的型號不在上面的列表中，使用下面的命令獲得你的tts命令，上面介紹過我的型號是L15A，所以我在console中輸入如下內容。

cdMiServiceexport MI_USER= 你的小米IDexport MI_PASS=你的密碼python3 ./micli.py list通過這個命令可以顯示自己賬號下的設備列表 ， 包含名稱、類型、DID、Token 等信息 ， 記錄下的你的智能音箱的IDexport MI_DID=你的智能音箱的IDpython3 ./micli.py spec xiaomi.wifispeaker.l15a

注意將xiaomi.wifispeaker.l15a中的l15a替換為你的型號，在輸出中獲取我的音箱命令，如下

文章插圖
我的音箱命令
從音箱命令中找到Intelligent_Speacker后的數字我的是7，找到_Play_Text后的數字我的是3
這樣組成了下面代碼config.py中的"L15A": "7-3" 。所以注意這里一定要找到你的命令代碼。將你的型號和tts命令添加打config.py中，如我上面列表中那樣。
完整代碼

#!/usr/bin/env python3import asyncioimport jsonimport reimport subprocessimport timefrom pathlib import Pathfrom aiohttp import ClientSessionfrom miservice import MiAccount, MiNAServicefrom wxtgptbot importWXTChatBot from config import (COOKIE_TEMPLATE,HARDWARE_COMMAND_DICT,KEY_WORD,LATEST_ASK_API,MI_ASK_SIMULATE_DATA,PROMPT,)from utils import calculate_tts_elapse, parse_cookie_stringclass MiGPT:def __init__(self,hardware,mi_user,mi_pass,openai_key,cookie="",use_command=False,mute_xiaoai=False,use_gpt3=False,use_chatgpt_api=False,api_base=None,verbose=False,)：self.mi_token_home = Path.home() / ".mi.token"self.hardware = hardwareself.mi_user = mi_userself.mi_pass = mi_passself.openai_key = openai_keyself.cookie_string = ""self.last_timestamp = int(time.time()*1000)# timestamp last call mi speakerself.session = Noneself.chatbot = None# a little slow to init we move it after xiaomi initself.user_id = ""self.device_id = ""self.service_token = ""self.cookie = cookieself.use_command = use_commandself.tts_command = HARDWARE_COMMAND_DICT.get(hardware, "7-3")self.conversation_id = Noneself.parent_id = Noneself.miboy_account = Noneself.mina_service = None# try to mute xiaoai configself.mute_xiaoai = mute_xiaoai# mute xiaomi in runtimeself.this_mute_xiaoai = mute_xiaoai# if use gpt3 apiself.use_gpt3 = use_gpt3self.use_chatgpt_api = use_chatgpt_apiself.api_base = api_baseself.verbose = verbose# this attr can be re set value in cliself.key_word = KEY_WORDself.prompt = PROMPTasync def init_all_data(self, session):await self.login_miboy(session)await self._init_data_hardware()with open(self.mi_token_home) as f:user_data = https://www.jianzixun.com/json.loads(f.read())self.user_id = user_data.get("userId")self.service_token = user_data.get("micoapi")[1]self._init_cookie()async def login_miboy(self, session):self.session = sessionself.account = MiAccount(session,self.mi_user,self.mi_pass,str(self.mi_token_home),)# Forced login to refresh to refresh tokenawait self.account.login("micoapi")self.mina_service = MiNAService(self.account)async def _init_data_hardware(self):if self.cookie:# if use cookie do not need initreturnhardware_data = await self.mina_service.device_list()for h in hardware_data:if h.get("hardware", "") == self.hardware:self.device_id = h.get("deviceID")print("設備id：",self.device_id)breakelse:raise Exception(f"we have no hardware: {self.hardware} please check")'''初始化cookie，調用小米api時需要'''def _init_cookie(self):if self.cookie:self.cookie = parse_cookie_string(self.cookie)else:self.cookie_string = COOKIE_TEMPLATE.format(device_id=self.device_id,service_token=self.service_token,user_id=self.user_id,)self.cookie = parse_cookie_string(self.cookie_string)#獲取小米音箱的最后一次的回答async def get_latest_ask_from_xiaoai(self):r = await self.session.get(LATEST_ASK_API.format(hardware=self.hardware, timestamp=str(int(time.time() * 1000))),cookies=parse_cookie_string(self.cookie),)return await r.json()def get_last_timestamp_and_record(self, data):if "data" in data:d= data.get("data")records = json.loads(d).get("records")if not records:return 0, Nonelast_record = records[0]timestamp = last_record.get("time")return timestamp, last_recordelse:return 0, Noneasync def do_tts(self, value, wait_for_finish=False):if not self.use_command:try:await self.mina_service.text_to_speech(self.device_id, value)except:# do nothing is okpasselse:#使用micli工具subprocess.check_output(["micli", self.tts_command, value])if wait_for_finish:elapse = calculate_tts_elapse(value)await asyncio.sleep(elapse)print("elapse:",elapse)while True:if not await self.get_if_xiaoai_is_playing():breakawait asyncio.sleep(2)await asyncio.sleep(2)print("回答完畢")#小米是否正在播報async def get_if_xiaoai_is_playing(self):#此函數沒有被找到playing_info = await self.mina_service.player_get_status(self.device_id)# WTF xiaomi apiis_playing = (json.loads(playing_info.get("data", {}).get("info", "{}")).get("status", -1)== 1)return is_playingasync def run_forever(self):print(f"Running xiaogpt now, 用`{'/'.join(KEY_WORD)}`開頭來提問")async with ClientSession() as session:await self.init_all_data(session)print("開始循環")while 1:if self.verbose:print(f"Now listening xiaoai new message timestamp: {self.last_timestamp}")try:r = await self.get_latest_ask_from_xiaoai()except Exception:# we try to init all againawait self.init_all_dat(session)r = await self.get_latest_ask_from_xiaoai()# spider ruleif not self.mute_xiaoai:await asyncio.sleep(1)else:await asyncio.sleep(0.3)new_timestamp, last_record = self.get_last_timestamp_and_record(r)print(new_timestamp, last_record)if new_timestamp > self.last_timestamp:self.last_timestamp = new_timestampquery = last_record.get("query", "")if query.startswith(tuple(self.key_word)):# only mute when your clause start's with the keywordself.this_mute_xiaoai = False# drop 幫我回答query = re.sub(rf"^({'|'.join(self.key_word)})", "", query)print("-" * 20)print("問題：" + query + "？")query = f"{query}，{PROMPT}"# waiting for xiaoai speaker doneif not self.mute_xiaoai:await asyncio.sleep(2)for i in range(8):if not await self.get_if_xiaoai_is_playing():print("小米結束回答")breakelse:print("小米正在回答")await asyncio.sleep(2)await self.do_tts("正在問GPT請耐心等待")await asyncio.sleep(0.5)try:print("以下是小愛的回答: ",last_record.get("answers")[0].get("tts", {}).get("text"),)except:print("小愛沒回")# message = await self.ask_gpt(query)message="以下是GPT的回答 "if "清除消息" in query:message="GPT 清除歷史消息"WXTChatBot.clear()else:message+=WXTChatBot.ask({"msg":query})# tts to xiaoai with ChatGPT answerprint("以下是GPT的回答: " + message)await self.do_tts(message, wait_for_finish=True)if self.mute_xiaoai:self.this_mute_xiaoai = Trueelse:if self.verbose:print("No new xiao ai record")if __name__=="__main__":app=MiGPT("型號","你的ID","你的密碼","")asyncio.run(app.run_forever())

這個代碼中需要非常注意的代碼時
message+=WXTChatBot.ask({“msg”:query})
WXTChatBot 這個模塊是我封裝的訪問chatgpt的代碼，請按照下面的介紹封裝一個你的模塊。
wxtchatbot模塊

importrequestsimportjsonimport timeclass WXTChatBot():baseurl="http://我的服務器ID:13456/"@classmethoddef ask(cls,msg):'''{"msg":"請做一首歐陽修風格的16字絕句"}'''t1=time.time()msg=json.dumps(msg,ensure_ascii=False)req = requests.post(WXTChatBot.baseurl+"ask",headers={"content-type":"application/json;charset=UTF-8"}, data=https://www.jianzixun.com/msg.encode('utf-8'),timeout=20)print("用時：",(time.time()-t1),'s')return req.content.decode("utf8")@classmethoddef clear(cls):'''清空歷史聊天'''req = requests.get(WXTChatBot.baseurl+"clearchat",timeout=3)

這里說明一下我的方式，由于我無法直接調用chatgpt api，所以我在一臺云端服務器（可以直接訪問chatgpt api），搭建了一個http服務，端口為13456，在服務器的ask接口中我使用openai提供的chatgpt例子調用了chatgpt，然后將內容返回給我。如果你能直接訪問openai服務器，可以在你的主機上直接訪問ChatGPT api
輔助模塊utils

import refrom http.cookies import SimpleCookiefrom requests.utils import cookiejar_from_dictdef parse_cookie_string(cookie_string):cookie = SimpleCookie()cookie.load(cookie_string)cookies_dict = {}cookiejar = Nonefor k, m in cookie.items():cookies_dict[k] = m.valuecookiejar = cookiejar_from_dict(cookies_dict, cookiejar=None, overwrite=True)return cookiejar_no_elapse_chars = re.compile(r"([「」『』《》“”'"()（）]|(?<!-)-(?!-))", re.UNICODE)def calculate_tts_elapse(text):# for simplicity, we use a fixed speedspeed = 4.25# this value is picked by trial and error# Exclude quotes and brackets that do not affect the total elapsed timereturn len(_no_elapse_chars.sub("", text)) / speed

這個模塊主要是計算文字播放的時間和解析小米的token
模塊config

LATEST_ASK_API = "https://userprofile.mina.mi.com/device_profile/v2/conversation?source=dialogu&hardware={hardware}×tamp={timestamp}&limit=2"COOKIE_TEMPLATE = "deviceId={device_id}; serviceToken={service_token}; userId={user_id}"HARDWARE_COMMAND_DICT = {"LX06": "5-1","L05B": "5-3","S12A": "5-1","LX01": "5-1","L06A": "5-1","LX04": "5-1","L05C": "5-3","L17A": "7-3","X08E": "7-3","LX05A": "5-1","LX5A": "5-1","L15A": "7-3",# add more here}MI_USER = ""MI_PASS = ""OPENAI_API_KEY = ""KEY_WORD = ["幫我", "請回答"]

這里需要注意的是，在這個模塊中我添加了一個”L15A”: “7-3″ ，其中L15A 是我的音箱型號，”7-3” 是miservice庫需要使用的文字轉音頻tts的命令，在我上面的介紹中介紹了你如何獲取到這個命令碼。
ChatGPT API
使用如下的代碼作為你訪問chapgpt的函數，函數中你需要使用你自己的openai_key，openai_key是在openai官網獲得的，每個用戶有18美元的免費額度。

import openaiclass ChatGPTBot():def __init__(self, session, openai_key, api_base=None):self.session = sessionself.history = []self.api_base = api_baseself.openai_key = openai_keyasync def ask(self, query):openai.api_key = self.openai_keyif self.api_base:openai.api_base = self.api_basems = []for h in self.history:ms.append({"role": "user", "content": h[0]})ms.append({"role": "assistant", "content": h[1]})ms.append({"role": "user", "content": f"{query}"})completion = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=ms)message = (completion["choices"][0].get("message").get("content").encode("utf8").decode())self.history.append([f"{query}", message])return message

代碼的主要邏輯是：
1 init_all_data 登陸智能音箱API ，根據你的ID，密碼獲取返回的token
2 根據你的硬件型號找到你的機器
_init_data_hardware這個函數用來根據我的型號L15A找到了我的智能音箱3 使用do_tts函數使用智能音箱的文本轉語音
4 使用get_if_xiaoai_is_playing這個函數判斷小米音箱自己的回答是否結束了
5 調用chatgpt api 獲取回答
6 通過米家tts接口讓智能音箱回答。
7 在config.py中定義了gpt問話的頭，這里定義為”請回答“，所以當你喚醒音箱后，使用”請回答“+你的問題，會觸發ChatGPT的回答，其他的像播放音樂啊不會觸發GPT回答
入口代碼
if __name__=="__main__":app=MiGPT("型號","你的ID","你的密碼","")asyncio.run(app.run_forever())入口代碼傳入你的型號，你的ID，你的密碼即可
如果無法訪問到ChatGPT的服務器
這時你需要要么有一臺自己的外網服務器，在這個服務器上搭建一個你的服務，來訪問chatgpt的api，就像我做的這樣，我定義的WXTChatBot模塊，實際上是用post http請求了我的云端服務器的rest api，云端服務器將請求轉發到chatgpt獲取回答。
【如何將智能音箱接入ChatGPT】如果你沒有自己的外網服務器的話，目前國內有很多用戶開放了非常多的免費體驗的套殼API，你可以使用他們的api封裝一下免費使用。我體驗過使用了多個免費api效果很不錯，可以在頭條里搜索幾個。

继夫的玩弄H辣文的小说|女人与拘性猛交视频|精品欧美高清不卡高清|一起做亏亏的事情的视频|啦啦啦在线视频观看|望月直播下载ios版本|国产日韩欧美一区二区三区

如何將智能音箱接入ChatGPT