🎤 CLARISSA Voice Roundtrip - Local Testing¶

Cross-Platform Voice Pipeline Demo

This notebook demonstrates the complete voice roundtrip:

🎤 Microphone → 📁 WAV File → 🗣️ Whisper STT → 🧠 Intent Parser → ⚡ Result

Works on: macOS, Windows, Linux
Requirements: Python 3.9+, microphone access

---

In [ ]:

# Colab Setup — API Keys & Dependencies
import sys, os
IN_COLAB = 'google.colab' in sys.modules

if IN_COLAB:
    !pip install -q anthropic openai
    from google.colab import userdata
    # Set keys from Colab Secrets (Settings → Secrets → Add)
    try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY')
    except: pass
    try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY')
    except: pass
    try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN')
    except: pass
    # Fallback: manual input
    if not os.environ.get('ANTHROPIC_API_KEY'):
        import getpass
        for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']:
            if not os.environ.get(key):
                try: os.environ[key] = getpass.getpass(f'{key}: ')
                except: pass

print(f'Environment: {"Colab" if IN_COLAB else "Local"}')
for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']:
    print(f'  {k}: {"✓" if os.environ.get(k) else "✗"}')

# Colab Setup — API Keys & Dependencies import sys, os IN_COLAB = 'google.colab' in sys.modules if IN_COLAB: !pip install -q anthropic openai from google.colab import userdata # Set keys from Colab Secrets (Settings → Secrets → Add) try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY') except: pass try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY') except: pass try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN') except: pass # Fallback: manual input if not os.environ.get('ANTHROPIC_API_KEY'): import getpass for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']: if not os.environ.get(key): try: os.environ[key] = getpass.getpass(f'{key}: ') except: pass print(f'Environment: {"Colab" if IN_COLAB else "Local"}') for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']: print(f' {k}: {"✓" if os.environ.get(k) else "✗"}')

1️⃣ Installation¶

Run this cell to install required packages:

In [ ]:

# Cross-platform audio recording + API clients
!pip install -q sounddevice scipy numpy openai anthropic httpx

print("✅ Packages installed!")

# Cross-platform audio recording + API clients !pip install -q sounddevice scipy numpy openai anthropic httpx print("✅ Packages installed!")

2️⃣ Configuration¶

Set your API keys (at least one required):

In [ ]:

import os

# Option A: Set directly (for quick testing)
# os.environ['OPENAI_API_KEY'] = 'sk-...'
# os.environ['ANTHROPIC_API_KEY'] = 'sk-ant-...'

# Option B: Load from .env file or environment
# Keys should already be set in your shell profile

# Check what's available
openai_ok = bool(os.getenv('OPENAI_API_KEY'))
anthropic_ok = bool(os.getenv('ANTHROPIC_API_KEY'))

print("🔑 API Key Status:")
print(f"   OpenAI:    {'✅ Set' if openai_ok else '❌ Not set (required for Whisper STT)'}")
print(f"   Anthropic: {'✅ Set' if anthropic_ok else '⚠️ Not set (optional for intent parsing)'}")

if not openai_ok:
    print("\n⚠️ OpenAI API key required for Whisper transcription!")
    print("   Set it with: os.environ['OPENAI_API_KEY'] = 'sk-...'")

import os # Option A: Set directly (for quick testing) # os.environ['OPENAI_API_KEY'] = 'sk-...' # os.environ['ANTHROPIC_API_KEY'] = 'sk-ant-...' # Option B: Load from .env file or environment # Keys should already be set in your shell profile # Check what's available openai_ok = bool(os.getenv('OPENAI_API_KEY')) anthropic_ok = bool(os.getenv('ANTHROPIC_API_KEY')) print("🔑 API Key Status:") print(f" OpenAI: {'✅ Set' if openai_ok else '❌ Not set (required for Whisper STT)'}") print(f" Anthropic: {'✅ Set' if anthropic_ok else '⚠️ Not set (optional for intent parsing)'}") if not openai_ok: print("\n⚠️ OpenAI API key required for Whisper transcription!") print(" Set it with: os.environ['OPENAI_API_KEY'] = 'sk-...'")

3️⃣ Audio Recording Module¶

Cross-platform microphone recording using sounddevice:

In [ ]:

import sounddevice as sd
import scipy.io.wavfile as wav
import numpy as np
import tempfile
import time
from pathlib import Path
from IPython.display import Audio, display, HTML

# Audio settings
SAMPLE_RATE = 16000  # Whisper expects 16kHz
CHANNELS = 1         # Mono

def list_audio_devices():
    """Show available audio input devices."""
    print("🎤 Available Audio Input Devices:")
    print("-" * 50)
    devices = sd.query_devices()
    for i, d in enumerate(devices):
        if d['max_input_channels'] > 0:
            default = " (DEFAULT)" if i == sd.default.device[0] else ""
            print(f"   [{i}] {d['name']}{default}")
    print("-" * 50)
    return sd.default.device[0]

def record_audio(duration_seconds: float = 5.0, device: int = None) -> tuple[np.ndarray, str]:
    """
    Record audio from microphone.
    
    Args:
        duration_seconds: Recording duration
        device: Audio device index (None = default)
    
    Returns:
        (audio_data, wav_path)
    """
    print(f"\n🔴 Recording for {duration_seconds} seconds...")
    print("   Speak now!")
    
    # Record
    audio = sd.rec(
        int(duration_seconds * SAMPLE_RATE),
        samplerate=SAMPLE_RATE,
        channels=CHANNELS,
        dtype='int16',
        device=device
    )
    
    # Show countdown
    for i in range(int(duration_seconds), 0, -1):
        print(f"   ⏱️ {i}...", end='\r')
        time.sleep(1)
    
    sd.wait()  # Wait for recording to complete
    print("\n✅ Recording complete!")
    
    # Save to WAV file
    wav_path = tempfile.mktemp(suffix='.wav')
    wav.write(wav_path, SAMPLE_RATE, audio)
    
    # Calculate stats
    audio_flat = audio.flatten()
    rms = np.sqrt(np.mean(audio_flat.astype(np.float32)**2))
    peak = np.max(np.abs(audio_flat))
    
    print(f"\n📊 Audio Stats:")
    print(f"   Duration: {len(audio)/SAMPLE_RATE:.1f}s")
    print(f"   RMS Level: {rms:.0f} (good if > 500)")
    print(f"   Peak Level: {peak} / 32767")
    print(f"   File: {wav_path}")
    
    if rms < 200:
        print("\n⚠️ Audio level very low - check microphone!")
    
    return audio, wav_path

# List devices
default_device = list_audio_devices()

import sounddevice as sd import scipy.io.wavfile as wav import numpy as np import tempfile import time from pathlib import Path from IPython.display import Audio, display, HTML # Audio settings SAMPLE_RATE = 16000 # Whisper expects 16kHz CHANNELS = 1 # Mono def list_audio_devices(): """Show available audio input devices.""" print("🎤 Available Audio Input Devices:") print("-" * 50) devices = sd.query_devices() for i, d in enumerate(devices): if d['max_input_channels'] > 0: default = " (DEFAULT)" if i == sd.default.device[0] else "" print(f" [{i}] {d['name']}{default}") print("-" * 50) return sd.default.device[0] def record_audio(duration_seconds: float = 5.0, device: int = None) -> tuple[np.ndarray, str]: """ Record audio from microphone. Args: duration_seconds: Recording duration device: Audio device index (None = default) Returns: (audio_data, wav_path) """ print(f"\n🔴 Recording for {duration_seconds} seconds...") print(" Speak now!") # Record audio = sd.rec( int(duration_seconds * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=CHANNELS, dtype='int16', device=device ) # Show countdown for i in range(int(duration_seconds), 0, -1): print(f" ⏱️ {i}...", end='\r') time.sleep(1) sd.wait() # Wait for recording to complete print("\n✅ Recording complete!") # Save to WAV file wav_path = tempfile.mktemp(suffix='.wav') wav.write(wav_path, SAMPLE_RATE, audio) # Calculate stats audio_flat = audio.flatten() rms = np.sqrt(np.mean(audio_flat.astype(np.float32)**2)) peak = np.max(np.abs(audio_flat)) print(f"\n📊 Audio Stats:") print(f" Duration: {len(audio)/SAMPLE_RATE:.1f}s") print(f" RMS Level: {rms:.0f} (good if > 500)") print(f" Peak Level: {peak} / 32767") print(f" File: {wav_path}") if rms < 200: print("\n⚠️ Audio level very low - check microphone!") return audio, wav_path # List devices default_device = list_audio_devices()

4️⃣ Speech-to-Text (Whisper API)¶

Transcribe audio using OpenAI's Whisper with reservoir domain vocabulary:

In [ ]:

import httpx
import io

# Domain vocabulary to improve recognition
DOMAIN_VOCABULARY = """
Reservoir simulation terms: permeability, porosity, water saturation, 
oil saturation, pressure, BHP, bottomhole pressure, OOIP, STOIIP,
waterflood, injector, producer, PROD1, INJ1, INJ2, INJ3, INJ4,
millidarcy, mD, psi, bar, bbl/day, STB, FOPT, FOPR, FWPT, FWPR, FWCT,
water cut, GOR, gas-oil ratio, layer, grid, cell, timestep,
3D visualization, cross-section, animation, ECLIPSE, OPM Flow
"""

async def transcribe_audio(wav_path: str) -> dict:
    """
    Transcribe audio file using OpenAI Whisper API.
    
    Returns:
        {'text': str, 'latency_ms': int, 'cost_usd': float}
    """
    api_key = os.getenv('OPENAI_API_KEY')
    if not api_key:
        raise ValueError("OPENAI_API_KEY not set!")
    
    print("\n🗣️ Transcribing with Whisper API...")
    start_time = time.time()
    
    with open(wav_path, 'rb') as f:
        audio_bytes = f.read()
    
    # Estimate duration for cost calculation
    audio_duration_s = len(audio_bytes) / (SAMPLE_RATE * 2)  # 16-bit = 2 bytes
    
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            "https://api.openai.com/v1/audio/transcriptions",
            headers={"Authorization": f"Bearer {api_key}"},
            files={"file": ("audio.wav", audio_bytes, "audio/wav")},
            data={
                "model": "whisper-1",
                "language": "en",
                "prompt": DOMAIN_VOCABULARY,
                "response_format": "json"
            }
        )
    
    latency_ms = int((time.time() - start_time) * 1000)
    cost_usd = (audio_duration_s / 60) * 0.006  # $0.006/minute
    
    if response.status_code != 200:
        raise Exception(f"Whisper API error: {response.text}")
    
    text = response.json().get('text', '').strip()
    
    print(f"\n📝 Transcription Result:")
    print(f"   Text: \"{text}\"")
    print(f"   Latency: {latency_ms}ms")
    print(f"   Cost: ${cost_usd:.4f}")
    
    return {
        'text': text,
        'latency_ms': latency_ms,
        'cost_usd': cost_usd
    }

# Sync wrapper
def transcribe_audio_sync(wav_path: str) -> dict:
    import asyncio
    return asyncio.get_event_loop().run_until_complete(transcribe_audio(wav_path))

print("✅ Transcription module ready!")

import httpx import io # Domain vocabulary to improve recognition DOMAIN_VOCABULARY = """ Reservoir simulation terms: permeability, porosity, water saturation, oil saturation, pressure, BHP, bottomhole pressure, OOIP, STOIIP, waterflood, injector, producer, PROD1, INJ1, INJ2, INJ3, INJ4, millidarcy, mD, psi, bar, bbl/day, STB, FOPT, FOPR, FWPT, FWPR, FWCT, water cut, GOR, gas-oil ratio, layer, grid, cell, timestep, 3D visualization, cross-section, animation, ECLIPSE, OPM Flow """ async def transcribe_audio(wav_path: str) -> dict: """ Transcribe audio file using OpenAI Whisper API. Returns: {'text': str, 'latency_ms': int, 'cost_usd': float} """ api_key = os.getenv('OPENAI_API_KEY') if not api_key: raise ValueError("OPENAI_API_KEY not set!") print("\n🗣️ Transcribing with Whisper API...") start_time = time.time() with open(wav_path, 'rb') as f: audio_bytes = f.read() # Estimate duration for cost calculation audio_duration_s = len(audio_bytes) / (SAMPLE_RATE * 2) # 16-bit = 2 bytes async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "https://api.openai.com/v1/audio/transcriptions", headers={"Authorization": f"Bearer {api_key}"}, files={"file": ("audio.wav", audio_bytes, "audio/wav")}, data={ "model": "whisper-1", "language": "en", "prompt": DOMAIN_VOCABULARY, "response_format": "json" } ) latency_ms = int((time.time() - start_time) * 1000) cost_usd = (audio_duration_s / 60) * 0.006 # $0.006/minute if response.status_code != 200: raise Exception(f"Whisper API error: {response.text}") text = response.json().get('text', '').strip() print(f"\n📝 Transcription Result:") print(f" Text: \"{text}\"") print(f" Latency: {latency_ms}ms") print(f" Cost: ${cost_usd:.4f}") return { 'text': text, 'latency_ms': latency_ms, 'cost_usd': cost_usd } # Sync wrapper def transcribe_audio_sync(wav_path: str) -> dict: import asyncio return asyncio.get_event_loop().run_until_complete(transcribe_audio(wav_path)) print("✅ Transcription module ready!")

5️⃣ Intent Parser¶

Parse transcribed text into structured intents using rules + LLM fallback:

In [ ]:

from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
import re
import json

class IntentType(Enum):
    VISUALIZE_PROPERTY = "visualize_property"
    QUERY_VALUE = "query_value"
    MODIFY_PARAMETER = "modify_parameter"
    RUN_SIMULATION = "run_simulation"
    NAVIGATE = "navigate"
    HELP = "help"
    CONFIRM = "confirm"
    CANCEL = "cancel"
    UNKNOWN = "unknown"

@dataclass
class Intent:
    type: IntentType
    confidence: float
    slots: Dict[str, Any] = field(default_factory=dict)
    raw_text: str = ""
    parse_method: str = "unknown"

def parse_with_rules(text: str) -> Optional[Intent]:
    """
    Rule-based intent parsing - works WITHOUT any API.
    """
    text_lower = text.lower().strip()
    slots = {}
    
    # Control commands (highest priority)
    if text_lower in ["stop", "cancel", "never mind", "abort"]:
        return Intent(IntentType.CANCEL, 1.0, {}, text, "rules")
    
    if text_lower in ["yes", "yeah", "confirm", "ok", "okay", "do it", "go ahead"]:
        return Intent(IntentType.CONFIRM, 1.0, {}, text, "rules")
    
    if text_lower == "help" or "how do i" in text_lower or "what can" in text_lower:
        return Intent(IntentType.HELP, 1.0, {}, text, "rules")
    
    # Visualization commands
    viz_triggers = ["show", "display", "visualize", "plot", "view", "see"]
    if any(t in text_lower for t in viz_triggers):
        # Extract property
        if "perm" in text_lower:
            slots["property"] = "permeability"
        elif "poro" in text_lower:
            slots["property"] = "porosity"
        elif "saturation" in text_lower or " sw" in text_lower:
            slots["property"] = "water_saturation"
        elif "pressure" in text_lower or "bhp" in text_lower:
            slots["property"] = "pressure"
        
        # Extract layer
        layer_match = re.search(r'layer\s*(\d+)', text_lower)
        if layer_match:
            slots["layer"] = int(layer_match.group(1))
        
        # Extract time
        time_match = re.search(r'(?:day|time|at)\s*(\d+)', text_lower)
        if time_match:
            slots["time_days"] = int(time_match.group(1))
        
        # Extract view type
        if "3d" in text_lower:
            slots["view_type"] = "3d"
        elif "animation" in text_lower:
            slots["view_type"] = "animation"
        
        confidence = 0.95 if slots else 0.75
        return Intent(IntentType.VISUALIZE_PROPERTY, confidence, slots, text, "rules")
    
    # Query commands
    query_triggers = ["what is", "how much", "tell me", "current"]
    if any(t in text_lower for t in query_triggers):
        if "oil rate" in text_lower or "fopr" in text_lower:
            slots["property"] = "oil_rate"
        elif "water rate" in text_lower or "fwpr" in text_lower:
            slots["property"] = "water_rate"
        elif "water cut" in text_lower or "fwct" in text_lower:
            slots["property"] = "water_cut"
        elif "pressure" in text_lower:
            slots["property"] = "pressure"
        elif "gor" in text_lower:
            slots["property"] = "gor"
        
        if slots:
            return Intent(IntentType.QUERY_VALUE, 0.90, slots, text, "rules")
    
    # Navigation
    if any(t in text_lower for t in ["go to", "navigate", "open"]):
        if "result" in text_lower:
            slots["target"] = "results"
        elif "model" in text_lower:
            slots["target"] = "model"
        if slots:
            return Intent(IntentType.NAVIGATE, 0.85, slots, text, "rules")
    
    # Run simulation
    if any(t in text_lower for t in ["run sim", "start sim", "execute"]):
        return Intent(IntentType.RUN_SIMULATION, 0.85, {}, text, "rules")
    
    return None  # No rule matched

async def parse_with_llm(text: str) -> Intent:
    """
    Parse intent using LLM (Claude or GPT-4).
    """
    system_prompt = """Parse this reservoir simulation voice command into JSON.
Available intents: visualize_property, query_value, modify_parameter, run_simulation, navigate, help, confirm, cancel
Extract slots: property, layer, time_days, view_type, well, target
Respond with ONLY valid JSON: {"intent": "...", "confidence": 0.0-1.0, "slots": {...}}"""
    
    # Try Claude first
    if os.getenv('ANTHROPIC_API_KEY'):
        try:
            import anthropic
            client = anthropic.Anthropic()
            response = client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=200,
                messages=[{"role": "user", "content": f"{system_prompt}\n\nCommand: \"{text}\""}]
            )
            result = json.loads(response.content[0].text)
            return Intent(
                IntentType(result['intent']),
                result['confidence'],
                result.get('slots', {}),
                text,
                "claude"
            )
        except Exception as e:
            print(f"   Claude failed: {e}")
    
    # Fall back to GPT-4
    if os.getenv('OPENAI_API_KEY'):
        try:
            import openai
            client = openai.OpenAI()
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": f'Command: "{text}"'}
                ],
                temperature=0.1,
                max_tokens=200
            )
            result_text = response.choices[0].message.content
            if "```" in result_text:
                result_text = result_text.split("```")[1].replace("json", "").strip()
            result = json.loads(result_text)
            return Intent(
                IntentType(result['intent']),
                result['confidence'],
                result.get('slots', {}),
                text,
                "openai"
            )
        except Exception as e:
            print(f"   OpenAI failed: {e}")
    
    return Intent(IntentType.UNKNOWN, 0.0, {}, text, "none")

async def parse_intent(text: str) -> Intent:
    """
    Parse intent: Rules first, then LLM fallback.
    """
    print(f"\n🧠 Parsing intent...")
    
    # Try rules first (instant, no API cost)
    result = parse_with_rules(text)
    if result:
        print(f"   ✅ Parsed with rules")
        return result
    
    # Fall back to LLM
    print(f"   ↪ Rules didn't match, trying LLM...")
    return await parse_with_llm(text)

def parse_intent_sync(text: str) -> Intent:
    import asyncio
    return asyncio.get_event_loop().run_until_complete(parse_intent(text))

print("✅ Intent parser ready!")

from enum import Enum from dataclasses import dataclass, field from typing import Dict, Any, Optional import re import json class IntentType(Enum): VISUALIZE_PROPERTY = "visualize_property" QUERY_VALUE = "query_value" MODIFY_PARAMETER = "modify_parameter" RUN_SIMULATION = "run_simulation" NAVIGATE = "navigate" HELP = "help" CONFIRM = "confirm" CANCEL = "cancel" UNKNOWN = "unknown" @dataclass class Intent: type: IntentType confidence: float slots: Dict[str, Any] = field(default_factory=dict) raw_text: str = "" parse_method: str = "unknown" def parse_with_rules(text: str) -> Optional[Intent]: """ Rule-based intent parsing - works WITHOUT any API. """ text_lower = text.lower().strip() slots = {} # Control commands (highest priority) if text_lower in ["stop", "cancel", "never mind", "abort"]: return Intent(IntentType.CANCEL, 1.0, {}, text, "rules") if text_lower in ["yes", "yeah", "confirm", "ok", "okay", "do it", "go ahead"]: return Intent(IntentType.CONFIRM, 1.0, {}, text, "rules") if text_lower == "help" or "how do i" in text_lower or "what can" in text_lower: return Intent(IntentType.HELP, 1.0, {}, text, "rules") # Visualization commands viz_triggers = ["show", "display", "visualize", "plot", "view", "see"] if any(t in text_lower for t in viz_triggers): # Extract property if "perm" in text_lower: slots["property"] = "permeability" elif "poro" in text_lower: slots["property"] = "porosity" elif "saturation" in text_lower or " sw" in text_lower: slots["property"] = "water_saturation" elif "pressure" in text_lower or "bhp" in text_lower: slots["property"] = "pressure" # Extract layer layer_match = re.search(r'layer\s*(\d+)', text_lower) if layer_match: slots["layer"] = int(layer_match.group(1)) # Extract time time_match = re.search(r'(?:day|time|at)\s*(\d+)', text_lower) if time_match: slots["time_days"] = int(time_match.group(1)) # Extract view type if "3d" in text_lower: slots["view_type"] = "3d" elif "animation" in text_lower: slots["view_type"] = "animation" confidence = 0.95 if slots else 0.75 return Intent(IntentType.VISUALIZE_PROPERTY, confidence, slots, text, "rules") # Query commands query_triggers = ["what is", "how much", "tell me", "current"] if any(t in text_lower for t in query_triggers): if "oil rate" in text_lower or "fopr" in text_lower: slots["property"] = "oil_rate" elif "water rate" in text_lower or "fwpr" in text_lower: slots["property"] = "water_rate" elif "water cut" in text_lower or "fwct" in text_lower: slots["property"] = "water_cut" elif "pressure" in text_lower: slots["property"] = "pressure" elif "gor" in text_lower: slots["property"] = "gor" if slots: return Intent(IntentType.QUERY_VALUE, 0.90, slots, text, "rules") # Navigation if any(t in text_lower for t in ["go to", "navigate", "open"]): if "result" in text_lower: slots["target"] = "results" elif "model" in text_lower: slots["target"] = "model" if slots: return Intent(IntentType.NAVIGATE, 0.85, slots, text, "rules") # Run simulation if any(t in text_lower for t in ["run sim", "start sim", "execute"]): return Intent(IntentType.RUN_SIMULATION, 0.85, {}, text, "rules") return None # No rule matched async def parse_with_llm(text: str) -> Intent: """ Parse intent using LLM (Claude or GPT-4). """ system_prompt = """Parse this reservoir simulation voice command into JSON. Available intents: visualize_property, query_value, modify_parameter, run_simulation, navigate, help, confirm, cancel Extract slots: property, layer, time_days, view_type, well, target Respond with ONLY valid JSON: {"intent": "...", "confidence": 0.0-1.0, "slots": {...}}""" # Try Claude first if os.getenv('ANTHROPIC_API_KEY'): try: import anthropic client = anthropic.Anthropic() response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=200, messages=[{"role": "user", "content": f"{system_prompt}\n\nCommand: \"{text}\""}] ) result = json.loads(response.content[0].text) return Intent( IntentType(result['intent']), result['confidence'], result.get('slots', {}), text, "claude" ) except Exception as e: print(f" Claude failed: {e}") # Fall back to GPT-4 if os.getenv('OPENAI_API_KEY'): try: import openai client = openai.OpenAI() response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f'Command: "{text}"'} ], temperature=0.1, max_tokens=200 ) result_text = response.choices[0].message.content if "```" in result_text: result_text = result_text.split("```")[1].replace("json", "").strip() result = json.loads(result_text) return Intent( IntentType(result['intent']), result['confidence'], result.get('slots', {}), text, "openai" ) except Exception as e: print(f" OpenAI failed: {e}") return Intent(IntentType.UNKNOWN, 0.0, {}, text, "none") async def parse_intent(text: str) -> Intent: """ Parse intent: Rules first, then LLM fallback. """ print(f"\n🧠 Parsing intent...") # Try rules first (instant, no API cost) result = parse_with_rules(text) if result: print(f" ✅ Parsed with rules") return result # Fall back to LLM print(f" ↪ Rules didn't match, trying LLM...") return await parse_with_llm(text) def parse_intent_sync(text: str) -> Intent: import asyncio return asyncio.get_event_loop().run_until_complete(parse_intent(text)) print("✅ Intent parser ready!")

6️⃣ Complete Voice Roundtrip¶

This is the main cell - run it to test!

In [ ]:

import asyncio

async def voice_roundtrip(duration: float = 5.0):
    """
    Complete voice processing pipeline:
    🎤 Record → 📁 WAV → 🗣️ Transcribe → 🧠 Parse → ⚡ Result
    """
    print("="*60)
    print("🎤 CLARISSA Voice Roundtrip")
    print("="*60)
    
    total_start = time.time()
    
    # Step 1: Record
    print("\n📌 Step 1: Recording Audio")
    audio, wav_path = record_audio(duration)
    
    # Playback for verification
    print("\n🔊 Playback (verify your recording):")
    display(Audio(wav_path))
    
    # Step 2: Transcribe
    print("\n📌 Step 2: Speech-to-Text (Whisper)")
    transcription = await transcribe_audio(wav_path)
    
    if not transcription['text']:
        print("\n❌ No speech detected - try again!")
        return None
    
    # Step 3: Parse Intent
    print("\n📌 Step 3: Intent Parsing")
    intent = await parse_intent(transcription['text'])
    
    # Results
    total_time = time.time() - total_start
    
    print("\n" + "="*60)
    print("📊 ROUNDTRIP RESULTS")
    print("="*60)
    print(f"\n🎤 You said: \"{transcription['text']}\"")
    print(f"\n🎯 Intent: {intent.type.value}")
    print(f"   Confidence: {intent.confidence:.0%}")
    print(f"   Slots: {intent.slots}")
    print(f"   Parse method: {intent.parse_method}")
    print(f"\n⏱️ Timing:")
    print(f"   Transcription: {transcription['latency_ms']}ms")
    print(f"   Total roundtrip: {total_time:.1f}s")
    print(f"\n💰 Cost: ${transcription['cost_usd']:.4f}")
    
    # Simulated response
    print("\n" + "="*60)
    print("💬 CLARISSA would respond:")
    print("="*60)
    
    if intent.type == IntentType.VISUALIZE_PROPERTY:
        prop = intent.slots.get('property', 'permeability')
        layer = intent.slots.get('layer', 'all layers')
        print(f'\n   "Displaying {prop} for {layer}..."')
    elif intent.type == IntentType.QUERY_VALUE:
        prop = intent.slots.get('property', 'value')
        print(f'\n   "The current {prop} is 1,234 STB/day."')
    elif intent.type == IntentType.HELP:
        print('\n   "You can say things like: show permeability, what is the water cut, go to results..."')
    elif intent.type == IntentType.CANCEL:
        print('\n   "Operation cancelled."')
    else:
        print(f'\n   "I understood: {intent.type.value}. Processing..."')
    
    return {
        'transcription': transcription,
        'intent': intent,
        'total_time_s': total_time
    }

# Sync wrapper for easy execution
def run_voice_test(duration: float = 5.0):
    return asyncio.get_event_loop().run_until_complete(voice_roundtrip(duration))

print("✅ Voice roundtrip function ready!")
print("\n👇 Run the next cell to test!")

import asyncio async def voice_roundtrip(duration: float = 5.0): """ Complete voice processing pipeline: 🎤 Record → 📁 WAV → 🗣️ Transcribe → 🧠 Parse → ⚡ Result """ print("="*60) print("🎤 CLARISSA Voice Roundtrip") print("="*60) total_start = time.time() # Step 1: Record print("\n📌 Step 1: Recording Audio") audio, wav_path = record_audio(duration) # Playback for verification print("\n🔊 Playback (verify your recording):") display(Audio(wav_path)) # Step 2: Transcribe print("\n📌 Step 2: Speech-to-Text (Whisper)") transcription = await transcribe_audio(wav_path) if not transcription['text']: print("\n❌ No speech detected - try again!") return None # Step 3: Parse Intent print("\n📌 Step 3: Intent Parsing") intent = await parse_intent(transcription['text']) # Results total_time = time.time() - total_start print("\n" + "="*60) print("📊 ROUNDTRIP RESULTS") print("="*60) print(f"\n🎤 You said: \"{transcription['text']}\"") print(f"\n🎯 Intent: {intent.type.value}") print(f" Confidence: {intent.confidence:.0%}") print(f" Slots: {intent.slots}") print(f" Parse method: {intent.parse_method}") print(f"\n⏱️ Timing:") print(f" Transcription: {transcription['latency_ms']}ms") print(f" Total roundtrip: {total_time:.1f}s") print(f"\n💰 Cost: ${transcription['cost_usd']:.4f}") # Simulated response print("\n" + "="*60) print("💬 CLARISSA would respond:") print("="*60) if intent.type == IntentType.VISUALIZE_PROPERTY: prop = intent.slots.get('property', 'permeability') layer = intent.slots.get('layer', 'all layers') print(f'\n "Displaying {prop} for {layer}..."') elif intent.type == IntentType.QUERY_VALUE: prop = intent.slots.get('property', 'value') print(f'\n "The current {prop} is 1,234 STB/day."') elif intent.type == IntentType.HELP: print('\n "You can say things like: show permeability, what is the water cut, go to results..."') elif intent.type == IntentType.CANCEL: print('\n "Operation cancelled."') else: print(f'\n "I understood: {intent.type.value}. Processing..."') return { 'transcription': transcription, 'intent': intent, 'total_time_s': total_time } # Sync wrapper for easy execution def run_voice_test(duration: float = 5.0): return asyncio.get_event_loop().run_until_complete(voice_roundtrip(duration)) print("✅ Voice roundtrip function ready!") print("\n👇 Run the next cell to test!")

🚀 Run the Test!¶

Example commands to try:

• "Show me the permeability"
• "What is the water cut?"
• "Display porosity in 3D"
• "Show layer 5 at day 1000"
• "Help"
• "Cancel"

In [ ]:

# 🎤 RUN THIS TO TEST!
# Speak clearly into your microphone after running

result = run_voice_test(duration=5.0)  # 5 seconds recording

# 🎤 RUN THIS TO TEST! # Speak clearly into your microphone after running result = run_voice_test(duration=5.0) # 5 seconds recording

7️⃣ Batch Testing with Test Utterances¶

Test the intent parser without microphone (uses text input):

In [ ]:

# Test utterances from CLARISSA test suite
TEST_UTTERANCES = [
    ("show me the permeability", "visualize_property", {"property": "permeability"}),
    ("display porosity in 3D", "visualize_property", {"property": "porosity", "view_type": "3d"}),
    ("show layer 3", "visualize_property", {"layer": 3}),
    ("what is the oil rate", "query_value", {"property": "oil_rate"}),
    ("tell me the water cut", "query_value", {"property": "water_cut"}),
    ("go to the results", "navigate", {"target": "results"}),
    ("help", "help", {}),
    ("stop", "cancel", {}),
    ("yes", "confirm", {}),
]

print("🧪 Intent Parser Test Suite")
print("="*70)

passed = 0
for text, expected_intent, expected_slots in TEST_UTTERANCES:
    intent = parse_with_rules(text)
    
    if intent is None:
        status = "❌ NO MATCH"
    elif intent.type.value == expected_intent:
        # Check key slots
        slots_ok = all(intent.slots.get(k) == v for k, v in expected_slots.items())
        if slots_ok:
            status = "✅ PASS"
            passed += 1
        else:
            status = f"⚠️ SLOTS: got {intent.slots}"
    else:
        status = f"❌ INTENT: got {intent.type.value}"
    
    print(f"{status:20} | \"{text}\"")

print("="*70)
print(f"\n📊 Results: {passed}/{len(TEST_UTTERANCES)} passed ({100*passed/len(TEST_UTTERANCES):.0f}%)")

# Test utterances from CLARISSA test suite TEST_UTTERANCES = [ ("show me the permeability", "visualize_property", {"property": "permeability"}), ("display porosity in 3D", "visualize_property", {"property": "porosity", "view_type": "3d"}), ("show layer 3", "visualize_property", {"layer": 3}), ("what is the oil rate", "query_value", {"property": "oil_rate"}), ("tell me the water cut", "query_value", {"property": "water_cut"}), ("go to the results", "navigate", {"target": "results"}), ("help", "help", {}), ("stop", "cancel", {}), ("yes", "confirm", {}), ] print("🧪 Intent Parser Test Suite") print("="*70) passed = 0 for text, expected_intent, expected_slots in TEST_UTTERANCES: intent = parse_with_rules(text) if intent is None: status = "❌ NO MATCH" elif intent.type.value == expected_intent: # Check key slots slots_ok = all(intent.slots.get(k) == v for k, v in expected_slots.items()) if slots_ok: status = "✅ PASS" passed += 1 else: status = f"⚠️ SLOTS: got {intent.slots}" else: status = f"❌ INTENT: got {intent.type.value}" print(f"{status:20} | \"{text}\"") print("="*70) print(f"\n📊 Results: {passed}/{len(TEST_UTTERANCES)} passed ({100*passed/len(TEST_UTTERANCES):.0f}%)")

---

📚 Summary¶

This notebook demonstrates CLARISSA's voice pipeline:

Stage	Technology	Latency	Cost
🎤 Capture	sounddevice (cross-platform)	~5s (recording)	Free
🗣️ STT	OpenAI Whisper API	~500-1500ms	$0.006/min
🧠 Intent	Rules + Claude/GPT-4	<100ms (rules)	Free (rules)

Next Steps¶

1. Local Whisper: For air-gapped deployments, use faster-whisper
2. VAD: Add Voice Activity Detection for automatic start/stop
3. Streaming: Process audio in real-time for lower latency

References¶

• ADR-028: Voice Input Architecture
• OpenAI Whisper API