diff --git a/history.md b/history.md index b6fe41c..165ec1b 100644 --- a/history.md +++ b/history.md @@ -185,3 +185,12 @@ The project now features a high-performance, Apple Silicon-optimized pipeline th - When a probable speaker switch is detected, the current buffered caption line is force-flushed early instead of waiting only for silence. - Added speaker-state resets on watchdog/hallucination/silence reset paths to avoid stale identity drift. - **Outcome:** Users can opt into faster caption segmentation at speaker boundaries while keeping default behavior unchanged. + +## Phase 23: Prompt Hardening + Anti-Drift Guardrails +- **Goal:** Reduce semantic drift in LLM post-correction and merge decisions. +- **Approach:** + - Rewrote post-correction prompt with strict "current line only" and "if uncertain, keep original" constraints. + - Expanded post-correction context to up to two previous lines but marked as reference-only. + - Added token-overlap acceptance guard (`--post-correct-min-overlap`) to reject LLM rewrites that diverge too far from source text. + - Updated merge-decider prompt to stricter JSON behavior and added overlap-based fallback to heuristic merge. +- **Outcome:** Improved protection against context bleed (e.g., replacing current line with prior sentence) while keeping optional LLM improvements. diff --git a/transcribe.py b/transcribe.py index 96198dd..c2030f1 100644 --- a/transcribe.py +++ b/transcribe.py @@ -274,26 +274,43 @@ def apply_rule_based_post_correction(text): corrected = corrected[0].upper() + corrected[1:] return corrected -def post_correct_with_local_llm(text, args, llm_state, context_hint=""): +def word_overlap_ratio(source_text, candidate_text): + """Compute token overlap ratio between source and candidate.""" + 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 post_correct_with_local_llm(text, args, llm_state, context_lines=None): """Optionally rewrite finalized English caption via a local Ollama model.""" if not args.post_correct_llm or not llm_state.get("available", True): return text + context_lines = context_lines or [] + prev2 = context_lines[-2] if len(context_lines) >= 2 else "" + prev1 = context_lines[-1] if len(context_lines) >= 1 else "" + if args.post_correct_debug: print(f"[POST-LLM][PRE]: {text}") prompt = ( - "You correct live English captions.\n" - "Rules:\n" - "- Preserve original meaning exactly.\n" - "- Fix grammar, punctuation, and natural wording.\n" - "- Remove filler words only when they do not change meaning.\n" - "- Keep output concise and in one line.\n" - "- Return only corrected text, no quotes or explanation.\n" + "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 (um, uh, you know, I mean, repeated starts).\n" + "3. Fix punctuation, casing, and obvious STT typos.\n" + "4. Keep the speaker's tone and intent; do not paraphrase for style.\n" + "5. If uncertain, return the original line unchanged.\n" + "6. Output ONLY one corrected line. No explanations, no quotes.\n" + "Context (reference only; never override current line):\n" + f"Previous line 1: {prev1}\n" + f"Previous line 2: {prev2}\n" + "Current line to correct:\n" ) - if context_hint: - prompt += f"Previous caption context: {context_hint}\n" - prompt += f"Caption: {text}\nCorrected:" + prompt += f"{text}\nCorrected:" try: response = requests.post( @@ -312,6 +329,13 @@ def post_correct_with_local_llm(text, args, llm_state, context_hint=""): if args.post_correct_debug: print(f"[POST-LLM][RAW]: {raw}") if cleaned: + overlap = word_overlap_ratio(text, cleaned) + if overlap < args.post_correct_min_overlap: + if args.post_correct_debug: + print( + f"[POST-LLM][REJECT]: low overlap ({overlap:.2f} < {args.post_correct_min_overlap:.2f}); keeping pre text" + ) + return text if args.post_correct_debug: print(f"[POST-LLM][POST]: {cleaned}") return cleaned @@ -354,13 +378,15 @@ def llm_decide_caption_merge(previous_text, current_text, candidate_text, args, return True, candidate_text prompt = ( - "You decide whether two live English caption lines should be merged.\n" - "Return strict JSON only in this shape:\n" - "{\"merge\": true|false, \"text\": \"final merged text or empty\"}\n" + "Decide if two consecutive captions should be merged.\n" + "Return JSON only:\n" + "{\"merge\": true|false, \"text\": \"...\"}\n" "Rules:\n" - "- Preserve meaning exactly.\n" - "- If uncertain, return merge=false.\n" - "- If merge=true, provide a single corrected line in text.\n" + "1. Merge only if current line clearly continues previous line.\n" + "2. Never drop unique information from either line.\n" + "3. If uncertain, return {\"merge\": false, \"text\": \"\"}.\n" + "4. If merge=true, output one corrected merged line.\n" + "5. Do not invent content.\n" f"Previous: {previous_text}\n" f"Current: {current_text}\n" f"HeuristicMerged: {candidate_text}\n" @@ -401,6 +427,9 @@ def llm_decide_caption_merge(previous_text, current_text, candidate_text, args, # Safety: reject large rewrites and keep heuristic merge. if len(llm_text) > max(2 * len(candidate_text), len(candidate_text) + 80): return True, candidate_text + merged_source = f"{previous_text} {current_text}".strip() + if word_overlap_ratio(merged_source, llm_text) < args.post_correct_min_overlap: + return True, candidate_text return True, llm_text except Exception as exc: if not llm_state.get("merge_warned", False): @@ -534,6 +563,7 @@ def main(): parser.add_argument("--post-correct-model", type=str, default="llama3.1:8b-instruct", help="Local Ollama model for post-correction") parser.add_argument("--post-correct-ollama-url", type=str, default="http://127.0.0.1:11434/api/generate", help="Ollama generate endpoint for local post-correction") parser.add_argument("--post-correct-llm-timeout", type=float, default=8.0, help="Timeout seconds for local LLM post-correction (default: 8.0)") + parser.add_argument("--post-correct-min-overlap", type=float, default=0.45, help="Minimum token overlap ratio required to accept LLM rewrite (default: 0.45)") parser.add_argument("--post-correct-debug", action="store_true", help="Print pre/post LLM correction text for debugging") parser.add_argument("--llm-merge-decider", action="store_true", help="Use local LLM to validate/refine smart-correct line merges") parser.add_argument("--llm-merge-timeout", type=float, default=1.5, help="Timeout seconds for LLM merge decision (default: 1.5)") @@ -819,13 +849,13 @@ def main(): english_for_translation = apply_glossary(english_for_translation, merged_glossary_pairs) if args.post_correct and english_for_translation: - context_hint = recent_en_lines[-1]["text"] if recent_en_lines else "" + context_lines = [item["text"] for item in recent_en_lines] english_for_translation = apply_rule_based_post_correction(english_for_translation) english_for_translation = post_correct_with_local_llm( english_for_translation, args, llm_post_correct_state, - context_hint=context_hint, + context_lines=context_lines, ) english_for_caption = english_for_translation