变身Siri！树莓派物理对讲机进化指南

1. 架构理念：单点真实源 (Single Source of Truth)

为了在资源极其受限的树莓派上实现语音助理，本方案摒弃了传统的“守护进程持续监听”模式，转而采用 “事件驱动 + CLI 直调” 的极客架构。

一脑多端：通过 Python 脚本劫持物理按键和音频外设，直接在本地调用 picoclaw agent 的单次运行模式 (CLI)。这意味着桌面的物理对讲机与云端的飞书机器人，共享同一个大模型大脑、同一套飞书凭证上下文、以及同一个本地沙盒工作区 (~/.picoclaw/workspace)。
极致轻量：只有在按下按钮的瞬间，底层组件才会被唤醒。松开按钮后，音频流迅速经过火山引擎 (STT) -> 本地 PicoClaw (LLM) -> 微软 Azure (TTS)，全程内存占用极低。

2. 核心踩坑记录与破局方案

在连接底层硬件音频与大模型纯文本的跨界开发中，我们遭遇并解决了以下几个极其隐蔽的系统级“暗坑”：

2.1 物理按键的“幽灵连击” (Mechanical Bouncing)

现象：按下一秒钟，程序却报出无数次 [系统忙碌]，甚至导致录音进程被瞬间强杀，引发致命错误。
破局：这是街机按钮金属微动开关的物理弹跳导致的。在 gpiozero 初始化时，强制加入硬件防抖参数 bounce_time=0.3，让系统在 300 毫秒内无视一切毛刺信号。

2.2 音频锁死与“吞字”现象 (Buffer Overrun)

现象：频繁报出 overrun!!!（缓冲区溢出），或者每次说话的头一两个字（如“北京”）总是录不进去；甚至脚本崩溃后，麦克风被僵尸进程永久锁死。
破局：
1. 开局核弹清理：在 Python 脚本顶部加入 killall -9 arecord mpg123，每次启动强杀孤儿进程，保证声卡绝对干净。
2. 超大缓冲区：给 arecord 加上 --buffer-time=250000 参数，彻底解决丢帧警告。
3. 视觉防呆设计：延迟点亮 LED 灯。让底层硬件通电稳定 0.5 秒后，再亮灯提示用户说话，完美避开硬件冷启动的“吞字”期。
4. 优雅释放：停止录音时使用 send_signal(signal.SIGINT) 而非直接 terminate()，确保 WAV 文件尾部正常闭合并释放 ALSA 硬件锁。

2.3 大模型输出的“精神污染”与 TTS 崩溃

现象：PicoClaw 的 CLI 输出包含了大量的 ASCII 字符画 Logo 和 [INFO] 系统日志；且大模型偶尔会输出 🦞 等 Emoji 或者 Markdown 表格，导致 edge-tts 语音合成引擎直接崩溃罢工。
破局：
1. 分层剥离：通过按行切片，丢弃最后一条 [INFO] 日志之前的所有终端输出。
2. 正则清洗：在送入发声引擎前，使用正则表达式 re.sub 暴力剔除所有非中英文字符和基础标点，确保语音朗读顺滑自然。

2.4 跨时区场景下的“时间幻觉”

现象：大模型没有系统时间概念，问及“明天”时会胡乱推测。
破局：系统提示词动态注入 (System Prompt Injection)。在 Python 组装查询文本时，利用 datetime 模块实时推算出美东时间与北京时间，并将其作为 [系统隐藏设定] 强制拼接在用户语音前。这样不仅消除了幻觉，还让大模型具备了完美的双时区业务处理能力。

2.5 安装依赖

当树莓派重刷系统（如切换至最新的 Debian 12 Bookworm Lite 版）后，原有环境会被彻底清空。请严格按照以下顺序执行，即可在 5 分钟内让对讲机满血复活。

2.5.1 安装系统级底层软件

更新软件源并安装基础的音频处理、进程管理工具及 Python 包管理器：

sudo apt update
# 安装音频播放器(mpg123)、录音驱动集(alsa-utils)、进程强杀工具(psmisc)和pip3
sudo apt install -y mpg123 alsa-utils psmisc python3-pip

2.5.2 安装 Python 核心驱动库

使用 pip3 安装所需的第三方库。 (注：在最新的树莓派系统中，系统为了保护全局环境限制了直接使用 pip，必须追加 --break-system-packages 参数强制安装至用户目录)：

pip3 install requests gpiozero edge-tts --break-system-packages

2.5.3 修复环境变量 (解决找不到命令的 Warning)

上一步安装的 edge-tts 会被存放在隐藏的用户级二进制目录中。必须将其加入系统环境变量雷达：

# 将用户私有 bin 目录追加到 PATH，并写入 bash 配置文件
echo 'export PATH="$HOME/.local/bin:$PATH"' >> ~/.bashrc
# 立即让配置生效
source ~/.bashrc

验证测试：终端输入 edge-tts --version，若正常输出版本号则配置成功。

2.5.4 恢复 PicoClaw 核心与工作区

对讲机的大脑需要被重新植入：

主程序归位：将编译好的 picoclaw 二进制文件移至 /usr/local/bin/，并赋予执行权限 (sudo chmod +x /usr/local/bin/picoclaw)。
工作区重建：重新创建 ~/.picoclaw/workspace/ 目录，并将之前备份的 skills 文件夹（如飞书插件、搜索插件）及配置文件原样拷入。

2.5.5 硬件声卡复健 (极易遗漏)

纯净系统下，外接 I2S 麦克风阵列（如 Google Voice HAT）默认处于物理静音状态：

重新挂载驱动：按官方文档修改 /boot/firmware/config.txt 开启 I2S，并配置 ~/.asoundrc 文件。
解除底层静音并调音：

alsamixer

按 F4 切换至 Capture (录音) 界面，选中麦克风按 Space 键解除 MM (静音) 状态，并将增益拉高。
按 F3 切换至 Playback (播放) 界面，使用方向键将播放音量降至 40%~50%，避免破音。

固化音量配置：执行 sudo alsactl store，确保重启后音量设置不丢失。

至此，软硬件环境已彻底复原，结合 Systemd 守护进程配置，系统即可重新投入全自动运行。

3. 终极整合代码 (`walkie_talkie.py`)

本代码已包含全部防弹锁、时区注入及异常捕获机制，可直接作为生产环境的守护脚本运行。

import subprocess
import os
import requests
import threading
import base64
import uuid
import re
import time
import signal
from datetime import datetime, timezone, timedelta
from gpiozero import Button, LED
from signal import pause

print("🧹 正在清理可能遗留的底层音频进程...")
# 启动前强制杀死遗留进程，释放 ALSA 硬件锁
subprocess.run(["killall", "-q", "-9", "arecord", "mpg123", "edge-tts"], stderr=subprocess.DEVNULL)

# ================= 1. 硬件与路径配置 =================
BUTTON_PIN = 23
LED_PIN = 25

# 采用 PicoClaw 官方标准工作区
WORKSPACE = os.path.expanduser("~/.picoclaw/workspace")
os.makedirs(WORKSPACE, exist_ok=True)
AUDIO_IN = os.path.join(WORKSPACE, "voice_in.wav")
AUDIO_OUT = os.path.join(WORKSPACE, "voice_out.mp3")

# ================= 2. 火山引擎 (极速版大模型 STT) 配置 =================
VOLC_APPID = "替换为你的_APP_ID"  
VOLC_TOKEN = "替换为你的_ACCESS_TOKEN"

# ================= 3. 全局状态初始化 =================
# 核心：bounce_time=0.3 解决物理按键抖动
button = Button(BUTTON_PIN, bounce_time=0.3)
led = LED(LED_PIN)
record_proc = None
is_processing = False  # 并发锁

def start_recording():
    global record_proc, is_processing
    if is_processing:
        print("⏳ [系统忙碌] 正在处理上一条指令...")
        return
        
    if os.path.exists(AUDIO_IN):
        os.remove(AUDIO_IN)
        
    print("\n⏳ [准备中] 正在唤醒麦克风底座...")
    
    # --buffer-time=250000 解决 overrun；DEVNULL 屏蔽中断报错
    cmd = ["arecord", "--buffer-time=250000", "-f", "S16_LE", "-r", "16000", "-c", "1", AUDIO_IN, "-q"]
    record_proc = subprocess.Popen(cmd, stderr=subprocess.DEVNULL)

    # 延迟亮灯交互：让硬件通电稳定，防止吞掉首字
    time.sleep(0.5)
    led.on()
    print("🟢 [录音中] 麦克风已就绪，请说话，松开结束...")

def stop_recording():
    global record_proc, is_processing
    if is_processing or record_proc is None:
        return
        
    # 缓冲收尾，防止尾音被切
    time.sleep(0.3)
    # 使用 SIGINT 优雅退出，确保 WAV 写入文件头并释放声卡
    record_proc.send_signal(signal.SIGINT)
    record_proc.wait()
    record_proc = None
    
    led.off()
    print("🛑 [已松开] 录音结束，移交处理中心...")
    
    is_processing = True
    threading.Thread(target=process_audio_pipeline).start()

def process_audio_pipeline():
    global is_processing
    led.blink(on_time=0.5, off_time=0.5) 

    try:
        if not os.path.exists(AUDIO_IN):
            print("❌ 致命错误：录音文件未生成！")
            return
            
        # ================= STT (火山引擎极速识别) =================
        print("🔄 请求火山引擎 STT...")
        user_text = ""
        try:
            with open(AUDIO_IN, "rb") as audio_file:
                base64_audio = base64.b64encode(audio_file.read()).decode('utf-8')
            
            headers = {
                "X-Api-App-Key": VOLC_APPID,
                "X-Api-Access-Key": VOLC_TOKEN,
                "X-Api-Resource-Id": "volc.bigasr.auc_turbo",
                "X-Api-Request-Id": str(uuid.uuid4()),
                "X-Api-Sequence": "-1",
                "Content-Type": "application/json"
            }
            payload = {
                "user": {"uid": VOLC_APPID},
                "audio": {"data": base64_audio},
                "request": {"model_name": "bigmodel"}
            }
            
            recognize_url = "https://openspeech.bytedance.com/api/v3/auc/bigmodel/recognize/flash"
            response = requests.post(recognize_url, json=payload, headers=headers, timeout=10)
            
            status_code = response.headers.get("X-Api-Status-Code")
            if status_code == '20000000':
                user_text = response.json().get("result", {}).get("text", "")
                if not user_text: user_text = "我听到声音了，但没听清内容。"
            elif status_code == '20000003': user_text = "你好像没说话？"
            else: user_text = "耳朵暂时罢工了。"

        except Exception as e:
            user_text = "网络连接出错了。"

        print(f"👤 你说: {user_text}")

        # ================= AI 思考 (PicoClaw 调度中心) =================
        print("🧠 唤醒 PicoClaw 本地进程...")
        try:
            # 动态双时区注入，消除时间幻觉
            utc_dt = datetime.now(timezone.utc)
            us_east_dt = utc_dt.astimezone(timezone(timedelta(hours=-4)))
            beijing_dt = utc_dt.astimezone(timezone(timedelta(hours=8)))
            
            weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
            time_context = f"[系统当前时间：美东 {us_east_dt.strftime('%m-%d %H:%M')}({weekdays[us_east_dt.weekday()]}) | 北京 {beijing_dt.strftime('%m-%d %H:%M')}({weekdays[beijing_dt.weekday()]})]"
            
            # 拼装包含工具调度规范的 Prompt
            voice_prompt = f"""{time_context}
用户指令：“{user_text}”

【系统强制执行规则（语音管家模式）】：
1. 【极简极速】你是一个干练、高效的私人语音助理。回复必须直奔主题，严格控制在 100 字以内，最多一到两句话。（注意：字数越少，系统的语音响应延迟越低！）
2. 【去AI味】直接给出答案！绝对不许说“为您查询到”、“根据我的搜索”、“测了一下发现”等机器味废话，也不要反问用户。
3. 【静默调用】涉及限行、天气、新闻等，必须静默调用内置工具，拿到数据后直接播报核心结论。
4. 【纯净发音】绝对禁止使用任何 Markdown（如 **、-、#）、表格、序号、Emoji 或生僻符号，只输出能直接用嘴流畅读出来的纯汉字和基本标点。"""

            cmd = ["picoclaw", "agent", "-m", voice_prompt]
            result = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT)
            
            # 剥离 PicoClaw CLI 的 ASCII Logo 与系统日志
            lines = result.split('\n')
            last_log_idx = -1
            for i, line in enumerate(lines):
                if "[INFO]" in line or "[WARN]" in line or "[ERROR]" in line:
                    last_log_idx = i
            
            if last_log_idx != -1 and last_log_idx < len(lines) - 1:
                ai_reply = "\n".join(lines[last_log_idx+1:]).strip()
            else:
                ai_reply = "\n".join([l for l in result.split('\n') if "██" not in l]).strip()

            if not ai_reply: ai_reply = "大脑执行完毕，未返回文本。"
                 
        except Exception as e:
            ai_reply = "连接大脑失败，请检查运行状态。"

        print(f"🤖 PicoClaw 原始回复: {ai_reply}")
        
        # ================= TTS 与 物理发声 =================
        print("🔊 清洗文本并合成语音...")
        
        # 终极清洗：剔除 Emoji 等会导致 TTS 崩溃的非法字符
        clean_text = re.sub(r'[^\u4e00-\u9fa5A-Za-z0-9，。！？、！\?,\.～~]', '', ai_reply)
        if not clean_text: clean_text = "回复中没有可读的文字。"

        tts_cmd = [
            "python3", "-m", "edge_tts", 
            "--text", clean_text, 
            "--voice", "zh-CN-XiaoxiaoNeural",
            "--write-media", AUDIO_OUT
        ]
        
        try:
            subprocess.run(tts_cmd, check=True, capture_output=True, text=True)
            led.on() 
            subprocess.run(["mpg123", AUDIO_OUT])
        except subprocess.CalledProcessError as e:
            print(f"❌ 语音生成失败！\n错误详情: {e.stderr}")
        
    except Exception as e:
        print(f"❌ 未捕获异常: {e}")
    finally:
        led.off() 
        is_processing = False
        print("✅ 系统已重置，等待下一次唤醒。")

button.when_pressed = start_recording
button.when_released = stop_recording

print("🚀 极客对讲机中枢系统已就绪！")
pause()

4. 终极部署：化身后台守护进程 (Daemonize)

为了让这个硬件助理彻底脱离 SSH 终端，实现“插电即用、断电重启自动恢复”，我们需要使用 Linux 原生的 systemd 将其注册为系统级服务。

⚠️ 避坑指南：绝对不要用 rc.local 或 crontab 来启动音频类的脚本。因为音频设备 (sound.target) 加载较慢，只有 systemd 能精准控制启动顺序，并且当脚本意外崩溃时自动拉起。

4.1 创建 systemd 服务文件

假设你的脚本放在了 /home/pi/walkie_talkie.py。在终端中执行以下命令创建配置文件：

sudo nano /etc/systemd/system/picoclaw-voice.service

将以下内容粘贴进去（注意：如果你的用户名不是 pi，或者脚本路径不同，请自行修改）：

[Unit]
Description=PicoClaw Voice Assistant (Hardware Gateway)
# 确保在网络和底层声卡硬件加载完毕后再启动本服务
After=network.target sound.target

[Service]
Type=simple
# 必须指定为 pi 用户，否则以 root 运行时，~/.picoclaw 路径会变成 /root/.picoclaw，导致配置全部失效
User=pi
WorkingDirectory=/home/pi
ExecStart=/usr/bin/python3 /home/pi/walkie_talkie.py

# 崩溃自动重启机制 (每5秒尝试重启一次)
Restart=always
RestartSec=5

# 【极其关键】注入环境变量！确保 Python 脚本底层的 subprocess 能找到 picoclaw、edge-tts 和 arecord
Environment="PATH=/home/pi/.local/bin:/usr/local/bin:/usr/bin:/bin"
Environment="PYTHONUNBUFFERED=1"

[Install]
WantedBy=multi-user.target

保存并退出（在 nano 中按 Ctrl+O -> Enter -> Ctrl+X）。

4.2 刷新系统并设置开机自启

配置写好后，告诉系统重载配置，并开启自启：

# 1. 重新加载 systemd 守护进程
sudo systemctl daemon-reload

# 2. 设置为开机自动启动
sudo systemctl enable picoclaw-voice.service

# 3. 立即启动该服务
sudo systemctl start picoclaw-voice.service

4.3 常用管理命令 (备忘录)

一旦注册为服务，你以后就可以像管理 Nginx 或 MySQL 一样，优雅地管理你的对讲机了：

查看当前运行状态 (能看到绿色的 active (running) 即为成功)： ```bash sudo systemctl status picoclaw-voice

* **手动停止**服务：
```bash
sudo systemctl stop picoclaw-voice

手动重启服务（当你修改了 Python 代码后，必须执行此命令生效）： ```bash sudo systemctl restart picoclaw-voice

* **查看实时运行日志**（替代原本在终端里看到的 print 输出，排错必备）：
```bash
# -f 代表实时滚动，类似于 tail -f
journalctl -u picoclaw-voice -f

补充说明：

这块拼图补齐之后，你的树莓派就彻底变成了一个“黑盒家电”。你可以放心地关掉电脑终端，拔掉树莓派的电源，把它搬到客厅或者床头柜插上电。开机大概等待一两分钟（等网络和后台进程加载完），你就可以直接拍下按钮和它聊天了！