# RealFlow: Real-Time Conversational AI Pipeline

**Last Updated:** 2026-01-24  
**Status:** ✅ Complete - Full real-time bidirectional voice conversation system  
**Official Name:** **RealFlow**

## Overview

**RealFlow** is THRPY's comprehensive real-time conversational AI pipeline that transforms simple audio input/output into a sophisticated multi-layer inference system. What started as an audio service has evolved into a complete "stream of consciousness" that processes multiple layers of understanding before responding.

## The RealFlow Pipeline

```
User Audio Input
    ↓
[STT] Speech-to-Text (Whisper ONNX)
    ↓
[Parallel Inference Layer]
    ├─ Vocal Inference (audio → emotion)
    └─ Text Inference (text → sentiment/emotion/topics)
    ↓
[Safety Layer] Safety Check (<50ms)
    ├─ Fast Gate (<20-50ms)
    └─ Full Assessment (<50ms)
    ↓
[LLM] Generate Response (streaming)
    ├─ If Safe: Normal therapeutic response
    └─ If Flagged: Restricted/crisis response
    ↓
[TTS] Text-to-Speech (Supertonic)
    ↓
User Audio Output (with synchronized word highlighting)
    ↓
[Repeat] "Stream of consciousness continues..."
```

## Why "RealFlow"?

**RealFlow** captures the essence of what this system does:
- **Real** - Real-time processing, no delays
- **Flow** - Continuous bidirectional conversation flow
- **Stream of Consciousness** - Multiple layers of inference happening simultaneously

It's more than an audio service—it's a complete conversational AI pipeline that thinks before it speaks.

### Evolvable Platform

**RealFlow is designed to evolve:**
- **Today**: Complete integrated pipeline with fixed components
- **Tomorrow**: Modular platform with plug-and-play customization
- **Core Value**: Real-time inferencing + Flexibility
- **Future**: Plugin system, custom inference layers, extensible architecture

The differentiators may change, but the core value remains: **real-time inferencing** with the flexibility to adapt and extend.

## Key Features

### 1. Multi-Layer Inference

**Vocal Inference** (Audio → Emotion)
- Analyzes prosody, tone, pitch from audio
- Extracts emotion labels (happy, sad, angry, etc.)
- Provides VAD scores (Valence, Arousal, Dominance)
- Latency: 50-150ms

**Text Inference** (Text → Sentiment/Emotion/Topics)
- Sentiment analysis (positive/negative/neutral)
- Emotion classification (joy, sadness, anger, fear, etc.)
- Topic extraction (anxiety, depression, relationships, etc.)
- Latency: 100-300ms

**Both run in parallel** for maximum speed.

### 2. Safety-First Architecture

**Fast Gate** (<20-50ms)
- Local crisis detection
- Keywords: self-harm, violence, emergency
- Result: `safe_to_start` boolean

**Full Assessment** (<50ms)
- Comprehensive risk analysis
- Risk level (0.0-1.0)
- Risk categories
- Crisis types
- Intervention protocols

**Safety Decision Logic:**
- If safe (`risk_level < 0.7` and not urgent): Normal therapeutic response
- If flagged: Restricted/crisis response with intervention protocol

### 3. Real-Time LLM Response

**Streaming Responses**
- First token: <200ms
- Continuous token streaming
- Real-time feedback to user

**Response Modes:**
- **Normal**: Full therapeutic response when safe
- **Restricted**: Crisis protocol response when flagged
- **Degraded**: Conservative response when safety unavailable

### 4. Synchronized Word Highlighting

**Perfect Timing**
- Uses AudioContext.currentTime for precise sync
- Updates at 60fps (requestAnimationFrame)
- Words highlight exactly when spoken

**Visual Feedback**
- Real-time word highlighting during TTS
- Word-level metadata (emotion, sentiment)
- Smooth transitions between words

## Performance Metrics

### Latency Breakdown

| Step | Latency | Notes |
|------|---------|-------|
| **STT** | 100-300ms | Whisper ONNX (real-time) |
| **Vocal Inference** | 50-150ms | Parallel with STT |
| **Text Inference** | 100-300ms | Parallel with STT |
| **Safety Check** | 20-50ms | Fast gate + full assessment |
| **LLM Response** | 200-1000ms | Streaming (first token <200ms) |
| **TTS** | 50-200ms | Supertonic (ultra-fast, ~167x real-time) |

**Total End-to-End**: ~500-2000ms (first response token)

### Real-Time Optimization

- **Parallel Processing**: Vocal + Text inference run simultaneously
- **Streaming**: LLM response streams tokens as they're generated
- **Fast Safety**: Safety check completes before LLM starts
- **Ultra-Fast TTS**: Supertonic generates audio at ~167x real-time

## Architecture Components

### Backend Services

**Unified Audio Streaming Service**
- `chat-api/app/unified_audio_streaming_service.py`
- Orchestrates entire RealFlow pipeline
- Handles STT, inference, safety, LLM, TTS

**Safety Client**
- `chat-api/app/safety_client.py`
- Fast gate + full assessment
- Crisis detection and intervention

**Chat Service**
- `chat-api/app/services/chat_service.py`
- LLM response generation
- Therapeutic response crafting

**Inference Services**
- `chat-api/app/services/enhanced_emotion_service.py` - Vocal inference
- `chat-api/app/services/realtime_text_analyzer.py` - Text inference

### Frontend Components

**WebSocket Hook**
- `frontend/src/hooks/useUnifiedAudioWebSocket.ts`
- Handles all RealFlow message types
- Manages bidirectional streaming

**Word Highlighting**
- `frontend/src/hooks/useStreamingAudioWithTiming.ts` - Streaming TTS
- `frontend/src/hooks/useSynchronousTTS.ts` - Non-streaming TTS
- Perfect synchronization with audio playback

## Message Flow

### Transcription with Inference

```json
{
  "type": "transcription",
  "text": "I've been feeling anxious lately",
  "vocal_inference": {
    "label": "sad",
    "confidence": 0.75,
    "valence": -0.6,
    "arousal": 0.7,
    "dominance": -0.3
  },
  "text_inference": {
    "sentiment": {
      "label": "negative",
      "score": 0.68
    },
    "emotion": {
      "primary": "anxiety",
      "top_emotions": [
        {"label": "anxiety", "score": 0.82}
      ]
    },
    "topics": [
      {"topic": "anxiety", "score": 0.89}
    ]
  },
  "safety_assessment": {
    "risk_level": 0.3,
    "is_urgent": false,
    "risk_categories": ["anxiety"],
    "confidence": 0.85
  },
  "llm_response": "I understand that anxiety can be really challenging...",
  "llm_ready": true,
  "is_final": true
}
```

### LLM Response Chunks (Streaming)

```json
{
  "type": "llm_chunk",
  "text": "I understand",
  "accumulated": "I understand",
  "safety_assessment": {...}
}
```

### Safety-Flagged Response

```json
{
  "type": "llm_chunk",
  "text": "I'm here to help. Let's make sure you're safe...",
  "accumulated": "I'm here to help. Let's make sure you're safe...",
  "safety_assessment": {
    "risk_level": 0.85,
    "is_urgent": true,
    "risk_categories": ["self_harm"],
    "intervention_required": true
  },
  "safety_flagged": true
}
```

## Usage Example

### Complete RealFlow Integration

```typescript
import { useUnifiedAudioWebSocket } from '../hooks/useUnifiedAudioWebSocket';

const { sendAudioInput, requestTTS } = useUnifiedAudioWebSocket({
  userId: "user123",
  onTranscription: (result) => {
    // Real-time transcription with inference
    console.log("Transcription:", result.text);
    console.log("Vocal emotion:", result.vocal_inference?.label);
    console.log("Text sentiment:", result.text_inference?.sentiment?.label);
    
    // Safety assessment
    if (result.safety_assessment) {
      console.log("Risk level:", result.safety_assessment.risk_level);
      console.log("Is urgent:", result.safety_assessment.is_urgent);
    }
    
    // LLM response (when ready)
    if (result.llm_ready && result.llm_response) {
      console.log("LLM response:", result.llm_response);
      // Automatically triggers TTS with synchronized word highlighting
    }
  },
  onLLMChunk: (chunk) => {
    // Real-time LLM response chunks (streaming)
    console.log("LLM chunk:", chunk.text);
    console.log("Accumulated:", chunk.accumulated);
  }
});

// Send audio input
// RealFlow automatically handles: STT → Inference → Safety → LLM → TTS
sendAudioInput(audioBytes, false);
```

## Safety Protocols

### Normal Flow (Safe)

1. User speaks → STT transcribes
2. Inference runs (vocal + text) in parallel
3. Safety check passes (`risk_level < 0.7`)
4. LLM generates normal therapeutic response
5. TTS speaks response with word highlighting

### Crisis Flow (Flagged)

1. User speaks → STT transcribes
2. Inference runs (vocal + text) in parallel
3. Safety check flags crisis (`risk_level >= 0.7` or `is_urgent`)
4. LLM generates restricted/crisis response
5. TTS speaks crisis protocol response
6. System logs crisis event for audit

### Degraded Mode (Safety Unavailable)

1. User speaks → STT transcribes
2. Inference runs (vocal + text) in parallel
3. Safety check unavailable → Uses degraded mode (`risk_level = 0.5`)
4. LLM generates conservative response (assumes unknown risk)
5. TTS speaks conservative response

## Configuration

### Enable/Disable RealFlow Components

**Safety Flow** (enabled by default):
```python
# In unified_audio_streaming_service.py
self.safety_client = get_safety_client()  # Enable safety → LLM flow
```

**Inference** (enabled by default):
```python
include_emotion = True  # Enable vocal + text inference
```

**TTS Mode**:
```python
# In config.py
TTS_MODE = "supertonic"  # Ultra-fast (~167x real-time)
# or
TTS_MODE = "piper"  # Fast and reliable
```

### Safety Thresholds

```python
# Risk level threshold for "safe" (default: 0.7)
is_safe = (
    safety_assessment.get("risk_level", 0.5) < 0.7
    and not safety_assessment.get("is_urgent", False)
)
```

## Troubleshooting

### LLM Response Not Appearing

**Check**:
1. Safety check is completing successfully
2. `is_safe` logic is passing
3. Chat service is initialized
4. LLM client is available

**Solutions**:
- Check backend logs for safety assessment results
- Verify `risk_level` is below threshold (0.7)
- Ensure `is_urgent` is false
- Check LLM client initialization

### Word Highlighting Not Synchronized

**Check**:
1. AudioContext is running (`audioContext.state === 'running'`)
2. Word timings are provided in audio chunks
3. `useStreamingAudioWithTiming` hook is used

**Solutions**:
- Ensure AudioContext is initialized before playback
- Verify timing metadata is included in audio chunks
- Use `useStreamingAudioWithTiming` for synchronized highlighting

### High Latency

**Check**:
1. Safety check latency (<50ms expected)
2. LLM response latency (<200ms first token expected)
3. Network latency

**Solutions**:
- Safety check should be <50ms (fast gate + full assessment)
- LLM streaming should start <200ms
- Total end-to-end should be <2000ms

## Related Documentation

- **Quick Packet**: `docs/REALFLOW_QUICK_PACKET.md` ⚡ (2-page visual overview)
- **KPIs & Differentiators**: `docs/REALFLOW_KPIS_AND_DIFFERENTIATORS.md` ⭐
- **Platform Vision**: `docs/REALFLOW_PLATFORM_VISION.md` 🚀
- **Vocal & Text Inference**: `docs/VOCAL_TEXT_INFERENCE_SETUP.md`
- **Safety → LLM Flow**: `docs/REALTIME_SAFETY_LLM_FLOW.md`
- **Voice Conversation Setup**: `docs/VOICE_CONVERSATION_SETUP.md`
- **Supertonic TTS**: `docs/SUPERTONIC_TTS_SETUP.md`
- **Real-Time Bidirectional**: `docs/REALTIME_BIDIRECTIONAL_VOICE.md`

## Summary

**RealFlow** is THRPY's complete real-time conversational AI pipeline that:

✅ Processes audio input with multi-layer inference  
✅ Ensures safety before responding  
✅ Generates therapeutic responses in real-time  
✅ Speaks back with synchronized word highlighting  
✅ Maintains natural conversation flow  

It's more than an audio service—it's a complete "stream of consciousness" that thinks before it speaks.

---

**Questions?** Check the code comments in `unified_audio_streaming_service.py` for detailed implementation notes.