Watching the 'Voice' of Generated Text: Catching a Silent Default-Model Swap Through Style Drift

import re
import statistics
 
POLITE_ENDINGS = ("です", "ます", "ました", "ません", "でしょう",
                   "ください", "ましょう", "でした", "います")
LEAD_CONJUNCTIONS = ("しかし", "そのため", "また", "さらに", "つまり", "ただし")
TEMPLATE_WORDS = ("この記事では", "本記事では", "いかがでしたか",
                  "徹底解説", "完全ガイド", "決定版", "について解説します")
# Noun-stop heuristic: the tail is not a verb / auxiliary / sentence-final particle
VERB_TAIL = re.compile(r"(る|た|だ|ない|です|ます|ました|でしょう|ください|う|く|す)$")
 
def split_sentences(text: str) -> list[str]:
    # Drop code blocks and headings, then split on the Japanese full stop
    text = re.sub(r"```.*?```", "", text, flags=re.S)
    text = re.sub(r"^#+\s.*$", "", text, flags=re.M)
    parts = re.split(r"(?<=。)", text)
    return [s.strip() for s in parts if len(s.strip()) >= 4]
 
def extract_fingerprint(text: str) -> dict[str, float]:
    sents = split_sentences(text)
    n = max(len(sents), 1)
    total_chars = sum(len(s) for s in sents)
 
    polite = sum(1 for s in sents if s.rstrip("。").endswith(POLITE_ENDINGS))
    taigen = sum(1 for s in sents
                 if not VERB_TAIL.search(s.rstrip("。")) and not s.rstrip("。").endswith(POLITE_ENDINGS))
    lead = sum(1 for s in sents if s.startswith(LEAD_CONJUNCTIONS))
    commas = sum(s.count("、") for s in sents)
    template_hits = sum(text.count(w) for w in TEMPLATE_WORDS)
 
    lengths = [len(s) for s in sents]
    return {
        "polite_ratio": polite / n,
        "mean_len": statistics.fmean(lengths),
        "len_stdev": statistics.pstdev(lengths) if len(lengths) > 1 else 0.0,
        "taigen_ratio": taigen / n,
        "lead_conj_ratio": lead / n,
        "comma_density": commas / n,
        "template_rate": template_hits / (total_chars / 1000 + 1e-9),  # per 1000 chars
    }

import json
 
def build_baseline(texts: list[str]) -> dict[str, dict[str, float]]:
    fps = [extract_fingerprint(t) for t in texts]
    keys = fps[0].keys()
    baseline = {}
    for k in keys:
        vals = [fp[k] for fp in fps]
        baseline[k] = {
            "mean": statistics.fmean(vals),
            "stdev": statistics.pstdev(vals) or 1e-6,  # avoid divide-by-zero
        }
    return baseline
 
# Build it from articles produced before the default model changed
baseline = build_baseline(reference_texts)
with open("style_baseline.json", "w", encoding="utf-8") as f:
    json.dump(baseline, f, ensure_ascii=False, indent=2)

def style_drift(text: str, baseline: dict, z_warn: float = 2.5,
                min_flags: int = 2) -> dict:
    fp = extract_fingerprint(text)
    z_scores, flags = {}, []
    for k, v in fp.items():
        b = baseline[k]
        z = (v - b["mean"]) / b["stdev"]
        z_scores[k] = round(z, 2)
        if abs(z) >= z_warn:
            flags.append(k)
    # Aggregate: root of the sum of squared z-scores (magnitude of multi-axis drift)
    aggregate = round((sum(z * z for z in z_scores.values())) ** 0.5, 2)
    return {
        "drifted": len(flags) >= min_flags,
        "flags": flags,
        "z_scores": z_scores,
        "aggregate": aggregate,
    }

from google import genai
 
client = genai.Client()
 
def generate_with_audit(prompt: str, model: str, baseline: dict) -> dict:
    resp = client.models.generate_content(model=model, contents=prompt)
    text = resp.text
    drift = style_drift(text, baseline)
    return {
        "text": text,
        "requested_model": model,            # the ID you asked for
        "served_model": getattr(resp, "model_version", "unknown"),  # what actually answered
        "style": drift,
    }
 
result = generate_with_audit(prompt, "gemini-flash-latest", baseline)
if result["style"]["drifted"] and result["requested_model"] != result["served_model"]:
    raise SystemExit(
        f"Style drift detected: flags={result['style']['flags']} / "
        f"requested={result['requested_model']} served={result['served_model']}"
    )

def publish_gate(result: dict) -> str:
    s = result["style"]
    if s["drifted"]:
        # Block publishing, route to review with a cause note
        return f"HOLD: style drift flags={s['flags']} aggregate={s['aggregate']}"
    if s["aggregate"] >= 4.0:
        # No single flag, but globally far — keep a monitoring log only
        return f"WATCH: aggregate={s['aggregate']}"
    return "OK"