"""Opt-in journal quality comparison. Not part of production generate. Usage from backend/: python journal_eval.py # synthetic, fake provider, contract only python journal_eval.py --live --profile-id python journal_eval.py --profile-ab # two synthetic writing profiles, prompts only unless --live Never writes private texts into the repository. Live quality stays unconfirmed until an explicit --live run succeeds. The harness never declares a winner. """ from __future__ import annotations import argparse import json import re import sys import time from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from journal_editorial import lexical_similarity VARIANT_BASELINE = "baseline_simple" VARIANT_PREVIOUS = "kansho_previous" VARIANT_CURRENT = "kansho_current" BASELINE_TEMPLATE = ( "Erstelle aus diesen Angaben einen ansprechenden persönlichen Tagebucheintrag. " "Bewahre alle Tatsachen und Unsicherheiten, erfinde nichts, korrigiere Sprache " "und schreibe im bereitgestellten persönlichen Stil.\n\n" "Persönlicher Stil:\n{{writing_profile}}\n\n" "Angaben:\n{{reconstruction}}\n" ) PREVIOUS_TEMPLATE = ( "Schreibe einen eigenständigen, gut lesbaren Tagebucheintrag in der Ich-Form von [[SELF]].\n" "Verwende ausschließlich die verifizierten Informationen aus sources[].text. " "Der Quellwortlaut ist keine Ausgabevorlage.\n" "Keine Dialogabschrift. Unsicherheiten im Wortlaut erhalten.\n" "Writing Profile:\n{{writing_profile}}\n\n" "Verified Artifact:\n{{reconstruction}}\n" ) PROFILE_A = ( "Core: kurze Sätze, trockener Schnitt, wenig Adjektive, kaum Reflexion. " "Wortwahl nüchtern, Rhythmus abgehackt." ) PROFILE_B = ( "Core: längere, ruhig fließende Sätze, beobachtend, leise Reflexion am Satzende. " "Wortwahl behutsam, Rhythmus getragen." ) FIXTURES: list[dict[str, Any]] = [ { "id": "prose_typos", "class": "already_narrative_with_errors", "source": ( "ich bin dan zum markt gegangen und da war es zimlich voll. " "vielleicht bleibe ich kürzer. ich wollte noch brot holen, hab es aber nicht gemacht. " "die rote tasche lag im auto." ), "typos": ("zimlich", "dan zum"), "must_keep": ("vielleicht", "nicht gemacht", "rote tasche"), "must_not_invent": ("traurig", "weil"), }, { "id": "notes_fragments", "class": "bullet_points_and_fragments", "source": "markt\nkirschen kaufen\nspäter hafen", "typos": (), "must_keep": ("markt", "kirschen", "hafen"), "must_not_invent": ("glücklich", "weil"), }, { "id": "plan_vs_done", "class": "plan_versus_completion", "source": "Ich wollte um elf zum Hafen. Stattdessen blieb ich zu Hause. Den Brief habe ich nicht abgeschickt.", "typos": (), "must_keep": ("wollte", "blieb", "nicht abgeschickt"), "must_not_invent": ("geschickt", "bin zum hafen"), }, { "id": "negation_uncertainty", "class": "negation_and_uncertainty", "source": "Vielleicht kommt der Techniker. Ich bin unsicher wegen des Tees. Den Kuchen habe ich nicht gebacken.", "typos": (), "must_keep": ("vielleicht", "unsicher", "nicht gebacken"), "must_not_invent": ("sicher", "gebacken"), }, { "id": "correction", "class": "correction_of_earlier_claim", "source": "Zuerst dachte ich, der Markt sei um neun. Später korrigierte ich mich: er war schon um acht zu Ende.", "typos": (), "must_keep": ("zuerst", "korrigierte", "acht"), "must_not_invent": ("neun zu ende",), }, { "id": "imprecise_time", "class": "imprecise_time", "source": "Gegen sechs bin ich aufgewacht. Irgendwann vorm Mittag war ich am Markt.", "typos": (), "must_keep": ("gegen sechs", "irgendwann"), "must_not_invent": ("genau 6:00", "12:00"), }, { "id": "outstanding_event", "class": "outstanding_event_among_everyday", "source": ( "gegen 6 uhr aufgewacht\n" "morgenroutine mit sprache, notizen und tee\n" "unsichere teepraeferenz\n" "techniker sollte um 11 uhr kommen\n" "kueche aufgeraeumt\n" "gegen 9 uhr gefruehstueckt\n" "kinder erst gegen 10 oder 11 uhr aufgestanden\n" "normaler strandtag\n" "ploetzlich seehunde im wasser gesehen" ), "typos": ("praeferenz", "aufgeraeumt", "gefruehstueckt"), "must_keep": ("normaler strandtag", "seehunde", "unsichere"), "must_not_invent": ("gluecklich", "schicksal"), "weight_tokens": ("seehunde", "normaler"), }, { "id": "recurring_people", "class": "recurring_people_and_projects", "source": ( "Am Vormittag sprach ich mit [[PERSON:01]] über Projekt [[PROJECT:01]]. " "Später half [[PERSON:01]] beim Aufräumen. [[PROJECT:01]] blieb liegen." ), "typos": (), "must_keep": ("[[PERSON:01]]", "[[PROJECT:01]]", "blieb liegen"), "must_not_invent": ("[[PERSON:02]]",), }, { "id": "incomplete_clause", "class": "incomplete_source_rebuildable", "source": "ich gieng zum laden. danach sprach ich mit dem und kam zurück. es war kald.", "typos": ("gieng", "kald"), "must_keep": ("laden", "kam zurück"), "must_not_invent": ("nachbarn", "freund"), "incomplete_ok": False, }, ] SYNTHETIC_PROSE = FIXTURES[0]["source"] SYNTHETIC_NOTES = FIXTURES[1]["source"] SYNTHETIC_TYPOS = FIXTURES[0]["typos"] def _words(text: str) -> list[str]: return re.findall(r"[A-Za-zÄÖÜäöüß]+", (text or "").lower()) def _sentences(text: str) -> list[str]: parts = [item.strip() for item in re.split(r"(?<=[.!?])\s+", text or "") if item.strip()] return parts or ([text.strip()] if (text or "").strip() else []) def score_output(source: str, output: str, *, profile: str = "", typos: tuple[str, ...] = ()) -> dict[str, Any]: """Diagnostic scores. Not a production gate and not a live-quality certificate.""" body = output or "" src_words = set(_words(source)) out_words = set(_words(body)) sentences = _sentences(body) paragraphs = [item for item in re.split(r"\n\s*\n", body) if item.strip()] avg_len = round(sum(len(item.split()) for item in sentences) / max(1, len(sentences)), 2) transitions = len(re.findall(r"(?i)\b(?:danach|später|dann|zuerst|schließlich|bis|plötzlich)\b", body)) profile_words = set(_words(profile)) style_overlap = round(len(out_words & profile_words) / max(1, len(profile_words)), 3) if profile_words else 0.0 kept = round(len(src_words & out_words) / max(1, len(src_words)), 3) extra = sorted(out_words - src_words - profile_words) lost = sorted(src_words - out_words) remaining_typos = [item for item in typos if item.lower() in body.lower()] from journal_editorial import incomplete_syntax_markers return { "spelling_typos_remaining": remaining_typos, "spelling_typos_fixed": [item for item in typos if item.lower() not in body.lower()], "readability_sentence_count": len(sentences), "readability_avg_sentence_words": avg_len, "readability_paragraphs": max(1, len(paragraphs)), "transitions": transitions, "style_token_overlap": style_overlap, "fact_token_keep": kept, "new_info_tokens": extra[:24], "lost_info_tokens": lost[:24], "lexical_similarity": lexical_similarity(source, body), "incomplete_syntax": incomplete_syntax_markers(body), "unresolved_privacy_tokens": bool(re.search(r"\[\[\s*(?:…|\.{2,})\s*\]\]", body)), } def variant_templates() -> dict[str, str]: from engine import load_active_prompt current = (load_active_prompt("mvp.journal_generate") or {}).get("template") or "" return { VARIANT_BASELINE: BASELINE_TEMPLATE, VARIANT_PREVIOUS: PREVIOUS_TEMPLATE, VARIANT_CURRENT: current, } def fixture_context(source: str, *, writing_profile: str | None = None) -> dict[str, str]: artifact = { "kind": "verified_artifact", "coverage": "all_selected_sources", "sources": [{"source_id": "u1", "role": "user", "text": source}], } from journal_editorial import narration_sources_text from journal_generation_policy import compile_selection, default_selection_ids from writing_profile_store import NEUTRAL_JOURNAL_STYLE compiled = compile_selection(default_selection_ids()) return { "reconstruction": narration_sources_text(artifact) or source, "writing_profile": writing_profile or NEUTRAL_JOURNAL_STYLE, "style_examples": "Keine historischen Stilbeispiele.", **compiled.instructions, "existing_text": "", "space_title": "Eval", } def synthetic_context(source: str) -> dict[str, str]: return fixture_context(source) def run_variant( name: str, template: str, context: dict[str, str], profile_id: str, *, live: bool, ) -> dict[str, Any]: from engine import execute_prompt from model_catalog import resolve_generate_metadata from prompt_budget import plan_journal_budget from providers import generate_provider prompt = { "id": f"eval-{name}", "slug": "mvp.journal_generate" if name == VARIANT_CURRENT else f"eval.{name}", "seed_revision": "eval", "prompt_type": "base", "required_feature": "ai_calls", "template": template, } started = time.perf_counter() config = generate_provider() window = resolve_generate_metadata(config) if config else None budget = plan_journal_budget(window, purpose="journal_generate") if window else None result = execute_prompt( prompt, profile_id, purpose="journal_generate", data_class="B", context=context, max_tokens=budget.reserved_output_tokens if budget else 1024, disable_context_compression=True, budget=budget, ) elapsed_ms = int((time.perf_counter() - started) * 1000) diag = result.get("diagnostics") or {} return { "variant": name, "live": live, "content": result.get("content") or "", "prompt_tokens": diag.get("prompt_tokens"), "completion_tokens": diag.get("completion_tokens"), "total_tokens": diag.get("total_tokens"), "cost": diag.get("cost"), "runtime_ms": elapsed_ms, "generate_ms": diag.get("generate_ms"), "model": (result.get("trace") or {}).get("model") or diag.get("actual_model"), "prompt_revision": diag.get("prompt_revision"), "fake_provider": not live, "rendered_contains_profile": (context.get("writing_profile") or "")[:40] in ((result.get("trace") or {}).get("intern") or ""), } def _offline_row(name: str, template: str, context: dict[str, str]) -> dict[str, Any]: from placeholders import resolve_template from privacy_gateway import _fake_complete rendered = resolve_template(template, context) started = time.perf_counter() content = _fake_complete("journal_generate", rendered) return { "variant": name, "live": False, "content": content, "prompt_tokens": None, "completion_tokens": None, "total_tokens": None, "cost": None, "runtime_ms": int((time.perf_counter() - started) * 1000), "model": "fake", "fake_provider": True, "rendered": rendered, "rendered_contains_profile": (context.get("writing_profile") or "")[:24] in rendered, } def _human_blind_pair(left: dict, right: dict) -> dict[str, Any]: """Side-by-side without naming a winner. Mapping stays in the machine report.""" return { "prompt_1": {"label": "Prompt 1", "text": left.get("content") or ""}, "prompt_2": {"label": "Prompt 2", "text": right.get("content") or ""}, "hidden_mapping": { "prompt_1": left.get("variant"), "prompt_2": right.get("variant"), }, "instruction": ( "Blind bewerten: Faktentreue, Unsicherheit, Sprache, Lesefluss, " "Gewichtung, Profiltreue, Eigenständigkeit. Keine automatische Siegeraussage." ), } def compare_synthetic( *, live: bool = False, profile_id: str | None = None, profile_ab: bool = False, ) -> dict[str, Any]: import placeholder_mvp # noqa: F401 templates = variant_templates() fixtures_out = [] for fixture in FIXTURES: context = fixture_context(fixture["source"]) rows = [] for name, template in templates.items(): if live: if not profile_id: raise SystemExit("--live requires --profile-id") row = run_variant(name, template, context, profile_id=profile_id, live=True) else: row = _offline_row(name, template, context) row["scores"] = score_output( fixture["source"], row["content"], profile=context.get("writing_profile") or "", typos=tuple(fixture.get("typos") or ()), ) rows.append(row) current = next((item for item in rows if item["variant"] == VARIANT_CURRENT), rows[-1]) baseline = next((item for item in rows if item["variant"] == VARIANT_BASELINE), rows[0]) fixtures_out.append( { "id": fixture["id"], "class": fixture["class"], "source": fixture["source"], "variants": rows, "human_blind": _human_blind_pair(current, baseline), } ) profile_ab_report = None if profile_ab: source = next(item["source"] for item in FIXTURES if item["id"] == "outstanding_event") ctx_a = fixture_context(source, writing_profile=PROFILE_A) ctx_b = fixture_context(source, writing_profile=PROFILE_B) current_template = templates[VARIANT_CURRENT] if live: if not profile_id: raise SystemExit("--live requires --profile-id") row_a = run_variant("profile_a", current_template, ctx_a, profile_id=profile_id, live=True) row_b = run_variant("profile_b", current_template, ctx_b, profile_id=profile_id, live=True) else: row_a = _offline_row("profile_a", current_template, ctx_a) row_b = _offline_row("profile_b", current_template, ctx_b) rendered_a = row_a.get("rendered") or "" rendered_b = row_b.get("rendered") or "" profile_ab_report = { "same_facts": ctx_a["reconstruction"] == ctx_b["reconstruction"], "prompts_differ": (PROFILE_A in rendered_a and PROFILE_B in rendered_b and PROFILE_A not in rendered_b) if not live else row_a.get("content") != row_b.get("content"), "profile_a_in_prompt": PROFILE_A[:20] in rendered_a if not live else None, "profile_b_in_prompt": PROFILE_B[:20] in rendered_b if not live else None, "human_blind": _human_blind_pair(row_a, row_b), "note": ( "Offline: beweist verschiedene Stilvorgaben im gerenderten Prompt, nicht Live-Prosa." if not live else "Live-A/B über das Privacy Gateway. Ob Rhythmus und Wortwahl divergieren, bewertet ein Mensch." ), } return { "live": live, "live_quality_confirmed": False, "winner_declared": False, "note": ( "Fake-Provider-Lauf: beweist den Vergleichsvertrag, nicht echte Modellprosa." if not live else "Live-Lauf über das Privacy Gateway. Qualitative Bewertung bleibt manuell. Keine automatische Siegeraussage." ), "source_kind": "synthetic", "fixtures": fixtures_out, "profile_ab": profile_ab_report, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Journal narration quality comparison (opt-in).") parser.add_argument("--live", action="store_true", help="Call the configured generate provider. Costs money.") parser.add_argument("--profile-id", help="Required with --live. Uses the Privacy Gateway.") parser.add_argument("--profile-ab", action="store_true", help="Compare two synthetic writing profiles.") parser.add_argument("--json", action="store_true", help="Print JSON instead of text.") args = parser.parse_args(argv) if args.live: import os if (os.environ.get("KANSHO_FAKE_PROVIDER") or "").strip() in {"1", "true", "yes"}: print("Refusing --live while KANSHO_FAKE_PROVIDER is set.", file=sys.stderr) return 2 if not args.profile_id: print("--live requires --profile-id", file=sys.stderr) return 2 else: import os os.environ.setdefault("KANSHO_FAKE_PROVIDER", "1") os.environ.setdefault("KANSHO_FAKE_DETECT", "1") from db import init_db init_db() report = compare_synthetic(live=bool(args.live), profile_id=args.profile_id, profile_ab=bool(args.profile_ab)) if args.json: print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 print(report["note"]) if not report["live"]: print("Live-Qualität: noch nicht bestätigt. Keine Siegeraussage.") for fixture in report["fixtures"]: print(f"\n[{fixture['id']} / {fixture['class']}]") for item in fixture["variants"]: scores = item["scores"] print( f" {item['variant']}: similarity={scores['lexical_similarity']} " f"keep={scores['fact_token_keep']} typos_left={scores['spelling_typos_remaining']} " f"incomplete={scores['incomplete_syntax']} tokens={item.get('total_tokens')} " f"cost={item.get('cost')} ms={item['runtime_ms']}" ) blind = fixture["human_blind"] print(" Blindvergleich: Prompt 1 vs Prompt 2 (Mapping nur im JSON).") print(f" {blind['instruction']}") if report.get("profile_ab"): ab = report["profile_ab"] print("\n[profile_ab]") print(f" same_facts={ab['same_facts']} prompts_differ={ab['prompts_differ']}") print(f" {ab['note']}") return 0 if __name__ == "__main__": raise SystemExit(main())