### Google Speech-to-Text API 结合唤醒词系统的实现 为了实现在Python中通过Google Speech-to-Text API来完成带有唤醒功能的语音识别，主要依赖于两个部分的工作：一是设置好能够监听特定唤醒词从而激活录音并发送给云端API的服务；二是处理来自该服务的数据流以获取文本输出。 对于前者——即创建一个有效的前端机制用于捕捉用户的音频输入直到检测到预设好的唤醒短语为止，在此之前保持静默状态不上传任何数据至服务器端。这通常涉及到使用专门设计用来做关键词侦测的小型嵌入式设备或是软件库[^1]。 一旦成功捕获到了指定的关键字，则开启麦克风录制一段完整的命令句，并将其编码成适合传输的形式（比如FLAC格式），之后利用`google-cloud-speech` SDK所提供的StreamingRecognize方法来进行在线解码操作： ```python from google.cloud import speech_v1p1beta1 as speech import pyaudio import queue import threading class MicrophoneStream(object): """Opens a recording stream as a generator yielding the audio chunks.""" def __init__(self, rate, chunk): self._rate = rate self._chunk = chunk # Create a thread-safe buffer of audio data self._buff = queue.Queue() self.closed = True def __enter__(self): self._audio_interface = pyaudio.PyAudio() self._audio_stream = self._audio_interface.open( format=pyaudio.paInt16, channels=1, rate=self._rate, input=True, frames_per_buffer=self._chunk, stream_callback=self._fill_buffer, ) self.closed = False return self def __exit__(self, type_, value, traceback): self._audio_stream.stop_stream() self._audio_stream.close() self.closed = True self._buff.put(None) self._audio_interface.terminate() def _fill_buffer(self, in_data, frame_count, time_info, status_flags): """Continuously collect data from the audio stream""" self._buff.put(in_data) return None, pyaudio.paContinue def generator(self): while not self.closed: chunk = self._buff.get() if chunk is None: return data = [chunk] yield b"".join(data) def listen_print_loop(responses): num_chars_printed = 0 for response in responses: if not response.results: continue result = response.results[0] if not result.alternatives: continue transcript = result.alternatives[0].transcript overwrite_chars = " " * (num_chars_printed - len(transcript)) if not result.is_final: sys.stdout.write(transcript + overwrite_chars + "\r") sys.stdout.flush() num_chars_printed = len(transcript) else: print(transcript + overwrite_chars) if re.search(r"\b(退出|停止)\b", transcript, re.I): print("Exiting..") break num_chars_printed = 0 client = speech.SpeechClient() config = speech.RecognitionConfig( encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16, sample_rate_hertz=RATE, language_code="cmn-Hans-CN", ) streaming_config = speech.StreamingRecognitionConfig(config=config) with MicrophoneStream(RATE, CHUNK) as stream: audio_generator = stream.generator() requests = ( speech.StreamingRecognizeRequest(audio_content=content) for content in audio_generator ) responses = client.streaming_recognize(streaming_config, requests) listen_print_loop(responses) ``` 值得注意的是，上述代码片段仅展示了如何连接到Google Cloud Speech API进行持续性的语音转文字工作流程的一部分。至于具体的唤醒词检测环节，则可能需要额外集成第三方解决方案或者是基于ESP32这样的硬件平台上的固件特性来达成更高效的本地化处理效果[^2]。