DEV Community: alfchee

The Hidden Costs of Real-Time: Latency vs Accuracy Trade-offs

alfchee — Fri, 03 Apr 2026 04:57:14 +0000

Hey developers! 🚀 After building a real-time speech-to-speech service with Nvidia Riva, I realized something crucial: real-time isn't just about speed—it's about making the right trade-offs. In this article, I want to share the hidden costs I discovered and how they shaped my engineering decisions.

The Illusion of "Real-Time"

When we say "real-time," we imagine instant responses. But in speech AI, there's always a tension between:

Latency: How fast we respond
Accuracy: How correct our output is
Stability: How consistent our service performs

Pick any two. All three? Impossible. That's the hidden cost nobody talks about.

The Buffer Size Dilemma

One of my first decisions was chunk size for audio streaming. Smaller chunks = faster response, but:

# Small chunks (64ms) - Lower latency, higher instability
audio_chunks = stream_audio(session_id, chunk_size=64)
# Processing: ~50ms per chunk
# Total latency: ~70ms

# Larger chunks (256ms) - Higher latency, more stable
audio_chunks = stream_audio(session_id, chunk_size=256)
# Processing: ~150ms per chunk  
# Total latency: ~200ms

With small chunks, I got complaints about "choppy" audio. With large chunks, users felt there was "lag." The solution? Adaptive buffering that adjusts based on network conditions.

GPU vs CPU: The Cost Calculation

I initially chose GPU because "it's faster." But the real cost analysis was more nuanced:

Factor	CPU (8 cores)	GPU (T4)
Per-session cost	$0.02/hr	$0.08/hr
Max concurrent	25	150
Latency	180ms	80ms
Dev complexity	Low	Medium

The GPU was 4x more expensive per session—but handled 6x more users with 2x better latency. For our scale, it was worth it. For yours? Run the numbers.

The Accuracy Trade-off Nobody Mentions

Here's the uncomfortable truth: faster models make more mistakes.

# Fast mode - Lower accuracy
config = RivaConfig(
    latency=50,  # ms
    word_boost=["um", "uh"],  # suppress filler words
)
# Result: "I um think it's working" → "I think it's working"

# Quality mode - Higher latency  
config = RivaConfig(
    latency=150,  # ms
    word_boost=[],
)
# Result: Full transcription with all filler words

Users prefer smooth, slightly inaccurate responses over accurate but choppy ones. The psychology of "feeling fast" often outweighs raw accuracy metrics.

What I Learned

Buffer adaptively - Static configurations fail in dynamic networks
Measure user perception, not just metrics - A 100ms response that feels smooth beats a 50ms response that glitches
Accept "good enough" - Perfect accuracy at 500ms latency loses to 95% accuracy at 100ms
Profile holistically - Single-component optimization means nothing if the pipeline still lags

The Bottom Line

Real-time systems aren't about eliminating latency—they're about managing expectations and making smart trade-offs. The best engineers don't chase impossible metrics; they design systems that prioritize the right compromises for their users.

From MVP to Production: Scaling a Speech AI Service

alfchee — Fri, 03 Apr 2026 04:53:13 +0000

Hey fellow builders! 🚀 After 7 articles diving into the technical details of my real-time speech service, I want to take a step back and share the bigger picture: how we scaled from a shaky MVP to a production-ready system handling 150+ concurrent users. This is the story that ties everything together.

The MVP Mindset (And Its Pitfalls)

Our first version worked—like most MVPs do. We had Flask handling WebSockets, Riva doing transcription, and basic TTS playback. Users could connect, speak, and get responses.

But "working" and "production-ready" are completely different categories:

# MVP code that "worked"
@app.websocket("/transcribe")
async def transcribe(websocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        result = riva.transcribe(data)  # Fire and forget
        await websocket.send_text(result)

This code has:

No error handling
No reconnection logic
No resource cleanup
No monitoring

It worked in demos. It would fail spectacularly in production.

The Migration to FastAPI (Article #1 Recap)

The first major shift was moving to FastAPI. But here's what nobody tells you: the code migration was the easy part.

The hard part was rethinking everything:

# Production-ready approach
@app.websocket("/transcribe/{session_id}")
async def transcribe(websocket: WebSocket, session_id: str):
    await websocket.accept()

    try:
        session = await create_session(session_id)
        async for message in websocket.iter_text():
            result = await process_audio(message, session)
            await websocket.send_text(result)
    except WebSocketDisconnect:
        logger.info(f"Session {session_id} disconnected")
    finally:
        await cleanup_session(session_id)  # Critical!

Lesson: Async isn't just syntax—it's a different mental model for resource management.

Adding Resilience Layer by Layer (Article #7)

Error handling evolved in phases:

Phase 1: Basic try-catch

try:
    result = riva.transcribe(audio)
except Exception as e:
    logger.error(f"Error: {e}")

Phase 2: Categorized errors

try:
    result = riva.transcribe(audio)
except RivaConnectionError as e:
    emit_reconnection_event(session_id)
except AudioFormatError as e:
    send_error_message(websocket, "Invalid audio format")
except RateLimitError:
    queue_request_for_retry(session_id)

Phase 3: Graceful degradation

async def transcribe_with_fallback(audio):
    try:
        return await riva.transcribe(audio)
    except RivaConnectionError:
        logger.warning("Riva unavailable, using fallback T2")
        return await fallback_service.transcribe(audio)

Lesson: Error handling isn't a layer you add—it's a journey that evolves with your system.

The Logging Evolution (Article #14)

In the MVP, we had print("here") scattered everywhere. Production required structure:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)

# Now we can correlate logs across services
logger = structlog.get_logger()
logger.info(
    "transcription_complete",
    session_id=session_id,
    latency_ms=latency,
    audio_chunks=chunk_count,
)

Lesson: Debugging production issues without structured logging is like trying to find a needle in a haystack—blindfolded.

Docker: From "It Works On My Machine" to Production (Article #9)

Our Dockerfile went through iterations:

# v1 - The MVP version
FROM python:3.9
COPY . /app
RUN pip install -r requirements.txt
CMD ["python", "app.py"]

# v2 - Production version
FROM nvidia/cuda:11.8-runtime-ubuntu22.04

# Install Python and dependencies
RUN apt-get update && apt-get install -y python3.9 python3-pip
COPY requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy only needed files
COPY ./app ./app

# Run as non-root user
RUN useradd -m appuser
USER appuser

CMD ["python", "-m", "uvicorn", "app:app", "--host", "0.0.0.0"]

Lesson: Docker isn't just packaging—it's defining your production environment.

Audio Quality: The Unexpected Challenge (Article #4)

We thought audio streaming was "solved." Then users started complaining about gaps and choppiness.

The solution involved:

Buffering with sequence numbers
Fade-out/fade-in for seamless transitions
Adaptive buffering based on network conditions

class AudioBuffer:
    def __init__(self):
        self.chunks = {}
        self.sequence = 0

    def add_chunk(self, audio_data: bytes, seq: int):
        self.chunks[seq] = audio_data
        self.sequence = max(self.sequence, seq)

    def get_ordered_audio(self) -> bytes:
        ordered = b"".join(
            self.chunks[i] 
            for i in sorted(self.chunks.keys())
        )
        return ordered

Lesson: The "last 10%" of quality takes 90% of the effort.

What I Would Do Different

If I could restart this journey:

Start with FastAPI - Don't accumulate technical debt with Flask
Add structured logging from day 1 - You will need it
Design for failure - Every component will fail; plan for it
Monitor everything - If you can't measure it, you can't improve it
Document decisions - Future you will thank present you

The Takeaway

Building an MVP is fun. Scaling it to production is where the real engineering happens. The 7 articles in this series covered the technical pieces, but the overarching lesson is:

Production systems aren't built—they're evolved.

Each challenge we faced—async programming, error handling, logging, GPU integration, audio quality, Docker—added another layer of resilience. That's what makes a service "production-ready."

Thanks for following this journey! What's your scaling story?

Docker Deployment for GPU-Accelerated Services

alfchee — Thu, 05 Mar 2026 00:25:12 +0000

Containerizing standard Python web apps is easy. Containerizing Python apps that need to talk to NVIDIA GPUs, manage gRPC streams, handle WebSockets at scale, and integrate with complex monitoring stacks? That's a different beast.

In this article, I'll share how we structured our Docker deployment for a GPU-accelerated Speech-to-Speech service, moving from a fragile "works on my machine" setup to a robust production infrastructure.

The Challenge: GPU & Environment Complexity

We faced three main challenges:

Dual Deployment Modes: We needed to support both "Cloud" (Nvidia NVCF) and "Self-Hosted" (On-premise GPU) modes from the same image.
Log Management: High-volume WebSocket traffic generates massive logs. We needed structured JSON for machines (Grafana/Loki) but readable text for developers.
Process Management: Uvicorn needs careful tuning for async workloads to avoid blocking the event loop.

1. The Entrypoint Pattern

Instead of a simple CMD ["uvicorn", ...], we implemented a robust run.sh entrypoint script. This allows us to handle environment setup before the application starts.

Dockerfile:

# ... builds setup ...
COPY run.sh .
RUN chmod +x run.sh
ENV FASTAPI_ENV=production
CMD ["./run.sh", "--env", "production"]

run.sh (Simplified):

#!/bin/bash
# Smart defaults based on environment
if [ "$FASTAPI_ENV" = "production" ]; then
    LOG_CONFIG="json"
    WORKERS=${WORKERS:-4}
else
    LOG_CONFIG="human"
    WORKERS=1
fi

# Launch Uvicorn with calculated arguments
exec uvicorn app:app \
    --host 0.0.0.0 \
    --workers $WORKERS \
    --log-config $LOG_CONFIG

This pattern gives us flexibility. We can flip a single environment variable FASTAPI_ENV and the container completely reconfigures its logging strategy and worker count.

2. Docker Compose Profiles

We use Docker Compose not just for local dev, but for defining deployment "flavors". By having separate compose files sharing the same core image, we document the requirements for each mode effectively.

docker-compose.cloud.yml (Minimal, API Keys only):

services:
  riva-s2s:
    image: riva-s2s:latest
    environment:
      - RIVA_DEPLOYMENT_MODE=cloud
      - RIVA_API_KEY=${RIVA_API_KEY}

docker-compose.selfhosted.yml (Heavy, internal GPU networking):

services:
  riva-s2s:
    image: riva-s2s:latest
    environment:
      - RIVA_DEPLOYMENT_MODE=self-hosted
      - RIVA_ASR_SERVER=10.0.1.5:50051
      - RIVA_TTS_SERVER=10.0.1.6:50051
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

3. Solving the Logging Nightmare

In a real-time WebSocket service, standard logging is useless. You get thousands of "Connected" / "Disconnected" messages. We implemented structured JSON logging that Loki can ingest.

Key Trick:
In your logging_config.py, check the LOG_FORMAT env var. If it's json, switch your formatter to python-json-logger.

# Docker sees this:
{"timestamp": "2026-03-04T12:00:00", "level": "INFO", "session_id": "abc-123", "event": "audio_chunk_received", "size_bytes": 4096}

This allows us to write Grafana queries like:
sum by (session_id) (rate({app="riva-s2s"} | json | event="audio_chunk_received"[1m]))

Benefits of this Approach

Immutability: The same Docker image runs in cloud prototyping and on-prem production.
Observability: JSON logging + Loki means we can trace a single audio packet through the entire stack.
Scalability: The run.sh script makes it trivial for Kubernetes or Docker Swarm to override worker counts without rebuilding.

Conclusion

Containerizing AI services isn't just about wrapping Python in Linux. It's about exposing the necessary knobs (Environment Variables) to control the complex underlying hardware and software stack without rebuilding the container. Treat your Dockerfile and entrypoint as first-class code.

Solving Audio Gaps in Real-Time Speech Translation

alfchee — Thu, 05 Mar 2026 00:20:47 +0000

When building a real-time Speech-to-Speech (S2S) translation service, latency is usually the enemy everyone talks about. But there's a silent killer (quite literally) that can ruin the user experience just as effectively: audio gaps.

In our journey migrating from Flask to FastAPI and implementing Nvidia Riva, we encountered a persistent issue where our synthesized audio had audible stuttering—specifically, 20ms gaps of silence between chunks. Here’s how we diagnosed and fixed it, turning a robotic output into a smooth, natural conversation.

The Problem: "Machine Gun" Audio

Our pipeline looked standard:

Receive user audio (WebSocket)
Transcribe (ASR) & Translate (NMT)
Synthesize speech (TTS)
Stream audio back to the client

But the output sounded like a machine gun. Words were clear, but the flow was choppy. Opening the raw audio dump in Audacity revealed the culprit: consistent 20-50ms gaps of silence inserted between every audio chunk returned by the TTS service.

What Wasn't The Cause

Network Latency: The gaps were present even when saving to a local file.
Frontend Playback: The gaps existed in the source PCM data.
Sample Rates: 24kHz in, 24kHz out. No mismatches.

Root Cause Analysis

After deep diving into the Riva TTS behavior, we found three contributing factors:

Padding by Design: The TTS model often pads the beginning and end of synthesized audio with silence.
Imperfect Silence: This "silence" wasn't always digital zero (0x00). It often contained low-amplitude noise (0x01, 0x02), meaning our simple if sample != 0 checks failed to detect it.
Fragmented Synthesis: We were sending text to the TTS engine too aggressively (sentence by sentence or even phrase by phrase). Each request generated its own padding, compounding the issue.

The Solution: A Three-Pronged Approach

1. Aggressive Silence Trimming

We moved from a simple "zero check" to a threshold-based trim. Since the audio is 16-bit PCM, values under 100 are effectively inaudible but prevent "perfect silence" detection.

# Before: Only removed perfect zeros (ineffective)
non_zero = np.where(audio_data != 0)[0]

# After: Remove near-silence (threshold-based)
SILENCE_THRESHOLD = 100
non_silent = np.where(np.abs(audio_data) > SILENCE_THRESHOLD)[0]

if len(non_silent) > 0:
    # Keep only the audible part
    trimmed_audio = audio_data[non_silent[0] : non_silent[-1] + 1]

This immediately removed about 10-30ms of "dead air" per chunk.

2. Audio Crossfading (Windowing)

Even after trimming, stitching two audio clips together can cause a "click" if the waveform jumps instantly from one amplitude to another. We implemented a 5ms linear fade-out on every chunk.

def apply_fade_out(audio_bytes: bytes, sample_rate: int = 24000, fade_ms: int = 5) -> bytes:
    audio_data = np.frombuffer(audio_bytes, dtype=np.int16)
    num_samples = int(sample_rate * (fade_ms / 1000))

    # Linear ramp from 1.0 to 0.0
    fade_curve = np.linspace(1.0, 0.0, num_samples)

    # Apply to the very end of the chunk
    audio_data[-num_samples:] = audio_data[-num_samples:] * fade_curve

    return audio_data.tobytes()

This acts like a micro-crossfade, ensuring every chunks ends at zero amplitude, eliminating the "click" sound at boundaries.

3. Smart Text Aggregation

Finally, we stopped sending every partial sentence to the TTS engine. We increased our text buffer size to accumulate more context before requesting synthesis.

Before: Flush to TTS every 3 segments.
After: Flush to TTS every 5 segments (or on punctuation).

Fewer requests = fewer boundaries = fewer gaps.

Results

Audio Gaps: Eliminated.
Playback Smoothness: Indistinguishable from a single continuous file.
Latency Cost: Negligible (<2ms processing overhead).

Key Takeaway

In real-time audio, how you handle the bytes is just as important as the model generating them. Models are imperfect; your DSP pipeline needs to clean up the mess. Don't trust "silence" to be zero, and never stitch audio without smoothing the edges.

Vue.js Composables for Speech-to-Speech Translation: Building Reactive Real-Time Audio Applications

alfchee — Mon, 16 Feb 2026 17:25:16 +0000

Introduction: Why Vue.js Composables Changed Everything

When I first set out to build a real-time speech-to-speech translation service frontend, I had already implemented a React version using hooks. "How different could Vue be?" I thought. Turns out, Vue's Composition API offered a refreshingly elegant approach to managing WebSocket connections, audio streams, and reactive state that made me rethink how I structure real-time applications.

In this article, I'll share the lessons I learned building Vue.js composables for real-time speech services, the challenges I faced, and the patterns that emerged from managing reactive WebSocket connections and audio playback. Whether you're coming from React, Angular, or are a Vue veteran, there's something here for everyone building real-time audio applications.

The Challenge: Real-Time Audio Streaming in the Browser

Before we dive into the code, let's understand what we're building:

WebSocket Connection Management: Establish and maintain a persistent connection to our FastAPI backend
Audio Capture: Access the user's microphone and stream audio chunks
Reactive State: Keep the UI in sync with connection status, transcription results, and errors
Lifecycle Management: Properly clean up resources when components unmount
Error Handling: Gracefully handle network failures, permission denials, and API errors

The beauty of Vue's Composition API is that we can encapsulate each of these concerns into reusable, testable composables.

Lesson 1: Composables Are More Than React Hooks with Different Syntax

Coming from React, I initially thought composables were just hooks with a different name. I was wrong. Here's what makes them different:

Vue Composables Have Direct Reactive References

// React Hook - needs useState
const [isConnected, setIsConnected] = useState(false);

// Vue Composable - direct reactive reference
const isConnected = ref(false);

This might seem like a small difference, but it has profound implications:

No dependency arrays: Vue's reactivity system tracks dependencies automatically
No stale closures: You're always working with the current value
Simpler mental model: Just mutate the .value property

Lifecycle is More Explicit

// React needs useEffect with cleanup
useEffect(() => {
  // Setup
  return () => {
    // Cleanup
  };
}, []);

// Vue has dedicated lifecycle hooks
onUnmounted(() => {
  // Cleanup
});

Lesson 2: Building the WebSocket Composable

Let me show you the core useTranscription composable I built. This handles all WebSocket communication with our FastAPI backend.

The Basic Structure

// composables/useTranscription.ts
import { ref, onUnmounted } from 'vue';
import { v4 as uuidv4 } from 'uuid';

export function useTranscription(apiUrl = 'ws://localhost:8000') {
  // Reactive state
  const ws = ref<WebSocket | null>(null);
  const sessionId = ref<string>('');
  const isConnected = ref(false);
  const isTranscribing = ref(false);
  const partialTranscript = ref('');
  const finalTranscript = ref('');
  const error = ref<string | null>(null);

  // Methods will go here...

  // Cleanup on unmount
  onUnmounted(() => {
    disconnect();
  });

  return {
    // State
    isConnected,
    isTranscribing,
    partialTranscript,
    finalTranscript,
    error,
    sessionId,

    // Methods
    connect,
    startTranscription,
    sendAudioChunk,
    stopTranscription,
    disconnect,
  };
}

Key Design Decision: Promise-Based Connection

One mistake I made initially was making the connection synchronous. When the UI tried to start recording immediately after connecting, the WebSocket wasn't ready. Here's how I solved it:

const connect = async (): Promise<void> => {
  return new Promise((resolve, reject) => {
    try {
      sessionId.value = uuidv4();
      const url = `${apiUrl}/transcribe/${sessionId.value}`;

      ws.value = new WebSocket(url);

      ws.value.onopen = () => {
        console.log('WebSocket connected');
      };

      ws.value.onmessage = (event) => {
        const message = JSON.parse(event.data);

        if (message.event === 'connected') {
          isConnected.value = true;
          resolve(); // ✅ Resolve when truly ready
        }
        // Handle other messages...
      };

      ws.value.onerror = (err) => {
        error.value = 'WebSocket connection error';
        reject(err);
      };

      // Timeout after 10 seconds
      setTimeout(() => {
        if (!isConnected.value) {
          reject(new Error('Connection timeout'));
        }
      }, 10000);
    } catch (err) {
      reject(err);
    }
  });
};

Why this matters: The UI can now reliably wait for the connection:

// In your component
const handleConnect = async () => {
  try {
    await connect(); // ✅ Waits for actual connection
    await startTranscription(); // ✅ Safe to start now
  } catch (err) {
    console.error('Connection failed:', err);
  }
};

Lesson 3: Reactive Message Handling

The beauty of Vue's reactivity shines when handling WebSocket messages. Here's how I structured message handling:

const handleTranscript = (message: TranscriptResult) => {
  if (message.type === 'partial') {
    partialTranscript.value = message.transcript || '';
  } else if (message.type === 'final') {
    finalTranscript.value += (message.transcript || '') + ' ';
    partialTranscript.value = ''; // Clear partial
  }
};

In the template, this just works:

<template>
  <div class="transcripts">
    <!-- Updates automatically when partialTranscript changes -->
    <p class="partial">{{ partialTranscript }}</p>

    <!-- Updates automatically when finalTranscript changes -->
    <p class="final">{{ finalTranscript }}</p>
  </div>
</template>

No manual updates, no forceUpdate, no setState callbacks. Vue's reactivity system handles everything.

Lesson 4: Audio Capture Composable

The second major composable handles microphone access. Here's where browser APIs meet Vue's reactivity:

// composables/useAudioCapture.ts
import { ref, onUnmounted } from 'vue';

export function useAudioCapture() {
  const isRecording = ref(false);
  const mediaRecorder = ref<MediaRecorder | null>(null);
  const audioStream = ref<MediaStream | null>(null);

  const startRecording = async (
    onAudioData: (data: Blob) => void
  ): Promise<void> => {
    try {
      // Request microphone access
      audioStream.value = await navigator.mediaDevices.getUserMedia({
        audio: {
          channelCount: 1, // Mono
          sampleRate: 16000, // 16kHz for speech
          echoCancellation: true,
          noiseSuppression: true,
        },
      });

      mediaRecorder.value = new MediaRecorder(audioStream.value, {
        mimeType: 'audio/webm',
        audioBitsPerSecond: 16000,
      });

      // Stream audio chunks
      mediaRecorder.value.ondataavailable = (event) => {
        if (event.data.size > 0) {
          onAudioData(event.data); // Callback to send via WebSocket
        }
      };

      // Capture in 100ms chunks for real-time streaming
      mediaRecorder.value.start(100);
      isRecording.value = true;
    } catch (err) {
      console.error('Microphone access denied:', err);
      throw err;
    }
  };

  const stopRecording = () => {
    if (mediaRecorder.value && isRecording.value) {
      mediaRecorder.value.stop();
      isRecording.value = false;
    }

    // Important: Stop all tracks to release microphone
    if (audioStream.value) {
      audioStream.value.getTracks().forEach((track) => track.stop());
      audioStream.value = null;
    }
  };

  // Automatic cleanup
  onUnmounted(() => {
    stopRecording();
  });

  return {
    isRecording,
    startRecording,
    stopRecording,
  };
}

Critical Lesson: Resource Cleanup

I learned this the hard way: failing to stop media tracks keeps the microphone active even after the component unmounts. Users see that scary "microphone is active" indicator in their browser, and they panic.

Always stop media tracks:

audioStream.value.getTracks().forEach((track) => track.stop());

Lesson 5: Bringing It Together in a Component

Here's how beautifully these composables work together:

<template>
  <div class="transcription-app">
    <div class="status">
      <span :class="{ active: isConnected }">
        {{ isConnected ? '🟢 Connected' : '🔴 Disconnected' }}
      </span>
      <span :class="{ active: isRecording }">
        {{ isRecording ? '🎤 Recording' : '⏸️ Stopped' }}
      </span>
    </div>

    <div class="controls">
      <button @click="handleStart" :disabled="isRecording">
        Start Recording
      </button>
      <button @click="handleStop" :disabled="!isRecording">
        Stop Recording
      </button>
    </div>

    <div v-if="error" class="error">⚠️ {{ error }}</div>

    <div class="transcripts">
      <p v-if="partialTranscript" class="partial">
        {{ partialTranscript }}
      </p>
      <p class="final">{{ finalTranscript }}</p>
    </div>
  </div>
</template>

<script setup lang="ts">
import { onMounted } from 'vue';
import { useTranscription } from '../composables/useTranscription';
import { useAudioCapture } from '../composables/useAudioCapture';

// Composables
const {
  isConnected,
  isTranscribing,
  partialTranscript,
  finalTranscript,
  error,
  connect,
  startTranscription,
  sendAudioChunk,
  stopTranscription,
} = useTranscription('ws://localhost:8000');

const { isRecording, startRecording, stopRecording } = useAudioCapture();

// Auto-connect on mount
onMounted(async () => {
  try {
    await connect();
  } catch (err) {
    console.error('Failed to connect:', err);
  }
});

// Start recording and transcription
const handleStart = async () => {
  try {
    await startTranscription({
      language_code: 'en-US',
      enable_automatic_punctuation: true,
    });

    await startRecording((audioBlob) => {
      sendAudioChunk(audioBlob);
    });
  } catch (err) {
    console.error('Failed to start:', err);
  }
};

// Stop everything
const handleStop = async () => {
  stopRecording();
  await stopTranscription();
};
</script>

<style scoped>
.status span.active {
  color: #00ff00;
}

.partial {
  color: #888;
  font-style: italic;
}

.final {
  color: #000;
  font-weight: bold;
}

.error {
  color: red;
  padding: 10px;
  background: #ffeeee;
}
</style>

Lesson 6: Composition API vs Options API

You might be wondering: "Why use the Composition API instead of the Options API?"

Here's what I discovered:

Composition API Wins:

Better code organization: Related logic stays together
Easier to test: Composables are just functions
TypeScript support: Inference works beautifully
Reusability: Share logic across components effortlessly

Options API was Better For:

Small, simple components: Sometimes data() and methods are clearer
Learning Vue: The structure is more explicit
Team familiarity: If your team knows Options API well

For real-time applications with complex state management, the Composition API is the clear winner.

Lesson 7: Error Handling and Edge Cases

Real-world applications need robust error handling. Here are the edge cases I encountered:

1. User Denies Microphone Permission

const handleStart = async () => {
  try {
    await startRecording(onAudioData);
  } catch (err) {
    if (err.name === 'NotAllowedError') {
      error.value = 'Microphone permission denied. Please allow access.';
    } else {
      error.value = 'Failed to start recording';
    }
  }
};

2. WebSocket Connection Lost

ws.value.onclose = (event) => {
  isConnected.value = false;
  isTranscribing.value = false;

  if (!event.wasClean) {
    error.value = 'Connection lost. Please refresh and try again.';
  }
};

3. Browser Compatibility

const checkBrowserSupport = () => {
  if (!('MediaRecorder' in window)) {
    throw new Error('MediaRecorder API not supported');
  }

  if (!('WebSocket' in window)) {
    throw new Error('WebSocket API not supported');
  }
};

Lesson 8: Performance Optimization

Real-time applications demand excellent performance. Here's what worked:

1. Debounce UI Updates for Partial Transcripts

import { ref, watch } from 'vue';
import { debounce } from 'lodash-es';

const debouncedPartial = ref('');

watch(partialTranscript, debounce((newValue) => {
  debouncedPartial.value = newValue;
}, 50));

2. Use `shallowRef` for Large Objects

import { shallowRef } from 'vue';

// Don't make every property reactive
const allTranscripts = shallowRef<TranscriptResult[]>([]);

3. Lazy Load Audio Components

<script setup lang="ts">
import { defineAsyncComponent } from 'vue';

const AudioVisualizer = defineAsyncComponent(
  () => import('./AudioVisualizer.vue')
);
</script>

Lesson 9: Testing Composables

The Composition API makes testing a joy. Here's how I test the WebSocket composable:

import { describe, it, expect, vi } from 'vitest';
import { useTranscription } from '../useTranscription';

describe('useTranscription', () => {
  it('should connect to WebSocket', async () => {
    const { connect, isConnected } = useTranscription();

    await connect();

    expect(isConnected.value).toBe(true);
  });

  it('should handle transcript messages', async () => {
    const { connect, partialTranscript } = useTranscription();

    await connect();

    // Simulate WebSocket message
    const message = {
      event: 'transcript',
      type: 'partial',
      transcript: 'Hello world',
    };

    // Emit message...

    expect(partialTranscript.value).toBe('Hello world');
  });
});

Lesson 10: What I'd Do Differently

Looking back, here's what I'd change:

1. Add Retry Logic Earlier

Instead of failing on connection errors, implement exponential backoff:

const connectWithRetry = async (maxRetries = 3) => {
  for (let i = 0; i < maxRetries; i++) {
    try {
      await connect();
      return;
    } catch (err) {
      if (i === maxRetries - 1) throw err;
      await new Promise(resolve => 
        setTimeout(resolve, Math.pow(2, i) * 1000)
      );
    }
  }
};

2. Use TypeScript from Day One

The type safety caught so many bugs during refactoring.

3. Implement Audio Queueing

For speech-to-speech translation, I should have queued audio output to prevent overlapping playback.

4. Add Connection State Machine

Managing connection states (connecting, connected, disconnecting, disconnected, error) would have simplified the logic.

Comparison: Vue Composables vs React Hooks

Having implemented both, here's my honest comparison:

Aspect	Vue Composables	React Hooks
Learning Curve	Gentler for beginners	Steeper (dependency arrays, closures)
Reactivity	Automatic, magical ✨	Manual via setState
Ref Management	Explicit `.value` access	Implicit
Lifecycle	Dedicated hooks	useEffect with deps
Code Reuse	Natural and intuitive	Requires careful hook design
TypeScript	Excellent inference	Good, but needs more annotations
Performance	Fine-grained reactivity	Virtual DOM diffing
Debugging	Vue DevTools is amazing	React DevTools is great

My verdict: For real-time applications, Vue's automatic reactivity saves significant mental overhead. React hooks are more explicit and give you more control, but require more boilerplate.

Real-World Performance Results

Here's what I achieved with this Vue implementation:

Connection establishment: ~200ms average
Audio chunk transmission: 100ms intervals, <10ms latency
UI updates: 60fps even with rapid transcript updates
Memory footprint: ~15MB for the entire app
Bundle size: 45KB (gzipped, excluding Vue)

Conclusion: The Joy of Vue Composables

Building real-time speech applications with Vue.js composables has been a revelation. The Composition API's elegant reactivity system, combined with TypeScript's type safety, creates a developer experience that's hard to beat.

Key Takeaways:

Composables promote clean separation of concerns - each composable handles one responsibility
Reactivity just works - no manual state updates needed
Lifecycle management is explicit - onUnmounted makes cleanup obvious
Promise-based patterns integrate seamlessly with async/await
Testing is straightforward - composables are just functions

If you're building real-time applications in Vue, embrace the Composition API. Your future self (and your team) will thank you.

Resources

GPU-Accelerated Speech AI: Integrating Nvidia Riva in Python

alfchee — Sat, 07 Feb 2026 18:07:38 +0000

Hey AI builders! 🚀 Today I want to share how I integrated Nvidia Riva into a real-time speech service and what I learned while optimizing a GPU-backed pipeline in Python. If you’re building low-latency ASR or S2S systems, this is one of those “level-up” integrations that will show off your engineering skills and deliver real user impact.

Why GPU Acceleration Was Non‑Negotiable

We started with CPU-based inference. It worked, but in real-time systems “it works” is not enough:

Latency ballooned during peak traffic
Concurrency hit a wall because each session hogged CPU
Scaling was expensive and still failed to feel real-time

Riva flipped the equation: more throughput, lower latency, and a cleaner streaming API. That’s when I knew this integration would be a cornerstone skill in my toolbox.

What Riva Brings to the Table

Riva is Nvidia’s GPU-accelerated speech SDK. In one platform you get:

ASR (speech-to-text)
NMT (translation)
TTS (text-to-speech)
Streaming-first architecture built for real-time apps

In short: I could finally build a pipeline that felt instant to users.

The Integration Pattern That Worked

The key is embracing gRPC streaming instead of trying to force batch logic into a realtime system.

1) Secure Auth + Service Setup

import riva.client

def build_riva_auth(api_key: str, function_id: str):
    return riva.client.Auth(
        use_ssl=True,
        uri="grpc.nvcf.nvidia.com:443",
        metadata_args=[
            ["function-id", function_id],
            ["authorization", f"Bearer {api_key}"]
        ]
    )

auth = build_riva_auth(api_key="YOUR_KEY", function_id="YOUR_FUNCTION_ID")
asr_service = riva.client.ASRService(auth)

2) Streaming Generator (The Heart of It)

import queue

class RivaStream:
    def __init__(self, asr_service):
        self.asr_service = asr_service
        self.audio_queue = queue.Queue()

    def audio_generator(self):
        while True:
            chunk = self.audio_queue.get()
            if chunk is None:
                break
            yield chunk

    def push(self, chunk: bytes):
        self.audio_queue.put(chunk)

    def close(self):
        self.audio_queue.put(None)

3) Streaming Config That Fits Real-Time

import riva.client

def build_streaming_config(lang="en"):
    return riva.client.StreamingRecognitionConfig(
        config=riva.client.RecognitionConfig(
            encoding=riva.client.AudioEncoding.LINEAR_PCM,
            sample_rate_hertz=16000,
            language_code=lang,
            max_alternatives=1,
            enable_automatic_punctuation=True,
            audio_channel_count=1,
        ),
        interim_results=True,
    )

4) Consuming the Stream

def stream_transcription(riva_stream, config):
    responses = riva_stream.asr_service.streaming_response_generator(
        audio_chunks=riva_stream.audio_generator(),
        streaming_config=config
    )

    for response in responses:
        if not response.results:
            continue
        for result in response.results:
            if result.alternatives:
                yield {
                    "text": result.alternatives[0].transcript,
                    "is_final": result.is_final,
                    "stability": result.stability
                }

That generator-based design gave me a clean, testable, and scalable integration point—exactly the kind of engineering pattern I’m proud to showcase.

The Real-Time WebSocket Bridge

I wired the Riva stream into FastAPI WebSockets to support live audio ingestion and low-latency results.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import json
import asyncio

app = FastAPI()

@app.websocket("/transcribe/{session_id}")
async def transcribe(session_id: str, websocket: WebSocket):
    await websocket.accept()
    riva_stream = RivaStream(asr_service)
    config = build_streaming_config(lang="en")

    async def send_results():
        for result in stream_transcription(riva_stream, config):
            await websocket.send_json({
                "event": "transcription",
                **result
            })

    try:
        task = asyncio.create_task(send_results())
        while True:
            audio_chunk = await websocket.receive_bytes()
            riva_stream.push(audio_chunk)
    except WebSocketDisconnect:
        pass
    finally:
        riva_stream.close()
        await task

Reliability: Where Production Skills Really Show

This is where I invested time to make it production-grade:

Retry logic with exponential backoff for gRPC failures
Timeouts to avoid stuck streams
Graceful disconnect handling to prevent leaked resources
Structured logs for tracing latency spikes

import grpc
from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientRivaClient:
    def __init__(self, api_key: str, function_id: str):
        self.api_key = api_key
        self.function_id = function_id
        self.service = self._create_service()

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def _create_service(self):
        auth = build_riva_auth(self.api_key, self.function_id)
        return riva.client.ASRService(auth)

    def reset(self):
        self.service = self._create_service()

Performance Tuning That Actually Matters

These are the optimizations that delivered real-world gains:

Chunk sizing: ~100ms frames kept latency low without spamming gRPC.
Connection reuse: pooling channels reduced handshake costs.
Backpressure-aware queues: prevented runaway memory usage.
GPU telemetry: kept utilization in a healthy range.

CHUNK_SIZE = 3200  # 16kHz * 2 bytes * 0.1s

Results I Was Proud to Share

Latency dropped by ~90% in streaming use cases
Concurrent sessions per instance jumped 10–15x
User experience felt instant—no more “wait for it” delays

These wins weren’t just technical; they proved I could design systems that scale in real-world constraints.

Final Takeaways

Integrating Riva wasn’t just about faster inference. It was about building an architecture that’s clean, resilient, and scalable. From gRPC streaming patterns to production-grade error recovery, this project sharpened the exact skills companies look for in real-time AI engineers.

If you’re considering GPU-accelerated speech AI, start small, embrace streaming early, and invest in reliability from day one. Your users will feel the difference—and your engineering profile will too.

Logging Strategies for Real-Time Applications: Session Tracking at Scale

alfchee — Sat, 31 Jan 2026 00:40:11 +0000

Hey builders! 👋 Let's talk about something that sounds boring but becomes absolutely critical in production: logging. When you're running hundreds of concurrent sessions, bad logging is the difference between finding bugs in minutes vs. spending days debugging.

Let me share how we built a logging system that actually helps instead of drowns you in noise.

The Real-Time Logging Challenge

Traditional logging advice doesn't work for real-time apps. Here's why:

Traditional app logging:

2024-01-15 10:30:45 INFO Processing request
2024-01-15 10:30:46 ERROR Failed to connect to database
2024-01-15 10:30:47 INFO Processing request

Real-time app with 100 concurrent sessions:

2024-01-15 10:30:45.123 INFO Processing audio
2024-01-15 10:30:45.124 INFO Processing audio
2024-01-15 10:30:45.125 ERROR Connection failed
2024-01-15 10:30:45.126 INFO Processing audio
2024-01-15 10:30:45.127 INFO Processing audio
2024-01-15 10:30:45.128 INFO Processing audio

Which session failed? Good luck finding out.

Strategy #1: Session-Based Logging

Every log entry MUST include session context:

import logging
import uuid
from contextvars import ContextVar
from typing import Optional

# Context variable for session tracking
session_context: ContextVar[Optional[str]] = ContextVar('session_context', default=None)

class SessionLoggerAdapter(logging.LoggerAdapter):
    """Logger that automatically includes session context"""

    def process(self, msg, kwargs):
        session_id = session_context.get()
        if session_id:
            return f'[{session_id}] {msg}', kwargs
        return msg, kwargs

def get_logger(name: str) -> SessionLoggerAdapter:
    """Get a session-aware logger"""
    base_logger = logging.getLogger(name)
    return SessionLoggerAdapter(base_logger, {})

# Usage in your endpoint
logger = get_logger(__name__)

@app.websocket("/transcribe/{session_id}")
async def transcribe_endpoint(websocket: WebSocket, session_id: str):
    # Set session context for this async task
    session_context.set(session_id)

    logger.info("Session started")  # Logs: [abc-123] Session started

    try:
        await process_transcription(websocket)
    except Exception as e:
        logger.error(f"Transcription failed: {e}")  # Logs: [abc-123] Transcription failed: ...
    finally:
        logger.info("Session ended")  # Logs: [abc-123] Session ended

Now every log line is traceable to a specific session!

Strategy #2: Structured Logging

Stop logging strings. Log structured data:

import logging
import json
from datetime import datetime
from typing import Any, Dict

class StructuredLogger:
    """Logger that outputs structured JSON"""

    def __init__(self, name: str):
        self.logger = logging.getLogger(name)

    def _log(self, level: int, event: str, **kwargs):
        """Log structured data as JSON"""
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "event": event,
            "session_id": session_context.get(),
            **kwargs
        }

        self.logger.log(level, json.dumps(log_data))

    def info(self, event: str, **kwargs):
        self._log(logging.INFO, event, **kwargs)

    def error(self, event: str, error: Exception = None, **kwargs):
        error_data = kwargs
        if error:
            error_data.update({
                "error_type": type(error).__name__,
                "error_message": str(error)
            })
        self._log(logging.ERROR, event, **error_data)

# Usage
logger = StructuredLogger(__name__)

logger.info(
    "audio_received",
    audio_size=len(audio_data),
    sample_rate=16000,
    channels=1
)

# Outputs:
# {"timestamp": "2024-01-15T10:30:45.123Z", "event": "audio_received", 
#  "session_id": "abc-123", "audio_size": 16000, "sample_rate": 16000, "channels": 1}

Now you can search logs by specific fields!

Strategy #3: Correlation IDs

Track requests across multiple services:

from contextvars import ContextVar
import uuid

# Correlation ID for tracking across services
correlation_id: ContextVar[Optional[str]] = ContextVar('correlation_id', default=None)

class CorrelatedLogger(StructuredLogger):
    """Logger with correlation ID support"""

    def _log(self, level: int, event: str, **kwargs):
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "event": event,
            "session_id": session_context.get(),
            "correlation_id": correlation_id.get(),
            **kwargs
        }

        self.logger.log(level, json.dumps(log_data))

@app.websocket("/transcribe/{session_id}")
async def transcribe_endpoint(websocket: WebSocket, session_id: str):
    # Generate correlation ID for this request
    corr_id = str(uuid.uuid4())
    correlation_id.set(corr_id)
    session_context.set(session_id)

    logger = CorrelatedLogger(__name__)
    logger.info("session_started")

    # When calling Riva service, pass correlation ID
    await riva_client.transcribe(
        audio_data,
        metadata={"correlation_id": corr_id}
    )

Now you can trace a request from client → your service → Riva → back!

Strategy #4: Performance Logging

Log performance metrics for every operation:

import time
from functools import wraps
from typing import Callable

def log_performance(operation: str):
    """Decorator to log operation performance"""
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            logger = CorrelatedLogger(func.__module__)
            start_time = time.time()

            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time

                logger.info(
                    f"{operation}_completed",
                    duration_ms=round(duration * 1000, 2),
                    success=True
                )

                return result

            except Exception as e:
                duration = time.time() - start_time

                logger.error(
                    f"{operation}_failed",
                    duration_ms=round(duration * 1000, 2),
                    success=False,
                    error=e
                )
                raise

        return wrapper
    return decorator

# Usage
@log_performance("audio_transcription")
async def transcribe_audio(audio: bytes, session_id: str) -> str:
    # Transcription logic
    return await riva_client.transcribe(audio)

# Logs:
# {"event": "audio_transcription_completed", "duration_ms": 245.67, "success": true}

Strategy #5: Log Levels by Component

Different components need different log levels:

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
            "format": "%(timestamp)s %(level)s %(name)s %(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "json",
            "stream": "ext://sys.stdout"
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "json",
            "filename": "logs/app.log",
            "maxBytes": 10485760,  # 10MB
            "backupCount": 5
        }
    },
    "loggers": {
        # Your app - verbose logging
        "app": {
            "level": "DEBUG",
            "handlers": ["console", "file"],
            "propagate": False
        },
        # Riva client - only warnings and errors
        "riva_client": {
            "level": "WARNING",
            "handlers": ["console", "file"],
            "propagate": False
        },
        # Third-party libraries - minimal logging
        "uvicorn": {
            "level": "INFO",
            "handlers": ["console"],
            "propagate": False
        },
        "grpc": {
            "level": "ERROR",
            "handlers": ["console"],
            "propagate": False
        }
    },
    "root": {
        "level": "INFO",
        "handlers": ["console", "file"]
    }
}

# Apply configuration
logging.config.dictConfig(LOGGING_CONFIG)

Strategy #6: Sampling for High-Volume Events

Don't log EVERY audio chunk - sample intelligently:

import random

class SampledLogger(CorrelatedLogger):
    """Logger with sampling support for high-frequency events"""

    def __init__(self, name: str, sample_rate: float = 0.01):
        super().__init__(name)
        self.sample_rate = sample_rate

    def sample(self, event: str, **kwargs):
        """Log with sampling"""
        if random.random() < self.sample_rate:
            self.info(event, sampled=True, **kwargs)

logger = SampledLogger(__name__, sample_rate=0.01)  # Log 1% of events

# Log every 100th audio chunk
logger.sample(
    "audio_chunk_processed",
    chunk_size=len(chunk),
    total_chunks=chunk_count
)

Strategy #7: Error Context Preservation

When errors happen, log EVERYTHING relevant:

import traceback
import sys

class ErrorContextLogger(CorrelatedLogger):
    """Logger with rich error context"""

    def error_with_context(
        self,
        event: str,
        error: Exception,
        **kwargs
    ):
        """Log error with full context"""

        # Get exception info
        exc_type, exc_value, exc_traceback = sys.exc_info()

        # Build error context
        error_context = {
            "error_type": type(error).__name__,
            "error_message": str(error),
            "error_code": getattr(error, 'code', None),
            "traceback": traceback.format_exc(),
            **kwargs
        }

        self.error(event, **error_context)

# Usage
logger = ErrorContextLogger(__name__)

try:
    await riva_client.transcribe(audio)
except Exception as e:
    logger.error_with_context(
        "transcription_failed",
        error=e,
        audio_size=len(audio),
        sample_rate=sample_rate,
        language=language,
        riva_endpoint=riva_client.endpoint
    )
    raise

Strategy #8: Log Aggregation & Search

Use ELK Stack or Loki for log aggregation:

# Docker Compose for Loki + Grafana
version: '3'
services:
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:latest
    volumes:
      - ./logs:/var/log
      - ./promtail-config.yaml:/etc/promtail/config.yaml
    command: -config.file=/etc/promtail/config.yaml

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

Now you can query logs with LogQL:

# Find all errors for a specific session
{job="transcription-service"} |= "session_id=abc-123" | json | level="ERROR"

# Find slow transcriptions
{job="transcription-service"} | json | duration_ms > 1000

# Count errors by type
sum by (error_type) (count_over_time({job="transcription-service"} | json | level="ERROR" [1h]))

Strategy #9: Monitoring Integration

Connect logs to metrics:

from prometheus_client import Counter, Histogram

# Metrics
transcription_requests = Counter(
    'transcription_requests_total',
    'Total transcription requests',
    ['session_id', 'language', 'status']
)

transcription_duration = Histogram(
    'transcription_duration_seconds',
    'Transcription duration',
    ['language']
)

class MonitoredLogger(ErrorContextLogger):
    """Logger integrated with metrics"""

    @log_performance("transcription")
    async def log_transcription(
        self,
        session_id: str,
        language: str,
        audio_size: int
    ):
        start_time = time.time()

        try:
            result = await transcribe(audio_data, language)

            # Log success
            self.info(
                "transcription_completed",
                audio_size=audio_size,
                language=language,
                result_length=len(result)
            )

            # Update metrics
            transcription_requests.labels(
                session_id=session_id,
                language=language,
                status="success"
            ).inc()

            transcription_duration.labels(
                language=language
            ).observe(time.time() - start_time)

            return result

        except Exception as e:
            # Log failure
            self.error_with_context(
                "transcription_failed",
                error=e,
                audio_size=audio_size,
                language=language
            )

            # Update metrics
            transcription_requests.labels(
                session_id=session_id,
                language=language,
                status="error"
            ).inc()

            raise

Best Practices

Always include session/correlation IDs - Makes debugging possible
Use structured logging - JSON is searchable and parseable
Sample high-frequency events - Don't fill disk with audio chunk logs
Log performance metrics - Know what's slow before users complain
Preserve error context - Log everything needed to debug
Set appropriate log levels - Debug in dev, Info in production
Rotate log files - Don't fill up disk
Centralize logs - Use log aggregation for multiple instances
Alert on log patterns - Error rate spikes should trigger alerts
Test your logging - Verify logs are useful during incidents

The Results

After implementing these logging strategies:

Mean Time to Resolution (MTTR) dropped from hours to minutes
Debug sessions became productive instead of frustrating
Production incidents were traceable across services
Performance bottlenecks became immediately visible
Customer support could look up exact session issues

Final Thoughts

Good logging is invisible when everything works, but invaluable when things break. The goal isn't to log everything - it's to log the right things at the right level with the right context.

Think of logs as breadcrumbs for future you. When you're debugging at 3 AM, you'll thank past you for logging that session ID.

What's your logging setup? Any horror stories about debugging without proper logs? Share below! 🚀

Building Resilient Real-Time Services: Error Handling at Scale

alfchee — Thu, 29 Jan 2026 21:40:49 +0000

Hey builders! 👋 Let's talk about something that separates hobby projects from production systems: error handling. When you're running a real-time service with hundreds of concurrent users, errors aren't edge cases—they're everyday reality.

Let me share how we built error handling that actually works at scale, and how it saved us countless 3 AM wake-up calls.

The Reality of Real-Time Systems

Here's what nobody tells you: in a real-time service, everything will fail:

Networks drop mid-stream
Clients disconnect without warning
GPUs run out of memory
External APIs timeout
Users send garbage data
Load spikes crash services

Your job isn't to prevent failures (you can't). Your job is to handle them gracefully.

The Error Categorization Framework

First rule: not all errors are created equal. We categorize every error into one of these buckets:

1. Connection Errors (CONN_*)

Problems with network connectivity or WebSocket state.

class ConnectionError(Exception):
    """Base class for connection-related errors"""
    pass

class CONN_CLIENT_DISCONNECT(ConnectionError):
    """Client closed connection normally"""
    code = "CONN_001"
    severity = "INFO"
    user_message = "Connection closed"

class CONN_TIMEOUT(ConnectionError):
    """Connection timed out"""
    code = "CONN_002"
    severity = "WARNING"
    user_message = "Connection timed out. Please reconnect."

class CONN_INVALID_MESSAGE(ConnectionError):
    """Received malformed message"""
    code = "CONN_003"
    severity = "ERROR"
    user_message = "Invalid message format"

2. Riva Errors (RIVA_*)

Problems with the AI service backend.

class RivaError(Exception):
    """Base class for Riva-related errors"""
    pass

class RIVA_CONNECTION_FAILED(RivaError):
    """Cannot connect to Riva service"""
    code = "RIVA_001"
    severity = "CRITICAL"
    user_message = "Service temporarily unavailable"
    retry_strategy = "exponential_backoff"

class RIVA_MODEL_NOT_FOUND(RivaError):
    """Requested model not available"""
    code = "RIVA_002"
    severity = "ERROR"
    user_message = "Language not supported"
    retry_strategy = "no_retry"

class RIVA_QUOTA_EXCEEDED(RivaError):
    """API quota exceeded"""
    code = "RIVA_003"
    severity = "WARNING"
    user_message = "Service capacity reached. Please try again later."
    retry_strategy = "rate_limit"

3. Audio Errors (AUDIO_*)

Problems with audio data or processing.

class AudioError(Exception):
    """Base class for audio-related errors"""
    pass

class AUDIO_INVALID_FORMAT(AudioError):
    """Audio format not supported"""
    code = "AUDIO_001"
    severity = "ERROR"
    user_message = "Invalid audio format. Please use PCM16."

class AUDIO_CHUNK_TOO_LARGE(AudioError):
    """Audio chunk exceeds size limit"""
    code = "AUDIO_002"
    severity = "WARNING"
    user_message = "Audio chunk too large"

class AUDIO_PROCESSING_FAILED(AudioError):
    """Failed to process audio"""
    code = "AUDIO_003"
    severity = "ERROR"
    user_message = "Failed to process audio"

The Error Handler Pattern

Here's the pattern that saved our sanity:

from typing import Optional, Callable
import logging
from dataclasses import dataclass

@dataclass
class ErrorContext:
    """Context information for error handling"""
    session_id: str
    user_id: Optional[str]
    endpoint: str
    error_code: str
    severity: str
    original_error: Exception
    retry_attempt: int = 0

class ErrorHandler:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.error_callbacks = {}

    def register_callback(self, error_class, callback: Callable):
        """Register a callback for specific error types"""
        self.error_callbacks[error_class] = callback

    async def handle(self, error: Exception, context: ErrorContext):
        """Central error handling"""
        # Log with context
        self.logger.log(
            level=self._get_log_level(context.severity),
            msg=f"[{context.session_id}] {context.error_code}: {str(error)}",
            extra={
                "session_id": context.session_id,
                "error_code": context.error_code,
                "severity": context.severity,
                "endpoint": context.endpoint,
                "retry_attempt": context.retry_attempt
            }
        )

        # Execute registered callback if exists
        callback = self.error_callbacks.get(type(error))
        if callback:
            await callback(error, context)

        # Return user-friendly error message
        return self._get_user_message(error)

    def _get_log_level(self, severity: str) -> int:
        levels = {
            "INFO": logging.INFO,
            "WARNING": logging.WARNING,
            "ERROR": logging.ERROR,
            "CRITICAL": logging.CRITICAL
        }
        return levels.get(severity, logging.ERROR)

    def _get_user_message(self, error: Exception) -> dict:
        """Convert error to user-friendly message"""
        return {
            "event": "error",
            "code": getattr(error, 'code', 'UNKNOWN'),
            "message": getattr(error, 'user_message', 'An error occurred')
        }

# Global error handler instance
error_handler = ErrorHandler()

Using the Error Handler in WebSocket Endpoints

Now let's put it to work:

@app.websocket("/transcribe/{session_id}")
async def transcribe_endpoint(websocket: WebSocket, session_id: str):
    context = ErrorContext(
        session_id=session_id,
        user_id=None,  # Set from auth
        endpoint="/transcribe"
    )

    try:
        await websocket.accept()

        # Main processing loop
        while True:
            try:
                message = await websocket.receive_text()
                data = json.loads(message)

                # Process message
                await process_message(session_id, data)

            except json.JSONDecodeError as e:
                # Invalid JSON
                error = CONN_INVALID_MESSAGE()
                context.error_code = error.code
                context.severity = error.severity
                context.original_error = e

                error_msg = await error_handler.handle(error, context)
                await websocket.send_json(error_msg)

            except AudioError as e:
                # Audio processing error
                context.error_code = e.code
                context.severity = e.severity
                context.original_error = e

                error_msg = await error_handler.handle(e, context)
                await websocket.send_json(error_msg)
                # Continue processing (don't close connection)

            except RivaError as e:
                # Riva service error
                context.error_code = e.code
                context.severity = e.severity
                context.original_error = e

                error_msg = await error_handler.handle(e, context)
                await websocket.send_json(error_msg)

                # Decide whether to retry or close
                if e.retry_strategy == "no_retry":
                    break
                elif e.retry_strategy == "exponential_backoff":
                    await asyncio.sleep(2 ** context.retry_attempt)
                    context.retry_attempt += 1

    except WebSocketDisconnect:
        # Normal disconnect - just log
        logger.info(f"[{session_id}] Client disconnected")

    except Exception as e:
        # Unexpected error - log and notify user
        context.error_code = "UNKNOWN"
        context.severity = "CRITICAL"
        context.original_error = e

        logger.exception(f"[{session_id}] Unexpected error: {e}")

        try:
            await websocket.send_json({
                "event": "error",
                "message": "An unexpected error occurred"
            })
        except:
            pass  # Connection might be dead

    finally:
        # Always cleanup
        await cleanup_session(session_id)

Graceful Degradation

When things fail, degrade gracefully instead of crashing:

class TranscriptionService:
    def __init__(self):
        self.primary_riva = RivaClient(primary=True)
        self.fallback_riva = RivaClient(primary=False)
        self.local_cache = TranscriptionCache()

    async def transcribe(self, audio: bytes, session_id: str):
        """Transcribe with fallbacks"""

        # Try primary Riva service
        try:
            return await self.primary_riva.transcribe(audio)
        except RIVA_CONNECTION_FAILED as e:
            logger.warning(f"[{session_id}] Primary Riva failed, trying fallback")

            # Try fallback service
            try:
                return await self.fallback_riva.transcribe(audio)
            except RIVA_CONNECTION_FAILED:
                logger.error(f"[{session_id}] Both Riva services failed")

                # Check cache for similar audio
                cached_result = self.local_cache.get_similar(audio)
                if cached_result:
                    logger.info(f"[{session_id}] Returning cached result")
                    return cached_result

                # All options exhausted
                raise RIVA_CONNECTION_FAILED(
                    "All transcription services unavailable"
                )

Monitoring Integration

Errors should trigger alerts and metrics:

from prometheus_client import Counter, Histogram

# Metrics
error_counter = Counter(
    'transcription_errors_total',
    'Total transcription errors',
    ['error_code', 'severity']
)

error_duration = Histogram(
    'error_handling_duration_seconds',
    'Time spent handling errors'
)

class MonitoredErrorHandler(ErrorHandler):
    async def handle(self, error: Exception, context: ErrorContext):
        with error_duration.time():
            # Count error
            error_counter.labels(
                error_code=context.error_code,
                severity=context.severity
            ).inc()

            # Handle error
            result = await super().handle(error, context)

            # Alert on critical errors
            if context.severity == "CRITICAL":
                await self.send_alert(error, context)

            return result

    async def send_alert(self, error: Exception, context: ErrorContext):
        """Send alert to monitoring system"""
        # Integration with PagerDuty, Slack, etc.
        pass

Client-Side Error Handling

Don't forget the client! Give users actionable feedback:

class TranscriptionClient {
    constructor(sessionId) {
        this.sessionId = sessionId;
        this.connect();
    }

    connect() {
        this.ws = new WebSocket(`ws://api.example.com/transcribe/${this.sessionId}`);

        this.ws.onmessage = (event) => {
            const data = JSON.parse(event.data);

            if (data.event === 'error') {
                this.handleError(data);
            } else {
                this.handleTranscription(data);
            }
        };
    }

    handleError(error) {
        const errorHandlers = {
            'CONN_TIMEOUT': () => {
                this.showNotification('Connection timeout. Reconnecting...', 'warning');
                this.reconnect();
            },
            'RIVA_QUOTA_EXCEEDED': () => {
                this.showNotification('Service busy. Please try again in a moment.', 'warning');
                this.retryAfterDelay(5000);
            },
            'AUDIO_INVALID_FORMAT': () => {
                this.showNotification('Microphone error. Please check your settings.', 'error');
                this.stopRecording();
            },
            'default': () => {
                this.showNotification('An error occurred. Please refresh the page.', 'error');
            }
        };

        const handler = errorHandlers[error.code] || errorHandlers['default'];
        handler();
    }
}

Best Practices

Always categorize errors: Use consistent error codes
Log with context: Include session IDs, user IDs, timestamps
Separate user messages from logs: Users don't need stack traces
Implement retry strategies: Not all errors should retry the same way
Monitor error rates: Set alerts for unusual patterns
Test error paths: Deliberately inject failures in tests
Document error codes: Create an error code reference for your team

The Results

After implementing this error handling framework:

Mean Time To Recovery (MTTR) dropped by 70%
User complaints about cryptic errors dropped to nearly zero
Production incidents became easier to debug
Team confidence in deploying changes increased dramatically

Final Thoughts

Good error handling is invisible when it works. Users don't know you just failed over to a backup service, retried a request, or pulled from cache. They just know things worked.

That's the goal: resilience that's transparent to users but visible to operators.

What's your approach to error handling in production? Share your battle stories below! 🚀

Async Python in Production: FastAPI vs Flask for Real-Time Services

alfchee — Wed, 21 Jan 2026 17:53:28 +0000

Hey Pythonistas! 👋 If you’ve ever tried to build a real-time app with Flask and found yourself tangled in threads, callbacks, or mysterious slowdowns, this article is for you. Today, I want to share the real-world lessons I learned about async programming, concurrency, and why FastAPI is a game-changer for modern, high-performance Python services.

Why Async Matters (and When It Doesn’t)

Let’s be honest: not every app needs async. But if you’re handling lots of simultaneous users, streaming data, or anything that can block (like network calls, file I/O, or waiting for a GPU), async is your best friend. It lets your app do more with less—less hardware, less waiting, less pain.

Flask: The Synchronous Wall

Flask is fantastic for simple APIs and quick prototypes. But when you try to scale up to real-time, concurrent workloads, you hit a wall:

Each request blocks a worker: If you’re waiting on a slow database or external API, that worker is stuck.
WebSockets are awkward: Flask needs extensions like Flask-SocketIO, which use threads or eventlet/gevent. Debugging and scaling become tricky.
No native async/await: You can’t just sprinkle async and await and expect magic.

FastAPI: Async, Await, and a Whole New World

FastAPI is built on Starlette and uses Python’s asyncio under the hood. This means you get:

True async endpoints: Use async def and await for non-blocking I/O.
WebSockets that scale: Native support, no hacks or monkey-patching.
Concurrency that feels natural: Handle hundreds of connections with a single process.

Let’s see the difference in practice.

Example: Handling a Long-Running Task

Flask (synchronous):

from flask import Flask
import time

app = Flask(__name__)

@app.route('/process')
def process():
    time.sleep(5)  # Simulate a blocking task
    return 'Done!'

If 10 users hit /process at once, you need 10 workers. Otherwise, users wait in line.

FastAPI (async):

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get('/process')
async def process():
    await asyncio.sleep(5)  # Non-blocking!
    return {'status': 'Done!'}

With async, your app can handle other requests while waiting. One worker, many happy users.

Real-Time WebSockets: The Async Advantage

With Flask-SocketIO, you’re juggling threads or greenlets. With FastAPI:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

No threads, no magic—just clean, readable async code.

Good Practices for Async Python in Production

Use async def everywhere you can: Especially for endpoints, database calls, and network I/O.
Never block the event loop: Avoid time.sleep(), heavy CPU work, or blocking libraries. Use await asyncio.sleep() or run CPU tasks in a thread pool.
Session Management: Track user sessions with unique IDs. Clean up resources when users disconnect.
Error Handling: Always wrap your async code in try/except blocks. Handle disconnects and unexpected errors gracefully.
Resource Cleanup: Use finally blocks to release resources, close connections, and avoid memory leaks.
Testing Async Code: Use pytest-asyncio or similar tools to write real async tests.

The Payoff: Why It’s Worth It

After switching to FastAPI and async patterns, our real-time service:

Handled 3x more users with the same hardware
Reduced average response times by 40%
Became easier to debug and extend
Had fewer production incidents (and happier developers!)

Final Thoughts

Async isn’t just a buzzword—it’s a superpower for modern Python apps. If you’re building real-time, high-concurrency services, FastAPI and async/await will make your life (and your users’ lives) so much better.

This article is based on real-world experience building production async Python services. All code is simplified for clarity but follows best practices.

From Flask to FastAPI: Why I Migrated My Real-Time Speech Service

alfchee — Wed, 21 Jan 2026 05:30:39 +0000

Hey fellow developers! 🚀 Let me tell you about one of the most impactful decisions I made in my career: migrating a critical real-time speech transcription service from Flask to FastAPI. If you're building anything that needs to handle concurrent connections, real-time data, or just wants to write more maintainable Python code, this story might just change how you approach your next project.

The Problem: Flask Was Holding Us Back

Picture this: We had a Flask-based speech-to-text service using Flask-SocketIO for WebSockets. It worked, but the cracks were showing:

Concurrency nightmares: Maxing out at ~50 concurrent sessions
Blocking operations everywhere: Every WebSocket message tied up a thread
Complex WebSocket handling: Flask-SocketIO added layers of abstraction that made debugging hell
No type safety: Manual validation led to runtime errors in production

Our users were complaining about dropped connections and slow response times. We needed a better foundation.

The Solution: FastAPI to the Rescue

FastAPI wasn't just an upgrade—it was a complete paradigm shift. Here's why it transformed our service:

1. Native Async Support That Actually Works

Before (Flask + Flask-SocketIO):

from flask import Flask
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app)

@socketio.on('start_transcription')
def handle_transcription(data):
    # This blocks the entire thread!
    result = riva_client.transcribe_audio(data['audio'])
    emit('transcription_result', result)

After (FastAPI):

from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="Speech Service", version="1.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"])

@app.websocket("/transcribe/{session_id}")
async def transcribe_endpoint(websocket: WebSocket, session_id: str):
    await websocket.accept()

    async for message in websocket.iter_text():
        data = json.loads(message)
        if data["type"] == "start_transcription":
            # Non-blocking async processing!
            await handle_transcription_async(session_id, data)

2. Performance That Scales

The numbers don't lie:

Metric	Flask (Before)	FastAPI (After)	Improvement
Concurrent Sessions	~50	150+	3x increase
Response Latency	150-200ms	80-120ms	35% faster
CPU Usage	High (blocking)	Low (async)	60% reduction
Memory Usage	Moderate	Optimized	25% reduction

3. Type Safety with Pydantic

Before:

def start_transcription(data):
    language = data.get('language', 'en-US')  # Runtime errors waiting to happen
    if 'audio' not in data:
        return {'error': 'Missing audio'}

After:

from pydantic import BaseModel
from typing import Optional

class TranscriptionRequest(BaseModel):
    language: str = "en-US"
    enable_automatic_punctuation: bool = True
    enable_interim_results: bool = True
    max_alternatives: int = 1

@app.post("/start-transcription")
async def start_transcription(request: TranscriptionRequest):
    # Type-safe, validated data - no more runtime surprises!
    return await process_transcription(request)

How to Start Your FastAPI Journey Right

1. Project Structure That Scales

speech-service/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI app instance
│   ├── config.py        # Settings management
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── transcription.py
│   │   └── health.py
│   ├── services/
│   │   ├── __init__.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── schemas.py   # Pydantic models
│   └── core/
│       ├── __init__.py
│       └── logging.py
├── tests/
├── requirements.txt
└── Dockerfile

2. Core Application Setup

A solid foundation is key. Here’s how to set up your main.py for scalability and maintainability.

main.py:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.routes import transcription, health
import logging

logger = logging.getLogger(__name__)

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    description="Real-time speech transcription service"
)

# CORS is essential for web clients to connect from different origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Routers keep your API endpoints organized and decoupled from the main app
app.include_router(health.router, prefix="/api/v1")
app.include_router(transcription.router, prefix="/api/v1")

@app.on_event("startup")
async def startup_event():
    # Use startup events to initialize resources that live for the entire application lifecycle.
    # This is the perfect place to warm up ML models, establish database connection pools,
    # or connect to other services.
    logger.info("Starting up the speech service...")

@app.on_event("shutdown")
async def shutdown_event():
    # Shutdown events are critical for graceful termination.
    # Always clean up resources like database connections or file handlers here
    # to prevent resource leaks.
    logger.info("Shutting down gracefully...")

3. Configuration Management

config.py:

from pydantic import BaseSettings

class Settings(BaseSettings):
    PROJECT_NAME: str = "Speech Service"
    VERSION: str = "1.0.0"
    API_V1_STR: str = "/api/v1"

    # Server settings
    HOST: str = "0.0.0.0"
    PORT: int = 8000

    # CORS
    ALLOWED_ORIGINS: list = ["http://localhost:3000", "https://yourapp.com"]

    # Service limits
    MAX_CONCURRENT_SESSIONS: int = 150
    SESSION_TIMEOUT_MINUTES: int = 30

    # Riva settings
    RIVA_SERVER: str = "grpc.nvcf.nvidia.com:443"
    RIVA_API_KEY: str

    class Config:
        env_file = ".env"

settings = Settings()

4. WebSocket Implementation Best Practices

Showing a bunch of code isn't enough. Let's break down why this implementation is robust and production-ready. These are the best practices that will save you from late-night debugging sessions:

Centralized Logic in a Service Layer: The TranscriptionService handles all business logic. The WebSocket endpoint is only responsible for receiving messages and managing the connection lifecycle. This separation of concerns is crucial for testing and maintainability.
Graceful Connection Handling: The entire connection's logic is wrapped in a try...except...finally block. This is non-negotiable for production services.
Specific Exception for Disconnects: We explicitly catch WebSocketDisconnect. This allows you to handle a client closing the connection normally, without treating it as an unexpected error.
Guaranteed Cleanup with finally: The finally block ensures that cleanup_session is always called, whether the client disconnects gracefully, an error occurs, or the connection times out. This prevents resource leaks (like orphaned sessions or lingering connections to Riva).
Clear Session Management: A unique session_id is used to track each client. This is fundamental for managing state, routing messages, and logging in a concurrent environment.
Asynchronous Operations: Every I/O-bound call (e.g., websocket.accept(), receive_text(), send_json(), and all service calls) uses await, ensuring the server remains non-blocking and can handle other clients while waiting.

Here is the code that puts all these practices into action:

routes/transcription.py:

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
import json
import logging
from app.services.transcription_service import TranscriptionService
from app.models.schemas import TranscriptionConfig

router = APIRouter()
transcription_service = TranscriptionService()
logger = logging.getLogger(__name__)

@router.websocket("/transcribe/{session_id}")
async def transcribe_websocket(websocket: WebSocket, session_id: str):
    await websocket.accept()

    try:
        # Initialize session
        session = await transcription_service.create_session(session_id, websocket)

        while True:
            # Receive JSON control messages or binary audio data
            message = await websocket.receive_text()
            data = json.loads(message)

            if data["type"] == "start_transcription":
                config = TranscriptionConfig(**data["config"])
                await transcription_service.start_transcription(session, config)
                await websocket.send_json({
                    "event": "transcription_started",
                    "session_id": session_id
                })

            elif data["type"] == "stop_transcription":
                await transcription_service.stop_transcription(session)
                break

    except WebSocketDisconnect:
        logger.info(f"Client {session_id} disconnected gracefully.")
    except Exception as e:
        logger.error(f"An unexpected error occurred in session {session_id}: {e}")
        await websocket.send_json({
            "event": "error",
            "message": "An unexpected error occurred. Please try reconnecting."
        })
    finally:
        # This block is CRITICAL. It guarantees that resources are released.
        logger.info(f"Cleaning up resources for session {session_id}.")
        await transcription_service.cleanup_session(session_id)

5. Service Layer for Business Logic

Your business logic is the heart of your application. Keep it clean, organized, and decoupled from the API layer.

services/transcription_service.py:

from typing import Dict, Optional
import asyncio
import logging
from fastapi import WebSocket
from app.core.riva_client import RivaClient
from app.models.schemas import TranscriptionConfig, Session

logger = logging.getLogger(__name__)

class TranscriptionService:
    def __init__(self):
        self.riva_client = RivaClient()
        self.active_sessions: Dict[str, Session] = {}
        # An asyncio.Lock is essential to prevent race conditions when modifying
        # shared state (like the active_sessions dict) from concurrent tasks.
        self._lock = asyncio.Lock()

    async def create_session(self, session_id: str, websocket: WebSocket) -> Session:
        async with self._lock:
            if session_id in self.active_sessions:
                raise ValueError(f"Session {session_id} already exists")

            session = Session(id=session_id, websocket=websocket)
            self.active_sessions[session_id] = session
            logger.info(f"Session {session_id} created.")
            return session

    async def start_transcription(self, session: Session, config: TranscriptionConfig):
        # Connect to Riva and start streaming
        await self.riva_client.start_transcription_stream(
            session_id=session.id,
            config=config
        )

        # Launching a background task with create_task is the key to non-blocking processing.
        # The main WebSocket loop can continue to handle other messages while this runs.
        session.processing_task = asyncio.create_task(self._process_results(session))

    async def _process_results(self, session: Session):
        """Background task to handle Riva streaming results and send them to the client."""
        try:
            async for result in self.riva_client.stream_results(session.id):
                # Forward results directly to the client's WebSocket
                await session.websocket.send_json(result)
        except Exception as e:
            logger.error(f"Error processing results for {session.id}: {e}")
            await session.websocket.send_json({"event": "error", "message": "Result processing failed."})

    async def cleanup_session(self, session_id: str):
        async with self._lock:
            if session_id in self.active_sessions:
                session = self.active_sessions.pop(session_id)
                if session.processing_task:
                    session.processing_task.cancel() # Ensure background tasks are stopped
                await self.riva_client.disconnect(session_id)
                logger.info(f"Successfully cleaned up session {session_id}.")

Key Takeaways for Your FastAPI Journey

1. Start with Type Safety

Pydantic models catch errors at the API boundary. Don't skip this—it's your first line of defense.

2. Structure for Scalability

Separate concerns: routes for API endpoints, services for business logic, models for data validation.

3. Async Everything

If you're building real-time services, async is non-negotiable. FastAPI makes it natural.

4. Handle WebSocket Lifecycle Properly

Connections drop, errors happen. Always implement proper cleanup and error handling.

5. Monitor and Observe

FastAPI's auto-generated docs are great, but add health checks, metrics, and structured logging.

The Results Speak for Themselves

After migration:

Zero downtime deployment
3x more concurrent users
35% faster response times
Development velocity increased by 50%
Runtime errors reduced by 80%

Ready to Make the Switch?

If you're still using Flask for real-time applications, I urge you: make the switch to FastAPI. The initial learning curve is worth every second. Your users will thank you, your team will thank you, and future-you will definitely thank you.

Start small: pick one service, migrate it to FastAPI, and measure the improvements. The results will convince you.

What's your experience with FastAPI? Have you migrated from Flask? Share in the comments—I'd love to hear your stories! 🚀

This article is based on real-world experience building production speech AI services. The code examples are simplified for clarity but demonstrate production-ready patterns.

Was I a target of social hacking?

alfchee — Sat, 16 Jan 2021 23:17:49 +0000

Few days ago I got a notification in LinkedIn about an invitation of a china man who saids to be a "Full-Stack Web Developer", I accepted it just to see what he wants to talk about, and what I got after the first message of greeting was a proposal of a deal:

that he wants to rent my Upwork account by 15% of his incomes by using my account.

I told to myself, "Am I reading well?", and I ask him what did he means with "rent" and why he didn't create a own account; and he explained that because of restrictions to US companies he can't activate an account, also told me that he would need my account credentials and that I was going to be the one that's going to withdraw from Upwork, retaining the 15% to myself. I stop to think about this and few minutes after he sent me the link of my Upwork profile telling me that currently I'm not using it, and then

Actually, I will pay for you $300 monthly if you rent your account, and If I earn over 2k+, I will pay 20% of my income

😑 "This is too good to be true", was the first thing I thought, so I checked his LinkedIn profile which has more than 200 contacts in between CEOs of many companies and startups and high profiles of developers, also has description of a lot of skills, but no projects, no GitHub, no links to any China website that proves he was doing something... so I asked for that, something that can demonstrate me he's a true developer, and how we can reach trust in each other. Also I shared to him this community thread about stolen Upwork account for a china man with a similar business model.

😏 - the expected, he avoided the topic, talking about friends with good profits about this kind of relationship, the usual is to avoid things instead of tell a lie, because lies has tracks. Time was going and as he wasn't getting any replies from me then tried to explain that he can't change credentials in Upwork, and added this lines

Upwork detect account's location and IP address, so If I use your account on my side, it will be blocked.

😕 and makes me think of how he was going to use my Upwork account? Why he wasn't using a VPN? Why he cannot get direct contract with such amazing list of contacts I envy?... and I asked.

😨 and after a few messages I got what he wanted to do

Actually, I should work remotely on your computer with Anydesk or Teamviewer,
You will know about Teamviewer and anydesk, it's remote accessing tools

What I considered the worst idea, for me, my security and my future in my current job and any next one. I just thought, that's an open access to my PC, my projects, my private life, my WiFi and any device on it... because, I'm a developer and you know that most of time, as users too, we will open a breach of security and someone with the plenty of the day to find it with a FULL ACCESS of TeamViewer or AnyDesk, it will.

👎 yeah... I told him that is not a deal for me and that I'm open to real jobs like software development, so I would like to hear about projects in which to work... 😶 no more replies from him.

Conclusion

I'm telling this story because I don't know how many people is being approached in this way and the promise of receive $300/month by doing nothing (more than give access to your accounts, PC included) is reasonable, and if there are many who accepted and lost more than the Upwork account... or in in deed I lose a deal (punishable by Upwork if they get to know) 😋.

If you know something about it, please let a comment.

Create a simple Vue component to use Google Maps

alfchee — Tue, 29 Dec 2020 01:21:45 +0000

There are some libraries over there that we may install in our projects to use Google Maps, but considering the amount of KB's to add and that is another dependence that sometimes can be deprecated some day, I got in the position to create my own component, also I didn't going to use too many features.

The code I'm going to share can be useful to use the basics of Google Maps, or even to start your own component and give it more features that can be useful for your projects.

Obtaining the API Key

To use Maps JavaScript API, you'll need an API Key, which is a unique identifier used to authenticate the requests associated with your project for payment accounting purposes.
To get an API key, do the following:

Go to the GCP Console: https://cloud.google.com/console/google/maps-apis/overview

Use the drop-down menu to select or create the project that you want a new API Key for
Click the button to open the sidebar and go to API & Services > Credentials

On the Credentials page, click on Create Credentials > API key. The API Key Created dialog will show the new API Key.

You can read more about the Google Maps API Key in the documentation: https://developers.google.com/maps/documentation/javascript/get-api-key

Now create an environment file at the root of the project, called .env, the content must be like

VUE_APP_MAPS_API_KEY="API_KEY"

Initializing the library

You can initialize the library with a function that returns a promise, which is resolved once the Google Maps library is downloaded in the browser and initialized.

Here is a basic example of how to create a map and add markers.

To actively use the Google Maps library in your Vue project, create a new script, /src/utils/gmaps.js.

const API_KEY = process.env.VUE_APP_MAPS_API_KEY
const CALLBACK_NAME = 'initMap'

let initialized = !!window.google
let resolveInitPromise
let rejectInitPromise

// This promise handles the initialization
// status of the google maps script
const initPromise = new Promise((resolve, reject) => {
  resolveInitPromise = resolve
  rejectInitPromise = reject
})

export default function init() {
  // if google maps is already init
  // the `initPromise` should be resolved
  if (initialized) return initPromise

  initialized = true
  // the callback function is called
  // by the Google maps script if it is
  // successfully loaded
  window[CALLBACK_NAME] = () => resolveInitPromise(window.google)

  // we inject a new tag into the Head
  // of the HTML to load the Google Maps script
  const script = document.createElement('script')
  script.async = true
  script.defer = true
  script.src = `https://maps.googleapis.com/maps/api/js?key=${API_KEY}&callback=${CALLBACK_NAME}`
  script.onerror = rejectInitPromise
  document.querySelector('head').appendChild(script)

  return initPromise
}

Creating a component to display the map

I'm gonna call the component GoogleMaps and for so create the component at /src/components/GoogleMaps.vue.

Let's add the markup at the template and the only CSS that's really mandatory

<template>
  <div class="map" ref="map"></div>
</template>

<script></script>

<style scoped>
.map {
  width: 100%;
  height: 400px;
}
</style>

That's all the required code at the markup, we may add more in dependence of the layout of our application.

The mounted lifecycle hook is asynchronous, and it waits for the Google Maps library to be initialized so that it can proceed to render the map using the drawMap method, adding a marker to represent the user's location on the map.

<script>
import gmapsInit from '@/utils/gmaps'
import { isNumber, isArray } from 'util'

export default {
  name: 'GoogleMaps',

  data() {
    return {
      google: null,
      map: null,
      innerMarkers: [],
      userMarker: null
    }
  },

  async mounted() {
    try {
      // init and wait for the Google script is mounted
      this.google = await gmapsInit()

      // if the location is already set, for example
      // when returning back to this view from another one
      if ('lat' in this.myLocation && this.myLocation.lat) {
        this.drawMap()
        // set the current location
        this.addMarker(this.myLocation)
      }
    } catch (err) {
      console.log('ERROR:', err)
    }
  },

  computed: {
    myLocation() {
      // if the coordinates is not set
      if (!('lat' in this.center) && !isNumber(this.center.lat)) {
        return null
      }
      // return the object expected by Google Maps
      return { lat: this.center.lat, lng: this.center.lng }
    }
  },

  methods: {
    drawMap() {
      if (this.myLocation.lat && this.myLocation.lng) {
        // creating the map object, displaying it in the $el DOM object
        this.map = new this.google.maps.Map(this.$refs['map'], {
          zoom: 18,
          center: this.myLocation
        })

        // center the canvas of the map to the location of the user
        this.map.setCenter(this.myLocation)
      }
    },

  }
}
</script>

We need to pass some data to the component, and I'm going to set as required prop for the component is center, which contains the coordinates for the center of the map and can receive a collection of markers via a prop with the same name.

<script>
  props: {
    center: {
      type: Object,
      required: true
    },
    markers: {
      type: Array
    }
  },
</script>

Then the last thing is to add reactivity in front of the change of the position of the current location of the user, for so is also required to add some methods to work with the markers.

<script>
  methods: {
        // add a marker with a blue dot to indicate the user location
    setUserMarker(location) {
      this.userMarker = new this.google.maps.Marker({
        position: location,
        map: this.map
      })
    },

    // Adds a marker to the map and push to the array
    addMarker(location) {
      // the marker positioned at `myLocation`
      const marker = new this.google.maps.Marker({
        position: location,
        map: this.map
      })
      this.innerMarkers.push(marker)
    },

    // Sets the map on all markers in the array
    setAllMarkersInMap(map) {
      for (let i = 0; i < this.innerMarkers.length; i++) {
        this.innerMarkers[i].setMap(map)
      }
    },

    // Removes the markers from the map, but keeps them in the array
    clearMarkers() {
      this.setAllMarkersInMap(null)
    },

    // Deletes all markers in the array by removing references to them
    deleteMarkers() {
      this.clearMarkers()
      this.innerMarkers = []
    }
  },

  watch: {
    marker: function(newVal) {
      if (isArray(newVal)) {
        // clear the markers
        this.clearMarkers()

        for (let i = 0; i < newVal.length; i++) {
          let position = newVal[i]
          if (
            'lat' in position &&
            isNumber(position.lat) &&
            isNumber(position.lng)
          ) {
            // set the current location
            this.addMarker(position)
          }
        }
      }
    }
  }
</script>

It is quite easy and now you can render maps in your application with a marker, with something like

<template>
  <GoogleMaps :center="{ ...coordinates }" />
<template>

Placing all the code for the GoogleMap component together

<template>
  <div class="map" ref="map"></div>
</template>

<script>
import gmapsInit from '@/utils/gmaps'
import { isNumber, isArray } from 'util'

export default {
  name: 'GoogleMaps',
  props: {
    center: {
      type: Object,
      required: true
    },
    markers: {
      type: Array
    }
  },
  async mounted() {
    try {
      // init and wait for the Google script is mounted
      this.google = await gmapsInit()

      // if the location is already set, for example
      // when returning back to this view from another one
      if ('lat' in this.myLocation && this.myLocation.lat) {
        this.drawMap()
        // set the current location
        this.addMarker(this.myLocation)
      }
    } catch (err) {
      console.log('ERROR:', err)
    }
  },
  data() {
    return {
      google: null,
      map: null,
      innerMarkers: [],
      userMarker: null
    }
  },
  computed: {
    myLocation() {
      // if the coordinates is not set
      if (!('lat' in this.center) && !isNumber(this.center.lat)) {
        return null
      }
      // return the object expected by Google Maps
      return { lat: this.center.lat, lng: this.center.lng }
    }
  },
  methods: {
    drawMap() {
      if (this.myLocation.lat && this.myLocation.lng) {
        // creating the map object, displaying it in the $el DOM object
        this.map = new this.google.maps.Map(this.$refs['map'], {
          zoom: 18,
          center: this.myLocation
        })

        // center the canvas of the map to the location of the user
        this.map.setCenter(this.myLocation)
      }
    },
    // add a marker with a blue dot to indicate the user location
    setUserMarker(location) {
      this.userMarker = new this.google.maps.Marker({
        position: location,
        map: this.map
      })
    },
    // Adds a marker to the map and push to the array
    addMarker(location) {
      // the marker positioned at `myLocation`
      const marker = new this.google.maps.Marker({
        position: location,
        map: this.map
      })
      this.innerMarkers.push(marker)
    },
    // Sets the map on all markers in the array
    setAllMarkersInMap(map) {
      for (let i = 0; i < this.innerMarkers.length; i++) {
        this.innerMarkers[i].setMap(map)
      }
    },
    // Removes the markers from the map, but keeps them in the array
    clearMarkers() {
      this.setAllMarkersInMap(null)
    },
    // Deletes all markers in the array by removing references to them
    deleteMarkers() {
      this.clearMarkers()
      this.innerMarkers = []
    }
  },
  watch: {
    marker: function(newVal) {
      if (isArray(newVal)) {
        // clear the markers
        this.clearMarkers()

        for (let i = 0; i < newVal.length; i++) {
          let position = newVal[i]
          if (
            'lat' in position &&
            isNumber(position.lat) &&
            isNumber(position.lng)
          ) {
            // set the current location
            this.addMarker(position)
          }
        }
      }
    }
  }
}
</script>

<style scoped>
.map {
  width: 100%;
  height: 400px;
}
</style>

Hoping it can be helpful for your project!

DEV Community: alfchee

The Hidden Costs of Real-Time: Latency vs Accuracy Trade-offs

The Illusion of "Real-Time"

The Buffer Size Dilemma

GPU vs CPU: The Cost Calculation

The Accuracy Trade-off Nobody Mentions

What I Learned

The Bottom Line

From MVP to Production: Scaling a Speech AI Service

The MVP Mindset (And Its Pitfalls)

The Migration to FastAPI (Article #1 Recap)

Adding Resilience Layer by Layer (Article #7)

Phase 1: Basic try-catch

Phase 2: Categorized errors

Phase 3: Graceful degradation

The Logging Evolution (Article #14)

Docker: From "It Works On My Machine" to Production (Article #9)

Audio Quality: The Unexpected Challenge (Article #4)

What I Would Do Different

The Takeaway

Docker Deployment for GPU-Accelerated Services

The Challenge: GPU & Environment Complexity

1. The Entrypoint Pattern

2. Docker Compose Profiles

3. Solving the Logging Nightmare

Benefits of this Approach

Conclusion

Solving Audio Gaps in Real-Time Speech Translation

The Problem: "Machine Gun" Audio

What Wasn't The Cause

Root Cause Analysis

The Solution: A Three-Pronged Approach

1. Aggressive Silence Trimming

2. Audio Crossfading (Windowing)

3. Smart Text Aggregation

Results

Key Takeaway

Vue.js Composables for Speech-to-Speech Translation: Building Reactive Real-Time Audio Applications

Introduction: Why Vue.js Composables Changed Everything

The Challenge: Real-Time Audio Streaming in the Browser

Lesson 1: Composables Are More Than React Hooks with Different Syntax

Vue Composables Have Direct Reactive References

Lifecycle is More Explicit

Lesson 2: Building the WebSocket Composable

The Basic Structure

Key Design Decision: Promise-Based Connection

Lesson 3: Reactive Message Handling

Lesson 4: Audio Capture Composable

Critical Lesson: Resource Cleanup

Lesson 5: Bringing It Together in a Component

Lesson 6: Composition API vs Options API

Composition API Wins:

Options API was Better For:

Lesson 7: Error Handling and Edge Cases

1. User Denies Microphone Permission

2. WebSocket Connection Lost

3. Browser Compatibility

Lesson 8: Performance Optimization

1. Debounce UI Updates for Partial Transcripts

2. Use shallowRef for Large Objects

3. Lazy Load Audio Components

Lesson 9: Testing Composables

Lesson 10: What I'd Do Differently

1. Add Retry Logic Earlier

2. Use TypeScript from Day One

3. Implement Audio Queueing

4. Add Connection State Machine

Comparison: Vue Composables vs React Hooks

Real-World Performance Results

Conclusion: The Joy of Vue Composables

Resources

GPU-Accelerated Speech AI: Integrating Nvidia Riva in Python

Why GPU Acceleration Was Non‑Negotiable

What Riva Brings to the Table

The Integration Pattern That Worked

1) Secure Auth + Service Setup

2) Streaming Generator (The Heart of It)

3) Streaming Config That Fits Real-Time

4) Consuming the Stream

The Real-Time WebSocket Bridge

2. Use `shallowRef` for Large Objects