feat(keyboard): harden polish/clipboard guards and ship 1.6.6 (build 59)
Keep marketing version at 1.6.6 and bump build to 59. Strengthen never-answer polish safeguards, clipboard reply-intent continuity, voice undo UX, device UITests harness, eval fixtures, and What's New assets.
This commit is contained in:
Rocky
@@ -0,0 +1,337 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replay complete production prompts against an explicit quality corpus.
|
||||
|
||||
This is deliberately separate from XCTest:
|
||||
- XCTest protects deterministic prompt assembly and local fallbacks.
|
||||
- this script sends the *complete* system prompt and complete user payload to
|
||||
the configured LLM, then records objective contract checks plus every output
|
||||
for human quality review.
|
||||
|
||||
It covers two protocols that must never be conflated:
|
||||
1. Dictation polish: the user message is the user's outbound draft. Questions
|
||||
must stay questions and never be answered.
|
||||
2. Clipboard command: the user message contains material and an ASR command.
|
||||
A reply command must produce a reply; a summary/review/translation must not.
|
||||
|
||||
Usage:
|
||||
python3 Scripts/llm_full_prompt_quality_eval.py --profile smoke
|
||||
python3 Scripts/llm_full_prompt_quality_eval.py --profile full --samples 2
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import concurrent.futures
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SHARED = ROOT / "OSGKeyboardShared"
|
||||
COMPOSER = SHARED / "Services" / "ClipboardCommandPromptComposer.swift"
|
||||
FIXTURES = ROOT / "Scripts" / "fixtures" / "llm_quality_matrix.json"
|
||||
QUESTION_EVAL = ROOT / "Scripts" / "polish_question_guard_eval.py"
|
||||
|
||||
|
||||
def load_question_eval():
|
||||
spec = importlib.util.spec_from_file_location("question_eval", QUESTION_EVAL)
|
||||
if spec is None or spec.loader is None:
|
||||
raise SystemExit(f"Cannot load {QUESTION_EVAL}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
Q = load_question_eval()
|
||||
|
||||
|
||||
def swift_string(name: str) -> str:
|
||||
source = COMPOSER.read_text()
|
||||
match = re.search(
|
||||
rf'private static let {re.escape(name)} = """(.*?)"""',
|
||||
source,
|
||||
re.DOTALL,
|
||||
)
|
||||
if not match:
|
||||
raise SystemExit(f"Could not extract {name} from {COMPOSER}")
|
||||
return match.group(1).strip()
|
||||
|
||||
|
||||
def sanitize_bias(bias: str) -> str:
|
||||
"""Mirror ClipboardCommandPromptComposer.sanitizeBias line by line."""
|
||||
markers = (
|
||||
"草稿", "不是对方", "不回答", "不作答", "代答", "接话",
|
||||
"draft", "do not answer", "never answer", "not a message from",
|
||||
)
|
||||
kept: list[str] = []
|
||||
for line in bias.splitlines():
|
||||
text = line.strip()
|
||||
if not text:
|
||||
if kept and kept[-1]:
|
||||
kept.append("")
|
||||
continue
|
||||
if not any(marker in text.lower() for marker in markers):
|
||||
kept.append(line)
|
||||
return "\n".join(kept).strip()
|
||||
|
||||
|
||||
def contains_reply_intent(instruction: str) -> bool:
|
||||
"""Mirror current production markers; this is reported, not asserted."""
|
||||
markers = (
|
||||
"回复", "回信", "回应", "答复", "帮我回", "回他", "回她",
|
||||
"回个", "回条", "回一下", "回下", "reply", "respond",
|
||||
"write back", "answer them", "answer him", "answer her",
|
||||
)
|
||||
lower = instruction.lower()
|
||||
return any(marker in lower for marker in markers)
|
||||
|
||||
|
||||
def clipboard_system(instruction: str, style_id: str) -> str:
|
||||
parts = [swift_string("chineseCore")]
|
||||
if contains_reply_intent(instruction):
|
||||
parts.append(swift_string("chineseReplyGuard"))
|
||||
bias = sanitize_bias(Q.style_prompt(style_id))
|
||||
if bias:
|
||||
parts += ["# 语气底色（弱偏置；口述指令优先）", bias[:800]]
|
||||
return "\n\n".join(parts)
|
||||
|
||||
|
||||
def resolved_material(case: dict) -> str:
|
||||
suffix = case.get("generated_suffix")
|
||||
if suffix:
|
||||
return case["material"] + suffix * int(case["generated_suffix_repeat"])
|
||||
return case["material"]
|
||||
|
||||
|
||||
def clipboard_user(case: dict) -> str:
|
||||
# Production ClipboardMaterialFilter.truncateSnapshot uses a hard 3000-char
|
||||
# prefix. The current implementation does not append a truncation marker.
|
||||
material = resolved_material(case)[:3000]
|
||||
return f"【材料】\n{material}\n\n【指令】\n{case['instruction'].strip()}"
|
||||
|
||||
|
||||
def is_english(text: str) -> bool:
|
||||
letters = len(re.findall(r"[A-Za-z]", text))
|
||||
cjk = len(re.findall(r"[\u3400-\u9fff]", text))
|
||||
return letters >= 12 and letters > cjk * 2
|
||||
|
||||
|
||||
def objective_checks(case: dict, output: str, protocol: str) -> list[str]:
|
||||
"""Checks only facts we can evaluate deterministically.
|
||||
|
||||
Passing these checks means “no detected contract breach”, not “excellent
|
||||
writing”. The report deliberately keeps the raw output for a Cursor-agent
|
||||
and human sample review.
|
||||
"""
|
||||
failures: list[str] = []
|
||||
normalized = output.strip()
|
||||
if not normalized:
|
||||
return ["empty_output"]
|
||||
if case.get("must_remain_question") and not Q.is_question_draft(normalized):
|
||||
failures.append("question_lost_or_answered")
|
||||
for token in case.get("required_all", []):
|
||||
if token not in normalized:
|
||||
failures.append(f"missing:{token}")
|
||||
for alternatives in case.get("required_one_of", []):
|
||||
if not any(token in normalized for token in alternatives):
|
||||
failures.append("missing_one_of:" + "|".join(alternatives))
|
||||
if case.get("required_any") and not any(
|
||||
token in normalized for token in case["required_any"]
|
||||
):
|
||||
failures.append("missing_any:" + "|".join(case["required_any"]))
|
||||
for token in case.get("forbidden_any", []):
|
||||
if token in normalized:
|
||||
failures.append(f"forbidden:{token}")
|
||||
for token in case.get("forbidden_exact", []):
|
||||
if normalized.casefold() == token.casefold():
|
||||
failures.append(f"forbidden_exact:{token}")
|
||||
if pattern := case.get("must_match_regex"):
|
||||
if not re.search(pattern, normalized, re.DOTALL):
|
||||
failures.append(f"format_mismatch:{pattern}")
|
||||
if case.get("language") == "en":
|
||||
if not is_english(normalized):
|
||||
failures.append("expected_english_output")
|
||||
if case.get("target_language_only") and re.search(r"[\u3400-\u9fff]", normalized):
|
||||
failures.append("intermediate_non_english_output")
|
||||
if protocol == "clipboard" and case["operation"] == "reply":
|
||||
# Replies need new user-side language, not a restatement of material.
|
||||
# This heuristic is intentionally advisory; the raw output is reviewed.
|
||||
material = resolved_material(case)[:3000].strip()
|
||||
if normalized == material:
|
||||
failures.append("reply_equals_material")
|
||||
return failures
|
||||
|
||||
|
||||
def prompt_fingerprint(system: str, user: str) -> dict:
|
||||
return {
|
||||
"system_chars": len(system),
|
||||
"user_chars": len(user),
|
||||
"system_sha256": hashlib.sha256(system.encode()).hexdigest()[:16],
|
||||
"user_sha256": hashlib.sha256(user.encode()).hexdigest()[:16],
|
||||
}
|
||||
|
||||
|
||||
def normal_job(case: dict, style_id: str, intensity: str) -> dict:
|
||||
text = case["input"]
|
||||
system = Q.build_prompt(style_id, intensity, text)
|
||||
# This mirrors normal dictation after the local skip gate. Skip outputs are
|
||||
# recorded instead of being sent, just like PolishingService.
|
||||
if Q.should_skip_llm(text, style_id):
|
||||
output = text
|
||||
error = None
|
||||
elapsed = 0.0
|
||||
source = "local_skip"
|
||||
else:
|
||||
started = time.monotonic()
|
||||
try:
|
||||
temperature = 0.65 if intensity == "heavy" and style_id in Q.FUN_STYLES else 0.1
|
||||
output = Q.call(Q.api_key, system, text, temperature=temperature)
|
||||
error = None
|
||||
except Exception as exc: # noqa: BLE001
|
||||
output = ""
|
||||
error = str(exc)
|
||||
elapsed = round(time.monotonic() - started, 3)
|
||||
source = "llm"
|
||||
return {
|
||||
"protocol": "normal_polish",
|
||||
"case_id": case["id"],
|
||||
"style": style_id,
|
||||
"intensity": intensity,
|
||||
"input": text,
|
||||
"system_prompt": system,
|
||||
"user_payload": text,
|
||||
"output": output,
|
||||
"source": source,
|
||||
"checks": objective_checks(case, output, "normal"),
|
||||
"error": error,
|
||||
"elapsed_seconds": elapsed,
|
||||
"prompt_fingerprint": prompt_fingerprint(system, text),
|
||||
}
|
||||
|
||||
|
||||
def clipboard_job(case: dict, style_id: str) -> dict:
|
||||
system = clipboard_system(case["instruction"], style_id)
|
||||
user = clipboard_user(case)
|
||||
started = time.monotonic()
|
||||
try:
|
||||
output = Q.call(Q.api_key, system, user, temperature=0.1)
|
||||
error = None
|
||||
except Exception as exc: # noqa: BLE001
|
||||
output = ""
|
||||
error = str(exc)
|
||||
return {
|
||||
"protocol": "clipboard_command",
|
||||
"case_id": case["id"],
|
||||
"style": style_id,
|
||||
"operation": case["operation"],
|
||||
"material": resolved_material(case),
|
||||
"instruction": case["instruction"],
|
||||
"system_prompt": system,
|
||||
"user_payload": user,
|
||||
"output": output,
|
||||
"source": "llm",
|
||||
"checks": objective_checks(case, output, "clipboard"),
|
||||
"error": error,
|
||||
"elapsed_seconds": round(time.monotonic() - started, 3),
|
||||
"prompt_fingerprint": prompt_fingerprint(system, user),
|
||||
}
|
||||
|
||||
|
||||
def print_summary(results: list[dict]) -> None:
|
||||
by_protocol: defaultdict[str, Counter] = defaultdict(Counter)
|
||||
by_case: defaultdict[str, Counter] = defaultdict(Counter)
|
||||
for result in results:
|
||||
status = "pass" if not result["checks"] and not result["error"] else "fail"
|
||||
by_protocol[result["protocol"]][status] += 1
|
||||
by_case[f"{result['protocol']}:{result['case_id']}"][status] += 1
|
||||
|
||||
print("\nObjective contract summary:")
|
||||
for protocol, counts in sorted(by_protocol.items()):
|
||||
print(f" {protocol}: pass={counts['pass']} fail={counts['fail']}")
|
||||
print("\nCases with a detected breach:")
|
||||
failures = 0
|
||||
for result in results:
|
||||
if result["checks"] or result["error"]:
|
||||
failures += 1
|
||||
reason = result["checks"] or [f"request_error:{result['error']}"]
|
||||
print(
|
||||
f" [{result['protocol']} | {result['style']} | {result['case_id']}] "
|
||||
f"{', '.join(reason)}\n"
|
||||
f" output={result['output']!r}"
|
||||
)
|
||||
if not failures:
|
||||
print(" none")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--profile", choices=("smoke", "full"), default="smoke")
|
||||
parser.add_argument("--samples", type=int, default=1)
|
||||
parser.add_argument("--workers", type=int, default=6)
|
||||
parser.add_argument("--output", default=".tmp/llm-full-prompt-quality.json")
|
||||
args = parser.parse_args()
|
||||
|
||||
fixtures = json.loads(FIXTURES.read_text())
|
||||
key_match = re.search(r'deepseek = "([^"]+)"', Q.KEYFILE.read_text())
|
||||
if not key_match:
|
||||
raise SystemExit("No DeepSeek key configured for live evaluation.")
|
||||
Q.api_key = key_match.group(1)
|
||||
|
||||
if args.profile == "smoke":
|
||||
normal_styles = ("builtin.chat", "builtin.dating", "user.emoji-chat")
|
||||
clipboard_styles = normal_styles
|
||||
normal_cases = fixtures["normal_polish"][:8]
|
||||
clipboard_cases = fixtures["clipboard_commands"][:12]
|
||||
else:
|
||||
normal_styles = tuple(Q.STYLES)
|
||||
clipboard_styles = ("builtin.light", "builtin.dating", "user.emoji-chat")
|
||||
normal_cases = fixtures["normal_polish"]
|
||||
clipboard_cases = fixtures["clipboard_commands"]
|
||||
|
||||
jobs = []
|
||||
for _ in range(args.samples):
|
||||
for case in normal_cases:
|
||||
for style_id in normal_styles:
|
||||
for intensity in ("light", "heavy"):
|
||||
jobs.append(("normal", case, style_id, intensity))
|
||||
for case in clipboard_cases:
|
||||
for style_id in clipboard_styles:
|
||||
jobs.append(("clipboard", case, style_id, None))
|
||||
|
||||
print(
|
||||
f"Running {len(jobs)} requests: profile={args.profile}, samples={args.samples}; "
|
||||
"each request includes the complete production-shaped system and user prompt."
|
||||
)
|
||||
results: list[dict] = []
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
|
||||
futures = []
|
||||
for protocol, case, style_id, intensity in jobs:
|
||||
if protocol == "normal":
|
||||
futures.append(pool.submit(normal_job, case, style_id, intensity))
|
||||
else:
|
||||
futures.append(pool.submit(clipboard_job, case, style_id))
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
result = future.result()
|
||||
results.append(result)
|
||||
flag = "PASS" if not result["checks"] and not result["error"] else "FAIL"
|
||||
print(
|
||||
f"[{flag:4}] {result['protocol']:17} {result['style']:16} "
|
||||
f"{result['case_id']} -> {result['output']!r}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
results.sort(key=lambda item: (
|
||||
item["protocol"], item["case_id"], item["style"], item.get("intensity", "")
|
||||
))
|
||||
destination = ROOT / args.output
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
destination.write_text(json.dumps(results, ensure_ascii=False, indent=2))
|
||||
print_summary(results)
|
||||
print(f"\nFull prompts, payloads, outputs, and fingerprints: {destination}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user