From 50f58dbc224a2296c77d3f088058062ef48bc6e0 Mon Sep 17 00:00:00 2001 From: Adolfo Reyna Date: Sun, 15 Mar 2026 22:05:05 -0400 Subject: [PATCH] feat: implement accumulative LLM paragraph engine and harden 4-process pipeline --- engine_distribute.py | 35 +++++++- engine_llm.py | 130 +++++++++++++++++++++++++++++ engine_transcribe.py | 193 +++++++++++++++++++++++-------------------- engine_translate.py | 175 ++++++++++++++++----------------------- history.md | 24 ++++++ main_v2.py | 67 +++++++++------ requirements.txt | 1 + 7 files changed, 401 insertions(+), 224 deletions(-) create mode 100644 engine_llm.py diff --git a/engine_distribute.py b/engine_distribute.py index 0810b1a..8329f2a 100644 --- a/engine_distribute.py +++ b/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 diff --git a/engine_llm.py b/engine_llm.py new file mode 100644 index 0000000..7691f47 --- /dev/null +++ b/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()) diff --git a/engine_transcribe.py b/engine_transcribe.py index 0f5ac8b..60183a2 100644 --- a/engine_transcribe.py +++ b/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__": diff --git a/engine_translate.py b/engine_translate.py index 4126826..7d70ab7 100644 --- a/engine_translate.py +++ b/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()) diff --git a/history.md b/history.md index 0d5dbbe..904edb6 100644 --- a/history.md +++ b/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. diff --git a/main_v2.py b/main_v2.py index 986d9ba..242dc0e 100644 --- a/main_v2.py +++ b/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) diff --git a/requirements.txt b/requirements.txt index d94fe0a..d6da72b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,3 +13,4 @@ pyannote.audio soundfile langdetect python-dotenv +openai