变身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 模块实时推算出美东时间与北京时间，并将其作为 [系统隐藏设定] 强制拼接在用户语音前。这样不仅消除了幻觉，还让大模型具备了完美的双时区业务处理能力。

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. 涉及实时信息（如限行、天气、新闻），必须优先调用内置工具获取最新数据。
2. 将结论转换为简短、自然的纯口语回复（对讲机风格）。
3. 绝对禁止包含 Markdown 符号、表格或复杂标点。"""

            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

补充说明：

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