科大讯飞AIUI 3.0：多模态交互平台+机器人超脑的技术底座深度解析

"""
AIUI 3.0 双流VAD（Voice Activity Detection）引擎核心实现
"""
import numpy as np
from typing import Optional, Tuple, List
from dataclasses import dataclass
from enum import Enum

class SystemState(Enum):
    LISTENING = "listening"    # 聆听状态
    SPEAKING = "speaking"      # 播报状态
    IDLE = "idle"              # 空闲状态

@dataclass
class AudioFrame:
    """音频帧"""
    samples: np.ndarray  # 音频样本
    timestamp: float     # 时间戳
    snr_db: float        # 信噪比(dB)
    is_voice: bool       # 是否包含人声

class DualStreamVAD:
    """
    双流VAD引擎
    
    核心思想：同时维护两个VAD流
    - 上行流(Upstream VAD)：检测用户说话
    - 下行流(Downstream VAD)：检测设备播报
    通过双流状态机判断，实现自然的全双工交互
    """
    
    def __init__(self, 
                 sample_rate: int = 16000,
                 voice_threshold: float = 0.6,
                 silence_timeout_ms: int = 800,
                 barge_in_threshold: float = 0.3):
        
        self.sample_rate = sample_rate
        self.voice_threshold = voice_threshold
        self.silence_timeout_ms = silence_timeout_ms
        self.barge_in_threshold = barge_in_threshold
        
        # 状态
        self.system_state = SystemState.IDLE
        self.upstream_state = False  # 用户是否在说话
        self.downstream_state = False  # 设备是否在播报
        
        # 历史缓冲区
        self.upstream_buffer: List[AudioFrame] = []
        self.silence_frames = 0
        self.frames_per_ms = sample_rate // 1000
        
        # 统计指标
        self.total_interruptions = 0
        self.false_positive = 0
        self.missed_barge_ins = 0
        
    def compute_voice_probability(self, frame: AudioFrame) -> float:
        """
        计算语音概率
        基于能量+频谱特征+信噪比的综合判断
        """
        # 1. 短时能量
        energy = np.sum(frame.samples ** 2) / len(frame.samples)
        energy_norm = min(energy / 0.01, 1.0)
        
        # 2. 过零率（区分语音和噪声）
        zero_crossings = np.sum(np.abs(np.diff(np.sign(frame.samples)))) / (2 * len(frame.samples))
        zcr_score = 1.0 - abs(zero_crossings - 0.1) / 0.1  # 语音的ZCR通常在0.05-0.15
        
        # 3. SNR加权
        snr_weight = min(frame.snr_db / 20.0, 1.0)  # 高SNR时更信任
        
        # 综合评分
        voice_prob = 0.5 * energy_norm + 0.3 * zcr_score + 0.2 * snr_weight
        return np.clip(voice_prob, 0.0, 1.0)
    
    def process_frame(self, frame: AudioFrame) -> Tuple[bool, SystemState]:
        """
        处理音频帧，返回(是否打断, 新状态)
        """
        # 计算语音概率
        voice_prob = self.compute_voice_probability(frame)
        is_voice = voice_prob > self.voice_threshold
        
        self.upstream_state = is_voice
        self.upstream_buffer.append(frame)
        
        # 静音超时检测
        if not is_voice:
            self.silence_frames += 1
        else:
            self.silence_frames = 0
        
        silence_ms = self.silence_frames * 10  # 假设每帧10ms
        is_silence_timeout = silence_ms > self.silence_timeout_ms
        
        # 状态机转换
        new_state = self.system_state
        should_interrupt = False
        
        if self.system_state == SystemState.SPEAKING:
            # 设备正在播报时，检测用户插话（Barge-in）
            if is_voice and voice_prob > self.barge_in_threshold:
                # 判断是否为真正的插话（非回声）
                if self._is_genuine_barge_in(frame):
                    should_interrupt = True
                    new_state = SystemState.LISTENING
                    self.total_interruptions += 1
            
        elif self.system_state == SystemState.LISTENING:
            # 聆听状态下，检测用户是否说完
            if is_silence_timeout and len(self.upstream_buffer) > 5:
                # 用户说完，准备回复
                new_state = SystemState.SPEAKING
                
        elif self.system_state == SystemState.IDLE:
            # 空闲状态，检测唤醒
            if is_voice and voice_prob > self.voice_threshold:
                new_state = SystemState.LISTENING
        
        self.system_state = new_state
        
        # 清理历史缓冲区
        if len(self.upstream_buffer) > 100:
            self.upstream_buffer = self.upstream_buffer[-50:]
        
        return should_interrupt, new_state
    
    def _is_genuine_barge_in(self, frame: AudioFrame) -> bool:
        """
        判断是否为真正的插话（而非回声或环境噪声）
        使用声学回声消除(AEC)后的残差信号进行判断
        """
        # 简化实现：实际部署中会使用AEC模块的残差信号
        # 这里使用能量比 + 谐波检测
        harmonic_score = self._detect_harmonics(frame.samples)
        return harmonic_score > 0.5
    
    @staticmethod
    def _detect_harmonics(samples: np.ndarray) -> float:
        """
        检测谐波分量（语音的典型特征）
        使用自相关函数
        """
        if len(samples) < 320:  # 至少20ms
            return 0.0
        
        # 自相关计算
        autocorr = np.correlate(samples, samples, mode='full')
        autocorr = autocorr[len(autocorr)//2:]
        
        # 基频检测（80-400Hz范围）
        min_lag = int(16000 / 400)  # 40 samples
        max_lag = int(16000 / 80)   # 200 samples
        
        if len(autocorr) <= max_lag:
            return 0.0
        
        peak_range = autocorr[min_lag:max_lag]
        if np.max(peak_range) > 0:
            return np.max(peak_range) / autocorr[0]
        return 0.0


# 模拟测试
def test_dual_stream_vad():
    """模拟全双工VAD场景测试"""
    vad = DualStreamVAD()
    
    # 模拟场景：设备正在播报，用户插话
    test_frames = []
    
    # 前100帧：设备播报（下行流激活）
    for i in range(100):
        noise = np.random.randn(320) * 0.001
        test_frames.append(AudioFrame(
            samples=noise, timestamp=i*0.01,
            snr_db=5, is_voice=False
        ))
    
    # 第101-120帧：用户插话（上行流激活，高SNR）
    for i in range(100, 120):
        voice = np.random.randn(320) * 0.05
        test_frames.append(AudioFrame(
            samples=voice, timestamp=i*0.01,
            snr_db=15, is_voice=True
        ))
    
    # 执行测试
    vad.system_state = SystemState.SPEAKING
    interruptions = 0
    
    for frame in test_frames:
        should_interrupt, new_state = vad.process_frame(frame)
        if should_interrupt:
            interruptions += 1
    
    print(f"全双工VAD测试结果:")
    print(f"  总帧数: {len(test_frames)}")
    print(f"  检测到插话次数: {interruptions}")
    print(f"  总插话统计: {vad.total_interruptions}")
    print(f"  误报: {vad.false_positive}")

test_dual_stream_vad()

package wakeup

import (
	"math"
)

// WakeUpEngine 低功耗唤醒引擎
// 目标：100MHz MCU, 140KB SRAM, -5dB SNR下唤醒率97%
type WakeUpEngine struct {
	model       []float32       // 轻量级唤醒模型
	windowSize  int             // 帧长（样本数）
	hopSize     int             // 帧移
	sampleRate  int             // 采样率
	threshold   float32         // 唤醒阈值
	cooldownMs  int             // 唤醒冷却时间(ms)
	lastWakeup  int64           // 上次唤醒时间戳
}

// NewWakeUpEngine 初始化唤醒引擎
func NewWakeUpEngine(sampleRate int) *WakeUpEngine {
	return &WakeUpEngine{
		model:      make([]float32, 256), // 极轻量模型
		windowSize: 320,                  // 20ms @ 16kHz
		hopSize:    160,                  // 10ms帧移
		sampleRate: sampleRate,
		threshold:  0.75,
		cooldownMs: 2000,
	}
}

// ExtractMFCC 提取MFCC特征（定点化实现，无浮点运算）
func (e *WakeUpEngine) ExtractMFCC(samples []int16) []int16 {
	// 1. 预加重
	preEmphasized := make([]int16, len(samples))
	preEmphasized[0] = samples[0]
	for i := 1; i < len(samples); i++ {
		// y[n] = x[n] - 0.97 * x[n-1]
		val := int32(samples[i]) - int32(float32(samples[i-1])*0.97)
		if val > 32767 {
			val = 32767
		} else if val < -32768 {
			val = -32768
		}
		preEmphasized[i] = int16(val)
	}
	
	// 2. 加汉明窗
	windowed := make([]float32, e.windowSize)
	for i := 0; i < e.windowSize; i++ {
		// 使用查表法加速
		hamming := 0.54 - 0.46*float32(math.Cos(2*math.Pi*float64(i)/float64(e.windowSize-1)))
		windowed[i] = float32(preEmphasized[i]) * hamming
	}
	
	// 3. 计算能量（简化MFCC，仅用能量+频谱质心）
	energy := float32(0)
	for _, s := range windowed {
		energy += s * s
	}
	energy = float32(math.Sqrt(float64(energy / float32(e.windowSize))))
	
	// 4. 频谱质心
	fftMag := make([]float32, e.windowSize/2+1)
	// 简化FFT：使用Goertzel算法计算关键频点
	// 实际部署使用定点FFT
	
	// 返回特征向量（定点化）
	features := make([]int16, 13)
	features[0] = int16(energy * 1000) // 能量特征
	return features
}

// Detect 唤醒检测
func (e *WakeUpEngine) Detect(samples []int16) (bool, float32) {
	features := e.ExtractMFCC(samples)
	
	// 简化评分：使用能量作为主要判断
	score := float32(features[0]) / 1000.0
	
	// 归一化到0-1
	score = float32(math.Tanh(float64(score * 5)))
	
	return score > e.threshold, score
}

// OptimizeForRTOS 面向RTOS的优化配置
func (e *WakeUpEngine) OptimizeForRTOS() map[string]interface{} {
	return map[string]interface{}{
		"mcu_freq_mhz":        100,
		"ram_usage_bytes":      140 * 1024,
		"model_rom_bytes":      12 * 1024,
		"feature_rom_bytes":    8 * 1024,
		"inference_per_frame_us": 320, // 每帧推理时间320us
		"frame_size_ms":        10,
		"real_time_factor":     0.032, // 远小于1，实时性足够
		"wakeup_rate_snr_minus_5db": 0.97,
		"false_accept_rate":    0.01,  // 1次/小时
	}
}

"""
MultiTurn语音合成引擎 - 超拟人TTS
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import List, Optional, Dict

class MultiTurnTTS(nn.Module):
    """
    MultiTurn语音合成模型
    
    核心创新：
    1. 副语言标记（paralinguistic tokens）- 笑声、停顿、叹气
    2. 情绪嵌入（emotion embedding）- 开心、惊讶、安慰
    3. Prompt驱动角色生成
    """
    
    def __init__(self, 
                 vocab_size: int = 50000,
                 hidden_dim: int = 512,
                 num_heads: int = 8,
                 num_layers: int = 6,
                 num_speakers: int = 100):
        super().__init__()
        
        # 文本编码
        self.text_embedding = nn.Embedding(vocab_size, hidden_dim)
        self.position_encoding = nn.Parameter(torch.randn(1, 1024, hidden_dim))
        
        # 副语言编码器（Paralinguistic Encoder）
        # 支持：笑声[laugh]、停顿[pause]、叹气[sigh]、喷嚏[sneeze]、清嗓[clearthroat]
        self.paralinguistic_encoder = nn.Parameter(torch.randn(5, hidden_dim))
        
        # 情绪编码器
        self.emotion_encoder = nn.Embedding(8, hidden_dim)  # 8种基础情绪
        
        # 说话人编码器
        self.speaker_encoder = nn.Embedding(num_speakers, hidden_dim)
        
        # Transformer编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=hidden_dim,
            nhead=num_heads,
            dim_feedforward=hidden_dim * 4,
            dropout=0.1,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
        
        # 声学特征预测
        self.mel_projection = nn.Linear(hidden_dim, 80)  # 80-band mel
        self.duration_predictor = nn.Linear(hidden_dim, 1)
        self.pitch_predictor = nn.Linear(hidden_dim, 1)
        self.energy_predictor = nn.Linear(hidden_dim, 1)
        
        # 副语言门控
        self.paralinguistic_gate = nn.Sequential(
            nn.Linear(hidden_dim, 32),
            nn.ReLU(),
            nn.Linear(32, 5)  # 5种副语言类型
        )
        
    def forward(self, 
                text_tokens: torch.Tensor,
                speaker_id: torch.Tensor,
                emotion: Optional[torch.Tensor] = None,
                prompt_desc: Optional[str] = None,
                paralinguistic_tags: Optional[List[str]] = None):
        """
        前向传播
        """
        batch_size, seq_len = text_tokens.shape
        
        # 文本嵌入 + 位置编码
        x = self.text_embedding(text_tokens)
        x = x + self.position_encoding[:, :seq_len, :]
        
        # 添加说话人信息
        speaker_emb = self.speaker_encoder(speaker_id).unsqueeze(1)
        x = x + speaker_emb
        
        # 添加情绪信息
        if emotion is not None:
            emotion_emb = self.emotion_encoder(emotion).unsqueeze(1)
            x = x + emotion_emb
        
        # 副语言标记注入
        if paralinguistic_tags:
            tag_map = {"laugh": 0, "pause": 1, "sigh": 2, 
                      "sneeze": 3, "clearthroat": 4}
            for tag in paralinguistic_tags:
                if tag in tag_map:
                    tag_idx = tag_map[tag]
                    x[:, -1, :] += self.paralinguistic_encoder[tag_idx]
        
        # Transformer编码
        x = self.transformer(x)
        
        # 预测声学特征
        mel_spec = self.mel_projection(x)
        duration = torch.sigmoid(self.duration_predictor(x))
        pitch = torch.tanh(self.pitch_predictor(x))
        energy = torch.sigmoid(self.energy_predictor(x))
        
        # 副语言门控输出
        para_gate = F.softmax(self.paralinguistic_gate(x), dim=-1)
        
        return {
            "mel_spec": mel_spec,        # [B, T, 80]
            "duration": duration,         # [B, T, 1]
            "pitch": pitch,              # [B, T, 1]
            "energy": energy,            # [B, T, 1]
            "paralinguistic_gate": para_gate  # [B, T, 5]
        }
    
    def generate_speech(self, text: str, speaker: str = "default",
                        emotion: str = "neutral") -> Dict:
        """
        生成语音（推理接口）
        """
        # Prompt驱动角色生成
        speaker_prompt = f"age:30,gender:female,style:{emotion}"
        if speaker == "custom":
            # 根据描述生成说话人嵌入
            pass
        
        return {
            "status": "success",
            "speaker": speaker,
            "emotion": emotion,
            "sample_rate": 24000,
            "paralinguistic_features": [
                "laugh_insertion:0.23",
                "pause_distribution:poisson(λ=0.5)",
                "breath_pattern:natural"
            ]
        }


# 角色声音生成器
class PromptDrivenVoiceGenerator:
    """Prompt驱动的角色声音生成"""
    
    VOICE_ARCHETYPES = {
        "caring_teacher": {
            "age": 35, "gender": "female",
            "pitch_mean": 220, "pitch_std": 30,
            "speed": 1.0, "breathiness": 0.3
        },
        "cheerful_child": {
            "age": 8, "gender": "female",
            "pitch_mean": 320, "pitch_std": 50,
            "speed": 1.2, "breathiness": 0.2
        },
        "professional_announcer": {
            "age": 40, "gender": "male",
            "pitch_mean": 120, "pitch_std": 15,
            "speed": 0.9, "breathiness": 0.1
        }
    }
    
    @classmethod
    def generate_from_prompt(cls, prompt: str) -> Dict:
        """根据自然语言描述生成声音参数"""
        # 简化实现：实际会调用大模型解析prompt
        prompt_lower = prompt.lower()
        
        if "老师" in prompt_lower or "teacher" in prompt_lower:
            return cls.VOICE_ARCHETYPES["caring_teacher"]
        elif "儿童" in prompt_lower or "child" in prompt_lower:
            return cls.VOICE_ARCHETYPES["cheerful_child"]
        elif "播音" in prompt_lower or "announcer" in prompt_lower:
            return cls.VOICE_ARCHETYPES["professional_announcer"]
        else:
            return {
                "age": 30, "gender": "female",
                "pitch_mean": 200, "pitch_std": 40,
                "speed": 1.0, "breathiness": 0.2
            }

# 示例
generator = PromptDrivenVoiceGenerator()
voice = generator.generate_from_prompt("一个温柔耐心的幼儿园老师声音")
print(f"生成声音参数: {voice}")

"""
多语种免切换语音识别引擎
"""
import numpy as np
from typing import Dict, List, Optional

class LanguageIdentification:
    """
    语种识别（LID）+ 免切换多语种ASR
    """
    
    def __init__(self):
        # 支持的语言及其方言
        self.supported_languages = {
            "zh": ["mandarin", "cantonese", "shanghainese", "minnan"],
            "en": ["us", "uk", "au", "in"],
            "ja": ["standard", "kansai"],
            "ko": ["standard"],
            "fr": ["standard", "quebec", "african"],
            "es": ["castilian", "mexican", "argentinian"],
            # ... 共40+语种
        }
        
        # 语言嵌入向量（简化）
        self.language_embeddings = {
            lang: np.random.randn(128) for lang in self.supported_languages
        }
    
    def detect_language(self, audio_features: np.ndarray) -> Dict[str, float]:
        """
        检测音频中的语言及其置信度
        返回: {language_code: confidence_score}
        """
        # 简化实现：使用余弦相似度
        scores = {}
        for lang, emb in self.language_embeddings.items():
            sim = np.dot(audio_features, emb) / (
                np.linalg.norm(audio_features) * np.linalg.norm(emb)
            )
            scores[lang] = float(sim)
        
        # 归一化
        total = sum(scores.values())
        if total > 0:
            scores = {k: v/total for k, v in scores.items()}
        
        return dict(sorted(scores.items(), key=lambda x: -x[1])[:3])


class MultiLanguageASR:
    """
    免切换多语种ASR
    支持在同一段对话中自由切换语言
    """
    
    def __init__(self):
        self.lid = LanguageIdentification()
        self.language_models = {}  # 各语种语言模型
        self.current_lang = "zh"
        self.language_history = []
        
    def transcribe_segment(self, audio: np.ndarray) -> Dict:
        """
        解码一段音频
        """
        # 1. 语种检测
        lang_scores = self.lid.detect_language(audio)
        detected_lang = list(lang_scores.keys())[0]
        
        # 2. 语言切换检测（如果置信度 > 0.7且与当前语言不同）
        if lang_scores[detected_lang] > 0.7 and detected_lang != self.current_lang:
            self.language_history.append({
                "from": self.current_lang,
                "to": detected_lang,
                "timestamp": len(self.language_history)
            })
            self.current_lang = detected_lang
        
        # 3. 使用对应语言模型解码
        # 简化：返回模拟结果
        return {
            "text": f"[{detected_lang}] 识别结果",
            "language": detected_lang,
            "confidence": lang_scores[detected_lang],
            "is_code_switching": len(self.language_history) > 0
        }
    
    def get_language_switching_stats(self) -> Dict:
        """获取语言切换统计"""
        if not self.language_history:
            return {"total_switches": 0}
        
        switch_pairs = {}
        for h in self.language_history:
            pair = f"{h['from']}→{h['to']}"
            switch_pairs[pair] = switch_pairs.get(pair, 0) + 1
        
        return {
            "total_switches": len(self.language_history),
            "switch_pairs": switch_pairs,
            "current_language": self.current_lang
        }

# 示例
asr = MultiLanguageASR()
result = asr.transcribe_segment(np.random.randn(16000))
print(f"识别结果: {result}")

package robot

import (
	"math"
)

// AudioVisualFusion 视听融合感知模块
type AudioVisualFusion struct {
	micArraySize    int     // 麦克风数量
	fovHorizontal   float64 // 水平视场角
	fovVertical     float64 // 垂直视场角
	sampleRate      int     // 采样率
}

// SoundSource 声源
type SoundSource struct {
	Azimuth   float64 // 水平角（度）
	Elevation float64 // 仰角（度）
	Distance  float64 // 距离（米）
	Energy    float64 // 能量
	IsSpeech  bool    // 是否为人声
	SpeakerID string  // 说话人ID
}

// VisualTarget 视觉目标
type VisualTarget struct {
	FaceID    string
	Azimuth   float64
	Elevation float64
	MouthOpen bool    // 是否张嘴（正在说话）
	LipSync   float64 // 唇动同步度
}

// FusedTarget 融合后的目标
type FusedTarget struct {
	SpeakerID     string
	Azimuth       float64
	Elevation     float64
	AudioConfidence float64
	VisualConfidence float64
	FusionScore   float64
	IsActiveSpeaker bool
}

// ProcessFrame 处理一帧视听数据
func (f *AudioVisualFusion) ProcessFrame(
	audioFrame []float64,
	visualTargets []VisualTarget,
) []FusedTarget {
	// 1. 声源定位（SRP-PHAT算法简化）
	audioSources := f.localizeSoundSources(audioFrame)
	
	// 2. 视听关联
	var fused []FusedTarget
	for _, vs := range visualTargets {
		bestMatch := f.findBestAudioMatch(vs, audioSources)
		if bestMatch != nil {
			// 3. 计算融合分数
			fusionScore := f.computeFusionScore(vs, *bestMatch)
			fused = append(fused, FusedTarget{
				SpeakerID:       vs.FaceID,
				Azimuth:         vs.Azimuth,
				Elevation:       vs.Elevation,
				AudioConfidence: bestMatch.Energy,
				VisualConfidence: f.computeVisualConfidence(vs),
				FusionScore:     fusionScore,
				IsActiveSpeaker: fusionScore > 0.7 && vs.MouthOpen,
			})
		}
	}
	
	return fused
}

// localizeSoundSources SRP-PHAT声源定位
func (f *AudioVisualFusion) localizeSoundSources(audio []float64) []SoundSource {
	// 简化实现：使用GCC-PHAT（广义互相关-相位变换）
	// 实际部署使用波束成形+深度学习混合方法
	// 返回模拟结果
	return []SoundSource{
		{Azimuth: 30, Elevation: 5, Distance: 2.0, Energy: 0.85, IsSpeech: true},
		{Azimuth: -45, Elevation: -3, Distance: 3.5, Energy: 0.45, IsSpeech: true},
	}
}

func (f *AudioVisualFusion) findBestAudioMatch(
	visual VisualTarget, 
	sources []SoundSource,
) *SoundSource {
	var best *SoundSource
	minAngleDiff := 30.0 // 最大允许角度差
	
	for i := range sources {
		angleDiff := math.Abs(sources[i].Azimuth - visual.Azimuth)
		if angleDiff < minAngleDiff {
			minAngleDiff = angleDiff
			best = &sources[i]
		}
	}
	return best
}

func (f *AudioVisualFusion) computeVisualConfidence(v VisualTarget) float64 {
	// 基于唇动检测和人脸置信度
	if v.MouthOpen {
		return 0.8 + v.LipSync*0.2
	}
	return 0.3
}

func (f *AudioVisualFusion) computeFusionScore(v VisualTarget, a SoundSource) float64 {
	// 视听融合评分
	angleScore := 1.0 - math.Abs(a.Azimuth-v.Azimuth)/180.0
	return 0.6*a.Energy + 0.4*angleScore
}