deepseek_voice_service.py
 端云协同语音交互服务（简化版）
 适用于ARM Cortex-M系列处理器 + 云推理的混合架构

import pyaudio
import requests
import json
import threading
from queue import Queue

 ========== 配置层 ==========
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
API_KEY = "your_api_key_here"
WAKE_WORD = "小猫"   唤醒词
SAMPLE_RATE = 16000
CHUNK_SIZE = 1024

 ========== 模块1：本地唤醒检测 ==========
class WakeWordDetector:
    """本地运行轻量级KWS模型，功耗<5mW"""
    def __init__(self, wake_word=WAKE_WORD):
        self.wake_word = wake_word
        self.is_awake = False
         实际生产环境使用TensorFlow Lite部署KWS模型
         此处简化为关键词匹配示意
        
    def detect(self, audio_chunk):
         实际调用：model.predict(audio_chunk) -> confidence
         简化为：检测到"小猫"字样即唤醒
        return self.is_awake

 ========== 模块2：语音识别与合成 ==========
class VoiceProcessor:
    """负责ASR语音转文本 + TTS文本转语音"""
    
    def asr(self, audio_data):
        """语音转文本：调用云端ASR服务"""
         实际使用：DeepSeek语音识别API 或 第三方ASR服务
        response = requests.post(
            "https://api.deepseek.com/v1/audio/transcriptions",
            files={"audio": audio_data},
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        return response.json().get("text", "")
    
    def tts(self, text):
        """文本转语音：返回音频字节流"""
        response = requests.post(
            "https://api.deepseek.com/v1/audio/speech",
            json={"input": text, "voice": "zh-female"},
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        return response.content

 ========== 模块3：大模型语义理解 ==========
class LLMInference:
    """DeepSeek大模型推理封装"""
    
    def chat(self, user_query, conversation_history):
        """调用DeepSeek进行对话推理"""
        messages = [
            {"role": "system", "content": "你是一个温暖的AI助手，帮助用户播放音乐、回答问题。"},
            conversation_history,
            {"role": "user", "content": user_query}
        ]
        
        payload = {
            "model": "deepseek-chat",
            "messages": messages,
            "temperature": 0.7,       控制回答的随机性
            "max_tokens": 500
        }
        
        response = requests.post(
            DEEPSEEK_API_URL,
            headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
            json=payload
        )
        
        return response.json()["choices"][0]["message"]["content"]

 ========== 主服务：端云协同编排 ==========
class CatAIHelper:
    """猫王AI助手核心服务——端云协同编排器"""
    
    def __init__(self):
        self.wake_word = WakeWordDetector()
        self.voice = VoiceProcessor()
        self.llm = LLMInference()
        self.history = []   多轮对话上下文
        self.audio_queue = Queue()
        
    def handle_conversation(self, user_text):
        """处理用户输入，生成回复"""
         Step 3: 大模型推理
        response_text = self.llm.chat(user_text, self.history)
        
         Step 4: 更新对话历史（用于多轮对话）
        self.history.append({"role": "user", "content": user_text})
        self.history.append({"role": "assistant", "content": response_text})
        
         Step 5: TTS语音合成
        audio_response = self.voice.tts(response_text)
        
        return audio_response, response_text
    
    def run(self):
        """主循环：持续监听 -> 唤醒检测 -> 语音采集 -> 云端推理 -> 语音输出"""
        print("猫王AI助手已启动，说‘小猫’唤醒...")
        
         Step 1: 初始化音频流
        audio = pyaudio.PyAudio()
        stream = audio.open(format=pyaudio.paInt16, channels=1,
                            rate=SAMPLE_RATE, input=True, frames_per_buffer=CHUNK_SIZE)
        
        while True:
             Step 2: 本地唤醒检测（持续低功耗监听）
            audio_chunk = stream.read(CHUNK_SIZE)
            if self.wake_word.detect(audio_chunk):
                print("唤醒成功，请说话...")
                
                 采集3秒音频
                user_audio = stream.read(3  SAMPLE_RATE)
                
                 Step 3: ASR语音转文本
                user_text = self.voice.asr(user_audio)
                print(f"用户说: {user_text}")
                
                 Step 4-5: 大模型推理 + TTS合成
                audio_resp, text_resp = self.handle_conversation(user_text)
                print(f"AI回复: {text_resp}")
                
                 Step 6: 播放音频响应
                 speaker.play(audio_resp)   实际需调用硬件音频输出

if __name__ == "__main__":
    helper = CatAIHelper()
    helper.run()