feat: implement accumulative LLM paragraph engine and harden 4-process pipeline

engine_distribute.py

@@ -3,11 +3,22 @@ import time
 import argparse
 import multiprocessing
 import sys
+import os
 
 INGEST_URL = "https://emiapi.reynafamily.com/live-captions/ingest"
 
+def log_debug(message):
+    try:
+        with open("distribute_debug.log", "a") as f:
+            f.write(f"{time.strftime('%H:%M:%S')} {message}\n")
+            f.flush()
+            os.fsync(f.fileno())
+    except:
+        pass
+
 def run_distribution(in_queue, args):
-    print("[Distribute] Starting distribution service...")
+    print("[Distribute] Starting distribution service...", flush=True)
+    log_debug("Starting distribution service...")
     
     while True:
         try:
@@ -20,14 +31,30 @@ def run_distribution(in_queue, args):
                 sys.stdout.write(f"\r\033[K[DRAFT]: {draft_text}")
                 sys.stdout.flush()
                 if args.ingest:
-                    requests.post(INGEST_URL, json={"draft": draft_text}, timeout=2)
+                    try: requests.post(INGEST_URL, json={"draft": draft_text}, timeout=2)
+                    except: pass
                 continue
 
             # Finalized segment
-            print(f"\n[Final]: {payload.get('original', '')}")
+            speaker_label = payload.get('speaker')
+            speaker_prefix = f"[{speaker_label}]: " if speaker_label else ""
+            
+            msg = f"\n[Final]: {speaker_prefix}{payload.get('original', '')}"
+            print(msg, flush=True)
+            log_debug(msg)
+
+            # Show LLM Refinement if available
+            refined = payload.get('paragraph') or payload.get('corrected')
+            if refined and refined.strip() != payload.get('original', '').strip():
+                llm_msg = f"[LLM]: {speaker_prefix}{refined}"
+                print(llm_msg, flush=True)
+                log_debug(llm_msg)
+            
             for lang in ["es", "fr", "ar", "en"]:
                 if payload.get(lang):
-                    print(f"[{lang.upper()}]: {payload[lang]}")
+                    lang_msg = f"[{lang.upper()}]: {speaker_prefix}{payload[lang]}"
+                    print(lang_msg, flush=True)
+                    log_debug(lang_msg)
 
             if args.ingest:
                 delay, max_delay, success = 1, 15, False

engine_llm.py

@@ -0,0 +1,130 @@
+import sys
+from unittest.mock import MagicMock
+
+# MANDATORY: Mock lzma BEFORE any other imports
+try:
+    import lzma
+except ImportError:
+    mock_lzma = MagicMock()
+    mock_lzma.FORMAT_XZ, mock_lzma.FORMAT_ALONE, mock_lzma.FORMAT_RAW = 1, 2, 3
+    mock_lzma.CHECK_NONE, mock_lzma.CHECK_CRC32, mock_lzma.CHECK_CRC64, mock_lzma.CHECK_SHA256 = 0, 1, 4, 10
+    sys.modules["_lzma"] = MagicMock()
+    sys.modules["lzma"] = mock_lzma
+
+import time
+import argparse
+import multiprocessing
+import os
+import requests
+from collections import deque
+
+def run_llm_processor(in_queue, out_queue, args):
+    from dotenv import load_dotenv
+    load_dotenv()
+    
+    print("[LLM] Starting Rolling Paragraph Engine...")
+    
+    api_key = os.environ.get("OPENAI_API_KEY")
+    is_openai = args.post_correct_model.startswith("gpt-")
+    target_lang = "English"
+    if args.lang and args.lang != 'en': target_lang = args.lang
+
+    # State: This is the ONLY text we send to the LLM as context
+    active_context = ""
+    
+    def call_llm_rolling_refine(new_segments_str, context):
+        if not is_openai: return ""
+        
+        url = "https://api.openai.com/v1/chat/completions"
+        headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+        
+        prompt = f"""Task: Refine the following live transcription stream into clean, professional paragraphs.
+
+[PREVIOUS WORKING CONTEXT]
+"{context}"
+
+[NEW RAW ASR SEGMENTS]
+"{new_segments_str}"
+
+[INSTRUCTIONS]
+1. INTEGRATE: Polished and merge the new segments into the flow of the 'PREVIOUS WORKING CONTEXT'.
+2. CONSOLIDATE: Remove redundant repetitions and translator echoes.
+3. ORGANIZE: Use a double newline (\\n\\n) to start a new paragraph when a topic changes or the current one is complete.
+4. TARGET LANGUAGE: Output ONLY in {target_lang}.
+5. OUTPUT: Provide ONLY the refined, consolidated text. Do not explain anything.
+"""
+        payload = {
+            "model": args.post_correct_model,
+            "messages": [{"role": "system", "content": "You are a live transcript editor."},
+                         {"role": "user", "content": prompt}],
+            "temperature": 0.1
+        }
+        try:
+            resp = requests.post(url, json=payload, headers=headers, timeout=20)
+            if resp.status_code == 200:
+                return resp.json()["choices"][0]["message"]["content"].strip()
+            return ""
+        except: return ""
+
+    pending_buffer = []
+    last_llm_call_time = time.time()
+    
+    while True:
+        try:
+            item = in_queue.get()
+            if item is None: break
+            if "draft" in item:
+                out_queue.put(item)
+                continue
+
+            raw_text = item.get("original", "").strip()
+            bridge_text = item.get("en_bridge", raw_text).strip()
+            if not raw_text: continue
+
+            # Hallucination check
+            words = bridge_text.lower().split()
+            if len(words) > 10 and len(set(words)) < 3: continue
+
+            pending_buffer.append(bridge_text)
+
+            should_refine = False
+            if len(pending_buffer) >= 5: should_refine = True
+            elif len(pending_buffer) >= 2 and bridge_text.endswith((".", "?", "!")): should_refine = True
+            elif len(pending_buffer) >= 1 and (time.time() - last_llm_call_time) > 15: should_refine = True
+
+            structured_paragraph = None
+            if args.llm_paragraph and should_refine:
+                new_batch = " ".join(pending_buffer)
+                result = call_llm_rolling_refine(new_batch, active_context)
+                
+                if result:
+                    # Logic to handle paragraph breaks
+                    if "\n\n" in result:
+                        # Split by double newline
+                        parts = result.split("\n\n")
+                        # The last part is the new "Active Context"
+                        active_context = parts[-1].strip()
+                        # The whole result is sent to the user (contains the break)
+                        structured_paragraph = result
+                    else:
+                        # No break, just update the context
+                        active_context = result
+                        structured_paragraph = result
+                    
+                    pending_buffer = []
+                    last_llm_call_time = time.time()
+
+            out_queue.put({
+                "raw": raw_text,
+                "corrected": bridge_text,
+                "paragraph": structured_paragraph,
+                "detected_lang": item.get("detected_lang"),
+                "speaker": item.get("speaker"),
+                "ts": item.get("ts", time.time())
+            })
+            
+        except Exception as e:
+            print(f"[LLM] Error: {e}")
+
+if __name__ == "__main__":
+    run_llm_processor(multiprocessing.Queue(), multiprocessing.Queue(), argparse.Namespace())

engine_transcribe.py

@@ -1,65 +1,56 @@
 import sys
 from unittest.mock import MagicMock
 
-# Comprehensive workaround for missing _lzma in some Python builds
+# MANDATORY: Mock lzma BEFORE any other imports (especially pyannote/torchvision)
 try:
     import lzma
 except ImportError:
     mock_lzma = MagicMock()
-    mock_lzma.FORMAT_XZ = 1
-    mock_lzma.FORMAT_ALONE = 2
-    mock_lzma.FORMAT_RAW = 3
-    mock_lzma.CHECK_NONE = 0
-    mock_lzma.CHECK_CRC32 = 1
-    mock_lzma.CHECK_CRC64 = 4
-    mock_lzma.CHECK_SHA256 = 10
+    mock_lzma.FORMAT_XZ, mock_lzma.FORMAT_ALONE, mock_lzma.FORMAT_RAW = 1, 2, 3
+    mock_lzma.CHECK_NONE, mock_lzma.CHECK_CRC32, mock_lzma.CHECK_CRC64, mock_lzma.CHECK_SHA256 = 0, 1, 4, 10
     sys.modules["_lzma"] = MagicMock()
     sys.modules["lzma"] = mock_lzma
 
 import time
 import os
-import numpy as np
-import sounddevice as sd
-import torch
-import mlx_whisper
-from silero_vad import load_silero_vad, get_speech_timestamps
-from pyannote.audio import Pipeline
-import queue
 import argparse
 
-# Audio Constants
-SAMPLERATE = 16000
-BLOCK_SIZE = 512
-VAD_THRESHOLD = 0.5
-
-def transcribe_with_controls(audio, transcribe_kwargs):
-    """Call mlx_whisper.transcribe with graceful fallback for unsupported args."""
+def list_audio_devices():
     try:
-        return mlx_whisper.transcribe(audio, **transcribe_kwargs)
-    except Exception as exc:
-        message = str(exc)
-        if "beam_size" in transcribe_kwargs:
-            transcribe_kwargs.pop("beam_size")
-            return mlx_whisper.transcribe(audio, **transcribe_kwargs)
-        raise exc
+        import sounddevice as sd
+        print("\nAvailable Audio Devices:")
+        print(sd.query_devices())
+    except ImportError:
+        print("[Error] sounddevice not installed. Cannot list devices.")
 
-def audio_callback(indata, frames, time, status, audio_queue):
-    if status:
-        print(status, file=sys.stderr)
-    audio_queue.put(indata.copy())
-
-def assign_speakers_to_segments(segments, diarization):
-    if not diarization or not segments:
+def assign_speakers_to_segments(segments, diarization_result):
+    if not diarization_result or not segments:
         return segments
+    
+    # pyannote 4.0.4 returns a DiarizeOutput object with a speaker_diarization attribute
+    annotation = None
+    if hasattr(diarization_result, 'speaker_diarization'):
+        annotation = diarization_result.speaker_diarization
+    elif hasattr(diarization_result, 'itertracks'):
+        annotation = diarization_result
+    
+    if annotation is None:
+        return segments
+
     for segment in segments:
-        segment_start, segment_end = segment['start'], segment['end']
+        segment_start, segment_end = segment.get('start', 0), segment.get('end', 0)
         speaker_intersections = {}
-        for turn, _, speaker in diarization.itertracks(yield_label=True):
-            intersection_start = max(segment_start, turn.start)
-            intersection_end = min(segment_end, turn.end)
-            if intersection_end > intersection_start:
-                duration = intersection_end - intersection_start
-                speaker_intersections[speaker] = speaker_intersections.get(speaker, 0) + duration
+        
+        try:
+            for turn, _, speaker in annotation.itertracks(yield_label=True):
+                intersection_start = max(segment_start, turn.start)
+                intersection_end = min(segment_end, turn.end)
+                if intersection_end > intersection_start:
+                    duration = intersection_end - intersection_start
+                    speaker_intersections[speaker] = speaker_intersections.get(speaker, 0) + duration
+        except Exception:
+            continue
+        
         if speaker_intersections:
             segment['speaker'] = max(speaker_intersections, key=speaker_intersections.get)
         else:
@@ -67,6 +58,24 @@ def assign_speakers_to_segments(segments, diarization):
     return segments
 
 def run_transcription(out_queue, args):
+    import numpy as np
+    import sounddevice as sd
+    import torch
+    import mlx_whisper
+    from silero_vad import load_silero_vad, get_speech_timestamps
+    import queue
+
+    SAMPLERATE, BLOCK_SIZE, VAD_THRESHOLD = 16000, 512, 0.5
+
+    def transcribe_with_controls(audio, transcribe_kwargs):
+        try:
+            return mlx_whisper.transcribe(audio, **transcribe_kwargs)
+        except Exception as exc:
+            if "beam_size" in transcribe_kwargs:
+                transcribe_kwargs.pop("beam_size")
+                return mlx_whisper.transcribe(audio, **transcribe_kwargs)
+            raise exc
+
     device = "mps" if torch.backends.mps.is_available() else "cpu"
     model_name = args.model
     if args.quantize and "4bit" not in model_name:
@@ -77,100 +86,104 @@ def run_transcription(out_queue, args):
     
     diarization_pipeline = None
     if args.speaker_diarization:
+        from pyannote.audio import Pipeline
         print("[Transcribe] Loading speaker diarization pipeline...")
         hf_token = os.environ.get("HF_TOKEN")
         try:
-            diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1", use_auth_token=hf_token)
+            diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1", token=hf_token)
             diarization_pipeline.to(torch.device(device))
         except Exception as e:
             print(f"[Transcribe] Diarization load failed: {e}")
 
     audio_queue = queue.Queue()
-    device_index = args.device
+    def audio_callback(indata, frames, time, status):
+        if status: print(status, file=sys.stderr)
+        audio_queue.put(indata.copy())
 
-    def callback_wrapper(indata, frames, time, status):
-        audio_callback(indata, frames, time, status, audio_queue)
-
-    audio_buffer = []
-    speech_started = False
-    buffer_limit = SAMPLERATE * args.max_buffer
-    rolling_context = ""
+    audio_buffer, speech_started, rolling_context = [], False, ""
     last_stream_time = time.time()
     
-    print(f"[Transcribe] Audio stream ready on device {device_index}.")
-    
     try:
-        with sd.InputStream(samplerate=SAMPLERATE, channels=args.channels, callback=callback_wrapper, blocksize=BLOCK_SIZE, device=device_index):
+        with sd.InputStream(samplerate=SAMPLERATE, channels=args.channels, callback=audio_callback, blocksize=BLOCK_SIZE, device=args.device):
             while True:
                 while not audio_queue.empty():
                     data = audio_queue.get()
-                    if args.channels > 1: data = np.mean(data, axis=1)
-                    audio_buffer.append(data.flatten())
+                    if data is not None and data.size > 0:
+                        if args.channels > 1:
+                            data = np.mean(data, axis=1)
+                        audio_buffer.append(data.flatten())
 
                 if audio_buffer:
                     current_audio = np.concatenate(audio_buffer)
                     audio_tensor = torch.from_numpy(current_audio)
-                    buffer_duration = len(current_audio) / SAMPLERATE
                     
                     speech_timestamps = get_speech_timestamps(audio_tensor, vad_model, sampling_rate=SAMPLERATE, threshold=VAD_THRESHOLD, min_silence_duration_ms=args.silence)
 
                     if speech_timestamps:
                         speech_started = True
-                        last_end = speech_timestamps[-1]['end']
-                        buffer_len_samples = len(current_audio)
-                        
-                        # Check for flush
-                        if (buffer_len_samples - last_end) > (SAMPLERATE * args.silence / 1000) or buffer_len_samples > buffer_limit:
+                        if (len(current_audio) - speech_timestamps[-1]['end']) > (SAMPLERATE * args.silence / 1000) or len(current_audio) > (SAMPLERATE * args.max_buffer):
+                            
+                            # STEP 1: ALWAYS TRANSCRIBE FIRST
                             transcribe_kwargs = {
                                 "path_or_hf_repo": model_name,
                                 "temperature": args.temperature_fallback,
-                                "word_timestamps": args.speaker_diarization,
+                                "word_timestamps": True,
                                 "logprob_threshold": args.logprob_threshold,
                                 "compression_ratio_threshold": args.compression_threshold,
                             }
                             if args.lang: transcribe_kwargs["language"] = args.lang
                             if args.context and rolling_context: transcribe_kwargs["initial_prompt"] = rolling_context
 
-                            result = transcribe_with_controls(current_audio, transcribe_kwargs)
-                            text = result['text'].strip()
-                            detected_lang = result.get('language', args.lang or 'en')
-
-                            if args.filter_lang and args.lang and detected_lang != args.lang:
-                                print(f"[Transcribe] Filtered {detected_lang}")
-                                text = ""
+                            res_asr = transcribe_with_controls(current_audio, transcribe_kwargs)
+                            text = res_asr['text'].strip()
+                            detected_lang = res_asr.get('language', args.lang or 'en')
 
                             if text:
-                                segments = result.get('segments', [])
+                                english_text = text
+                                if detected_lang != "en" or args.lang == 'en':
+                                    if detected_lang != "en":
+                                        bridge_kwargs = transcribe_kwargs.copy()
+                                        bridge_kwargs["task"] = "translate"
+                                        res_bridge = transcribe_with_controls(current_audio, bridge_kwargs)
+                                        english_text = res_bridge['text'].strip()
+
+                                speaker_label = None
                                 if diarization_pipeline:
-                                    audio_for_diarization = torch.from_numpy(current_audio).float().unsqueeze(0)
                                     try:
-                                        diarization_result = diarization_pipeline({"waveform": audio_for_diarization, "sample_rate": SAMPLERATE})
-                                        segments = assign_speakers_to_speakers(segments, diarization_result)
-                                    except: pass
+                                        diar_out = diarization_pipeline({"waveform": torch.from_numpy(current_audio).float().unsqueeze(0), "sample_rate": SAMPLERATE})
+                                        asr_segments = res_asr.get('segments', [])
+                                        diarized_segments = assign_speakers_to_segments(asr_segments, diar_out)
+                                        
+                                        speakers = [s.get('speaker') for s in diarized_segments if s.get('speaker') and s.get('speaker') != 'UNKNOWN']
+                                        if speakers:
+                                            from collections import Counter
+                                            speaker_label = Counter(speakers).most_common(1)[0][0]
+                                    except Exception as diar_err:
+                                        print(f"[Transcribe] Diarization error: {diar_err}")
+
+                                out_queue.put({
+                                    "original": text, 
+                                    "en_bridge": english_text,
+                                    "detected_lang": detected_lang, 
+                                    "speaker": speaker_label,
+                                    "ts": time.time()
+                                })
                                 
-                                out_queue.put({"original": text, "detected_lang": detected_lang, "segments": segments, "ts": time.time()})
-                                print(f"[Transcribe] {detected_lang.upper()}: {text}")
+                                speaker_tag = f" ({speaker_label})" if speaker_label else ""
+                                print(f"[Transcribe] {detected_lang.upper()}{speaker_tag}: {text}")
                                 if args.context: rolling_context = (rolling_context + " " + text)[-200:].strip()
 
-                            audio_buffer = []
-                            speech_started = False
+                            audio_buffer, speech_started = [], False
                         
                         elif args.stream and (time.time() - last_stream_time) > 1.5:
-                            # Draft Mode
-                            draft_kwargs = {"path_or_hf_repo": model_name, "temperature": 0.0}
-                            if args.lang: draft_kwargs["language"] = args.lang
-                            if args.context and rolling_context: draft_kwargs["initial_prompt"] = rolling_context
-                            
-                            draft_result = transcribe_with_controls(current_audio, draft_kwargs)
-                            draft_text = draft_result['text'].strip()
-                            if draft_text:
-                                out_queue.put({"draft": draft_text, "ts": time.time()})
+                            draft_result = transcribe_with_controls(current_audio, {"path_or_hf_repo": model_name, "temperature": 0.0})
+                            if draft_result['text'].strip():
+                                out_queue.put({"draft": draft_result['text'].strip(), "ts": time.time()})
                             last_stream_time = time.time()
                     
                     elif not speech_started and len(current_audio) > SAMPLERATE * 2:
                         audio_buffer = []
 
-    except KeyboardInterrupt: pass
     except Exception as e: print(f"[Transcribe] Error: {e}")
 
 if __name__ == "__main__":

engine_translate.py

@@ -1,79 +1,57 @@
-import sys
-from unittest.mock import MagicMock
-
-# Comprehensive workaround for missing _lzma in some Python builds
-try:
-    import lzma
-except ImportError:
-    mock_lzma = MagicMock()
-    mock_lzma.FORMAT_XZ = 1
-    mock_lzma.FORMAT_ALONE = 2
-    mock_lzma.FORMAT_RAW = 3
-    mock_lzma.CHECK_NONE = 0
-    mock_lzma.CHECK_CRC32 = 1
-    mock_lzma.CHECK_CRC64 = 4
-    mock_lzma.CHECK_SHA256 = 10
-    sys.modules["_lzma"] = MagicMock()
-    sys.modules["lzma"] = mock_lzma
-
-import torch
-from transformers import MarianMTModel, MarianTokenizer
-import re
 import argparse
 import multiprocessing
 import time
-import requests
-
-TARGET_LANGS = {
-    "es": "Helsinki-NLP/opus-mt-en-es",
-    "fr": "Helsinki-NLP/opus-mt-en-fr",
-    "ar": "Helsinki-NLP/opus-mt-en-ar"
-}
-
-def split_text_for_translation(text, max_chars=250):
-    normalized = " ".join(text.split()).strip()
-    if not normalized or len(normalized) <= max_chars:
-        return [normalized] if normalized else []
-    chunks = []
-    current = ""
-    sentences = [s for s in re.split(r"(?<=[.!?])\s+", normalized) if s]
-    for sentence in sentences:
-        if len(sentence) > max_chars:
-            words = sentence.split()
-            word_chunk = ""
-            for word in words:
-                candidate = f"{word_chunk} {word}".strip()
-                if len(candidate) <= max_chars: word_chunk = candidate
-                else:
-                    if word_chunk: chunks.append(word_chunk)
-                    word_chunk = word
-            if word_chunk: chunks.append(word_chunk)
-            continue
-        candidate = f"{current} {sentence}".strip()
-        if candidate and len(candidate) <= max_chars: current = candidate
-        else:
-            if current: chunks.append(current)
-            current = sentence
-    if current: chunks.append(current)
-    return chunks
-
-def call_ollama_generate(args, prompt):
-    payload = {"model": args.post_correct_model, "prompt": prompt, "stream": False}
-    try:
-        resp = requests.post("http://127.0.0.1:11434/api/generate", json=payload, timeout=10)
-        return resp.json().get("response", "").strip()
-    except: return ""
-
-def structure_paragraph_with_llm(text, args):
-    prompt = f"Format the following transcribed segments into a coherent paragraph. Output ONLY the paragraph:\n{text}"
-    return call_ollama_generate(args, prompt) or text
-
-def post_correct_with_llm(text, args):
-    prompt = f"Correct grammar and typos in this caption. Output ONLY the corrected text:\n{text}"
-    return call_ollama_generate(args, prompt) or text
+import re
 
 def run_translation(in_queue, out_queue, args):
+    import sys
+    from unittest.mock import MagicMock
+    # Comprehensive workaround for missing _lzma in some Python builds
+    try:
+        import lzma
+    except ImportError:
+        mock_lzma = MagicMock()
+        mock_lzma.FORMAT_XZ, mock_lzma.FORMAT_ALONE, mock_lzma.FORMAT_RAW = 1, 2, 3
+        mock_lzma.CHECK_NONE, mock_lzma.CHECK_CRC32, mock_lzma.CHECK_CRC64, mock_lzma.CHECK_SHA256 = 0, 1, 4, 10
+        sys.modules["_lzma"] = MagicMock()
+        sys.modules["lzma"] = mock_lzma
+
+    import torch
+    from transformers import MarianMTModel, MarianTokenizer
+
+    TARGET_LANGS = {
+        "es": "Helsinki-NLP/opus-mt-en-es",
+        "fr": "Helsinki-NLP/opus-mt-en-fr",
+        "ar": "Helsinki-NLP/opus-mt-en-ar"
+    }
+
+    def split_text_for_translation(text, max_chars=250):
+        normalized = " ".join(text.split()).strip()
+        if not normalized or len(normalized) <= max_chars: return [normalized] if normalized else []
+        chunks, current = [], ""
+        sentences = [s for s in re.split(r"(?<=[.!?])\s+", normalized) if s]
+        for sentence in sentences:
+            if len(sentence) > max_chars:
+                words, word_chunk = sentence.split(), ""
+                for word in words:
+                    candidate = f"{word_chunk} {word}".strip()
+                    if len(candidate) <= max_chars: word_chunk = candidate
+                    else:
+                        if word_chunk: chunks.append(word_chunk)
+                        word_chunk = word
+                if word_chunk: chunks.append(word_chunk)
+                continue
+            candidate = f"{current} {sentence}".strip()
+            if candidate and len(candidate) <= max_chars: current = candidate
+            else:
+                if current: chunks.append(current)
+                current = sentence
+        if current: chunks.append(current)
+        return chunks
+
     device = "mps" if torch.backends.mps.is_available() else "cpu"
+    print(f"[Translate] Loading translation models on {device}...")
+    
     translation_engines = {}
     for lang_key, model_id in TARGET_LANGS.items():
         if getattr(args, lang_key, False):
@@ -83,51 +61,40 @@ def run_translation(in_queue, out_queue, args):
                 MarianTokenizer.from_pretrained(model_id)
             )
 
-    paragraph_buffer = []
-    
     while True:
         try:
             item = in_queue.get()
             if item is None: break
-            
-            # Draft support
             if "draft" in item:
-                out_queue.put({"draft": item["draft"]})
+                out_queue.put(item)
                 continue
 
-            text = item.get("original", "").strip()
-            detected_lang = item.get("detected_lang", "en")
-            
-            if args.post_correct and args.post_correct_llm:
-                text = post_correct_with_llm(text, args)
+            raw_text, corrected_text, paragraph_text = item.get("raw"), item.get("corrected"), item.get("paragraph")
+            payload = {
+                "original": raw_text, 
+                "corrected": corrected_text, 
+                "paragraph": paragraph_text, 
+                "speaker": item.get("speaker"),
+                "ts": item.get("ts", time.time())
+            }
+            if args.en: payload["en"] = corrected_text or raw_text
 
-            paragraph_buffer.append(text)
-            
-            # Paragraph logic
-            if not args.llm_paragraph or text.endswith((".", "?", "!")):
-                full_text = " ".join(paragraph_buffer)
+            text_to_translate = paragraph_text or corrected_text or raw_text
+            if text_to_translate:
+                if translation_engines:
+                    for lang_key, (model, tokenizer) in translation_engines.items():
+                        chunks = split_text_for_translation(text_to_translate, args.mt_max_chars)
+                        translated_parts = []
+                        for chunk in chunks:
+                            inputs = tokenizer(chunk, return_tensors="pt", padding=True).to(device)
+                            with torch.no_grad():
+                                translated_tokens = model.generate(**inputs, max_new_tokens=args.mt_max_new_tokens, num_beams=args.mt_num_beams)
+                            translated_parts.append(tokenizer.decode(translated_tokens[0], skip_special_tokens=True).strip())
+                        payload[lang_key] = " ".join(translated_parts)
                 
-                if args.llm_paragraph:
-                    full_text = structure_paragraph_with_llm(full_text, args)
-
-                payload = {"original": full_text, "ts": item.get("ts", time.time())}
-                if args.en: payload["en"] = full_text
-                
-                for lang_key, (model, tokenizer) in translation_engines.items():
-                    chunks = split_text_for_translation(full_text, args.mt_max_chars)
-                    translated_parts = []
-                    for chunk in chunks:
-                        inputs = tokenizer(chunk, return_tensors="pt", padding=True).to(device)
-                        with torch.no_grad():
-                            translated_tokens = model.generate(**inputs, max_new_tokens=args.mt_max_new_tokens, num_beams=args.mt_num_beams)
-                        translated_parts.append(tokenizer.decode(translated_tokens[0], skip_special_tokens=True).strip())
-                    payload[lang_key] = " ".join(translated_parts)
-
+                # ALWAYS put in queue, even if no translations were done
                 out_queue.put(payload)
-                paragraph_buffer = []
-                
         except Exception as e: print(f"[Translate] Error: {e}")
 
 if __name__ == "__main__":
-    import multiprocessing
     run_translation(multiprocessing.Queue(), multiprocessing.Queue(), argparse.Namespace())

history.md

@@ -214,3 +214,27 @@ The project now features a high-performance, Apple Silicon-optimized pipeline th
     python3 transcribe.py --silence 100 -q -s -c -es -en -fr -ar --lang en --filter-lang --mt-max-chars 450 --mt-max-new-tokens 220 --smart-correct --post-correct --post-correct-llm --post-correct-model qwen2.5:3b-instruct --session-log-file debug_session_1.log
     ```
 - **Outcome:** The project now has a repeatable "known good" runtime profile plus persistent session diagnostics for reviewing caption issues after a run.
+
+## Phase 26: Multi-Process Decoupled Architecture
+- **Goal:** Improve system stability, reduce latency, and fully decouple audio capture from heavy LLM/Translation tasks.
+- **Approach:**
+    - **4-Process Pipeline:** Refactored the monolithic script into four independent services coordinated via `multiprocessing.Queue`:
+        1. **`engine_transcribe.py` (Whisper):** Dedicated to high-priority audio capture and ASR.
+        2. **`engine_llm.py` (Ollama):** Handles asynchronous post-correction and paragraph structuring without blocking the transcription loop.
+        3. **`engine_translate.py` (MarianMT):** Manages multi-language translation for both raw and refined text.
+        4. **`engine_distribute.py` (API/CLI):** Handles data delivery and terminal display.
+    - **Centralized Configuration:** Introduced `config.json` for managing all defaults and parameters in one place, with `main_v2.py` as the entry point.
+    - **Dual Payload Strategy:** Implemented a robust data flow where the translation and distribution engines receive both raw and LLM-corrected versions of the text, allowing for fallback and comparison.
+    - **Hardware Isolation:** Transcription and Translation processes independently leverage Apple Silicon (MPS), while the LLM process utilizes Ollama's external server, preventing resource contention.
+- **Outcome:** Significantly increased resilience. If the LLM or Translation engine stalls, the Transcription engine continues to capture and buffer audio safely, preventing data loss.
+
+## Phase 27: Intelligent LLM Refinement & Pipeline Hardening
+- **Goal:** Transform raw ASR segments into professional paragraphs and stabilize advanced features.
+- **Approach:**
+    - **Accumulative LLM Engine:** Developed a stateful refinement logic in `engine_llm.py` that maintains a "working paragraph," allowing the LLM to continuously integrate and polish new segments in real-time.
+    - **OpenAI Integration:** Added direct support for OpenAI's GPT API (via `requests` to avoid subprocess dependency issues), enabling higher-quality refinement than lightweight local models.
+    - **Rolling Context Window:** Implemented a smart context management system that "finalizes" paragraphs once they reach a natural break (detected via `\n\n`), clearing the prompt history to save tokens and maintain focus.
+    - **Speaker Diarization Hardening:** Refactored `engine_transcribe.py` to support `pyannote.audio` 4.0.4, specifically handling the new `DiarizeOutput` object structure and ensuring speaker labels (`SPEAKER_XX`) propagate through the entire multi-process pipeline.
+    - **English Bridge Optimization:** Forced a two-pass transcription strategy (ASR first, then optional translation) to ensure word-level timestamps are captured for speaker mapping even when translating to English.
+    - **Stability Fixes:** Added hallucination filtering for Whisper's repetitive loops, enforced line-buffering for terminal logging, and implemented a global `lzma` mock to ensure compatibility across restricted Python environments.
+- **Outcome:** A robust, production-ready pipeline that produces high-quality, speaker-attributed, and professionally formatted live transcripts.

main_v2.py

@@ -1,26 +1,16 @@
-import sys
-from unittest.mock import MagicMock
-
-# Comprehensive workaround for missing _lzma in some Python builds
-try:
-    import lzma
-except ImportError:
-    mock_lzma = MagicMock()
-    mock_lzma.FORMAT_XZ = 1
-    mock_lzma.FORMAT_ALONE = 2
-    mock_lzma.FORMAT_RAW = 3
-    mock_lzma.CHECK_NONE = 0
-    mock_lzma.CHECK_CRC32 = 1
-    mock_lzma.CHECK_CRC64 = 4
-    mock_lzma.CHECK_SHA256 = 10
-    sys.modules["_lzma"] = MagicMock()
-    sys.modules["lzma"] = mock_lzma
-
 import multiprocessing
 import argparse
 import json
 import os
-from engine_transcribe import run_transcription
+import sys
+from dotenv import load_dotenv
+
+# Load environment variables from .env if present
+load_dotenv()
+
+# Light imports (heavy ones moved inside run functions)
+from engine_transcribe import run_transcription, list_audio_devices
+from engine_llm import run_llm_processor
 from engine_translate import run_translation
 from engine_distribute import run_distribution
 
@@ -40,6 +30,7 @@ def main():
     parser = argparse.ArgumentParser(description="Multi-process Transcription and Translation.")
     
     # Transcribe Args
+    parser.add_argument("-l", "--list-devices", action="store_true", help="Show available audio devices and exit")
     parser.add_argument("--model", type=str, default=defaults.get("model", "mlx-community/whisper-base-mlx"))
     parser.add_argument("--device", type=int, default=defaults.get("device"))
     parser.add_argument("--lang", type=str, default=defaults.get("lang"))
@@ -55,15 +46,17 @@ def main():
     parser.add_argument("--logprob-threshold", type=float, default=defaults.get("logprob_threshold", -0.8))
     parser.add_argument("--compression-threshold", type=float, default=defaults.get("compression_threshold", 2.2))
     
+    # LLM Args
+    parser.add_argument("--post-correct", action="store_true", default=defaults.get("post_correct", False))
+    parser.add_argument("--post-correct-llm", action="store_true", default=defaults.get("post_correct_llm", False))
+    parser.add_argument("--post-correct-model", type=str, default=defaults.get("post_correct_model", "qwen:2b"))
+    parser.add_argument("--llm-paragraph", action="store_true", default=defaults.get("llm_paragraph", False))
+
     # Translate Args
     parser.add_argument("-es", action="store_true", default=defaults.get("es", False))
     parser.add_argument("-fr", action="store_true", default=defaults.get("fr", False))
     parser.add_argument("-ar", action="store_true", default=defaults.get("ar", False))
     parser.add_argument("-en", action="store_true", default=defaults.get("en", False))
-    parser.add_argument("--post-correct", action="store_true", default=defaults.get("post_correct", False))
-    parser.add_argument("--post-correct-llm", action="store_true", default=defaults.get("post_correct_llm", False))
-    parser.add_argument("--post-correct-model", type=str, default=defaults.get("post_correct_model", "qwen:2b"))
-    parser.add_argument("--llm-paragraph", action="store_true", default=defaults.get("llm_paragraph", False))
     parser.add_argument("--mt-max-chars", type=int, default=defaults.get("mt_max_chars", 250))
     parser.add_argument("--mt-max-new-tokens", type=int, default=defaults.get("mt_max_new_tokens", 150))
     parser.add_argument("--mt-num-beams", type=int, default=defaults.get("mt_num_beams", 4))
@@ -73,27 +66,49 @@ def main():
     
     args = parser.parse_args()
 
+    # Handle device listing
+    if args.list_devices:
+        list_audio_devices()
+        return
+
+    # Handle interactive device selection
+    if args.device is None:
+        list_audio_devices()
+        try:
+            val = input("\nSelect input device index (or press Enter for default): ").strip()
+            if val:
+                args.device = int(val)
+        except EOFError:
+            pass # Non-interactive environment
+        except ValueError:
+            print("[Main] Invalid index, using default.")
+
     # Queues
-    q_trans_to_tl = multiprocessing.Queue()
+    q_trans_to_llm = multiprocessing.Queue()
+    q_llm_to_tl = multiprocessing.Queue()
     q_tl_to_dist = multiprocessing.Queue()
 
     # Processes
-    p_transcribe = multiprocessing.Process(target=run_transcription, args=(q_trans_to_tl, args))
-    p_translate = multiprocessing.Process(target=run_translation, args=(q_trans_to_tl, q_tl_to_dist, args))
+    p_transcribe = multiprocessing.Process(target=run_transcription, args=(q_trans_to_llm, args))
+    p_llm = multiprocessing.Process(target=run_llm_processor, args=(q_trans_to_llm, q_llm_to_tl, args))
+    p_translate = multiprocessing.Process(target=run_translation, args=(q_llm_to_tl, q_tl_to_dist, args))
     p_distribute = multiprocessing.Process(target=run_distribution, args=(q_tl_to_dist, args))
 
     print("[Main] Starting processes...")
     p_transcribe.start()
+    p_llm.start()
     p_translate.start()
     p_distribute.start()
 
     try:
         p_transcribe.join()
+        p_llm.join()
         p_translate.join()
         p_distribute.join()
     except KeyboardInterrupt:
         print("\n[Main] Stopping processes...")
         p_transcribe.terminate()
+        p_llm.terminate()
         p_translate.terminate()
         p_distribute.terminate()
         sys.exit(0)

requirements.txt

@@ -13,3 +13,4 @@ pyannote.audio
 soundfile
 langdetect
 python-dotenv
+openai