本文介紹一下某米音箱如何接入目前大火的ChatGPT ,本文的主要內容是整理Github上yihong0618作者的教程而來，經過歸納整理，并做了小幅的改動。這里對作者表示感謝，希望大家都能成功的改造自己的音箱。
某米音箱是一個開發的平臺，支持個人和企業開發者定制音頻產品和定制化操作，在官網可以查看具體的教程。
下面開始詳細介紹如何接入chatgpt
查看你的賬戶ID
登陸某米官網，登陸你的賬號后在菜單欄會顯示你的賬號ID，這個ID也可以在App的個人信息里找到，記錄下你的賬戶名和密碼。
獲取你的賬戶ID
獲取你的機器型號
在你的智能音箱的底部的標簽上找到你的機器型號，我的機器是Pro 2代，在小米的底部標簽上可以找到如下內容
產品名稱：智能音箱
型號: L15A
記錄下你的型號我這里是L15A
安裝Python/ target=_blank class=infotextkey>Python庫miservice
建議不要直接使用pip3 install miservice這種形式安裝，這種安裝方式的版本低一些，少幾個查詢智能音箱當前狀態的函數。
在Github上搜索MiService庫，作者是yihong0608，將代碼clone下來，使用如下方式安裝
sudo pip3 install .
庫中介紹了一些查詢你的機器信息和使用智能音箱文字轉音頻的命令，需要的同學，也可以體驗一下，取決你的智能音箱型號，可能很多命令是無效的，建議根據這個庫的介紹找到你的音箱對應的文字轉音頻的命令，上面介紹過我的型號是L15A，所以我根據教程輸入
python3 micli.py spec xiaomi.wifispeaker.l15a
獲取我的音箱命令
我的音箱命令
從音箱命令中找到Intelligent_Speacker后的數字我的是7，找到_Play_Text后的數字我的是3
這樣組成了下面代碼config.py中的"L15A": "7-3"。所以注意這里一定要找到你的命令代碼。
完整代碼
#!/usr/bin/env python3
import asyncio
import json
import re
import subprocess
import time
from pathlib import Path
from aiohttp import ClientSession
from miservice import MiAccount, MiNAService
from wxtgptbot import WXTChatBot
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_string
class 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,
):
# TODO !!!! refactor so many of this shit
self.mi_token_home = Path.home() / ".mi.token"
self.hardware = hardware
self.mi_user = mi_user
self.mi_pass = mi_pass
self.openai_key = openai_key
self.cookie_string = ""
self.last_timestamp = int(time.time()*1000) # timestamp last call mi speaker
self.session = None
self.chatbot = None # a little slow to init we move it after xiaomi init
self.user_id = ""
self.device_id = ""
self.service_token = ""
self.cookie = cookie
self.use_command = use_command
self.tts_command = HARDWARE_COMMAND_DICT.get(hardware, "7-3")
self.conversation_id = None
self.parent_id = None
self.miboy_account = None
self.mina_service = None
# try to mute xiaoai config
self.mute_xiaoai = mute_xiaoai
# mute xiaomi in runtime
self.this_mute_xiaoai = mute_xiaoai
# if use gpt3 api
self.use_gpt3 = use_gpt3
self.use_chatgpt_api = use_chatgpt_api
self.api_base = api_base
self.verbose = verbose
# this attr can be re set value in cli
self.key_word = KEY_WORD
self.prompt = PROMPT
async 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 = json.loads(f.read())
self.user_id = user_data.get("userId")
self.service_token = user_data.get("micoapi")[1]
self._init_cookie()
#await self._init_first_data_and_chatbot()
async def login_miboy(self, session):
self.session = session
self.account = MiAccount(
session,
self.mi_user,
self.mi_pass,
str(self.mi_token_home),
)
# Forced login to refresh to refresh token
await 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 init
return
hardware_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)
break
else:
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, None
last_record = records[0]
timestamp = last_record.get("time")
return timestamp, last_record
else:
return 0, None
async 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 ok
pass
else:
#使用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():
break
await 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 api
is_playing = (
json.loads(playing_info.get("data", {}).get("info", "{}")).get("status", -1)
== 1
)
return is_playing
async 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 again
await self.init_all_dat(session)
r = await self.get_latest_ask_from_xiaoai()
# spider rule
if 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_timestamp
query = last_record.get("query", "")
if query.startswith(tuple(self.key_word)):
# only mute when your clause start's with the keyword
self.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 done
if not self.mute_xiaoai:
await asyncio.sleep(2)
for i in range(8):
if not await self.get_if_xiaoai_is_playing():
print("小米結束回答")
break
else:
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 answer
print("以下是GPT的回答: " + message)
await self.do_tts(message, wait_for_finish=True)
if self.mute_xiaoai:
self.this_mute_xiaoai = True
else:
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的代碼，請按照下面的介紹封裝一個你的模塊。
輔助模塊utils
import re
from http.cookies import SimpleCookie
from requests.utils import cookiejar_from_dict
def parse_cookie_string(cookie_string):
cookie = SimpleCookie()
cookie.load(cookie_string)
cookies_dict = {}
cookiejar = None
for k, m in cookie.items():
cookies_dict[k] = m.value
cookiejar = 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 speed
speed = 4.25 # this value is picked by trial and error
# Exclude quotes and brackets that do not affect the total elapsed time
return len(_no_elapse_chars.sub("", text)) / speed
模塊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的命令，在我上面的介紹中介紹了你如何獲取到這個命令碼。
代碼的主要邏輯是：
1 init_all_data 登陸智能音箱API，根據你的ID，密碼獲取返回的token
2 根據你的硬件型號找到你的機器
_init_data_hardware 這個函數用來根據我的型號L15A找到了我的智能音箱
3 使用do_tts函數使用智能音箱的文本轉語音
4 使用get_if_xiaoai_is_playing這個函數判斷小米音箱自己的回答是否結束了
5 在config.py中定義了gpt問話的頭，這里定義為”請回答“，所以當你喚醒音箱后，使用”請回答“+你的問題，會觸發ChatGPT的回答，其他的像播放音樂啊不會觸發GPT回答
入口代碼
if __name__=="__main__":
app=MiGPT("型號","你的ID","你的密碼","")
asyncio.run(app.run_forever())
入口代碼傳入你的型號，你的ID，你的密碼即可
ChatGPT API
使用如下的代碼作為你訪問chapgpt的函數，函數中你需要使用你自己的openai_key，openai_key是在openai官網獲得的，每個用戶有18美元的免費額度。
import openai
from xiaogpt.bot.base_bot import BaseBot
class ChatGPTBot(BaseBot):
def __init__(self, session, openai_key, api_base=None):
self.session = session
self.history = []
self.api_base = api_base
self.openai_key = openai_key
async def ask(self, query):
openai.api_key = self.openai_key
if self.api_base:
openai.api_base = self.api_base
ms = []
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
如果無法訪問到ChatGPT的服務器
這時你需要要么有一臺自己的外網服務器，在這個服務器上搭建一個你的服務，來訪問chatgpt的api，就像我做的這樣，我定義的WXTChatBot模塊，實際上是用post http請求了我的云端服務器的rest api，云端服務器將請求轉發到chatgpt獲取回答。
如果你沒有自己的外網服務器的話，目前國內有很多用戶開放了非常多的免費體驗的套殼API，你可以使用他們的api封裝一下免費使用。我體驗過使用了多個免費api效果很不錯，可以在頭條里搜索幾個。
總結：
借助于智能音箱的開發性，我們一起diy試試吧
