首先我们安装一下固件，配置好开发环境。

去官网下载最新的Debian镜像，最新的镜像链接可以在官方wiki上找到，我选择用github方式下载server版的镜像。因为板载内存仅有1个H，如果带有图形环境的话再进行项目开发会比较吃力。

选择图里的V2.4_07-12_server_WalnutPi-1b_6.1.31_debian12.img.rar就可以。

接下来，我们需要一个烧录工具，我推荐使用便携版的balenaEtcher。虽然这个工具文件较大，但操作非常简单，一键即可完成烧录。而且便携版不需要安装，使用起来也很方便。

随后，找一张容量在16G以上的内存卡，插入电脑，打开balenaEtcher，选择相应的镜像和内存卡，点击烧录，等待过程完成即可。

完成烧录后，把内存卡插入开发板，开发板上电，这样系统就算完成了烧录工作。

由于我没有HDMI显示器和键盘，因此无法按照官方WIKI那样使用桌面系统。因此，在首次启动时，我需要采用无头方式来配置无线网络。好在核桃派的接口非常齐全，不但有串口接口可以实现无头登录，拓展板上更集合了网线接口，可以直接使用有线网来ssh登录，非常方便。这里我们就使用有线网的方式来进行配置。

把板子插上网线后，可以通过登录路由器管理界页面来找到核桃派的IP地址。但是考虑到每个人的路由器不一样，或许有些路由器并不支持此功能，因此我们可以使用一个简单的小软件Advanced IP Scanner来扫描整个网段的设备，用来找到核桃派的ip

找到ip后，在电脑上打开cmd，在里面输入以下ssh命令，就可以连接到核桃派。我们使用pi用户登录，密码同样为pi。

接下来连接无线wifi。网上大多教程使用的都是nmcli这个工具，但其实有个更加方便地工具可以使用，nmtui，它还可以自动连接已连接的网络。

sudo nmtui

选择第二个选项激活网络连接，找到自己家的WiFi按回车输入密码即可完成连接。

这时候我们就可以用sudo ifconfig查看到无线网卡的ip，就可以拔掉网线，使用wifi登录ssh了。

开发环境搭建

整个项目是在python下进行开发的。为了不污染系统环境，我们会使用虚拟环境来进行开发。有linux python开发经验的小伙伴一定会知道，linux平台中很多python包并不是通过pip的方式进行安装的，而是通过apt包管理器安装在系统环境中的，这是为了有权限对硬件或是内核进行操作。因此，创建虚拟环境时需要导入系统环境包。

由于一部分包需要用到apt安装，另一部分需要用到sudo安装，而有一些包之间还会存在冲突和兼容问题，因此我把所有需要安装的包和创建虚拟环境的命令都写在了一个bash中，只需要一键运行这个bash，就可以自动完成环境配置工作。

#!/bin/bash



sudo apt update

sudo apt full-upgrade -y

sudo apt autoremove -y



cd ~

mkdir jarvis_walnutpi

cd jarvis_walnutpi



sudo apt install gcc python3-dev -y

sudo apt install python3.11-venv -y

python -m venv venv --system-site-packages

source venv/bin/activate



pip install -U "fastapi[standard]"



sudo apt install python3-pyaudio -y



pip install -U SpeechRecognition

sudo apt install flac -y



pip install -U opencv-python

pip install -U torchvision==0.15.2

pip install -U ultralytics



pip install -U "g4f[all]"



pip install -U edge-tts

pip install -U pydub

sudo apt install ffmpeg libavcodec-extra -y

这些包安装是需要一些时间1的，通过ssh安装会存在一个风险，一旦出现ssh掉线，安装也会被终止。为了避免这种情况，我们可以使用screen工具来新建一个窗口，在这个窗口内执行安装程序，这样即使我们关闭窗口，安装也不会中止。

sudo apt install screen

screen -R setup

接下来运行上面的sh文件，我用了半个小时左右，完成后我们就完成了全部环境配置工作。

音频输入

由于核桃派上的3.5mm接口仅可以做音频输出使用，不能输入，因此这个项目我计划使用电脑来进行音频输入，输入完成后将录音的bytes数据通过FastAPI发送到核桃派，接着核桃派进行语音识别，图像视频，大语言模型调用，TTS合成音频，播放。大预言返回的文本结果可以作为返回值返回给PC。相当于将PC仅作为一个网络麦克风使用。

那首先我们先把PC这块语音输入的问题给解决了。我们在pc上使用的库同样也是pyaudio，用pydub来进行测试。先安装所需的库：

pip install pyaudio

pip install pydub

pip install numpy

然后我写了下面代码用来获取声音。

import time

import pyaudio

import pydub

import numpy as np

import io





def listen(_min_db=200, _end_delay=1.0, _timeout=60):

    _CHANNELS = 1

    _RATE = 16000

    _CHUNK = 1024

    _FORMAT = pyaudio.paInt16



    _p = pyaudio.PyAudio()

    _stream = _p.open(

        format=_FORMAT,

        channels=_CHANNELS,

        rate=_RATE,

        input=True,

        frames_per_buffer=_CHUNK,

    )



    print("开始计时")



    _buffer = io.BytesIO()



    _flag_record = True  # 判断是否继续录音

    _flag_start = False  # 开始录音节点

    _flag_end = False  # 判断声音小了



    _time_wait = time.time()

    _time_start = 0

    _time_end = 0



    while _flag_record:

        _data = _stream.read(_CHUNK, exception_on_overflow=False)

        _buffer.write(_data)

        _db = np.max(np.frombuffer(_data, dtype=np.short))

        print(_db)



        if _flag_start == False and _db > _min_db:

            _flag_start = True

            print("开始录音")

            _time_start = time.time()



        if _flag_start:

            if _flag_end == False and _db < _min_db:

                _flag_end = True

                _time_end = time.time()

                print(f"间隔{_end_delay}s后开始检测是否还是小声")

                continue



            if _flag_end == True and _db > _min_db:

                _flag_end = False

                print("大声，继续录音")

                continue



            if _flag_end == True and time.time() - _time_end > _end_delay:

                _flag_record = False

                print("结束录音")

                continue



        if time.time() - _time_wait > _timeout:  # 超时直接退出

            _flag_record = False

            print("超时")

            continue



    _stream.stop_stream()

    _stream.close()

    _p.terminate()



    _buffer.seek(0)

    _audio = pydub.AudioSegment.from_file(

        _buffer,

        format="raw",

        frame_rate=_RATE,

        channels=_CHANNELS,

        sample_width=_p.get_sample_size(_FORMAT),

    )

    _audio = _audio.set_frame_rate(16000)

    _wav = io.BytesIO()

    _audio.export(_wav, format="wav")

    return _wav





if __name__ == "__main__":

    import pydub.playback as playback



    save = False



    result = listen()



    sound = pydub.AudioSegment.from_file(result, format="wav")

    playback.play(sound)



    if save:

        with open("audio.wav", "wb") as file:

            file.write(result.getbuffer())

代码运行起来后程序会循环检测麦克风的声压，当升压超过阈值后，也就是定义的_min_db=200，则开始录音；当超过一段时间音量小于阈值后，则认为说话结束，停止录音。

录音仅用到pyaudio，pydub是用来把声音数据和数据头组装起来形成wav文件的。在下面的测试运行代码中使用了pydub的播放功能来把录到的音频再播放出来，可以用作测试。

如果电脑的录音和播放功能正常，运行这段代码，就可以完成上述功能测试。如果把save改成1，还可以得到保存好的wav录音文件。

语音识别

下面我们回到核桃派上进行开发。

得到录音后，下一步要进行语音识别。语音识别我使用的是常用的python库speech_recognition。这个库中有多个模型可以使用，我使用的是老版的recognize_google模型。这个模型是唯一一个不需要API KEY的云端模型，非常适合核桃派这样迷你的开发板使用。

import speech_recognition as sr



r = sr.Recognizer()





def detect(wav_bytesio):

    try:

        audio_source = sr.AudioFile(wav_bytesio)

        with audio_source as source:

            audio = r.record(source)

        text = r.recognize_google(audio, language="cmn-Hans-CN")

    except sr.UnknownValueError as _error:

        print("Google could not understand audio")

        print(_error)

        text_input = None

    except sr.RequestError as _error:

        print("Could not request results from Google")

        print(_error)

        text_input = None

    return text





if __name__ == "__main__":

    import io



    with open("audio.wav", mode="rb") as f:

        wav_bytes = f.read()

    wav_bytesio = io.BytesIO(wav_bytes)



    text = detect(wav_bytesio)

    print(text)

后面的测试代码是从本地读取一个文件，然后识别出文字来打印出来。我们可以把上一步中录好的文件传到核桃派上进行测试。

细心的小伙伴可能会注意到，为什么在上一期和这一期中我们实际传递的是一个wav文件，但却并没有真的保存文件，而是使用一个io.BytesIO Object来进行传递？这是因为如果真的去保存读取中间文件的话会导致性能下降，同时带来对TF卡的不必要读写，降低TF卡寿命。而很多方法接收的参数都是一个文件object，因此我们使用io.BytesIO就可以在内存中直接生成一个文件object，这样所有的数据传递都在内存中就可以完成。因此整个项目中需要传递文件的地方我也都会使用io.BytesIO来进行处理。

图像识别

这一部分图像识别，是整个项目中唯一一块AI功能并不使用云端API，而是所有计算预测都在本地完成的部分。这一块会对内存和核心算力有一定的考验。我们使用yolov8n这个最小的模型来进行物品识别。

首先还是这个功能模块的代码：

from ultralytics import YOLO





class yolov8:

    def __init__(self, _model="yolov8n.pt"):

        self._model = YOLO(_model)



    def detect(self, _frame, _show=False, _conf=0.5):

        _results = self._model.predict(_frame, stream=True, show=_show, conf=_conf)

        for _result in _results:

            try:

                _detections = []

                for _box in _result.boxes:

                    _class_id = _result.names[_box.cls[0].item()]

                    _cords = _box.xyxy[0].tolist()

                    _cords = [round(x) for x in _cords]

                    _confi = round(_box.conf[0].item(), 2)

                    # print("Object type:", _class_id)

                    # print("Coordinates:", _cords)

                    # print("Probability:", _confi)

                    # print("---")

                    _detections.append(

                        {"name": _class_id, "cords": _cords, "conf": _confi}

                    )

            except Exception as error:

                print(error)

        return _detections





if __name__ == "__main__":

    import requests

    from PIL import Image

    import io



    bot = yolov8()



    img_raw = requests.get("https://ultralytics.com/images/bus.jpg").content

    img = Image.open(io.BytesIO(img_raw))

    print(bot.detect(img))

后面测试代码部分的功能是，从官方下载官方的测试图片，然后对该图片进行物体识别。识别返回的结果是一个列表，每个列表中的元素就是识别到的一个物体，这个物体是以字典的形式存储的，字典中存有这个物体的名字，位置，以及识别的置信度。

GPT接入

GPT接入我们可以使用大名鼎鼎的GPT4FREE项目来实现。这个项目的api与openai的api基本一致，可以很方便的将代码进行移植。好处是，使用这个库并不需要申请api，可以直接调用。同样的，我将方法封装成一个新的库，并在下面写了个测试小例程：

from g4f.client import Client





class ai:

    def __init__(

        self,

        model="gpt-3.5-turbo",

        init_prompt="你是一名助手，你的任务是回答日常问题的解决方法，并提供有用的建议。",

        proxy=None,

    ):

        self.model = model

        self.init_prompt = init_prompt

        self.messages = [{"role": "system", "content": self.init_prompt}]

        self.proxy = proxy



        self.client = Client()



    def chat(self, input):

        if input == "clean":

            self.messages = [{"role": "system", "content": self.init_prompt}]

            return "History cleaned"

        self.messages.append({"role": "user", "content": input})

        response = self.client.chat.completions.create(

            model=self.model,

            messages=self.messages,

            proxy=self.proxy,

        )

        self.messages.append(

            {

                "role": response.choices[0].message.role,

                "content": response.choices[0].message.content,

            }

        )

        return self.messages[-1]["content"]





if __name__ == "__main__":

    bot = ai()

    while True:

        content = input(">说点什么：")

        gpt_reply = bot.chat(content)

        print(f"AI回答：{gpt_reply}")

这个项目中我们并不需要很聪明的模型，但是需要模型稳定，可连接，因此使用源最多的GPT-3.5模型。调用方法非常简单，生成实例时可以直接传入system提示词，以此改变ai的用途。直接运行一下代码，就可以在terminal中与之对话；输入clean就可以清空上下文。

语音输出

语音输出使用edge-tts来实现。这个库原来仅支持async的方法，但是在和FastAPI一起使用时则会出现冲突问题。因此作者增加了sync的方式运行，具体的pull可以参考这里：

https://github.com/rany2/edge-tts/pull/215

由于我们也需要使用FastAPI，因此使用最新的sync方法来实现tts功能。

import edge_tts

import pydub

import io





def tts(text, actor="zh-CN-XiaoyiNeural", fmt="mp3"):

    _communicate = edge_tts.Communicate(text, actor)

    _mp3 = io.BytesIO()

    for _chunk in _communicate.stream_sync():

        if _chunk["type"] == "audio":

            _mp3.write(_chunk["data"])

        elif _chunk["type"] == "WordBoundary":

            # print(f"WordBoundary: {_chunk}")

            pass

    _mp3.seek(0)

    if fmt == "mp3":

        return _mp3

    if fmt == "wav":

        _audio = pydub.AudioSegment.from_file(_mp3, format="mp3")

        _audio = _audio.set_frame_rate(16000)

        _wav = io.BytesIO()

        _audio.export(_wav, format="wav")

        return _wav





if __name__ == "__main__":

    import pydub.playback as playback



    fmt = "mp3"

    save = False



    while True:

        text_in = input(">说点什么：")

        if fmt == "mp3":

            result = tts(text_in, actor="zh-CN-XiaoyiNeural", fmt="mp3")

        elif fmt == "wav":

            result = tts(text_in, actor="zh-CN-XiaoyiNeural", fmt="wav")

        else:

            raise ("fmt should be either mp3 or wav")



        sound = pydub.AudioSegment.from_file(result, format=fmt)

        playback.play(sound)



        if save:

            with open("audio." + fmt, "wb") as file:

                file.write(result.getbuffer())

这个代码的测试部分我写的是可以把输入的任何文字转换成语音，并直接播放出来，如果有需要也可以储存下来。找个耳机或者音箱插在核桃派的3.5mm的接口上，运行这个代码，输入内容，应该很快就可以听到输出的声音。

因为默认的输出格式是mp3，但是在python中大部分情况下使用的音频格式是wav，因为无需解码可以直接读取数据，因此我在这个方法里还引入了pydub，可以选择以wav的方式输出。返回的内i容同样是io.Bytesio Object.

功能整合

到此为止我们已经实现了所有的独立功能模块，最后只需要把他们整合在一起，用FastAPI来连接起PC和核桃派就可以。

图像识别之前测试的时候我们使用的是官方的图片，当然可以把这个图片地址修改成我们网络摄像头的静态捕捉地址，这样就可以从网络摄像头获取到最新帧。

而网络摄像头如果大家没有现成的，可以直接使用ESP32CAM模块，在上面烧写官方的CameraWebServer.ino示例就可以：
CameraWebServer.ino

之前的每个模块我们都把他保存为单独的文件，方便调用。首先是核桃派这边，我们需要在核桃派上建立FastAPI服务器，当PC访问后，得到PC发来的音频数据，然后开始一系列后面的处理，最后返回GPT的回复。

from typing import Annotated

from fastapi import FastAPI, File

from starlette.responses import Response



import io

import speech_detect

import img_detect

import requests

from PIL import Image

import ai

import tts

import pydub

import pydub.playback as playback



bot = ai.ai(

    init_prompt="你是一个拥有视力的人类助手。假设你拥有视力，可以看到东西。你的任务是根据user提供的你所看到的内容，回答user的问题。请在接下来的对话中隐藏你是计算机程序这件事，并且模仿的像一个具有视力的人类助手。不要提及你是一个语言模型，不要说提及你无法看到任何东西。"

)

bot_img = img_detect.yolov8()





def main(audio):

    text_speech = speech_detect.detect(audio)

    text_detect = ""

    img_raw = requests.get("http://192.168.50.232/capture?_cb=0").content

    img = Image.open(io.BytesIO(img_raw))

    for i in bot_img.detect(img):

        text_detect += i["name"] + ","

    if not text_detect:

        text_detect = "什么都没有"

    else:

        text_detect = text_detect[:-1]



    text = (

        "你看到了你的面前有以下这些东西："

        + text_detect

        + "。请用中文，以拥有视力的人类助手的身份回答这个问题："

        + text_speech

    )

    print(text)

    gpt_reply = bot.chat(text)



    for i in "#*":

        gpt_reply = gpt_reply.replace(i, "")

    result = tts.tts(gpt_reply)



    sound = pydub.AudioSegment.from_file(result)

    playback.play(sound)

    return gpt_reply





app = FastAPI()





@app.post("/uploadfile")

def create_file(file: Annotated[bytes, File()]):

    received_audio = io.BytesIO(file)

    ret = main(received_audio)

    return ret

    # return Response(ret.getbuffer())





if __name__ == "__main__":

    import uvicorn



    uvicorn.run(app, host="0.0.0.0", port=8000)

而在PC这边，我们需要录音，然后将录音数据通过post方法发送给FastAPI服务器，最后将收到的返回内容打印出来。

import listen

import requests



# import pydub.playback as playback

# import pydub

# import io



while True:

    files = {"file": listen.listen(_min_db=200)}

    response = requests.post("http://192.168.50.158:8000/uploadfile", files=files)

    print(response.content.decode(encoding="UTF-8"))



    # sound = pydub.AudioSegment.from_file(io.BytesIO(response.content))

    # playback.play(sound)

整个流程中最重要的内容就是GPT的输入和输出，因此我们把输入内容在核桃派上打印出来，而把输出内容在PC上打印出来。这时候两边同时运行，我们可以测试一下效果：

这是核桃派的输出，其中打印的内容是GPT的输入内容：

这是电脑的输出，其中打印的内容是GPT的输出内容：

评测到此项目就结束了。整体来说，这是一块小型的linux开发板，主打低成本，低功耗，用来做一些小项目非常合适。值得大大赞赏的是，官方文档教程非常细致，对新手极度友好，这在国产品牌中是非常难能可贵的一点。树莓派之所以能够壮大，而其他各路派都是昙花一现，就是因为树莓派对水平层次不齐的爱好者非常友好。核桃派能重视这一点是非常不容易的，希望可以坚持下去，真正成为可以和树莓派竞争的一股力量。

最后的最后，附上整个项目的源代码。

*附件：jarvis_walnutpi.zip