生成文の『文体の癖』を数値で見張る — 既定モデルの差し替えを文体ドリフトで検知する設計

import re
import statistics
 
POLITE_ENDINGS = ("です", "ます", "ました", "ません", "でしょう",
                   "ください", "ましょう", "でした", "います")
LEAD_CONJUNCTIONS = ("しかし", "そのため", "また", "さらに", "つまり", "ただし")
TEMPLATE_WORDS = ("この記事では", "本記事では", "いかがでしたか",
                  "徹底解説", "完全ガイド", "決定版", "について解説します")
# 体言止め判定: 末尾が動詞・助動詞・終助詞でないことを近似する
VERB_TAIL = re.compile(r"(る|た|だ|ない|です|ます|ました|でしょう|ください|う|く|す)$")
 
def split_sentences(text: str) -> list[str]:
    # コードブロックと見出し記号を落としてから句点で分割する
    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),  # 1000字あたり
    }

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,  # ゼロ割回避
        }
    return baseline
 
# 既定モデルが変わる前の安定した時期の記事で作る
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)
    # 集約スコア: 各 z の二乗和の平方根（多次元のズレの大きさ）
    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,            # 自分が指定したID
        "served_model": getattr(resp, "model_version", "unknown"),  # 実際に応答した版
        "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"文体ドリフト検知: 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"]:
        # 公開は止め、原因メモを添えてレビュー行きにする
        return f"HOLD: 文体ドリフト flags={s['flags']} aggregate={s['aggregate']}"
    if s["aggregate"] >= 4.0:
        # 個別フラグは立たないが全体的に遠い → 監視ログだけ残す
        return f"WATCH: aggregate={s['aggregate']}"
    return "OK"