""" Labeler B: LLM-judge labels. Per DESIGN.md ยง4: Labeler B is Claude via Anthropic API, fresh context, provided with the 20 FVS entry identification sections plus the document, asked for per-frame binary judgments. Execution order: runs AFTER curator labels (A) are frozen, BEFORE detector run (C). Writes labels/labels_llm.json. """ from __future__ import annotations import json import os import re import subprocess import sys import urllib.error import urllib.request from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[2] STUDY_DIR = REPO_ROOT / "fvs_eval" / "validation_study" CORPUS_DIR = STUDY_DIR / "corpus" LABELS_DIR = STUDY_DIR / "labels" FRAME_LIB_DIR = REPO_ROOT / "data" / "frame_library" FRAME_IDS = [f"FVS-{i:03d}" for i in range(1, 21)] def _extract_identification_section(md_path: Path) -> str: """Extract the Identification section + short-definition one-liner from an FVS markdown entry. The entry structure (see FVS-001 as reference): # Frame Name **FVS entry:** FVS-NNN ... metadata ... ## Identification (paragraph of definition, followed by what-it-makes-visible / invisible, examples, adjacents, honest-limits) We take the entry name, FVS id, and the Identification section up to the next ## heading. Long entries are truncated at ~800 chars to keep the labeling prompt bounded. """ with open(md_path, "r", encoding="utf-8") as f: text = f.read() # Name from first H1 name_match = re.search(r"^# (.+)$", text, flags=re.MULTILINE) name = name_match.group(1).strip() if name_match else md_path.stem # FVS id from metadata id_match = re.search(r"\*\*FVS entry:\*\*\s*(FVS-\d{3})", text) fvs_id = id_match.group(1) if id_match else md_path.stem.split("_")[0] # Identification section body ident_match = re.search( r"## Identification\s*\n(.*?)(?=\n## |\Z)", text, flags=re.DOTALL, ) body = ident_match.group(1).strip() if ident_match else "" # Keep only the first paragraph of Identification (the definition) # plus any "What this frame makes visible/invisible" bullet list. # Truncate to ~800 chars to keep the prompt bounded. if len(body) > 800: cut = body[:800] last_period = cut.rfind(".") if last_period > 500: body = cut[:last_period + 1] else: body = cut return f"**{fvs_id} {name}.** {body}" def build_frame_library_reference() -> str: """Build the FVS library reference block for the labeler prompt.""" lines = [] for fvs_id in FRAME_IDS: # Find the markdown file for this id candidates = list(FRAME_LIB_DIR.glob(f"{fvs_id}_*.md")) if not candidates: lines.append(f"**{fvs_id}.** (entry not found)") continue lines.append(_extract_identification_section(candidates[0])) return "\n\n".join(lines) LABELING_PROMPT_TEMPLATE = """You are a careful analytical reader tasked with labeling a document against the Frame Vocabulary Standard (FVS), a taxonomy of 20 named framing failures in AI-generated and analytical text. Your task: for each of the 20 FVS frames, make a binary judgment about whether the document EXHIBITS that frame as a real, structurally visible pattern in its text. A frame "exhibits" when its definition describes a genuine organizing or evident feature of the document, not merely when a single keyword matches. Multi-label judgments are expected; several frames may apply. Meta-side frames (FVS-003, FVS-004, FVS-005, FVS-006, FVS-013, FVS-020) describe mechanisms about AI interactions or reader posture rather than document text properties; for these, judge "exhibits: true" only if the text itself shows the frame as a feature (e.g., FVS-013 Oracle Frame applies only if the document explicitly positions itself as authoritative in a way that reinforces reader-treats-as-oracle dynamics). ## FVS Library Reference {frame_library} ## Document to label Title: {doc_title} Source: {doc_source} {doc_text} ## Output format Respond with a single JSON object and nothing else. Keys are FVS IDs; values are objects with "exhibits" (boolean) and "reasoning" (one short sentence). Example: {{ "FVS-001": {{"exhibits": false, "reasoning": "Document covers multiple perspectives with balanced density."}}, "FVS-002": {{"exhibits": true, "reasoning": "Highly fluent prose with zero epistemic sourcing."}}, ... }} Include all 20 FVS IDs. Be concrete in reasoning; cite specific features of the document when possible. """ def _call_anthropic_json(prompt: str, max_tokens: int = 4000) -> dict: """Call Anthropic API and parse JSON from the response.""" api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: raise RuntimeError("ANTHROPIC_API_KEY not set") # Use Opus 4.7 if available, else Sonnet 4.6, else Haiku. candidates = ["claude-opus-4-7", "claude-sonnet-4-6", "claude-haiku-4-5-20251001"] last_err = None for cand in candidates: payload = { "model": cand, "max_tokens": max_tokens, "temperature": 0.0, # labeling task: deterministic "messages": [{"role": "user", "content": prompt}], } req = urllib.request.Request( "https://api.anthropic.com/v1/messages", method="POST", data=json.dumps(payload).encode("utf-8"), headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=120) as resp: data = json.loads(resp.read()) text = "".join( b.get("text", "") for b in data.get("content", []) if b.get("type") == "text" ) # Find the JSON object in the response m = re.search(r"\{.*\}", text, flags=re.DOTALL) if not m: raise RuntimeError(f"no JSON found in response: {text[:200]}") parsed = json.loads(m.group(0)) return { "labels": parsed, "model": data.get("model"), "requested_model": cand, "stop_reason": data.get("stop_reason"), "input_tokens": data.get("usage", {}).get("input_tokens"), "output_tokens": data.get("usage", {}).get("output_tokens"), "raw_text": text, } except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") if e.fp else str(e) last_err = f"{e.code} {e.reason}: {body[:300]}" continue except Exception as e: last_err = str(e) continue raise RuntimeError(f"All candidates failed; last: {last_err}") def main(): # Load corpus manifest with open(CORPUS_DIR / "manifest.json", "r", encoding="utf-8") as f: manifest = json.load(f) frame_library_ref = build_frame_library_reference() print(f"Frame library reference: {len(frame_library_ref)} chars") llm_labels = { "version": 1, "labeler": "llm_judge", "labeler_note": ( "Claude via Anthropic API. Temperature 0.0. Fresh context per " "document (single-turn per doc). Prompt included FVS library " "reference (identification sections for all 20 frames). " "Produced AFTER curator labels frozen, BEFORE detector run." ), "frame_universe": FRAME_IDS, "labels": {}, "call_meta": {}, } for doc in manifest["documents"]: doc_id = doc["doc_id"] doc_path = CORPUS_DIR / f"{doc_id}.txt" if not doc_path.exists(): print(f" {doc_id}: SKIP (file missing)") continue doc_text = doc_path.read_text(encoding="utf-8") prompt = LABELING_PROMPT_TEMPLATE.format( frame_library=frame_library_ref, doc_title=doc.get("title", doc_id), doc_source=doc.get("source", "unknown"), doc_text=doc_text, ) print(f" labeling {doc_id} (prompt {len(prompt)} chars)...") try: result = _call_anthropic_json(prompt) except Exception as e: print(f" FAILED: {e}") continue parsed = result["labels"] # Normalize: make sure all 20 FVS IDs are present normalized = {} for fid in FRAME_IDS: if fid in parsed and isinstance(parsed[fid], dict): normalized[fid] = { "exhibits": bool(parsed[fid].get("exhibits", False)), "reasoning": str(parsed[fid].get("reasoning", ""))[:500], } else: normalized[fid] = { "exhibits": False, "reasoning": "(not returned by labeler)", } llm_labels["labels"][doc_id] = normalized llm_labels["call_meta"][doc_id] = { "model_requested": result["requested_model"], "model_served": result["model"], "input_tokens": result["input_tokens"], "output_tokens": result["output_tokens"], "stop_reason": result["stop_reason"], } n_exhibits = sum(1 for v in normalized.values() if v["exhibits"]) print(f" {n_exhibits}/20 frames marked exhibits") out_path = LABELS_DIR / "labels_llm_v3.json" with open(out_path, "w", encoding="utf-8") as f: json.dump(llm_labels, f, indent=2) print(f"\nWritten: {out_path}") print(f"Total documents labeled: {len(llm_labels['labels'])}") if __name__ == "__main__": main()