來養(yǎng)一只羊駝寶寶吧？！快來Duo S上跑你的第一個生成式AI

OpenAI的創(chuàng)始成員Andrej Karpathy近日在一個周末內(nèi)訓(xùn)練了一個微型LLaMA 2模型，并成功將其移植到C語言中。這個項目被他命名為Baby LLaMA 2，令人驚嘆的是，推理代碼僅有500行。

在RISC-V挑戰(zhàn)賽中，我們期望在一個輕量級的RISC-V開發(fā)板上把這個模型運行起來，所以就有了這個賽題：Baby LLaMA 2 on Duo 速度優(yōu)化

先看看效果：

賽題回顧

讓 Baby LLaMA 2 運行在 Milk-V Duo 這樣的小板子上是很有挑戰(zhàn)的事情。本次競賽旨在提升 Baby LLaMA 2 在 Milk-V Duo 平臺上的性能，目標(biāo)是實現(xiàn)更高的每秒 Token 處理速度。參賽者需要運用輕量級技術(shù)和編譯器優(yōu)化策略，結(jié)合麥克風(fēng)語音輸入或命令行輸入提示詞等多種方式，開發(fā)一個能夠講故事的機器人 Demo。該 Demo 應(yīng)通過揚聲器進行輸出，并可借鑒小米米兔講故事機器人的原型設(shè)計。

賽題地址：https://rvspoc.org/s2311/

實機教程與演示:讓DuoS成為孩子的“故事王”

通過外接SPI顯示屏、麥克風(fēng)、音頻輸出設(shè)備，Duo團隊實現(xiàn)了一個簡易的場景Demo。（源碼附在最后）

主要分為以下四個部分：

1、通過麥克風(fēng)采集語音

2、經(jīng)過語音轉(zhuǎn)文字ASR模型實現(xiàn)語音實時轉(zhuǎn)換

3、大模型實現(xiàn)“講故事”實時交互

4、通過文字轉(zhuǎn)語音TTS模型實現(xiàn)語音實時從揚聲器播放“故事”

硬件連接方法:

需要設(shè)置和使用到的硬件主要有：duo s、SPI顯示屏、麥克風(fēng)、音頻輸出、按鍵、Wifi、UART串口、type-c（type-c這里只做供電使用，連接板子均通過串口實現(xiàn)）

1、SPI顯示屏連接

將SPI顯示屏背面的引腳對應(yīng)的接口和duo s板卡的引腳對應(yīng)

duo s整體引腳圖如下：

SPI屏幕對應(yīng)的引腳

整體引腳對應(yīng)連接圖如下：

# 清屏cat /dev/zero > /dev/fb0# 花屏cat /dev/random > /dev/fb0

2、麥克風(fēng)連接

使用USB聲卡，注意麥克風(fēng)和音頻輸出對應(yīng)孔的正確連接

# 錄音命令(Ctrl+C結(jié)束錄音)：arecord -f dat -c 1 -r 16000 XXXX.wav

3、音頻輸出連接

使用USB聲卡，注意麥克風(fēng)和揚聲器輸入孔的正確對應(yīng)

# 播放錄音:aplay XXXX.wav

4、按鍵連接

按鍵連接引腳如下：（只需要將引腳和duo s對應(yīng)功能的引腳連接即可）

5、Wifi連接

這里通過一個一鍵運行腳本進行設(shè)置Wifi

vi wifi-duo-s.sh
# 執(zhí)行i進入編輯模式，將以下內(nèi)容寫入########################################!/bin/bash# 提示用戶輸入WiFi的SSID和密碼read -p "請輸入WiFi的SSID: " ssidread -p "請輸入WiFi的密碼: " password# 編輯 /etc/wpa_supplicant.conf 文件cat < /etc/wpa_supplicant.confctrl_interface=/var/run/wpa_supplicantap_scan=1update_config=1
network={ ssid="$ssid" psk="$password" key_mgmt=WPA-PSK}EOF# 重啟網(wǎng)絡(luò)wpa_supplicant -B -i wlan0 -c /etc/wpa_supplicant.confecho "WiFi configuration completed."######################################## ESC退出編輯模式，:wq保存退出
# 執(zhí)行shsh wifi-duo-s.sh

6、UART串口連接

將USB-TTL的引腳對應(yīng)duo s的引腳，對應(yīng)關(guān)系如下：

duo s的對應(yīng)引腳如下：

USB-TTL引腳如下：

注：關(guān)于麥克風(fēng)和音頻輸出的其他相關(guān)設(shè)置命令（可選）

# 查看錄音設(shè)備arecord -l# 查看播放設(shè)備aplay -l# 查看具體設(shè)備號的信息（假設(shè)設(shè)備號為3）amixer contents -c 3
# 麥克風(fēng)音量設(shè)置（name根據(jù)具體而定，常規(guī)是這個）amixer -Dhw:0 cset name='ADC Capture Volume' 24# 揚聲器播放音量設(shè)置（假設(shè)音量設(shè)置為24）# 兩種方式:（假設(shè)設(shè)備號為3）amixer -Dhw:3 cset name='Speaker Playback Volume' 24amixer cset -c 3 numid=6 24

軟件使用方法

1、連接WiFi

sh wifi.sh# 輸入連接wifi的ssid和pwd

2、安裝運行依賴包

pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple some-package

3、運行故事機baby llama

python asr_chat-llama-baby.py# 在輸出Wait for key press后，通過按鍵輸入語音（按住即開始語音輸入，松開即結(jié)束語音輸入）

注：baby llama源碼如下

# -*- coding: UTF-8 -*-import http.clientimport urllib.parseimport jsonimport subprocessimport timeimport httpximport requests
subprocess.Popen(['chmod', '+x', 'stable_demo'])subprocess.Popen(['./stable_demo'])print('Asr chat tts begin....')appKey = 'P918jP30TLJNHi3Q'#'P918jP30TLJNHi3Q'#s9NZm8ozBKyX63vK' #'RxkHgzYYYYLIP4OD'token = '31b129713beb46b8b0db321a005ecb0d'
# Chat ConfigurationAPI_KEY = "ebb785194c713e7b419ca8742277d414.hCBC11QCZvC5N0YK"BASE_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions"history = [{"role": "system", "content": "您好！"}]# Aliyun# url = 'https://nls-gateway-cn-shanghai.aliyuncs.com/stream/v1/asr'host = 'nls-gateway-cn-shanghai.aliyuncs.com'
def record_on_gpio(pin): is_pressed = False audioFilepath = './output.wav' while True: try: with open('/sys/class/gpio/gpio{}/value'.format(pin), 'r') as gpio_file: value = gpio_file.read().strip() #print('get key value {}',value) if value == '1' and not is_pressed: # 按鍵按下時開始錄音 recording_process = subprocess.Popen(['arecord', '-f', 'dat', '-c', '1', '-r', '16000', 'output.wav']) is_pressed = True print("Recording started.")
if value == '0' and is_pressed: subprocess.Popen(['killall', 'arecord']) recording_process.wait() # 等待錄音進程結(jié)束 is_pressed = False print("Recording stopped.") return audioFilepath except Exception as e: print("Error:", e)
def process_chunk(chunk,response_accumulator): if chunk.strip() == "[DONE]": return True, None try: data = json.loads(chunk) # print('process_chunk data:', data) if 'choices' in data and data['choices']: for choice in data['choices']: if 'delta' in choice and 'content' in choice['delta'] and choice['delta']['content']: result = choice['delta']['content'] # print('process_chunk result:', result) response_accumulator.append(result) return False, result except Exception as e: print(f"處理數(shù)據(jù)塊時出錯: {e}") return False, None


def chat(query, history): history += [{"role": "user", "content": query}] data = { "model": "glm-4", "messages": history, "temperature": 0.3, "stream": True, }
headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" }
response_accumulator = [] response = requests.post(BASE_URL, data=json.dumps(data), headers=headers, stream=True) send_to_lvgl(f"[CLEAR]{query}: ") try: for chunk in response.iter_lines(): if chunk: chunk_str = chunk.decode("utf-8") if chunk_str.startswith("data: "): chunk_str = chunk_str[len("data: "):]
done, result = process_chunk(chunk_str,response_accumulator) # print('result is', result)
chunk_str = "data: " + chunk_str # print("Get response:", chunk_str) if result: send_to_lvgl(result)
if done: tts_text = ''.join(response_accumulator) tts_to_play(tts_text)

except Exception as e: print(f"Error: {str(e)}")

def send_to_lvgl(text): pipe_name = '/tmp/query_pipe' try: with open(pipe_name, 'w') as pipe: pipe.write(text) pipe.flush() except Exception as e: print(f"LVGL send error: {e}")

def process(request, token, audioFile) : # 讀取音頻 print('process {} {}'.format(request, audioFile)) with open(audioFile, mode = 'rb') as f: audioContent = f.read()
host = 'nls-gateway-cn-shanghai.aliyuncs.com'
# 設(shè)置HTTPS請求頭部 httpHeaders = { 'X-NLS-Token': token, 'Content-type': 'application/octet-stream', 'Content-Length': len(audioContent) }
conn = http.client.HTTPSConnection(host)
conn.request(method='POST', url=request, body=audioContent, headers=httpHeaders)
response = conn.getresponse() print('Response status and response reason:') print(response.status ,response.reason)
try: body = json.loads(response.read()) text = body['result'] print('Recognized Text:', text) story = makeLLAMAStory(text) print('[makeLLAMAStory] return {}'.format(story)) #send_to_lvgl(story) tts_to_play(story) #chat_response = chat(text, history) #print('Chat Response:', chat_response) except ValueError: print('The response is not json format string')
conn.close()
def makeLLAMAStory(text): print('[makeLLAMAStory] {}'.format(text)) recording_process = subprocess.Popen(['./runq-fast-gcc', 'stories15M_q80.bin', '-t', '0.8', '-n', '256', '-i', text], stdout=subprocess.PIPE, stderr=subprocess.PIPE) return_value, stderr = recording_process.communicate() return return_value.decode('utf-8')
def oneloop(): print('Wait for key press') audioFilepath = record_on_gpio(499)
#print('Wait for first audio') format = 'pcm' sampleRate = 16000 enablePunctuationPrediction = True enableInverseTextNormalization = True enableVoiceDetection = False
# 設(shè)置RESTful請求參數(shù) asrurl = f'https://{host}/stream/v1/asr' request = asrurl + '?appkey=' + appKey request = request + '&format=' + format request = request + '&sample_rate=' + str(sampleRate)
if enablePunctuationPrediction : request = request + '&enable_punctuation_prediction=' + 'true'
if enableInverseTextNormalization : request = request + '&enable_inverse_text_normalization=' + 'true'
if enableVoiceDetection : request = request + '&enable_voice_detection=' + 'true'
print('Request: ' + request)
process(request, token, audioFilepath)
def tts_to_play(text, file_path='response.wav'): ttsurl = f'https://{host}/stream/v1/tts' text_encoded = urllib.parse.quote_plus(text) tts_request = f"{ttsurl}?appkey={appKey}&token={token}&text={text_encoded}&format=wav&sample_rate=16000"
conn = http.client.HTTPSConnection(host) conn.request('GET', tts_request) response = conn.getresponse() body = response.read() if response.status == 200 and response.getheader('Content-Type') == 'audio/mpeg': with open(file_path, 'wb') as f: f.write(body) print('TTS audio saved successfully') subprocess.Popen(['aplay', file_path]) else: print('TTS request failed:', body) conn.close()

while True: try: oneloop() except Exception as e: print(e)