diff --git a/config.json b/config.json index 5f17899..d5f10a6 100644 --- a/config.json +++ b/config.json @@ -18,6 +18,11 @@ "post_correct": false, "post_correct_llm": false, "post_correct_model": "qwen:2b", + "post_correct_ollama_url": "http://127.0.0.1:11434/api/generate", + "post_correct_llm_timeout": 8.0, + "post_correct_keep_alive": "30m", + "post_correct_min_overlap": 0.45, + "post_correct_debug": false, "llm_paragraph": false, "only_translate_llm": false, "temperature_fallback": [0.0, 0.2, 0.4, 0.6, 0.8, 1.0], diff --git a/engine_distribute.py b/engine_distribute.py index f585903..b55fa93 100644 --- a/engine_distribute.py +++ b/engine_distribute.py @@ -7,6 +7,21 @@ import os INGEST_URL = "https://emiapi.reynafamily.com/live-captions/ingest" +def post_with_retries(payload, timeout, max_attempts=3): + delay = 1 + for attempt in range(1, max_attempts + 1): + try: + response = requests.post(INGEST_URL, json=payload, timeout=timeout) + if response.status_code == 200: + return True + print(f"[Distribute Error] {response.status_code} on attempt {attempt}/{max_attempts}.") + except Exception as exc: + print(f"[Distribute Error] {exc} on attempt {attempt}/{max_attempts}.") + if attempt < max_attempts: + time.sleep(delay) + delay = min(delay * 2, 5) + return False + def log_debug(message): try: with open("distribute_debug.log", "a") as f: @@ -31,8 +46,7 @@ def run_distribution(in_queue, args): sys.stdout.write(f"\r\033[K[DRAFT]: {draft_text}") sys.stdout.flush() if args.ingest: - try: requests.post(INGEST_URL, json={"draft": draft_text}, timeout=2) - except: pass + post_with_retries({"draft": draft_text}, timeout=2, max_attempts=2) continue # Finalized segment @@ -75,18 +89,9 @@ def run_distribution(in_queue, args): should_send = False if should_send: - delay, max_delay, success = 1, 15, False - while not success: - try: - response = requests.post(INGEST_URL, json=ingest_payload, timeout=5) - if response.status_code == 200: success = True - else: print(f"[Distribute Error] {response.status_code}. Retrying...") - except Exception as e: - print(f"[Distribute Error] {e}. Retrying...") - - if not success: - time.sleep(delay) - delay = min(delay * 2, max_delay) + success = post_with_retries(ingest_payload, timeout=5, max_attempts=3) + if not success: + log_debug(f"Dropped ingest payload after retries: {ingest_payload}") except Exception as e: print(f"[Distribute] Error: {e}") diff --git a/engine_llm.py b/engine_llm.py index 7691f47..0e29f77 100644 --- a/engine_llm.py +++ b/engine_llm.py @@ -15,6 +15,7 @@ import time import argparse import multiprocessing import os +import re import requests from collections import deque @@ -26,18 +27,74 @@ def run_llm_processor(in_queue, out_queue, args): 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 + llm_state = {"line_warned": False, "paragraph_warned": False} + + def normalize_english_caption(text): + normalized = " ".join((text or "").split()).strip() + if normalized and normalized[0].isalpha(): + normalized = normalized[0].upper() + normalized[1:] + return normalized + + def apply_rule_based_post_correction(text): + corrected = normalize_english_caption(text) + corrected = re.sub(r"\s+", " ", corrected).strip() + corrected = re.sub(r"\s+([,.;!?])", r"\1", corrected) + corrected = re.sub(r"([,.;!?]){2,}", r"\1", corrected) + corrected = re.sub(r"\b(uh+|um+|erm+|ah+|hmm+)\b", "", corrected, flags=re.IGNORECASE) + corrected = re.sub(r"\s+", " ", corrected).strip() + corrected = re.sub(r"\b(\w+)\s+\1\b", r"\1", corrected, flags=re.IGNORECASE) + return normalize_english_caption(corrected) + + def word_overlap_ratio(source_text, candidate_text): + src_tokens = set(re.findall(r"[a-z0-9']+", source_text.lower())) + cand_tokens = set(re.findall(r"[a-z0-9']+", candidate_text.lower())) + if not src_tokens: + return 1.0 + return len(src_tokens & cand_tokens) / len(src_tokens) + + def call_openai(messages): + response = requests.post( + "https://api.openai.com/v1/chat/completions", + json={"model": args.post_correct_model, "messages": messages, "temperature": 0.1}, + headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, + timeout=getattr(args, "post_correct_llm_timeout", 8.0), + ) + response.raise_for_status() + return response.json()["choices"][0]["message"]["content"].strip() + + def call_ollama(prompt, temperature): + response = requests.post( + getattr(args, "post_correct_ollama_url", "http://127.0.0.1:11434/api/generate"), + json={ + "model": args.post_correct_model, + "prompt": prompt, + "stream": False, + "options": {"temperature": temperature}, + "keep_alive": getattr(args, "post_correct_keep_alive", "30m"), + }, + timeout=getattr(args, "post_correct_llm_timeout", 8.0), + ) + response.raise_for_status() + return response.json().get("response", "").strip() + + def call_llm(messages, prompt, temperature, warn_key): + try: + if is_openai: + if not api_key: + raise RuntimeError("OPENAI_API_KEY is not set") + return call_openai(messages) + return call_ollama(prompt, temperature) + except Exception as exc: + if not llm_state[warn_key]: + print(f"[LLM] {warn_key.replace('_', ' ').title()} unavailable ({exc}). Falling back to deterministic text.") + llm_state[warn_key] = True + return "" # State: This is the ONLY text we send to the LLM as context active_context = "" + recent_lines = deque(maxlen=2) 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] @@ -50,21 +107,50 @@ def run_llm_processor(in_queue, out_queue, args): 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}. +4. TARGET LANGUAGE: Output ONLY in English. 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 "" + messages = [ + {"role": "system", "content": "You are a live transcript editor."}, + {"role": "user", "content": prompt}, + ] + return call_llm(messages, prompt, 0.1, "paragraph_warned") + + def post_correct_line(text): + if not getattr(args, "post_correct", False): + return normalize_english_caption(text) + + corrected = apply_rule_based_post_correction(text) + if not getattr(args, "post_correct_llm", False): + return corrected + + prev2 = recent_lines[-2] if len(recent_lines) >= 2 else "" + prev1 = recent_lines[-1] if len(recent_lines) >= 1 else "" + prompt = ( + "You are a real-time English caption corrector for live speech.\n" + "Task:\n" + "Clean ONLY the current caption line.\n" + "Hard rules:\n" + "1. Preserve meaning exactly. Never replace current content with prior context.\n" + "2. Remove disfluencies and false starts.\n" + "3. Fix punctuation, casing, and obvious STT typos.\n" + "4. If uncertain, return the original line unchanged.\n" + "5. Output ONLY one corrected English line.\n" + "Context (reference only):\n" + f"Previous line 1: {prev1}\n" + f"Previous line 2: {prev2}\n" + "Current line to correct:\n" + f"{corrected}\n" + "Corrected:" + ) + messages = [ + {"role": "system", "content": "You are a real-time English caption corrector."}, + {"role": "user", "content": prompt}, + ] + candidate = normalize_english_caption(call_llm(messages, prompt, 0.1, "line_warned")) + if candidate and word_overlap_ratio(corrected, candidate) >= getattr(args, "post_correct_min_overlap", 0.45): + return candidate + return corrected pending_buffer = [] last_llm_call_time = time.time() @@ -81,15 +167,18 @@ def run_llm_processor(in_queue, out_queue, args): bridge_text = item.get("en_bridge", raw_text).strip() if not raw_text: continue + corrected_text = post_correct_line(bridge_text) + # Hallucination check - words = bridge_text.lower().split() + words = corrected_text.lower().split() if len(words) > 10 and len(set(words)) < 3: continue - pending_buffer.append(bridge_text) + pending_buffer.append(corrected_text) + recent_lines.append(corrected_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) >= 2 and corrected_text.endswith((".", "?", "!")): should_refine = True elif len(pending_buffer) >= 1 and (time.time() - last_llm_call_time) > 15: should_refine = True structured_paragraph = None @@ -110,13 +199,16 @@ def run_llm_processor(in_queue, out_queue, args): # No break, just update the context active_context = result structured_paragraph = result - - pending_buffer = [] - last_llm_call_time = time.time() + else: + structured_paragraph = new_batch + active_context = new_batch + + pending_buffer = [] + last_llm_call_time = time.time() out_queue.put({ "raw": raw_text, - "corrected": bridge_text, + "corrected": corrected_text, "paragraph": structured_paragraph, "detected_lang": item.get("detected_lang"), "speaker": item.get("speaker"), diff --git a/engine_transcribe.py b/engine_transcribe.py index 60183a2..a19a90e 100644 --- a/engine_transcribe.py +++ b/engine_transcribe.py @@ -100,7 +100,7 @@ def run_transcription(out_queue, args): if status: print(status, file=sys.stderr) audio_queue.put(indata.copy()) - audio_buffer, speech_started, rolling_context = [], False, "" + audio_buffer, speech_started, rolling_context, rolling_context_en = [], False, "", "" last_stream_time = time.time() try: @@ -127,7 +127,7 @@ def run_transcription(out_queue, args): transcribe_kwargs = { "path_or_hf_repo": model_name, "temperature": args.temperature_fallback, - "word_timestamps": True, + "word_timestamps": args.speaker_diarization, "logprob_threshold": args.logprob_threshold, "compression_ratio_threshold": args.compression_threshold, } @@ -139,13 +139,26 @@ def run_transcription(out_queue, args): detected_lang = res_asr.get('language', args.lang or 'en') if text: + if args.filter_lang and args.lang and detected_lang != args.lang: + print(f"[Transcribe] Discarding segment (Detected: {detected_lang}, Expected: {args.lang})") + audio_buffer, speech_started = [], False + continue + 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() + if detected_lang != "en": + bridge_kwargs = { + "path_or_hf_repo": model_name, + "task": "translate", + "temperature": args.temperature_fallback, + "logprob_threshold": args.logprob_threshold, + "compression_ratio_threshold": args.compression_threshold, + } + if args.lang: + bridge_kwargs["language"] = args.lang + if args.context and rolling_context_en: + bridge_kwargs["initial_prompt"] = rolling_context_en + res_bridge = transcribe_with_controls(current_audio, bridge_kwargs) + english_text = res_bridge.get("text", "").strip() or text speaker_label = None if diarization_pipeline: @@ -171,7 +184,9 @@ def run_transcription(out_queue, args): 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() + if args.context: + rolling_context = (rolling_context + " " + text)[-200:].strip() + rolling_context_en = (rolling_context_en + " " + english_text)[-200:].strip() audio_buffer, speech_started = [], False diff --git a/engine_translate.py b/engine_translate.py index 0cb1fc4..04ca4eb 100644 --- a/engine_translate.py +++ b/engine_translate.py @@ -85,7 +85,7 @@ def run_translation(in_queue, out_queue, args): else: text_to_translate = paragraph_text or corrected_text or raw_text if args.en: - payload["en"] = corrected_text or raw_text + payload["en"] = text_to_translate if text_to_translate and translation_engines: for lang_key, (model, tokenizer) in translation_engines.items(): diff --git a/history.md b/history.md index 904edb6..349fc05 100644 --- a/history.md +++ b/history.md @@ -238,3 +238,20 @@ The project now features a high-performance, Apple Silicon-optimized pipeline th - **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. + +## Phase 28: `main_v2` Bridge Contract Fixes + Live Reliability Review +- **Goal:** Repair stage-contract regressions in the new multi-process pipeline and preserve live throughput under failure. +- **Review Findings:** + - The Whisper English bridge in `engine_transcribe.py` reused source-language prompt context for `task="translate"` instead of a dedicated English context, which could bias or degrade translated bridge text. + - `--filter-lang` was exposed in `main_v2.py` but not enforced in the new transcription engine, so wrong-language speech could still be bridged and propagated downstream. + - `engine_llm.py` told the paragraph refiner to output the source language even though downstream Marian models require English input, breaking `llm_paragraph` for non-English sources. + - `engine_translate.py` could translate one English variant while publishing a different `[EN]` line, making the visible source text diverge from what target-language MT actually used. + - `engine_distribute.py` retried ingest forever in the queue consumer, allowing a network outage to stall the whole live pipeline. + - `main_v2.py` exposed post-correction settings that the new pipeline did not fully honor, making CLI behavior drift from the documented workflow. +- **Approach:** + - Restored a dedicated rolling English prompt context for the Whisper bridge pass and rebuilt bridge kwargs explicitly instead of copying source-ASR kwargs. + - Reinstated `--filter-lang` enforcement before bridge generation and reduced unnecessary ASR cost by only asking Whisper for word timestamps when diarization is enabled. + - Updated the LLM stage so paragraph refinement always preserves English bridge text, added deterministic post-correction plus optional LLM line cleanup, and provided a safe fallback paragraph when the LLM backend is unavailable. + - Aligned published English output with the exact text chosen for translation and added the missing Ollama-related config/plumbing to `main_v2`. + - Replaced infinite ingest retry loops with bounded retry logic so failed delivery degrades gracefully instead of freezing caption flow. +- **Outcome:** The `main_v2` pipeline now keeps an English-first contract across transcription, refinement, and Marian translation while behaving more predictably under API outages and mixed-language input. diff --git a/main_v2.py b/main_v2.py index 987cad1..20a1c78 100644 --- a/main_v2.py +++ b/main_v2.py @@ -50,6 +50,11 @@ def main(): 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("--post-correct-ollama-url", type=str, default=defaults.get("post_correct_ollama_url", "http://127.0.0.1:11434/api/generate")) + parser.add_argument("--post-correct-llm-timeout", type=float, default=defaults.get("post_correct_llm_timeout", 8.0)) + parser.add_argument("--post-correct-keep-alive", type=str, default=defaults.get("post_correct_keep_alive", "30m")) + parser.add_argument("--post-correct-min-overlap", type=float, default=defaults.get("post_correct_min_overlap", 0.45)) + parser.add_argument("--post-correct-debug", action="store_true", default=defaults.get("post_correct_debug", False)) parser.add_argument("--llm-paragraph", action="store_true", default=defaults.get("llm_paragraph", False)) # Translate Args