1. 项目概述

宇树机器人G1是一款功能强大的智能机器人平台，支持丰富的二次开发接口。本文将详细介绍如何为G1机器人实现完整的语音对话功能，包括：

• 语音打断：在机器人说话时随时打断
• 停止功能：立即停止当前语音输出
• 待命模式：进入低功耗监听状态
• 激活唤醒：通过特定唤醒词激活机器人
• 多话筒支持：兼容有线话筒和无线蓝牙话筒

2. 环境准备

2.1 硬件要求

• 宇树机器人G1
• USB有线话筒或蓝牙无线话筒
• 扬声器或耳机

2.2 软件依赖

# 安装必要的Python库
pip install pyaudio
pip install speechrecognition
pip install pydub
pip install pyttsx3
pip install pyserial
pip install sounddevice
pip install numpy

3. 核心功能实现

3.1 语音识别模块

import speech_recognition as sr
import threading
import queue
import time

class VoiceRecognizer:
    def __init__(self, mic_type='usb'):
        """
        初始化语音识别器
        :param mic_type: 'usb' 或 'bluetooth'
        """
        self.recognizer = sr.Recognizer()
        self.mic_type = mic_type
        self.is_listening = False
        self.interrupt_flag = False
        self.audio_queue = queue.Queue()
        self.stop_event = threading.Event()
        
        # 配置麦克风
        if mic_type == 'usb':
            self.mic = sr.Microphone(device_index=0)
        else:
            # 蓝牙麦克风配置
            self.mic = sr.Microphone(device_index=self._find_bluetooth_mic())
    
    def _find_bluetooth_mic(self):
        """查找蓝牙麦克风设备索引"""
        mic_list = sr.Microphone.list_microphone_names()
        for idx, name in enumerate(mic_list):
            if 'bluetooth' in name.lower() or 'bt' in name.lower():
                return idx
        return 0  # 默认使用第一个麦克风
    
    def start_listening(self, callback):
        """开始监听语音输入"""
        self.is_listening = True
        self.stop_event.clear()
        
        def listen_thread():
            with self.mic as source:
                self.recognizer.adjust_for_ambient_noise(source, duration=1)
                
                while self.is_listening and not self.stop_event.is_set():
                    if self.interrupt_flag:
                        time.sleep(0.1)
                        continue
                    
                    try:
                        print("🎤 正在监听...")
                        audio = self.recognizer.listen(
                            source, 
                            timeout=3, 
                            phrase_time_limit=5
                        )
                        
                        # 将音频数据放入队列
                        self.audio_queue.put(audio)
                        
                        # 在新线程中处理识别
                        threading.Thread(
                            target=self._process_audio,
                            args=(audio, callback),
                            daemon=True
                        ).start()
                        
                    except sr.WaitTimeoutError:
                        continue
                    except Exception as e:
                        print(f"监听错误: {e}")
        
        threading.Thread(target=listen_thread, daemon=True).start()
    
    def _process_audio(self, audio, callback):
        """处理音频识别"""
        try:
            text = self.recognizer.recognize_google(audio, language='zh-CN')
            if text and callback:
                callback(text)
        except sr.UnknownValueError:
            print("无法识别语音")
        except sr.RequestError as e:
            print(f"语音识别服务错误: {e}")
    
    def interrupt(self):
        """打断当前语音识别"""
        self.interrupt_flag = True
        time.sleep(0.5)  # 短暂暂停
        self.interrupt_flag = False
        print("⏸️ 语音识别已打断")
    
    def stop(self):
        """停止语音识别"""
        self.is_listening = False
        self.stop_event.set()
        print("⏹️ 语音识别已停止")
    
    def standby(self):
        """进入待命模式"""
        self.is_listening = False
        print("💤 进入待命模式")
    
    def activate(self, wake_word="小宇"):
        """激活机器人"""
        def check_wake_word(text):
            if wake_word in text:
                self.is_listening = True
                print(f"🔊 已激活，唤醒词: {wake_word}")
                return True
            return False
        return check_wake_word

3.2 语音合成模块

import pyttsx3
import threading

class VoiceSynthesizer:
    def __init__(self):
        self.engine = pyttsx3.init()
        self.is_speaking = False
        self.stop_speaking = False
        self.current_thread = None
        
        # 配置语音参数
        self.engine.setProperty('rate', 150)  # 语速
        self.engine.setProperty('volume', 0.9)  # 音量
        voices = self.engine.getProperty('voices')
        for voice in voices:
            if 'chinese' in voice.name.lower():
                self.engine.setProperty('voice', voice.id)
                break
    
    def speak(self, text, callback=None):
        """语音合成输出"""
        if self.is_speaking:
            print("⚠️ 正在说话，请等待或打断")
            return False
        
        self.is_speaking = True
        self.stop_speaking = False
        
        def speak_thread():
            try:
                # 注册回调
                def on_start(name):
                    print(f"开始说话: {name}")
                
                def on_word(name, location, length):
                    if self.stop_speaking:
                        self.engine.stop()
                
                def on_end(name, completed):
                    self.is_speaking = False
                    print(f"说话结束，完成: {completed}")
                    if callback:
                        callback(completed)
                
                self.engine.connect('started-utterance', on_start)
                self.engine.connect('started-word', on_word)
                self.engine.connect('finished-utterance', on_end)
                
                # 开始说话
                self.engine.say(text)
                self.engine.runAndWait()
                
            except Exception as e:
                print(f"语音合成错误: {e}")
                self.is_speaking = False
        
        self.current_thread = threading.Thread(target=speak_thread, daemon=True)
        self.current_thread.start()
        return True
    
    def interrupt_speech(self):
        """打断当前语音输出"""
        if self.is_speaking:
            self.stop_speaking = True
            self.engine.stop()
            self.is_speaking = False
            print("⏸️ 语音输出已打断")
            return True
        return False
    
    def stop_speech(self):
        """停止语音输出"""
        self.interrupt_speech()
        print("⏹️ 语音输出已停止")
    
    def set_voice_parameters(self, rate=None, volume=None, voice_id=None):
        """设置语音参数"""
        if rate:
            self.engine.setProperty('rate', rate)
        if volume:
            self.engine.setProperty('volume', volume)
        if voice_id:
            self.engine.setProperty('voice', voice_id)

3.3 主控制模块

import json
import os
from datetime import datetime

class G1VoiceController:
    def __init__(self, config_file='config.json'):
        self.recognizer = None
        self.synthesizer = VoiceSynthesizer()
        self.config = self._load_config(config_file)
        self.conversation_history = []
        self.current_state = 'standby'  # standby, listening, speaking
        
        # 状态映射
        self.state_actions = {
            'standby': self._enter_standby,
            'listening': self._enter_listening,
            'speaking': self._enter_speaking
        }
    
    def _load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            'mic_type': 'usb',
            'wake_word': '小宇',
            'response_timeout': 10,
            'max_history': 50,
            'log_file': 'voice_log.json',
            'bluetooth_mac': None
        }
        
        if os.path.exists(config_file):
            with open(config_file, 'r', encoding='utf-8') as f:
                user_config = json.load(f)
                default_config.update(user_config)
        
        return default_config
    
    def initialize(self):
        """初始化系统"""
        print("🤖 初始化宇树G1语音控制系统...")
        
        # 初始化语音识别器
        self.recognizer = VoiceRecognizer(self.config['mic_type'])
        
        # 设置唤醒词检测
        wake_checker = self.recognizer.activate(self.config['wake_word'])
        
        def recognition_callback(text):
            """语音识别回调"""
            print(f"识别到: {text}")
            self._log_conversation('user', text)
            
            # 检查唤醒词
            if wake_checker(text):
                self.wake_up()
                return
            
            # 处理用户输入
            response = self._process_input(text)
            if response:
                self.speak(response)
        
        # 开始监听
        self.recognizer.start_listening(recognition_callback)
        print("✅ 系统初始化完成")
    
    def _process_input(self, text):
        """处理用户输入并生成响应"""
        # 这里可以集成AI对话模型（如ChatGPT、文心一言等）
        # 示例：简单的命令响应
        responses = {
            '停止': '好的，已停止',
            '待命': '进入待命模式',
            '激活': '已激活，请说',
            '时间': f'现在时间是 {datetime.now().strftime("%H:%M")}',
            '日期': f'今天是 {datetime.now().strftime("%Y年%m月%d日")}'
        }
        
        for key in responses:
            if key in text:
                return responses[key]
        
        # 默认响应
        return f"我听到你说：{text}"
    
    def speak(self, text):
        """语音输出"""
        if self.current_state == 'speaking':
            print("⚠️ 正在说话，请稍候")
            return False
        
        self._change_state('speaking')
        
        def speak_callback(completed):
            self._log_conversation('robot', text)
            if completed:
                self._change_state('listening')
            else:
                self._change_state('standby')
        
        return self.synthesizer.speak(text, speak_callback)
    
    def interrupt(self):
        """打断当前操作"""
        if self.current_state == 'speaking':
            self.synthesizer.interrupt_speech()
            print("⏸️ 已打断语音输出")
        elif self.recognizer:
            self.recognizer.interrupt()
            print("⏸️ 已打断语音识别")
        
        self._change_state('listening')
        return True
    
    def stop(self):
        """停止所有语音活动"""
        self.synthesizer.stop_speech()
        if self.recognizer:
            self.recognizer.stop()
        
        self._change_state('standby')
        print("⏹️ 已停止所有语音活动")
        return True
    
    def standby(self):
        """进入待命模式"""
        if self.recognizer:
            self.recognizer.standby()
        
        self._change_state('standby')
        print("💤 已进入待命模式")
        return True
    
    def wake_up(self):
        """从待命模式唤醒"""
        if self.recognizer:
            self.recognizer.is_listening = True
        
        self._change_state('listening')
        print("🔊 已唤醒，正在监听...")
        return True
    
    def _change_state(self, new_state):
        """改变系统状态"""
        old_state = self.current_state
        self.current_state = new_state
        
        if new_state in self.state_actions:
            self.state_actions[new_state]()
        
        print(f"🔄 状态变更: {old_state} -> {new_state}")
    
    def _enter_standby(self):
        """进入待命状态"""
        pass
    
    def _enter_listening(self):
        """进入监听状态"""
        pass
    
    def _enter_speaking(self):
        """进入说话状态"""
        pass
    
    def _log_conversation(self, role, content):
        """记录对话历史"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'role': role,
            'content': content
        }
        
        self.conversation_history.append(entry)
        
        # 限制历史记录长度
        if len(self.conversation_history) > self.config['max_history']:
            self.conversation_history = self.conversation_history[-self.config['max_history']:]
        
        # 保存到文件
        self._save_log()
    
    def _save_log(self):
        """保存日志到文件"""
        try:
            with open(self.config['log_file'], 'w', encoding='utf-8') as f:
                json.dump(self.conversation_history, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"保存日志失败: {e}")
    
    def get_status(self):
        """获取系统状态"""
        return {
            'state': self.current_state,
            'mic_type': self.config['mic_type'],
            'is_speaking': self.synthesizer.is_speaking,
            'is_listening': self.recognizer.is_listening if self.recognizer else False,
            'history_count': len(self.conversation_history)
        }

3.4 蓝牙话筒支持模块

import bluetooth
import socket

class BluetoothMicrophone:
    def __init__(self, mac_address=None):
        self.mac_address = mac_address
        self.socket = None
        self.connected = False
    
    def discover_devices(self):
        """发现附近的蓝牙设备"""
        print("🔍 搜索蓝牙设备...")
        devices = bluetooth.discover_devices(lookup_names=True)
        
        mic_devices = []
        for addr, name in devices:
            if 'mic' in name.lower() or 'headset' in name.lower():
                mic_devices.append({'address': addr, 'name': name})
                print(f"找到话筒: {name} ({addr})")
        
        return mic_devices
    
    def connect(self, mac_address=None):
        """连接到蓝牙话筒"""
        if mac_address:
            self.mac_address = mac_address
        
        if not self.mac_address:
            print("❌ 未指定蓝牙MAC地址")
            return False
        
        try:
            # 创建RFCOMM socket
            self.socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
            
            # 连接到设备
            self.socket.connect((self.mac_address, 1))
            self.connected = True
            
            print(f"✅ 已连接到蓝牙话筒: {self.mac_address}")
            return True
            
        except Exception as e:
            print(f"❌ 连接失败: {e}")
            self.connected = False
            return False
    
    def disconnect(self):
        """断开蓝牙连接"""
        if self.socket:
            self.socket.close()
            self.socket = None
        
        self.connected = False
        print("📴 蓝牙已断开")
    
    def send_audio_config(self, sample_rate=16000, channels=1):
        """发送音频配置到蓝牙设备"""
        if not self.connected:
            print("❌ 未连接蓝牙设备")
            return False
        
        try:
            config = {
                'sample_rate': sample_rate,
                'channels': channels,
                'format': 'int16'
            }
            
            config_str = json.dumps(config)
            self.socket.send(config_str.encode())
            
            print(f"✅ 音频配置已发送: {config}")
            return True
            
        except Exception as e:
            print(f"❌ 发送配置失败: {e}")
            return False

4. 完整示例代码

4.1 主程序入口

import sys
import signal

def main():
    """主程序"""
    print("=" * 50)
    print("宇树机器人G1 - 语音对话系统")
    print("=" * 50)
    
    # 创建控制器
    controller = G1VoiceController('config.json')
    
    # 初始化系统
    controller.initialize()
    
    # 注册信号处理
    def signal_handler(sig, frame):
        print("\n\n👋 正在关闭系统...")
        controller.stop()
        sys.exit(0)
    
    signal.signal(signal.SIGINT, signal_handler)
    
    # 命令行交互
    print("\n📝 可用命令:")
    print("  speak <文本>  - 机器人说话")
    print("  interrupt    - 打断当前语音")
    print("  stop         - 停止所有语音活动")
    print("  standby      - 进入待命模式")
    print("  wake         - 唤醒机器人")
    print("  status       - 查看系统状态")
    print("  exit         - 退出程序")
    print("\n💬 开始语音交互...\n")
    
    while True:
        try:
            cmd = input(">>> ").strip().lower()
            
            if cmd.startswith('speak '):
                text = cmd[6:]
                controller.speak(text)
                
            elif cmd == 'interrupt':
                controller.interrupt()
                
            elif cmd == 'stop':
                controller.stop()
                
            elif cmd == 'standby':
                controller.standby()
                
            elif cmd == 'wake':
                controller.wake_up()
                
            elif cmd == 'status':
                status = controller.get_status()
                print(f"状态: {status}")
                
            elif cmd == 'exit':
                print("👋 再见！")
                controller.stop()
                break
                
            else:
                print("❌ 未知命令")
                
        except KeyboardInterrupt:
            signal_handler(None, None)
        except Exception as e:
            print(f"❌ 错误: {e}