谷歌Gemini 3.8 Live实时对话模型深度解析——97种语言+视觉理解+异步工具调用，AI语音交互进入"边聊边干"时代

package main

import (
    "bytes"
    "encoding/binary"
    "fmt"
    "io"
    "log"
    "net/http"
    "time"

    "github.com/gorilla/websocket"
)

// AudioConfig 定义音频参数
type AudioConfig struct {
    InputSampleRate  int // 16kHz
    OutputSampleRate int // 24kHz
    BitDepth         int // 16位
    Channels         int // 单声道
    FrameSize        int // 帧大小（毫秒）
}

// DefaultAudioConfig 默认配置
var DefaultAudioConfig = AudioConfig{
    InputSampleRate:  16000,
    OutputSampleRate: 24000,
    BitDepth:         16,
    Channels:         1,
    FrameSize:        20, // 20ms帧
}

// LiveAPIClient Gemini Live API客户端
type LiveAPIClient struct {
    conn      *websocket.Conn
    audioCfg  AudioConfig
    apiKey    string
    buffer    bytes.Buffer
}

// NewLiveAPIClient 创建新的API客户端
func NewLiveAPIClient(apiKey string) *LiveAPIClient {
    return &LiveAPIClient{
        apiKey:   apiKey,
        audioCfg: DefaultAudioConfig,
    }
}

// Connect 建立WebSocket连接
func (c *LiveAPIClient) Connect(url string) error {
    header := http.Header{}
    header.Set("Authorization", "Bearer "+c.apiKey)
    
    conn, _, err := websocket.DefaultDialer.Dial(url, header)
    if err != nil {
        return fmt.Errorf("websocket连接失败: %w", err)
    }
    c.conn = conn
    return nil
}

// StreamAudioChunk 流式发送音频块
func (c *LiveAPIClient) StreamAudioChunk(pcmData []byte) error {
    // 构建音频消息
    msg := AudioMessage{
        Type:      "audio_input",
        Format:    "pcm16",
        SampleRate: c.audioCfg.InputSampleRate,
        Data:      pcmData,
    }
    return c.conn.WriteJSON(msg)
}

// ReceiveAudioStream 接收音频流响应
func (c *LiveAPIClient) ReceiveAudioStream() (<-chan []byte, <-chan error) {
    audioCh := make(chan []byte, 100)
    errCh := make(chan error, 1)
    
    go func() {
        defer close(audioCh)
        defer close(errCh)
        
        for {
            var msg ServerMessage
            if err := c.conn.ReadJSON(&msg); err != nil {
                errCh <- fmt.Errorf("读取消息失败: %w", err)
                return
            }
            
            switch msg.Type {
            case "audio_output":
                audioCh <- msg.AudioData
            case "turn_complete":
                // turnComplete不意味着推理结束
                // 需检查interaction_status
                if msg.InteractionStatus == "IDLE" {
                    return
                }
            case "error":
                errCh <- fmt.Errorf("服务端错误: %s", msg.Error)
                return
            }
        }
    }()
    
    return audioCh, errCh
}

// AudioMessage 音频消息结构
type AudioMessage struct {
    Type       string `json:"type"`
    Format     string `json:"format"`
    SampleRate int    `json:"sample_rate"`
    Data       []byte `json:"data"`
}

// ServerMessage 服务端消息结构
type ServerMessage struct {
    Type              string `json:"type"`
    AudioData         []byte `json:"audio_data,omitempty"`
    TurnComplete      bool   `json:"turn_complete,omitempty"`
    InteractionStatus string `json:"interaction_status,omitempty"`
    Error             string `json:"error,omitempty"`
}

// CalculateFrameSize 计算帧大小（字节）
func (c *AudioConfig) CalculateFrameSize() int {
    msPerSecond := 1000
    bytesPerSample := c.BitDepth / 8
    samplesPerFrame := c.InputSampleRate * c.FrameSize / msPerSecond
    return samplesPerFrame * bytesPerSample * c.Channels
}

func main() {
    client := NewLiveAPIClient("YOUR_API_KEY")
    
    if err := client.Connect("wss://gemini-live-api.google.com/v1/live"); err != nil {
        log.Fatal("连接失败:", err)
    }
    defer client.conn.Close()
    
    log.Printf("已连接，帧大小: %d 字节", client.audioCfg.CalculateFrameSize())
    
    audioCh, errCh := client.ReceiveAudioStream()
    
    select {
    case audio := <-audioCh:
        log.Printf("收到音频数据: %d 字节", len(audio))
    case err := <-errCh:
        log.Fatal("流错误:", err)
    case <-time.After(30 * time.Second):
        log.Println("超时")
    }
}

import asyncio
import numpy as np
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ThinkingLevel(Enum):
    """可配置的推理深度级别"""
    LOW = "low"       # 低推理：快速响应
    MEDIUM = "medium" # 中等推理：平衡
    HIGH = "high"     # 高推理：深度多步

@dataclass
class ExtendedThinkingConfig:
    """Extended Thinking配置"""
    thinking_level: ThinkingLevel = ThinkingLevel.MEDIUM
    max_thinking_tokens: int = 4096
    early_verbal_cue: bool = True  # "让我确认一下..."类提示
    live_progress_narration: bool = True  # 实时进度叙述

class ExtendedThinkingPipeline:
    """
    Gemini 3.8 Live Extended Thinking推理流水线
    
    核心设计：推理流与语音流并行，互不阻塞
    """
    
    def __init__(self, config: ExtendedThinkingConfig):
        self.config = config
        self.thinking_buffer = []  # 推理中间状态缓冲区
        self.speech_buffer = []    # 语音输出缓冲区
        self.is_reasoning = False
        
    async def process_with_extended_thinking(
        self, 
        user_input: np.ndarray,
        visual_context: Optional[np.ndarray] = None
    ) -> asyncio.Queue:
        """
        并行处理推理和语音生成
        
        返回一个队列，包含交替的推理进度和语音输出
        """
        output_queue = asyncio.Queue()
        
        async def reasoning_stream():
            """推理流——在后台运行"""
            self.is_reasoning = True
            
            # 阶段1: 快速初步理解（生成早期语音提示）
            preliminary = await self._quick_understand(user_input, visual_context)
            if self.config.early_verbal_cue:
                await output_queue.put({
                    "type": "speech",
                    "content": self._generate_early_cue(preliminary),
                    "is_progress": False
                })
            
            # 阶段2: 分步推理（可配置深度）
            steps = self._decompose_reasoning_steps(
                user_input, 
                self.config.thinking_level
            )
            
            for i, step in enumerate(steps):
                step_result = await self._execute_reasoning_step(step)
                self.thinking_buffer.append(step_result)
                
                if self.config.live_progress_narration:
                    # 在推理过程中同步输出进度叙述
                    narration = self._generate_progress_narration(
                        step_index=i, 
                        total_steps=len(steps),
                        step_result=step_result
                    )
                    await output_queue.put({
                        "type": "speech",
                        "content": narration,
                        "is_progress": True
                    })
            
            # 阶段3: 综合推理结果
            final_result = await self._synthesize_results(self.thinking_buffer)
            await output_queue.put({
                "type": "speech",
                "content": self._generate_final_response(final_result),
                "is_progress": False
            })
            
            self.is_reasoning = False
        
        asyncio.create_task(reasoning_stream())
        return output_queue
    
    def _generate_early_cue(self, preliminary: dict) -> str:
        """生成早期语言提示，如"让我确认一下..." """
        cue_templates = [
            "让我确认一下这个信息...",
            "好的，我来查一下相关数据...",
            "这个问题需要几步处理，我先梳理一下...",
            "让我看看最新的情况..."
        ]
        # 根据初步理解选择最合适的提示
        confidence = preliminary.get("confidence", 0.5)
        if confidence > 0.8:
            return cue_templates[0]
        elif confidence > 0.5:
            return cue_templates[1]
        else:
            return cue_templates[2]
    
    def _decompose_reasoning_steps(
        self, 
        input_data: np.ndarray,
        level: ThinkingLevel
    ) -> list:
        """根据推理深度级别分解推理步骤"""
        step_map = {
            ThinkingLevel.LOW: 2,     # 低深度：2步
            ThinkingLevel.MEDIUM: 4,  # 中等深度：4步
            ThinkingLevel.HIGH: 8     # 高深度：8步
        }
        num_steps = step_map[level]
        return [f"reasoning_step_{i}" for i in range(num_steps)]
    
    async def _quick_understand(self, audio: np.ndarray, visual: Optional[np.ndarray]) -> dict:
        """快速初步理解——用于决定早期提示"""
        await asyncio.sleep(0.05)  # 模拟快速推理
        return {"confidence": 0.7, "intent": "complex_query"}
    
    async def _execute_reasoning_step(self, step: str) -> dict:
        """执行单步推理"""
        await asyncio.sleep(0.1)
        return {"step": step, "result": f"intermediate_result_{step}"}
    
    async def _synthesize_results(self, buffer: list) -> dict:
        """综合所有推理步骤结果"""
        return {"final": True, "steps_completed": len(buffer)}
    
    def _generate_progress_narration(self, step_index: int, total_steps: int, step_result: dict) -> str:
        """生成实时进度叙述"""
        return f"正在处理第{step_index+1}步（共{total_steps}步）..."
    
    def _generate_final_response(self, result: dict) -> str:
        """生成最终回答"""
        return f"综合分析了{result['steps_completed']}个维度，我的结论如下..."

# 使用示例
async def main():
    config = ExtendedThinkingConfig(
        thinking_level=ThinkingLevel.HIGH,
        early_verbal_cue=True,
        live_progress_narration=True
    )
    pipeline = ExtendedThinkingPipeline(config)
    
    # 模拟用户语音输入（16kHz PCM音频）
    dummy_audio = np.zeros(16000, dtype=np.int16)
    
    output_queue = await pipeline.process_with_extended_thinking(dummy_audio)
    
    # 消费输出流
    while pipeline.is_reasoning or not output_queue.empty():
        try:
            item = await asyncio.wait_for(output_queue.get(), timeout=1.0)
            print(f"[{'进度' if item['is_progress'] else '回答'}] {item['content']}")
        except asyncio.TimeoutError:
            break

asyncio.run(main())

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// ToolResult 异步工具调用结果
type ToolResult struct {
    ToolID    string      `json:"tool_id"`
    Result    interface{} `json:"result"`
    Error     error       `json:"error,omitempty"`
    Timestamp time.Time   `json:"timestamp"`
}

// AsyncToolManager 异步工具管理器
type AsyncToolManager struct {
    mu           sync.RWMutex
    pendingTools map[string]context.CancelFunc
    results      chan ToolResult
    activeCount  int32
}

// NewAsyncToolManager 创建异步工具管理器
func NewAsyncToolManager() *AsyncToolManager {
    return &AsyncToolManager{
        pendingTools: make(map[string]context.CancelFunc),
        results:      make(chan ToolResult, 100),
    }
}

// ExecuteToolAsync 异步执行工具调用
// 关键设计：不阻塞对话流，通过回调/轮询获取结果
func (m *AsyncToolManager) ExecuteToolAsync(
    ctx context.Context,
    toolID string,
    toolFunc func(context.Context) (interface{}, error),
) error {
    m.mu.Lock()
    if _, exists := m.pendingTools[toolID]; exists {
        m.mu.Unlock()
        return fmt.Errorf("工具 %s 正在执行中", toolID)
    }
    
    toolCtx, cancel := context.WithCancel(ctx)
    m.pendingTools[toolID] = cancel
    m.mu.Unlock()
    
    go func() {
        defer func() {
            m.mu.Lock()
            delete(m.pendingTools, toolID)
            m.mu.Unlock()
        }()
        
        result, err := toolFunc(toolCtx)
        m.results <- ToolResult{
            ToolID:    toolID,
            Result:    result,
            Error:     err,
            Timestamp: time.Now(),
        }
    }()
    
    return nil
}

// GetResult 非阻塞获取工具结果
func (m *AsyncToolManager) GetResult(timeout time.Duration) (*ToolResult, bool) {
    select {
    case result := <-m.results:
        return &result, true
    case <-time.After(timeout):
        return nil, false
    }
}

// CancelTool 取消正在执行的工具
func (m *AsyncToolManager) CancelTool(toolID string) bool {
    m.mu.RLock()
    cancel, exists := m.pendingTools[toolID]
    m.mu.RUnlock()
    
    if exists {
        cancel()
        return true
    }
    return false
}

// VoiceAgent 语音智能体——集成异步工具调用
type VoiceAgent struct {
    toolManager *AsyncToolManager
    dialogueCh  chan string
}

// HandleUserRequest 处理用户请求——对话与工具并行
func (a *VoiceAgent) HandleUserRequest(ctx context.Context, userSpeech string) {
    // 1. 立即回复用户确认（不等待工具完成）
    a.dialogueCh <- "好的，我来查询相关信息，请稍等..."
    
    // 2. 在后台异步执行工具
    toolID := fmt.Sprintf("tool_%d", time.Now().UnixNano())
    _ = a.toolManager.ExecuteToolAsync(ctx, toolID,
        func(ctx context.Context) (interface{}, error) {
            // 模拟耗时API调用（如查询数据库）
            select {
            case <-time.After(3 * time.Second):
                return map[string]interface{}{
                    "query_result": "模拟数据查询结果",
                    "timestamp":    time.Now().Unix(),
                }, nil
            case <-ctx.Done():
                return nil, ctx.Err()
            }
        })
    
    // 3. 对话继续——模型可以发出更多语音回复
    a.dialogueCh <- "同时我还发现了一些相关信息..."
    
    // 4. 用户无需等待，可以继续提问
    a.dialogueCh <- "您可以继续提问，结果出来后我会通知您。"
    
    // 5. 后台结果就绪后，无缝融入当前对话
    go func() {
        if result, ok := a.toolManager.GetResult(5 * time.Second); ok {
            if result.Error == nil {
                a.dialogueCh <- fmt.Sprintf("查询结果已就绪: %v", result.Result)
            }
        }
    }()
}

import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass
class LanguageProfile:
    """语言特征配置"""
    code: str                  # 语言代码，如 "zh-CN"
    name: str                  # 语言名称
    phoneme_set: set           # 音素集合
    prosodic_features: dict    # 韵律特征
    weight: float = 1.0        # 当前对话中的权重

class MultilingualRouter:
    """
    多语言路由引擎
    
    核心能力：
    - 实时检测音频中的语言
    - 对话中的语言无缝切换
    - 多语言混合输入处理
    """
    
    def __init__(self):
        self.supported_languages: Dict[str, LanguageProfile] = {}
        self.active_languages: List[str] = []
        self.confidence_threshold = 0.75
        self._init_language_profiles()
    
    def _init_language_profiles(self):
        """初始化97种语言配置（以下是部分示例）"""
        languages = [
            ("zh-CN", "中文（简体）"),
            ("en-US", "英语（美国）"),
            ("ja-JP", "日语"),
            ("ko-KR", "韩语"),
            ("es-ES", "西班牙语"),
            ("fr-FR", "法语"),
            ("de-DE", "德语"),
            ("pt-BR", "葡萄牙语"),
            ("ar-SA", "阿拉伯语"),
            ("hi-IN", "印地语"),
            ("ru-RU", "俄语"),
            ("id-ID", "印度尼西亚语"),
            ("th-TH", "泰语"),
            ("vi-VN", "越南语"),
        ]
        
        for code, name in languages:
            # 每个语言初始化特征向量（实际系统使用深度神经网络提取）
            self.supported_languages[code] = LanguageProfile(
                code=code,
                name=name,
                phoneme_set=self._generate_phoneme_set(code),
                prosodic_features=self._generate_prosodic_features(code)
            )
    
    def detect_language(self, audio_frame: np.ndarray) -> tuple:
        """
        检测音频帧的语言
        
        使用声学特征和语言模型的联合推理
        """
        # 提取声学特征（模拟MFCC提取）
        acoustic_features = self._extract_acoustic_features(audio_frame)
        
        scores = {}
        for code, profile in self.supported_languages.items():
            # 计算余弦相似度（实际使用神经语言分类器）
            similarity = self._compute_language_similarity(
                acoustic_features, profile
            )
            scores[code] = similarity
        
        # 取最高分
        best_lang = max(scores, key=scores.get)
        best_score = scores[best_lang]
        
        return best_lang, best_score
    
    def _extract_acoustic_features(self, audio: np.ndarray) -> np.ndarray:
        """提取声学特征（简化的MFCC模拟）"""
        # 实际系统使用13维MFCC + delta + delta-delta
        frame_size = 400  # 25ms @ 16kHz
        hop_size = 160    # 10ms
        
        if len(audio) < frame_size:
            return np.zeros(39)
        
        # 简化的特征提取
        features = []
        for start in range(0, len(audio) - frame_size, hop_size):
            frame = audio[start:start + frame_size]
            # 应用汉明窗
            window = np.hamming(frame_size)
            windowed = frame * window
            # FFT并提取能量
            spectrum = np.abs(np.fft.rfft(windowed))
            features.append(spectrum[:13])  # 取前13个频带
        
        return np.mean(features, axis=0)
    
    def _compute_language_similarity(
        self, features: np.ndarray, profile: LanguageProfile
    ) -> float:
        """计算特征与语言配置的相似度"""
        # 模拟相似度计算
        profile_features = profile.prosodic_features.get("mfcc_mean", 
            np.random.randn(13))
        
        # 余弦相似度
        if len(features) != len(profile_features):
            return 0.0
        
        dot_product = np.dot(features, profile_features)
        norm_product = np.linalg.norm(features) * np.linalg.norm(profile_features)
        
        if norm_product == 0:
            return 0.0
        
        return float(dot_product / norm_product)
    
    def _generate_phoneme_set(self, code: str) -> set:
        """生成语言音素集"""
        phoneme_sets = {
            "zh-CN": {"p", "pʰ", "t", "tʰ", "k", "kʰ", "m", "n", "ŋ",
                      "f", "s", "ʂ", "ɕ", "x", "h", "l", "i", "u", "y",
                      "a", "o", "ə", "e", "ai", "ei", "ao", "ou", "an"},
            "en-US": {"p", "b", "t", "d", "k", "g", "f", "v", "θ", "ð",
                      "s", "z", "ʃ", "ʒ", "h", "m", "n", "ŋ", "l", "r",
                      "w", "j", "iː", "ɪ", "eɪ", "ɛ", "æ", "ɑː", "ɒ",
                      "ɔː", "oʊ", "ʊ", "uː", "ʌ", "ɜː", "aɪ", "aʊ", "ɔɪ"},
        }
        return phoneme_sets.get(code, set())
    
    def _generate_prosodic_features(self, code: str) -> dict:
        """生成语言的韵律特征"""
        # 每种语言有独特的韵律模式
        base = {
            "mfcc_mean": np.random.randn(13),
            "pitch_range": np.random.uniform(60, 300),
            "speaking_rate": np.random.uniform(3, 8),
        }
        return base
    
    def handle_code_switching(
        self, audio_stream: np.ndarray
    ) -> List[Dict]:
        """
        处理对话中的语言切换
        
        返回时间戳标注的语码转换序列
        """
        segment_duration_ms = 500  # 每500ms检测一次
        sample_rate = 16000
        segment_samples = int(segment_duration_ms * sample_rate / 1000)
        
        segments = []
        current_lang = None
        
        for start in range(0, len(audio_stream), segment_samples):
            segment = audio_stream[start:start + segment_samples]
            if len(segment) < segment_samples // 2:
                break
            
            detected_lang, confidence = self.detect_language(segment)
            
            if confidence >= self.confidence_threshold:
                if detected_lang != current_lang:
                    segments.append({
                        "timestamp_ms": start * 1000 // sample_rate,
                        "from_lang": current_lang,
                        "to_lang": detected_lang,
                        "confidence": confidence
                    })
                    current_lang = detected_lang
        
        return segments

# 使用示例
router = MultilingualRouter()
test_audio = np.random.randn(16000 * 5)  # 5秒音频
switches = router.handle_code_switching(test_audio)
print(f"检测到 {len(switches)} 次语言切换")
for s in switches:
    print(f"  {s['timestamp_ms']}ms: {s['from_lang']} -> {s['to_lang']} ({s['confidence']:.2f})")

import numpy as np
from scipy import signal
from typing import Tuple

class SynthIDWatermark:
    """
    SynthID音频水印嵌入器
    
    核心原理：在音频的频谱域中嵌入不可感知的伪随机信号，
    该信号在重新编码/压缩后仍然可检测。
    """
    
    def __init__(self, sample_rate: int = 24000):
        self.sample_rate = sample_rate
        self.watermark_key = self._generate_watermark_key()
        
    def _generate_watermark_key(self) -> np.ndarray:
        """生成水印密钥——伪随机相位序列"""
        np.random.seed(42)  # 固定的种子用于可复现检测
        # 使用经过心理声学模型优化的伪随机序列
        key_length = 1024
        phase = np.exp(2j * np.pi * np.random.rand(key_length))
        return phase
    
    def _psychoacoustic_mask(
        self, spectrum: np.ndarray
    ) -> np.ndarray:
        """
        心理声学掩蔽阈值计算
        
        根据人耳听觉特性，计算在哪些频段可以嵌入水印
        而不被感知
        """
        # Bark尺度映射
        n_fft = len(spectrum)
        bark_scale = np.linspace(0, 1, n_fft) ** 0.5
        
        # 计算掩蔽阈值（简化版）
        mask = np.zeros_like(spectrum)
        for i in range(len(spectrum)):
            # 每个频点的掩蔽由其邻近频点的能量决定
            spread = np.exp(-0.5 * ((np.arange(n_fft) - i) / 10) ** 2)
            mask[i] = np.max(spectrum * spread) * 0.01
        
        return mask
    
    def embed_watermark(
        self, audio: np.ndarray
    ) -> np.ndarray:
        """
        在音频中嵌入SynthID水印
        
        流程：
        1. STFT变换到频域
        2. 计算心理声学掩蔽阈值
        3. 在不可感知的频段嵌入密钥
        4. ISTFT变换回时域
        """
        # STFT参数
        n_fft = 512
        hop_length = 128
        
        # STFT变换
        f, t, Zxx = signal.stft(
            audio, 
            fs=self.sample_rate,
            nperseg=n_fft,
            noverlap=n_fft - hop_length
        )
        
        magnitude = np.abs(Zxx)
        phase = np.angle(Zxx)
        
        # 计算心理声学掩蔽
        mask = self._psychoacoustic_mask(magnitude.mean(axis=1))
        
        # 水印嵌入强度（基于掩蔽阈值动态调整）
        embedding_strength = 0.005
        
        # 在频谱中嵌入水印
        watermark = np.outer(
            self.watermark_key[:len(f)],
            np.ones(t.shape[0])
        )
        
        # 只在掩蔽阈值允许的频段嵌入
        valid_freqs = mask > 0.01
        magnitude_modified = magnitude.copy()
        magnitude_modified[valid_freqs] *= (
            1 + embedding_strength * np.real(watermark[valid_freqs])
        )
        
        # ISTFT还原到时域
        Zxx_modified = magnitude_modified * np.exp(1j * phase)
        _, watermarked_audio = signal.istft(
            Zxx_modified,
            fs=self.sample_rate,
            nperseg=n_fft,
            noverlap=n_fft - hop_length
        )
        
        return watermarked_audio
    
    def detect_watermark(self, audio: np.ndarray) -> Tuple[bool, float]:
        """
        检测音频中的SynthID水印
        
        返回：(是否检测到, 置信度分数)
        """
        n_fft = 512
        hop_length = 128
        
        f, t, Zxx = signal.stft(
            audio,
            fs=self.sample_rate,
            nperseg=n_fft,
            noverlap=n_fft - hop_length
        )
        
        magnitude = np.abs(Zxx)
        
        # 提取频谱模式
        pattern = magnitude.mean(axis=1)
        pattern = pattern / (np.linalg.norm(pattern) + 1e-10)
        
        # 与密钥模式进行互相关检测
        key_pattern = np.abs(self.watermark_key[:len(f)])
        key_pattern = key_pattern / (np.linalg.norm(key_pattern) + 1e-10)
        
        correlation = np.correlate(pattern, key_pattern, mode='same')
        confidence = float(np.max(np.abs(correlation)))
        
        # 检测阈值
        detection_threshold = 0.3
        is_detected = confidence > detection_threshold
        
        return is_detected, confidence

import asyncio
import websockets
import json

class GeminiLiveClient:
    """Gemini Live API客户端示例"""
    
    def __init__(self, api_key: str, model: str = "gemini-3.8-live"):
        self.api_key = api_key
        self.model = model
        self.ws_url = f"wss://generativelanguage.googleapis.com/ws/" \
                      f"google.ai.generativelanguage.v1alpha.GenerativeService." \
                      f"BidiGenerateContent?key={api_key}"
    
    async def start_session(self):
        """启动实时会话"""
        async with websockets.connect(self.ws_url) as ws:
            # 发送配置
            setup_payload = {
                "setup": {
                    "model": f"models/{self.model}",
                    "system_instruction": {
                        "parts": [{"text": "你是一个友好的AI助手。"}]
                    },
                    "config": {
                        "audio_config": {
                            "input_audio_format": "PCM16_16000",
                            "output_audio_format": "PCM16_24000"
                        }
                    }
                }
            }
            await ws.send(json.dumps(setup_payload))
            
            # 处理双向流
            async def send_audio():
                """发送音频流"""
                while True:
                    # 从麦克风读取音频块
                    audio_chunk = await self._read_microphone()
                    msg = {
                        "real_time_input": {
                            "audio": audio_chunk
                        }
                    }
                    await ws.send(json.dumps(msg))
            
            async def receive_responses():
                """接收响应流"""
                async for message in ws:
                    response = json.loads(message)
                    if "audio_output" in response:
                        await self._play_audio(response["audio_output"])
                    if "text_transcript" in response:
                        print(response["text_transcript"], end="")
            
            await asyncio.gather(send_audio(), receive_responses())
    
    async def _read_microphone(self):
        """读取麦克风输入（模拟）"""
        import sounddevice as sd
        chunk = sd.rec(int(16000 * 0.1), samplerate=16000, channels=1)
        return chunk.tobytes()
    
    async def _play_audio(self, audio_data):
        """播放音频输出（模拟）"""
        import sounddevice as sd
        import numpy as np
        audio = np.frombuffer(audio_data, dtype=np.int16)
        sd.play(audio, samplerate=24000)

from enum import Enum
from typing import Optional

class ConversationState(Enum):
    """实时对话状态机状态"""
    IDLE = "idle"                    # 空闲
    LISTENING = "listening"          # 正在监听
    PROCESSING = "processing"        # 正在处理（含推理）
    SPEAKING = "speaking"            # 正在发言
    EXTENDED_THINKING = "extended_thinking"  # 扩展推理（后台）
    BACKGROUND_TOOL = "background_tool"      # 后台工具执行
    INTERRUPTED = "interrupted"      # 被打断

class LiveConversationStateMachine:
    """
    实时对话状态机
    
    管理以下关键状态转换：
    - 用户说话时的打断处理
    - Extended Thinking的并行推理状态
    - 后台异步工具调用的生命周期
    - 多模态输入的协调（语音+视觉）
    """
    
    def __init__(self):
        self.state = ConversationState.IDLE
        self.interaction_status = "IDLE"
        
    def transition(self, event: str) -> Optional[ConversationState]:
        transitions = {
            ConversationState.IDLE: {
                "user_speech_start": ConversationState.LISTENING
            },
            ConversationState.LISTENING: {
                "user_speech_end": ConversationState.PROCESSING,
                "extended_query_detected": ConversationState.EXTENDED_THINKING
            },
            ConversationState.PROCESSING: {
                "response_ready": ConversationState.SPEAKING,
                "tool_needed": ConversationState.BACKGROUND_TOOL,
                "user_interrupt": ConversationState.INTERRUPTED
            },
            ConversationState.SPEAKING: {
                "user_interrupt": ConversationState.LISTENING,
                "turn_complete": ConversationState.IDLE,
                "extended_thinking_needed": ConversationState.EXTENDED_THINKING
            },
            ConversationState.EXTENDED_THINKING: {
                "intermediate_result": ConversationState.SPEAKING,
                "reasoning_complete": ConversationState.SPEAKING,
                "user_interrupt": ConversationState.LISTENING
            },
            ConversationState.BACKGROUND_TOOL: {
                "tool_complete": ConversationState.PROCESSING,
                "user_interrupt": ConversationState.LISTENING
            },
            ConversationState.INTERRUPTED: {
                "user_resume": ConversationState.LISTENING
            }
        }
        
        if self.state in transitions and event in transitions[self.state]:
            self.state = transitions[self.state][event]
        return self.state