Google Gemini 3.8 Live & Extended Thinking Deep Dive — 97 Languages, Async Tool Calls, and Real-Time Visual Grounding Redefine Voice AI

package main

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

    "github.com/gorilla/websocket"
)

type AudioConfig struct {
    InputSampleRate  int
    OutputSampleRate int
    BitDepth         int
    Channels         int
    FrameSize        int
}

var DefaultAudioConfig = AudioConfig{
    InputSampleRate:  16000,
    OutputSampleRate: 24000,
    BitDepth:         16,
    Channels:         1,
    FrameSize:        20,
}

type LiveAPIClient struct {
    conn     *websocket.Conn
    audioCfg AudioConfig
    apiKey   string
    buffer   bytes.Buffer
}

func NewLiveAPIClient(apiKey string) *LiveAPIClient {
    return &LiveAPIClient{
        apiKey:   apiKey,
        audioCfg: DefaultAudioConfig,
    }
}

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 connection failed: %w", err)
    }
    c.conn = conn
    return nil
}

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)
}

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("read message failed: %w", err)
                return
            }
            
            switch msg.Type {
            case "audio_output":
                audioCh <- msg.AudioData
            case "turn_complete":
                if msg.InteractionStatus == "IDLE" {
                    return
                }
            case "error":
                errCh <- fmt.Errorf("server error: %s", msg.Error)
                return
            }
        }
    }()
    
    return audioCh, errCh
}

type AudioMessage struct {
    Type       string `json:"type"`
    Format     string `json:"format"`
    SampleRate int    `json:"sample_rate"`
    Data       []byte `json:"data"`
}

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"`
}

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("connection failed:", err)
    }
    defer client.conn.Close()
    
    log.Printf("Connected, frame size: %d bytes", 
        client.audioCfg.CalculateFrameSize())
    
    audioCh, errCh := client.ReceiveAudioStream()
    
    select {
    case audio := <-audioCh:
        log.Printf("Received audio: %d bytes", len(audio))
    case err := <-errCh:
        log.Fatal("stream error:", err)
    case <-time.After(30 * time.Second):
        log.Println("timeout")
    }
}

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

class ThinkingLevel(Enum):
    """Configurable reasoning depth levels"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@dataclass
class ExtendedThinkingConfig:
    """Extended Thinking configuration"""
    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 reasoning pipeline
    
    Core design: reasoning stream and speech stream run in parallel,
    never blocking each other
    """
    
    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:
        """Process user input with parallel reasoning and speech generation"""
        output_queue = asyncio.Queue()
        
        async def reasoning_stream():
            """Reasoning stream — runs in background"""
            self.is_reasoning = True
            
            # Phase 1: Quick preliminary understanding (generates early verbal cue)
            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
                })
            
            # Phase 2: Step-by-step reasoning (configurable depth)
            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
                    })
            
            # Phase 3: Synthesize reasoning results
            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:
        """Generate early verbal cues like 'Let me check that...' """
        cue_templates = [
            "Let me check that information...",
            "Let me look up the relevant data...",
            "This needs a few steps, let me work through it...",
            "Let me find the latest on this..."
        ]
        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:
        """Decompose reasoning steps based on thinking depth level"""
        step_map = {
            ThinkingLevel.LOW: 2,
            ThinkingLevel.MEDIUM: 4,
            ThinkingLevel.HIGH: 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"Processing step {step_index+1} of {total_steps}..."
    
    def _generate_final_response(self, result: dict) -> str:
        return f"After analyzing {result['steps_completed']} dimensions, here's my conclusion..."

async def main():
    config = ExtendedThinkingConfig(
        thinking_level=ThinkingLevel.HIGH,
        early_verbal_cue=True,
        live_progress_narration=True
    )
    pipeline = ExtendedThinkingPipeline(config)
    
    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)
            tag = "PROGRESS" if item['is_progress'] else "RESPONSE"
            print(f"[{tag}] {item['content']}")
        except asyncio.TimeoutError:
            break

asyncio.run(main())

package main

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

type ToolResult struct {
    ToolID    string      `json:"tool_id"`
    Result    interface{} `json:"result"`
    Error     error       `json:"error,omitempty"`
    Timestamp time.Time   `json:"timestamp"`
}

type AsyncToolManager struct {
    mu           sync.RWMutex
    pendingTools map[string]context.CancelFunc
    results      chan ToolResult
}

func NewAsyncToolManager() *AsyncToolManager {
    return &AsyncToolManager{
        pendingTools: make(map[string]context.CancelFunc),
        results:      make(chan ToolResult, 100),
    }
}

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("tool %s is already running", 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
}

func (m *AsyncToolManager) GetResult(timeout time.Duration) (*ToolResult, bool) {
    select {
    case result := <-m.results:
        return &result, true
    case <-time.After(timeout):
        return nil, false
    }
}

type VoiceAgent struct {
    toolManager *AsyncToolManager
    dialogueCh  chan string
}

func (a *VoiceAgent) HandleUserRequest(ctx context.Context, userSpeech string) {
    // 1. Immediate acknowledgment (no waiting for tool completion)
    a.dialogueCh <- "Sure, let me look into that for you..."
    
    // 2. Execute tools in background
    toolID := fmt.Sprintf("tool_%d", time.Now().UnixNano())
    _ = a.toolManager.ExecuteToolAsync(ctx, toolID,
        func(ctx context.Context) (interface{}, error) {
            select {
            case <-time.After(3 * time.Second):
                return map[string]interface{}{
                    "query_result": "simulated data",
                    "timestamp":    time.Now().Unix(),
                }, nil
            case <-ctx.Done():
                return nil, ctx.Err()
            }
        })
    
    // 3. Continue conversation — model can produce more speech
    a.dialogueCh <- "Meanwhile, I also found some related information..."
    a.dialogueCh <- "Feel free to keep asking — I'll update you when results come in."
    
    // 4. Background results seamlessly merge into conversation
    go func() {
        if result, ok := a.toolManager.GetResult(5 * time.Second); ok {
            if result.Error == nil {
                a.dialogueCh <- fmt.Sprintf("Results are ready: %v", result.Result)
            }
        }
    }()
}

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

@dataclass
class LanguageProfile:
    code: str
    name: str
    phoneme_set: set
    prosodic_features: dict

class MultilingualRouter:
    """
    Multilingual routing engine
    
    Core capabilities:
    - Real-time language detection from audio
    - Seamless mid-conversation language switching
    - Mixed-language input processing
    """
    
    def __init__(self):
        self.supported_languages: Dict[str, LanguageProfile] = {}
        self.confidence_threshold = 0.75
        self._init_language_profiles()
    
    def _init_language_profiles(self):
        languages = [
            ("zh-CN", "Chinese (Simplified)"),
            ("en-US", "English (US)"),
            ("ja-JP", "Japanese"),
            ("ko-KR", "Korean"),
            ("es-ES", "Spanish"),
            ("fr-FR", "French"),
            ("de-DE", "German"),
            ("pt-BR", "Portuguese"),
            ("ar-SA", "Arabic"),
            ("hi-IN", "Hindi"),
            ("ru-RU", "Russian"),
        ]
        
        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:
        """Detect language from audio frame using acoustic features"""
        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)
        return best_lang, scores[best_lang]
    
    def _extract_acoustic_features(self, audio: np.ndarray) -> np.ndarray:
        """Extract simplified MFCC-like acoustic features"""
        frame_size = 400
        hop_size = 160
        
        if len(audio) < frame_size:
            return np.zeros(13)
        
        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
            spectrum = np.abs(np.fft.rfft(windowed))
            features.append(spectrum[:13])
        
        return np.mean(features, axis=0)
    
    def _compute_language_similarity(
        self, features: np.ndarray, profile: LanguageProfile
    ) -> float:
        """Compute cosine similarity between features and profile"""
        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)
        
        return float(dot_product / norm_product) if norm_product > 0 else 0.0
    
    def _generate_phoneme_set(self, code: str) -> set:
        phoneme_sets = {
            "zh-CN": {"p", "pʰ", "t", "tʰ", "k", "kʰ", "m", "n", "ŋ",
                      "a", "o", "ə", "e", "ai", "ei", "ao", "ou"},
            "en-US": {"p", "b", "t", "d", "k", "g", "f", "v", "θ", "ð",
                      "s", "z", "ʃ", "ʒ", "h", "m", "n", "ŋ", "l", "r",
                      "iː", "ɪ", "eɪ", "ɛ", "æ", "ɑː", "ɒ", "ɔː", "oʊ",
                      "ʊ", "uː", "ʌ", "ɜː", "aɪ", "aʊ", "ɔɪ"},
        }
        return phoneme_sets.get(code, set())
    
    def _generate_prosodic_features(self, code: str) -> dict:
        return {
            "mfcc_mean": np.random.randn(13),
            "pitch_range": float(np.random.uniform(60, 300)),
            "speaking_rate": float(np.random.uniform(3, 8)),
        }
    
    def handle_code_switching(self, audio_stream: np.ndarray) -> List[Dict]:
        """Handle mid-conversation language switching"""
        segment_duration_ms = 500
        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": float(confidence)
                    })
                    current_lang = detected_lang
        
        return segments

# Usage example
router = MultilingualRouter()
test_audio = np.random.randn(16000 * 5)
switches = router.handle_code_switching(test_audio)
print(f"Detected {len(switches)} language 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 Audio Watermark Embedder
    
    Core principle: embed imperceptible pseudo-random signals in the
    spectral domain that remain detectable after re-encoding/compression.
    """
    
    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:
        """Generate watermark key — pseudo-random phase sequence"""
        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:
        """Compute psychoacoustic masking threshold"""
        n_fft = len(spectrum)
        mask = np.zeros_like(spectrum)
        for i in range(n_fft):
            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:
        """Embed SynthID watermark into audio using STFT domain"""
        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)
        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])
        )
        
        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]:
        """Detect SynthID watermark in audio"""
        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)))
        
        return confidence > 0.3, confidence

import asyncio
import websockets
import json

class GeminiLiveClient:
    """Gemini Live API client example"""
    
    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):
        """Start a real-time session"""
        async with websockets.connect(self.ws_url) as ws:
            setup_payload = {
                "setup": {
                    "model": f"models/{self.model}",
                    "system_instruction": {
                        "parts": [{"text": "You are a helpful AI assistant."}]
                    },
                    "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):
    """Real-time conversation state machine"""
    IDLE = "idle"
    LISTENING = "listening"
    PROCESSING = "processing"
    SPEAKING = "speaking"
    EXTENDED_THINKING = "extended_thinking"
    BACKGROUND_TOOL = "background_tool"
    INTERRUPTED = "interrupted"

class LiveConversationStateMachine:
    """
    Real-time conversation state machine
    
    Manages critical state transitions:
    - Interruption handling while user speaks
    - Extended Thinking parallel reasoning state
    - Background async tool lifecycle
    - Multimodal input coordination (voice + vision)
    """
    
    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,
                "tool_complete": ConversationState.SPEAKING
            },
            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