rodunia/llm-research-app
0
1"""Generate human-readable claim extraction examples from real runs.2 3Creates Markdown documentation showing:4- Original generated text excerpts with block structure5- Extracted claims with offsets, triggers, claim_kind, block_kind6- Verification that all offsets are exact substrings7 8Usage:9 python scripts/make_claim_examples.py --n 3 --out docs/claim_extraction_examples.md10 11 # With custom materials12 python scripts/make_claim_examples.py --prefer-materials digital_ad,faq,blog_post_promo13 14 # From specific claims directory15 python scripts/make_claim_examples.py --claims-dir analysis/claims --outputs-dir outputs16"""17 18import json19import random20import argparse21from pathlib import Path22from typing import List, Dict, Any, Optional, Tuple23import csv24 25 26def classify_text_kind(text: str) -> str:27 """Classify text as 'prompt', 'output', or 'unknown'.28 29 Uses deterministic heuristics to detect prompt/template language.30 31 Args:32 text: Text to classify33 34 Returns:35 One of: 'prompt', 'output', 'unknown'36 """37 # Check first 500 chars for prompt markers38 header = text[:500].lower()39 40 # Prompt markers (high confidence)41 prompt_markers = [42 "you are an elite content marketing strategist",43 "you are writing",44 "your mission:",45 "compliance framework",46 "authorized claims",47 "prohibited claims",48 "mandatory disclaimers",49 "output format",50 "before writing, mentally confirm",51 "hard rules:",52 "use only the information provided below",53 "do not invent or infer"54 ]55 56 for marker in prompt_markers:57 if marker in header:58 return "prompt"59 60 # Output markers (typical generated content)61 output_markers = [62 "experience the",63 "discover",64 "introducing the",65 "## ", # Markdown headings in blog posts66 "headline:",67 "primary text:",68 "description:",69 "**q", # FAQ questions70 ]71 72 has_output_markers = any(marker in header for marker in output_markers)73 74 # If it has output markers and no prompt markers, it's likely output75 if has_output_markers:76 return "output"77 78 return "unknown"79 80 81def find_output_text(run_id: str, outputs_dir: Path) -> Optional[Path]:82 """Find output text file for a run_id using discovery strategy.83 84 Prioritizes actual generated outputs over prompts/templates.85 86 Args:87 run_id: Run identifier88 outputs_dir: Root outputs directory89 90 Returns:91 Path to output file or None if not found92 """93 # Strategy A: Look for per_run.json artifacts (most reliable)94 per_run_json = Path("analysis/per_run.json")95 if per_run_json.exists():96 try:97 with open(per_run_json, 'r') as f:98 per_run_data = json.load(f)99 for record in per_run_data:100 if record.get('run_id') == run_id:101 artifacts = record.get('artifacts', {})102 output_path = artifacts.get('output_path')103 if output_path and Path(output_path).exists():104 text = Path(output_path).read_text(encoding='utf-8')105 if classify_text_kind(text) == "output":106 return Path(output_path)107 except (json.JSONDecodeError, KeyError):108 pass109 110 # Strategy B: Direct lookup with _output.txt suffix (common pattern)111 output_patterns = [112 outputs_dir / f"{run_id}_output.txt",113 outputs_dir / f"{run_id}.txt",114 ]115 116 for path in output_patterns:117 if path.exists():118 text = path.read_text(encoding='utf-8')119 kind = classify_text_kind(text)120 if kind == "output":121 return path122 elif kind == "prompt":123 continue # Skip prompts, keep searching124 125 # Strategy C: Search known output directories126 output_dirs = [127 outputs_dir,128 outputs_dir / "comprehensive_test" / "test_b_materials",129 Path("results/outputs"),130 Path("results/text"),131 ]132 133 for out_dir in output_dirs:134 if not out_dir.exists():135 continue136 137 # Look for files matching run_id pattern138 for pattern in [f"{run_id}_output.txt", f"{run_id}.txt", f"*{run_id[:12]}*.txt"]:139 matches = list(out_dir.glob(pattern))140 for match in matches:141 # Skip prompt files142 if "prompt" in match.name.lower():143 continue144 145 text = match.read_text(encoding='utf-8')146 kind = classify_text_kind(text)147 if kind == "output":148 return match149 150 # Strategy D: Bounded recursive search (cap at 200 files)151 all_txt_files = list(outputs_dir.glob("**/*.txt"))[:200]152 153 for file_path in all_txt_files:154 # Skip obvious prompt files155 if "prompt" in file_path.name.lower():156 continue157 158 # Check if filename contains run_id159 if run_id[:12] in file_path.name or run_id in file_path.name:160 text = file_path.read_text(encoding='utf-8')161 kind = classify_text_kind(text)162 if kind == "output":163 return file_path164 165 return None166 167 168def load_or_generate_claims(169 run_id: str,170 output_path: Optional[Path],171 claims_dir: Path172) -> List[Dict[str, Any]]:173 """Load existing claims or generate from output text.174 175 Args:176 run_id: Run identifier177 output_path: Path to output text (if found)178 claims_dir: Directory containing claim JSONs179 180 Returns:181 List of claim records182 """183 # Try loading existing claims184 claims_file = claims_dir / f"{run_id}.json"185 if claims_file.exists():186 with open(claims_file, 'r', encoding='utf-8') as f:187 claims = json.load(f)188 # Filter to ensure we have v2.0 claims with block_kind189 if claims and all('block_kind' in c for c in claims):190 return claims191 192 # Generate claims if output text exists193 if output_path and output_path.exists():194 # Import claim extractor195 import sys196 sys.path.insert(0, '.')197 from analysis.claim_extractor import extract_claim_candidates198 199 full_text = output_path.read_text(encoding='utf-8')200 201 # Infer material type from path or use unknown202 material_type = "unknown"203 if "digital_ad" in str(output_path) or "ad" in str(output_path):204 material_type = "digital_ad.j2"205 elif "faq" in str(output_path):206 material_type = "faq.j2"207 elif "blog" in str(output_path):208 material_type = "blog_post_promo.j2"209 210 run_metadata = {211 "run_id": run_id,212 "product_id": "unknown",213 "material_type": material_type,214 "engine": "unknown",215 "temperature": 0.6,216 "time_of_day": "unknown",217 "repetition_id": 1218 }219 220 claims = extract_claim_candidates(full_text, run_metadata, include_meta=False)221 return claims222 223 return []224 225 226def select_example_runs(227 outputs_dir: Path,228 claims_dir: Path,229 n: int,230 prefer_materials: List[str],231 seed: int,232 require_block_kinds: bool233) -> List[Tuple[str, Path, List[Dict[str, Any]], str]]:234 """Select representative runs for examples.235 236 Args:237 outputs_dir: Outputs directory238 claims_dir: Claims directory239 n: Number of examples240 prefer_materials: Preferred material types241 seed: Random seed242 require_block_kinds: If True, only select v2.0 claims243 244 Returns:245 List of (run_id, output_path, claims, material_type) tuples246 """247 random.seed(seed)248 249 # Scan outputs directory for _output.txt files (skip prompts)250 output_files = [251 f for f in outputs_dir.glob("*.txt")252 if "prompt" not in f.name.lower()253 ]254 255 # Also check for comprehensive test outputs (skip prompts)256 comprehensive_outputs = [257 f for f in outputs_dir.glob("comprehensive_test/**/*.txt")258 if "prompt" not in f.name.lower()259 ]260 261 all_outputs = output_files + comprehensive_outputs262 random.shuffle(all_outputs)263 264 selected = []265 materials_found = set()266 267 for output_path in all_outputs:268 if len(selected) >= n:269 break270 271 # Verify this is an output, not a prompt272 text = output_path.read_text(encoding='utf-8')273 text_kind = classify_text_kind(text)274 if text_kind == "prompt":275 continue # Skip prompts276 277 # Infer material type from path278 path_str = str(output_path).lower()279 material_type = None280 281 if any(m in path_str for m in ["digital_ad", "ad_output"]):282 material_type = "digital_ad"283 elif "faq" in path_str:284 material_type = "faq"285 elif "blog" in path_str:286 material_type = "blog_post"287 288 # Skip if we already have this material type289 if material_type and material_type in materials_found:290 continue291 292 # Try to extract run_id from filename293 filename = output_path.stem294 run_id = filename.replace("_output", "").replace("_prompt", "")295 296 # Load or generate claims297 claims = load_or_generate_claims(run_id, output_path, claims_dir)298 299 # Filter by requirements300 if require_block_kinds and claims:301 if not all('block_kind' in c and 'claim_kind' in c for c in claims):302 continue303 304 # Skip if no claims305 if not claims:306 continue307 308 # Add to selected309 if material_type:310 selected.append((run_id, output_path, claims, material_type))311 materials_found.add(material_type)312 313 return selected314 315 316def extract_claim_aware_excerpt(317 full_text: str,318 claims: List[Dict[str, Any]],319 max_chars: int320) -> str:321 """Extract excerpt using claim offsets to show relevant context.322 323 Args:324 full_text: Full text325 claims: List of claim records with char_span326 max_chars: Maximum characters327 328 Returns:329 Excerpt showing claims in context330 """331 if len(full_text) <= max_chars:332 return full_text333 334 # Find claim span range335 if claims:336 char_spans = [c.get('char_span', (0, 0)) for c in claims if c.get('char_span')]337 if char_spans:338 min_start = min(s[0] for s in char_spans)339 max_end = max(s[1] for s in char_spans)340 341 # Expand window by ±200 chars342 window_start = max(0, min_start - 200)343 window_end = min(len(full_text), max_end + 200)344 345 # If window is still too large, truncate346 if window_end - window_start <= max_chars:347 return full_text[window_start:window_end]348 349 # Fallback: show beginning and end350 half = max_chars // 2 - 50351 return full_text[:half] + "\n\n[... middle section omitted ...]\n\n" + full_text[-half:]352 353 354def truncate_text(text: str, max_chars: int) -> str:355 """Truncate text to max_chars with ellipsis if needed.356 357 Args:358 text: Input text359 max_chars: Maximum characters360 361 Returns:362 Truncated text363 """364 if len(text) <= max_chars:365 return text366 367 # Show beginning and end368 half = max_chars // 2 - 50369 return text[:half] + "\n\n[... middle section omitted ...]\n\n" + text[-half:]370 371 372def format_claim_table_row(claim: Dict[str, Any]) -> str:373 """Format a claim as a Markdown table row.374 375 Args:376 claim: Claim record377 378 Returns:379 Markdown table row380 """381 claim_id = claim.get('claim_id', 'N/A')382 claim_kind = claim.get('claim_kind', 'N/A')383 block_kind = claim.get('block_kind', 'N/A')384 triggers = ', '.join(claim.get('trigger_types', []))385 char_span = claim.get('char_span', (0, 0))386 sentence = claim.get('sentence', '').replace('|', '\\|').replace('\n', ' ')[:80]387 388 return f"| `{claim_id[:20]}...` | {claim_kind} | {block_kind} | {triggers} | {char_span} | {sentence}... |"389 390 391def verify_offsets(392 claims: List[Dict[str, Any]],393 full_text: str394) -> Tuple[int, List[str]]:395 """Verify that all claim char_spans are exact substrings.396 397 Args:398 claims: List of claim records399 full_text: Original full text400 401 Returns:402 (num_verified, warnings) tuple403 """404 verified = 0405 warnings = []406 407 for claim in claims:408 sentence = claim.get('sentence', '')409 char_span = claim.get('char_span')410 if not char_span:411 warnings.append(f"Claim {claim.get('claim_id')} missing char_span")412 continue413 414 start, end = char_span415 if start >= len(full_text) or end > len(full_text):416 warnings.append(f"Claim {claim.get('claim_id')} has out-of-bounds char_span: {char_span}")417 continue418 419 extracted = full_text[start:end]420 if extracted == sentence:421 verified += 1422 else:423 warnings.append(424 f"Claim {claim.get('claim_id')} char_span mismatch:\n"425 f" Expected: {sentence[:50]}...\n"426 f" Got: {extracted[:50]}..."427 )428 429 return verified, warnings430 431 432def generate_markdown_example(433 run_id: str,434 output_path: Optional[Path],435 claims: List[Dict[str, Any]],436 material_type: str,437 max_excerpt_chars: int,438 max_claims: int,439 example_num: int440) -> str:441 """Generate Markdown section for one example.442 443 Args:444 run_id: Run identifier445 output_path: Path to output text (or None if not found)446 claims: List of claim records447 material_type: Material type name448 max_excerpt_chars: Max chars for text excerpt449 max_claims: Max claims to show450 example_num: Example number (1, 2, 3, ...)451 452 Returns:453 Markdown string454 """455 md = []456 md.append(f"## Example {example_num} — {material_type.replace('_', ' ').title()}")457 458 # Extract metadata from first claim459 if claims:460 product = claims[0].get('product', 'unknown')461 engine = claims[0].get('engine', 'unknown')462 extractor_version = claims[0].get('extractor_version', 'unknown')463 md.append(f"**Run ID:** `{run_id}` ")464 md.append(f"**Product:** {product} | **Engine:** {engine} ")465 md.append(f"**Extractor:** {extractor_version}")466 else:467 md.append(f"**Run ID:** `{run_id}`")468 469 md.append("")470 471 # Text excerpt472 if output_path and output_path.exists():473 full_text = output_path.read_text(encoding='utf-8')474 475 # Verify this is actually output, not a prompt476 text_kind = classify_text_kind(full_text)477 if text_kind == "prompt":478 md.append("### Generated Text Excerpt")479 md.append("")480 md.append("⚠️ **WARNING:** Located file appears to be a prompt/template, not generated output.")481 md.append("Showing extracted claims only (offsets may not match).")482 md.append("")483 else:484 # Use claim-aware excerpt to show relevant context485 excerpt = extract_claim_aware_excerpt(full_text, claims, max_excerpt_chars)486 487 md.append("### Generated Text Excerpt (Verbatim Model Output)")488 md.append("")489 md.append("```")490 md.append(excerpt)491 md.append("```")492 md.append("")493 494 # Verify offsets495 verified, warnings = verify_offsets(claims, full_text)496 if warnings:497 md.append("**Offset Verification:**")498 md.append(f"- {verified}/{len(claims)} claims verified")499 if warnings:500 md.append(f"- {len(warnings)} warnings (see debug output)")501 md.append("")502 else:503 md.append("### Generated Text Excerpt")504 md.append("")505 md.append("⚠️ **Generated output text file not found for this run_id.**")506 md.append("")507 md.append("_Showing extracted claims only (from analysis/claims/*.json)._")508 md.append("")509 510 # Claims table511 md.append("### Extracted Claims (Verbatim)")512 md.append("")513 514 if claims:515 # Show up to max_claims516 display_claims = claims[:max_claims]517 518 md.append("| Claim ID | Claim Kind | Block Kind | Triggers | Char Span | Sentence |")519 md.append("|----------|------------|------------|----------|-----------|----------|")520 521 for claim in display_claims:522 md.append(format_claim_table_row(claim))523 524 md.append("")525 526 # Summary stats527 product_claims = sum(1 for c in claims if c.get('claim_kind') == 'product_claim')528 disclaimer_claims = sum(1 for c in claims if c.get('claim_kind') == 'disclaimer')529 meta_claims = sum(1 for c in claims if c.get('claim_kind') == 'meta')530 531 md.append("**Summary:**")532 md.append(f"- Total extracted claims: {len(claims)}")533 md.append(f"- Product claims: {product_claims}")534 md.append(f"- Disclaimer claims: {disclaimer_claims}")535 md.append(f"- Meta claims: {meta_claims}")536 if len(claims) > max_claims:537 md.append(f"- _(Showing {max_claims} of {len(claims)} claims)_")538 md.append("- **Note:** All sentences are exact substrings (offset-traceable)")539 else:540 md.append("_No claims extracted_")541 542 md.append("")543 md.append("---")544 md.append("")545 546 return '\n'.join(md)547 548 549def write_json_preview(550 run_id: str,551 claims: List[Dict[str, Any]],552 out_dir: Path,553 max_claims: int = 2554) -> Optional[Path]:555 """Write JSON preview of claims for technical appendix.556 557 Args:558 run_id: Run identifier559 claims: List of claim records560 out_dir: Output directory (docs/examples/)561 max_claims: Max claims to include562 563 Returns:564 Path to JSON file or None565 """566 if not claims:567 return None568 569 out_dir.mkdir(parents=True, exist_ok=True)570 json_file = out_dir / f"run_{run_id[:12]}_claims_preview.json"571 572 preview_claims = claims[:max_claims]573 with open(json_file, 'w', encoding='utf-8') as f:574 json.dump(preview_claims, f, indent=2, ensure_ascii=False)575 576 return json_file577 578 579def main():580 """Main entry point."""581 parser = argparse.ArgumentParser(582 description="Generate claim extraction examples from real runs"583 )584 parser.add_argument(585 '--claims-dir',586 default='analysis/claims',587 help='Directory containing claim JSONs'588 )589 parser.add_argument(590 '--outputs-dir',591 default='outputs',592 help='Directory containing output text files'593 )594 parser.add_argument(595 '--n',596 type=int,597 default=3,598 help='Number of examples to generate'599 )600 parser.add_argument(601 '--prefer-materials',602 default='digital_ad,faq,blog_post',603 help='Comma-separated material types to prefer'604 )605 parser.add_argument(606 '--out',607 default='docs/claim_extraction_examples.md',608 help='Output Markdown file'609 )610 parser.add_argument(611 '--seed',612 type=int,613 default=42,614 help='Random seed for deterministic selection'615 )616 parser.add_argument(617 '--max-excerpt-chars',618 type=int,619 default=900,620 help='Max characters for text excerpt'621 )622 parser.add_argument(623 '--max-claims',624 type=int,625 default=6,626 help='Max claims to show per example'627 )628 parser.add_argument(629 '--require-block-kinds',630 action='store_true',631 default=True,632 help='Require v2.0 claims with block_kind/claim_kind'633 )634 parser.add_argument(635 '--write-json-previews',636 action='store_true',637 help='Write JSON previews to docs/examples/'638 )639 640 args = parser.parse_args()641 642 # Parse preferred materials643 prefer_materials = [m.strip() for m in args.prefer_materials.split(',')]644 645 # Setup paths646 claims_dir = Path(args.claims_dir)647 outputs_dir = Path(args.outputs_dir)648 out_file = Path(args.out)649 650 print("Claim Extraction Example Generator")651 print("=" * 60)652 print(f"Claims directory: {claims_dir}")653 print(f"Outputs directory: {outputs_dir}")654 print(f"Preferred materials: {prefer_materials}")655 print(f"Output file: {out_file}")656 print()657 658 # Create claims dir if it doesn't exist (for on-the-fly generation)659 claims_dir.mkdir(parents=True, exist_ok=True)660 661 # Select examples662 print("Selecting example runs...")663 selected = select_example_runs(664 outputs_dir=outputs_dir,665 claims_dir=claims_dir,666 n=args.n,667 prefer_materials=prefer_materials,668 seed=args.seed,669 require_block_kinds=args.require_block_kinds670 )671 672 if not selected:673 print("ERROR: No suitable examples found!")674 print(" - Check that outputs_dir contains .txt files")675 print(" - Or generate claims first: python -m analysis.evaluate")676 return 1677 678 print(f"✓ Selected {len(selected)} examples")679 print()680 681 # Self-checks682 print("Self-checks:")683 for run_id, output_path, claims, material_type in selected:684 print(f" - Run {run_id[:12]}: {material_type}")685 print(f" Output text: {'✓ found' if output_path and output_path.exists() else '✗ not found'}")686 if claims:687 extractor_version = claims[0].get('extractor_version', 'unknown')688 print(f" Extractor version: {extractor_version}")689 print(f" Claims count: {len(claims)}")690 691 # Verify offsets if text available692 if output_path and output_path.exists():693 full_text = output_path.read_text(encoding='utf-8')694 verified, warnings = verify_offsets(claims, full_text)695 if warnings:696 print(f" ⚠ Offset warnings: {len(warnings)}")697 for warning in warnings[:2]: # Show first 2698 print(f" {warning.split(chr(10))[0]}")699 else:700 print(f" ✓ All {verified} offsets verified")701 print()702 703 # Generate Markdown704 print("Generating Markdown examples...")705 out_file.parent.mkdir(parents=True, exist_ok=True)706 707 with open(out_file, 'w', encoding='utf-8') as f:708 # Header709 f.write("# Claim Extraction Examples\n\n")710 f.write("Real examples from the LLM research pipeline, showing structure-aware claim extraction (v2.0).\n\n")711 f.write("**All excerpts below are verbatim segments of model-generated outputs (not prompts).** \n")712 f.write("Claims are exact substrings; offsets are shown for traceability.\n\n")713 f.write("**Features demonstrated:**\n")714 f.write("- Block-aware parsing (headlines, Q/A, disclaimers)\n")715 f.write("- Claim kind tagging (product_claim vs disclaimer)\n")716 f.write("- Anchor-based trigger detection (numeric, guarantee, medical, financial, comparative)\n")717 f.write("- Exact char_span offsets (all sentences are verifiable substrings)\n\n")718 f.write("---\n\n")719 720 # Examples721 for i, (run_id, output_path, claims, material_type) in enumerate(selected, 1):722 example_md = generate_markdown_example(723 run_id=run_id,724 output_path=output_path,725 claims=claims,726 material_type=material_type,727 max_excerpt_chars=args.max_excerpt_chars,728 max_claims=args.max_claims,729 example_num=i730 )731 f.write(example_md)732 733 # Optional JSON preview734 if args.write_json_previews and claims:735 json_path = write_json_preview(736 run_id=run_id,737 claims=claims,738 out_dir=out_file.parent / "examples",739 max_claims=2740 )741 if json_path:742 print(f" ✓ JSON preview: {json_path}")743 744 print(f"✓ Generated examples: {out_file}")745 print()746 747 # Final sanity report748 print("=" * 60)749 print("SANITY REPORT")750 print("=" * 60)751 print()752 753 all_passed = True754 for i, (run_id, output_path, claims, material_type) in enumerate(selected, 1):755 print(f"Example {i} — {material_type}")756 print(f" Run ID: {run_id[:20]}...")757 print(f" Output path: {output_path if output_path else 'NOT FOUND'}")758 759 if output_path and output_path.exists():760 full_text = output_path.read_text(encoding='utf-8')761 text_kind = classify_text_kind(full_text)762 print(f" Text kind: {text_kind} {'✓ (expected: output)' if text_kind == 'output' else '⚠ WARNING'}")763 764 if text_kind == "prompt":765 print(f" ❌ FAILED: Found prompt instead of output!")766 all_passed = False767 768 # Check extractor version769 if claims:770 extractor_version = claims[0].get('extractor_version', 'unknown')771 print(f" Extractor version: {extractor_version} {'✓' if extractor_version == 'v2.0' else '⚠'}")772 773 # Verify offsets774 verified, warnings = verify_offsets(claims, full_text)775 match_rate = verified / len(claims) if claims else 0776 print(f" Offset match rate: {verified}/{len(claims)} ({match_rate:.1%})")777 778 if match_rate < 0.98:779 print(f" ⚠ WARNING: Match rate below 98%!")780 if warnings:781 print(f" First mismatch: {warnings[0][:80]}...")782 all_passed = False783 else:784 print(f" ✓ All offsets verified")785 else:786 print(f" ⚠ WARNING: Output file not found")787 all_passed = False788 789 print()790 791 print("=" * 60)792 if all_passed:793 print("✅ ALL SANITY CHECKS PASSED")794 else:795 print("⚠ SOME SANITY CHECKS FAILED - Review warnings above")796 print("=" * 60)797 print()798 799 print("✅ Done! Examples ready for documentation.")800 return 0 if all_passed else 1801 802 803if __name__ == "__main__":804 import sys805 sys.exit(main())806