STT代码向树莓派4B发送转录文本时遇500序列化请求异常

核心错误原因 #

你遇到的Error: 500 Exception serializing request!是因为在Google Cloud Speech的StreamingRecognizeRequest中添加了未定义的location参数，该请求对象仅接受官方规定的字段（如audio_content），额外参数会导致序列化失败。同时，transcript变量在生成请求的阶段还未被识别出来，属于提前引用未定义变量。

第一步：修复STT请求序列化错误 #

修改笔记本端代码中生成语音识别请求的部分，移除非法参数并修正请求类的引用：

原错误代码：

requests = (
    stream.StreamingRecognizeRequest(audio_content=content, location=transcript)
    for content in audio_generator
)

修正后：

requests_list = (
    speech.StreamingRecognizeRequest(audio_content=content)
    for content in audio_generator
)

注：需使用导入的speech模块中的StreamingRecognizeRequest类，而非自定义的MicrophoneStream实例。

第二步：优化端到端流程（满足语音输入→STT→树莓派查天气→返回→TTS播报） #

当前代码还存在两个关键问题：

1. 树莓派端app.run()是阻塞函数，后续的while True循环永远不会执行
2. 未实现树莓派返回天气信息给笔记本、笔记本TTS播报的逻辑

树莓派端修改（接收地点后直接返回天气） #

修改receive_location接口，接收地点后立即查询天气并返回结果，同时保留全局变量记录当前地点：

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

current_location = 'Tarragona'  # 初始地点

def get_weather(location):
    api_key = '<你的OpenWeatherMap API密钥>'
    base_url = 'https://api.openweathermap.org/data/2.5/weather'
    params = {'q': location, 'appid': api_key, 'units': 'metric'}
    response = requests.get(base_url, params=params)
    if response.status_code == 200:
        data = response.json()
        temperature = data['main']['temp']
        description = data['weather'][0]['description']
        return {
            'status': 'success',
            'weather': f'The weather in {location} is {description} with a temperature of {temperature} degrees Celsius'
        }
    else:
        return {'status': 'error', 'weather': 'Weather data not available'}

@app.route('/location', methods=['POST'])
def receive_location():
    global current_location
    data = request.get_json()
    location = data.get('location')
    if not location:
        return jsonify({'status': 'error', 'message': 'Invalid location data'}), 400
    current_location = location
    weather_result = get_weather(location)
    return jsonify(weather_result), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=True)  # 开启debug方便调试

笔记本端修改（添加TTS播报逻辑） #

1. 安装TTS库：pip install pyttsx3
2. 修改send_location_to_pi函数，接收树莓派返回的天气信息并播报：

import queue
import re
import sys
import os
from google.cloud import speech
import pyaudio
import requests
import pyttsx3

# 初始化TTS引擎
tts_engine = pyttsx3.init()

# Set your Google Cloud Speech API credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C://Users//danim//OneDrive//Desktop//programming//workpls-401909-66721bf5ced8.json"

# Audio recording parameters
RATE = 48000
CHUNK = int(RATE / 10)  # 100ms

class MicrophoneStream:
    """Opens a recording stream as a generator yielding the audio chunks."""
    def __init__(self: object, rate: int = RATE, chunk: int = CHUNK) -> None:
        self._rate = rate
        self._chunk = chunk
        self._buff = queue.Queue()
        self.closed = True

    def __enter__(self: object) -> object:
        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: object, type: object, value: object, traceback: object) -> None:
        self._audio_stream.stop_stream()
        self._audio_stream.close()
        self.closed = True
        self._buff.put(None)
        self._audio_interface.terminate()

    def _fill_buffer(self: object, in_data: object, frame_count: int, time_info: object, status_flags: object) -> object:
        self._buff.put(in_data)
        return None, pyaudio.paContinue

    def generator(self: object) -> object:
        while not self.closed:
            chunk = self._buff.get()
            if chunk is None:
                return
            data = [chunk]
            while True:
                try:
                    chunk = self._buff.get(block=False)
                    if chunk is None:
                        return
                    data.append(chunk)
                except queue.Empty:
                    break
            yield b"".join(data)

def send_location_to_pi(location):
    pi_url = 'http://192.168.1.128:8080/location'  # 移除IP地址的<>符号
    data = {'location': location}
    try:
        response = requests.post(pi_url, json=data)
        if response.status_code == 200:
            result = response.json()
            print(f"Location sent to Pi successfully. Weather: {result['weather']}")
            # 播报天气信息
            tts_engine.say(result['weather'])
            tts_engine.runAndWait()
        else:
            print(f"Failed to send location to Pi. Status code: {response.status_code}")
    except requests.exceptions.RequestException as e:
        print(f"Connection error: {e}")

def listen_print_loop(responses: object) -> str:
    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(exit|quit)\b", transcript, re.I):
                print("Exiting..")
                break
            # 仅在识别结果最终确定时发送地点
            send_location_to_pi(transcript)
            num_chars_printed = 0

if __name__ == "__main__":
    try:
        audio_client = speech.SpeechClient()
        config = speech.RecognitionConfig(
            encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
            sample_rate_hertz=RATE,
            language_code='en-US',
        )
        with MicrophoneStream(RATE, CHUNK) as stream:
            audio_generator = stream.generator()
            requests_list = (
                speech.StreamingRecognizeRequest(audio_content=content)
                for content in audio_generator
            )
            responses = audio_client.streaming_recognize(config=config, requests=requests_list)
            listen_print_loop(responses)
    except Exception as e:
        print(f"Error: {e}")

额外注意事项 #

1. 确保笔记本和树莓派在同一局域网内，树莓派的IP地址正确（移除URL中的<>符号）
2. 替换树莓派代码中的OpenWeatherMap API密钥
3. 若pyttsx3无法正常工作，可尝试安装pywin32（Windows平台）或使用其他TTS库

内容的提问来源于stack exchange，提问作者Daniel Masso Pardo