1. 引言

宇树机器人G1是一款功能强大的通用人形机器人，其开放的SDK为开发者提供了广阔的二次开发空间。语音交互是机器人实现自然、高效人机交互的核心能力。本文将详细介绍如何在宇树G1机器人上实现一套完整的语音对话系统，涵盖语音对话、打断对话、停止对话、进入待命、激活对话等核心功能，并提供有线话筒与无线话筒两种方案的完整代码实现。

通过本文，你将能够为你的G1机器人赋予“耳朵”和“嘴巴”，使其能够听懂指令、进行多轮对话，并灵活响应用户的交互意图。

2. 环境准备与依赖安装

2.1 硬件准备

• 宇树机器人G1：确保机器人系统已启动，网络连接正常。
• 音频输入设备：
  • 有线话筒方案：USB接口的麦克风，直接插入机器人主控计算机的USB口。
  • 无线话筒方案：支持蓝牙或2.4G无线连接的麦克风，需确保与机器人主控计算机配对成功。
• 音频输出设备：机器人内置扬声器或外接音箱。

2.2 软件环境

• 操作系统：Ubuntu 20.04/22.04 (推荐，与官方SDK兼容性最佳)
• Python版本：Python 3.8+
• 宇树G1 SDK：确保已安装并配置好unitree_sdk2。

2.3 Python依赖包安装

在机器人主控计算机上执行以下命令安装必要的Python库：

pip install pyaudio wave numpy sounddevice soundfile
pip install openai-whisper # 可选，用于本地语音识别
pip install speechrecognition
pip install pydub
pip install pyttsx3
# 如果使用在线ASR服务（如百度、阿里云、Azure等），还需安装对应SDK
# pip install baidu-aip

3. 核心功能模块设计

整个语音对话系统可分为以下几个核心模块：

4. 完整代码实现

4.1 主程序框架：voice_controller.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
宇树G1机器人语音对话控制器
功能：语音对话、打断、停止、待命、激活、支持有线/无线话筒
作者：CSDN博主
"""

import threading
import time
import queue
import json
from enum import Enum
import logging

# 配置日志
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RobotState(Enum):
    """机器人对话状态枚举"""
    STANDBY = "standby"      # 待命状态，等待唤醒
    LISTENING = "listening"  # 聆听状态，正在接收语音
    PROCESSING = "processing" # 处理中，ASR或NLU
    SPEAKING = "speaking"    # 播报状态，TTS输出中
    STOPPED = "stopped"      # 强制停止状态

class VoiceController:
    """语音对话主控制器"""
    
    def __init__(self, mic_type='wired'):
        """
        初始化语音控制器
        :param mic_type: 话筒类型，'wired' 或 'wireless'
        """
        self.state = RobotState.STANDBY
        self.mic_type = mic_type
        self.audio_queue = queue.Queue(maxsize=10)
        self.stop_event = threading.Event()
        self.interrupt_event = threading.Event()
        
        # 初始化各模块
        self.audio_input = self._init_audio_input(mic_type)
        self.vad_detector = VADetector()
        self.asr_engine = ASREngine()
        self.tts_engine = TTSEngine()
        self.dialog_manager = DialogManager()
        
        # 唤醒词和打断词
        self.wake_words = ["小宇", "机器人", "嗨G1"]
        self.interrupt_words = ["停", "停止", "闭嘴", "打断"]
        self.standby_words = ["待命", "休息", "退下"]
        
        logger.info(f"语音控制器初始化完成，话筒类型: {mic_type}")
    
    def _init_audio_input(self, mic_type):
        """初始化音频输入设备"""
        if mic_type == 'wired':
            return WiredMicrophone()
        elif mic_type == 'wireless':
            return WirelessMicrophone()
        else:
            raise ValueError(f"不支持的话筒类型: {mic_type}")
    
    def start(self):
        """启动语音对话系统"""
        logger.info("启动语音对话系统...")
        self.stop_event.clear()
        self.interrupt_event.clear()
        
        # 启动音频采集线程
        audio_thread = threading.Thread(target=self._audio_capture_loop, daemon=True)
        audio_thread.start()
        
        # 启动主处理线程
        main_thread = threading.Thread(target=self._main_loop, daemon=True)
        main_thread.start()
        
        logger.info("语音对话系统已启动，等待唤醒...")
        return audio_thread, main_thread
    
    def stop(self):
        """停止整个语音系统"""
        logger.info("停止语音对话系统...")
        self.stop_event.set()
        self.state = RobotState.STOPPED
        self.tts_engine.stop()  # 停止当前播报
    
    def interrupt(self):
        """打断当前对话"""
        logger.info("收到打断指令")
        self.interrupt_event.set()
        if self.state == RobotState.SPEAKING:
            self.tts_engine.interrupt()
            self.state = RobotState.STANDBY
            self.tts_engine.speak("已打断")
    
    def enter_standby(self):
        """进入待命状态"""
        logger.info("进入待命状态")
        self.state = RobotState.STANDBY
        self.tts_engine.speak("已进入待命模式")
    
    def activate(self):
        """激活对话（从待命状态唤醒）"""
        logger.info("激活对话")
        self.state = RobotState.LISTENING
        self.tts_engine.speak("我在，请说")
    
    def _audio_capture_loop(self):
        """音频采集循环"""
        try:
            self.audio_input.start_stream()
            logger.info("音频采集已开始")
            
            while not self.stop_event.is_set():
                audio_data = self.audio_input.read_chunk()
                if audio_data is not None:
                    # 简单的VAD检测，避免空音频入队
                    if self.vad_detector.is_speech(audio_data):
                        self.audio_queue.put(audio_data)
                time.sleep(0.01)
                
        except Exception as e:
            logger.error(f"音频采集异常: {e}")
        finally:
            self.audio_input.stop_stream()
    
    def _main_loop(self):
        """主处理循环"""
        while not self.stop_event.is_set():
            try:
                # 状态机处理
                if self.state == RobotState.STANDBY:
                    self._handle_standby()
                elif self.state == RobotState.LISTENING:
                    self._handle_listening()
                elif self.state == RobotState.PROCESSING:
                    self._handle_processing()
                elif self.state == RobotState.SPEAKING:
                    self._handle_speaking()
                
                time.sleep(0.05)
                
            except Exception as e:
                logger.error(f"主循环异常: {e}")
                time.sleep(1)
    
    def _handle_standby(self):
        """待命状态处理：检测唤醒词"""
        if not self.audio_queue.empty():
            audio_data = self.audio_queue.get()
            text = self.asr_engine.recognize(audio_data)
            
            if text and any(word in text for word in self.wake_words):
                logger.info(f"检测到唤醒词: {text}")
                self.activate()
    
    def _handle_listening(self):
        """聆听状态处理：收集语音并识别"""
        audio_chunks = []
        silence_count = 0
        max_silence = 20  # 静音帧数阈值
        
        while silence_count < max_silence and not self.interrupt_event.is_set():
            if not self.audio_queue.empty():
                audio_data = self.audio_queue.get()
                if self.vad_detector.is_speech(audio_data):
                    audio_chunks.append(audio_data)
                    silence_count = 0
                else:
                    silence_count += 1
            time.sleep(0.01)
        
        if audio_chunks and not self.interrupt_event.is_set():
            # 合并音频并识别
            full_audio = b''.join(audio_chunks)
            text = self.asr_engine.recognize(full_audio)
            
            if text:
                logger.info(f"识别结果: {text}")
                self.state = RobotState.PROCESSING
                self._process_user_input(text)
    
    def _process_user_input(self, text):
        """处理用户输入文本"""
        # 检查是否为控制指令
        if any(word in text for word in self.interrupt_words):
            self.interrupt()
            return
        
        if any(word in text for word in self.standby_words):
            self.enter_standby()
            return
        
        # 正常对话处理
        response = self.dialog_manager.get_response(text)
        
        if response:
            self.state = RobotState.SPEAKING
            self.tts_engine.speak(response)
            # 播报完成后回到聆听状态（除非被打断）
            if not self.interrupt_event.is_set():
                self.state = RobotState.LISTENING
        else:
            self.state = RobotState.LISTENING
    
    def _handle_processing(self):
        """处理中状态：短暂等待"""
        time.sleep(0.1)
    
    def _handle_speaking(self):
        """播报状态：检查是否被打断"""
        if self.interrupt_event.is_set():
            self.interrupt_event.clear()
            self.state = RobotState.STANDBY

# 主程序入口
if __name__ == "__main__":
    # 创建控制器（可选择话筒类型）
    controller = VoiceController(mic_type='wired')  # 或 'wireless'
    
    try:
        # 启动系统
        threads = controller.start()
        
        # 保持主线程运行
        while True:
            cmd = input("输入命令 (stop/exit退出): ").strip().lower()
            if cmd in ['stop', 'exit', 'quit']:
                controller.stop()
                break
            elif cmd == 'interrupt':
                controller.interrupt()
            elif cmd == 'standby':
                controller.enter_standby()
            elif cmd == 'activate':
                controller.activate()
            time.sleep(0.1)
            
    except KeyboardInterrupt:
        logger.info("收到中断信号，停止程序...")
        controller.stop()
    except Exception as e:
        logger.error(f"程序运行异常: {e}")
    finally:
        logger.info("程序结束")

4.2 音频输入模块：audio_input.py

import pyaudio
import numpy as np
import sounddevice as sd
import threading
import time

class WiredMicrophone:
    """有线话筒采集（基于PyAudio）"""
    
    def __init__(self, rate=16000, chunksize=1024, channels=1):
        self.rate = rate
        self.chunksize = chunksize
        self.channels = channels
        self.p = pyaudio.PyAudio()
        self.stream = None
        
    def start_stream(self):
        """开始音频流"""
        self.stream = self.p.open(
            format=pyaudio.paInt16,
            channels=self.channels,
            rate=self.rate,
            input=True,
            frames_per_buffer=self.chunksize
        )
    
    def read_chunk(self):
        """读取一个音频块"""
        if self.stream:
            try:
                data = self.stream.read(self.chunksize, exception_on_overflow=False)
                return data
            except Exception as e:
                print(f"读取音频数据失败: {e}")
                return None
        return None
    
    def stop_stream(self):
        """停止音频流"""
        if self.stream:
            self.stream.stop_stream()
            self.stream.close()

class WirelessMicrophone:
    """无线话筒采集（基于sounddevice，支持蓝牙）"""
    
    def __init__(self, rate=16000, chunksize=1024):
        self.rate = rate
        self.chunksize = chunksize
        self.buffer = []
        self.lock = threading.Lock()
        self.callback_running = False
        
    def _audio_callback(self, indata, frames, time, status):
        """音频回调函数"""
        if status:
            print(f"音频流状态: {status}")
        with self.lock:
            self.buffer.append(indata.copy())
    
    def start_stream(self):
        """开始音频流"""
        self.buffer = []
        self.callback_running = True
        
        # 查找无线音频设备
        devices = sd.query_devices()
        wireless_device = None
        for i, dev in enumerate(devices):
            if 'bluetooth' in dev['name'].lower() or 'wireless' in dev['name'].lower():
                wireless_device = i
                break
        
        if wireless_device is None:
            print("警告: 未找到无线音频设备，使用默认设备")
            wireless_device = sd.default.device[0]
        
        # 启动音频流
        self.stream = sd.InputStream(
            device=wireless_device,
            samplerate=self.rate,
            channels=1,
            callback=self._audio_callback,
            blocksize=self.chunksize
        )
        self.stream.start()
    
    def read_chunk(self):
        """读取一个音频块"""
        with self.lock:
            if len(self.buffer) > 0:
                # 合并缓冲区中的数据
                if len(self.buffer) > 1:
                    data = np.concatenate(self.buffer, axis=0)
                else:
                    data = self.buffer[0]
                self.buffer = []
                
                # 转换为字节数据
                audio_bytes = (data * 32767).astype(np.int16).tobytes()
                return audio_bytes
        return None
    
    def stop_stream(self):
        """停止音频流"""
        if hasattr(self, 'stream') and self.stream:
            self.stream.stop()
            self.stream.close()
        self.callback_running = False

4.3 语音活动检测：vad.py

import numpy as np
import webrtcvad
import collections

class VADetector:
    """语音活动检测器"""
    
    def __init__(self, mode=3, sample_rate=16000):
        """
        :param mode: 0-3，越大越激进（3最严格）
        :param sample_rate: 采样率，支持8000, 16000, 32000, 48000
        """
        self.vad = webrtcvad.Vad(mode)
        self.sample_rate = sample_rate
        self.frame_duration = 30  # 毫秒
        self.frame_size = int(sample_rate * self.frame_duration / 1000)
        
    def is_speech(self, audio_bytes):
        """检测音频中是否包含语音"""
        if len(audio_bytes) < self.frame_size * 2:
            return False
        
        # 将音频分割成帧
        frames = self._frame_generator(audio_bytes)
        
        # 统计语音帧比例
        speech_frames = 0
        total_frames = 0
        
        for frame in frames:
            is_speech = self.vad.is_speech(frame, self.sample_rate)
            if is_speech:
                speech_frames += 1
            total_frames += 1
        
        # 如果超过30%的帧是语音，则认为包含语音
        return total_frames > 0 and (speech_frames / total_frames) > 0.3
    
    def _frame_generator(self, audio_bytes):
        """生成音频帧"""
        n = len(audio_bytes)
        offset = 0
        
        while offset + self.frame_size * 2 <= n:
            yield audio_bytes[offset:offset + self.frame_size * 2]
            offset += self.frame_size * 2

4.4 语音识别模块：asr.py

import speech_recognition as sr
import whisper
import io
import wave

class ASREngine:
    """语音识别引擎（支持多种后端）"""
    
    def __init__(self, engine_type='google'):
        """
        :param engine_type: 'google'|'whisper'|'baidu'|'azure'
        """
        self.engine_type = engine_type
        self.recognizer = sr.Recognizer()
        
        if engine_type == 'whisper':
            # 加载Whisper模型（首次使用会自动下载）
            self.whisper_model = whisper.load_model("base")
    
    def recognize(self, audio_bytes):
        """识别音频数据"""
        if not audio_bytes:
            return None
        
        try:
            if self.engine_type == 'google':
                return self._recognize_google(audio_bytes)
            elif self.engine_type == 'whisper':
                return self._recognize_whisper(audio_bytes)
            elif self.engine_type == 'baidu':
                return self._recognize_baidu(audio_bytes)
            else:
                #