Spaces:
Sleeping
Sleeping
Download scripts/agentic_lev.py from bearda/croissantminer: direct link, hf CLI and curl.
- Browser
- Download file 56 kB
-
https://huggingface.co/spaces/bearda/croissantminer/resolve/main/scripts/agentic_lev.py
- Command line
-
hf download hf://spaces/bearda/croissantminer/scripts/agentic_lev.py
-
curl -L -o agentic_lev.py https://huggingface.co/spaces/bearda/croissantminer/resolve/main/scripts/agentic_lev.py
56 kB
| #!/usr/bin/env python3 | |
| """CroissantMiner LEV (Locate-Extract-Verify) pipeline. | |
| LOCATE Reuse Phase 1 Gemini-Flash triage (section names per field group) + | |
| Phase 0 section chunks; build a targeted text buffer per group with | |
| fallback keyword matching if triage is empty. | |
| EXTRACT Five independent LLM calls (one per field group) with specialist | |
| system prompts and a {value, evidence} schema. | |
| VERIFY Cross-document enrichment (HuggingFace + Semantic Scholar), schema | |
| validation, confidence scoring. | |
| Output schema (canonical): | |
| {"dataset_id", "model", "extraction" (flat 30 fields), "usage", "valid", | |
| "_meta": {provenance, pipeline="agentic_lev", extraction_details, | |
| group_stats, paper_stats, ...}} | |
| Usage: | |
| python scripts/agentic_lev.py --backbone sonnet-4-5 --paper AI4Math_MathVista | |
| python scripts/agentic_lev.py --backbone gpt-5.4 --dev-only | |
| python scripts/agentic_lev.py --backbone sonnet-4-5 --batch # Anthropic only | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import re | |
| import time | |
| import urllib.parse | |
| import urllib.request | |
| from collections import defaultdict | |
| from pathlib import Path | |
| from dotenv import load_dotenv | |
| from _agentic_helpers import ( | |
| ROOT, | |
| SYSTEM_PROMPT, | |
| CANONICAL_FIELDS, | |
| call_llm, | |
| estimate_cost, | |
| fill_canonical, | |
| get_paper_text, | |
| meta_block, | |
| normalize_field_names, | |
| output_dir_for, | |
| parse_json_response, | |
| resolve_backbone, | |
| strip_references, | |
| write_canonical_output, | |
| ) | |
| load_dotenv(ROOT / ".env") | |
| SCRIPT_NAME = "scripts/agentic_lev.py" | |
| MAX_TOKENS_GROUP = 4096 | |
| MAX_TOKENS_GROUP_V7 = 8192 # increased cap for v7 to avoid evidence-quote truncation | |
| MAX_TOKENS_LOCATOR = 2048 | |
| LOCATOR_CHAR_CAP = 200_000 # cap full-paper input to locator at 200K chars | |
| GROUP_CHAR_CAP = 60_000 | |
| GROUP_MIN_CONTEXT = 2_000 | |
| PHASE0_DIR = ROOT / "data" / "agentic" / "phase0" | |
| PHASE1_DIR = ROOT / "data" / "agentic" / "phase1" | |
| with open(ROOT / "data" / "agentic" / "dev_test_split.json") as f: | |
| SPLIT = json.load(f) | |
| # ββ Field groups and prompts ββββββββββββββββββββββββββββββββββββββ | |
| FIELD_GROUPS = { | |
| "core": { | |
| "fields": ["name", "description", "url", "license", "creator", | |
| "publisher", "datePublished", "inLanguage", "citeAs", "isLiveDataset"], | |
| "triage_key": None, | |
| "fallback_sections": ["abstract", "introduction", "data availability"], | |
| }, | |
| "collection": { | |
| "fields": ["rai:dataCollection", "rai:dataCollectionType", | |
| "rai:dataCollectionMissingData", "rai:dataCollectionRawData", | |
| "rai:dataCollectionTimeframe"], | |
| "triage_key": "G3_collection", | |
| "fallback_sections": ["method", "data collection", "dataset", "corpus"], | |
| }, | |
| "annotation": { | |
| "fields": ["rai:dataAnnotationProtocol", "rai:dataAnnotationPlatform", | |
| "rai:dataAnnotationAnalysis", "rai:annotationsPerItem", | |
| "rai:annotatorDemographics", "rai:machineAnnotationTools"], | |
| "triage_key": "G4_annotation", | |
| "fallback_sections": ["annotation", "labeling", "human evaluation", "crowdsourc"], | |
| }, | |
| "impact": { | |
| "fields": ["rai:dataBiases", "rai:dataLimitations", "rai:dataSocialImpact", | |
| "rai:personalSensitiveInformation", "rai:dataUseCases", | |
| "rai:dataReleaseMaintenancePlan"], | |
| "triage_key": "G5_rai", | |
| "fallback_sections": ["ethic", "limitation", "broader impact", "discussion", "bias", "social"], | |
| }, | |
| "processing": { | |
| "fields": ["rai:dataImputationProtocol", "rai:dataManipulationProtocol", | |
| "rai:dataPreprocessingProtocol"], | |
| "triage_key": "G6_processing", | |
| "fallback_sections": ["preprocess", "data preparation", "filtering", "cleaning"], | |
| }, | |
| } | |
| GROUP_ORDER = ["core", "collection", "annotation", "impact", "processing"] | |
| ALL_FIELDS: list[str] = [] | |
| for _g in GROUP_ORDER: | |
| ALL_FIELDS.extend(FIELD_GROUPS[_g]["fields"]) | |
| GROUP_SYSTEM_PROMPTS = { | |
| "core": """You are an expert at extracting general metadata from ML dataset papers. | |
| Extract ONLY the fields listed below. Use null for any field not explicitly stated in the paper. | |
| Do not guess or fabricate values. | |
| Field definitions: | |
| - name: Dataset name as stated in the paper | |
| - description: Brief description (1-3 sentences) | |
| - url: URL where the dataset can be accessed | |
| - license: Distribution license (e.g., MIT, CC-BY-4.0). Return null if not stated. | |
| - creator: Dataset authors or creating team | |
| - publisher: The organization that funded or released the dataset; this may be a conference or shared task in some cases | |
| - datePublished: The date the dataset was released to users (YYYY or YYYY-MM-DD); may differ from the arxiv submission date | |
| - inLanguage: Content language(s), ISO codes (e.g., "en") | |
| - citeAs: Recommended citation format if provided in the paper. If absent, return null β the pipeline fills this from external sources downstream. | |
| - isLiveDataset: "Yes" if actively updated, "No" if static, null if unknown""", | |
| "collection": """You are an expert at extracting data collection metadata from ML dataset papers. | |
| Extract ONLY the fields listed below. Use null for any field not explicitly discussed. | |
| Field definitions (from the Croissant RAI specification): | |
| - rai:dataCollection: Description of the data collection process | |
| - rai:dataCollectionType: Choose from: Surveys, Secondary Data analysis, Physical data collection, Direct measurement, Document analysis, Manual Human Curator, Software Collection, Experiments, Web Scraping, Web API, Focus groups, Self-reporting, Customer feedback data, User-generated content data, Passive Data Collection, Others. | |
| Prefer "Web Scraping" over "Web API" when data was harvested without authenticated endpoints. Prefer "Secondary Data analysis" when data was reused from an existing corpus rather than freshly gathered. | |
| - rai:dataCollectionMissingData: How missing data was handled. Only if explicitly discussed. | |
| - rai:dataCollectionRawData: Description of the raw/source data | |
| - rai:dataCollectionTimeframe: When data was collected (start/end dates)""", | |
| "annotation": """You are an expert at extracting annotation metadata from ML dataset papers. | |
| Extract ONLY the fields listed below. Use null for any field not discussed. | |
| This is ONLY about the labeling/annotation process. Do NOT include data collection or preprocessing. | |
| Field definitions (from the Croissant RAI specification): | |
| - rai:dataAnnotationProtocol: How labels/ratings were created: task, instructions, workforce, QC | |
| - rai:dataAnnotationPlatform: Platform used (e.g., Amazon MTurk, Label Studio) | |
| - rai:dataAnnotationAnalysis: Annotation quality analysis: agreement metrics, validation | |
| - rai:annotationsPerItem: Number of human labels per data item | |
| - rai:annotatorDemographics: Demographics of annotators | |
| - rai:machineAnnotationTools: ML tools used in annotation (e.g., NER tools, auto-labelers)""", | |
| "impact": """You are an expert at extracting responsible AI metadata from ML dataset papers. | |
| Extract ONLY the fields listed below. Use null for any field not explicitly discussed. | |
| Look in Ethics, Limitations, Broader Impact, and Discussion sections. | |
| Field definitions (from the Croissant RAI specification): | |
| - rai:dataBiases: Description of biases in the dataset, if applicable | |
| - rai:dataLimitations: Known limitations (e.g., data generalization limits, quality issues) and non-recommended uses | |
| - rai:dataSocialImpact: Discussion of social implications, if applicable | |
| - rai:personalSensitiveInformation: Any sensitive human attribute(s) collected as part of this dataset (e.g., gender, socio-economic status, geography, language, age, culture, experience) | |
| - rai:dataUseCases: Dataset use case(s) (e.g., Training, Testing, Validation, Fine-tuning) and usage guidelines | |
| - rai:dataReleaseMaintenancePlan: Versioning information in terms of the updating timeframe, the maintainers, and the deprecation policies""", | |
| "processing": """You are an expert at extracting data processing metadata from ML dataset papers. | |
| Extract ONLY the fields listed below. Use null for any field not discussed. | |
| Distinguish carefully between: preprocessing (making data ML-ready), manipulation (post-preprocessing changes), and imputation (filling missing values). | |
| Field definitions (from the Croissant RAI specification): | |
| - rai:dataImputationProtocol: How missing values were imputed/filled | |
| - rai:dataManipulationProtocol: Post-preprocessing modifications: augmentation, balancing, sampling | |
| - rai:dataPreprocessingProtocol: Steps to make data ML-ready: filtering, cleaning, normalization""", | |
| } | |
| GROUP_USER_PROMPT = """Extract the following fields from the paper section below. | |
| For each field, provide "value" and "evidence" (a brief quote from the text supporting your extraction). | |
| For null fields: {{"value": null, "evidence": null}} | |
| Return ONLY valid JSON with these keys: %s | |
| PAPER SECTION: | |
| %s""" | |
| # v2 user prompt: anti-null-bias with few-shot null examples. Phase 3 Fix A | |
| # from the 2026-05-02 V2/LEV iteration plan. Targets failure mode F1 | |
| # (null-gold hallucination, ~60%% of LEV-C2 round-0 score=3 cells). | |
| # Variant v5: round-0 prompt + multi-aspect enumeration emphasis (Fix D). | |
| # NOT compounded with Fix A (anti-null prompt) since the v4 compound made things | |
| # significantly worse (Ξ=-0.061, p=0.004 reverted). v5 = Fix B (locator gating) | |
| # + Fix D (multi-type enumeration in extractor prompt). | |
| GROUP_USER_PROMPT_V5 = """Extract the following fields from the paper section below. | |
| For each field, provide "value" and "evidence" (a brief quote from the text supporting your extraction). | |
| For null fields: {{"value": null, "evidence": null}} | |
| CRITICAL FOR MULTI-ASPECT FIELDS: be EXHAUSTIVE. | |
| - rai:dataCollectionType: many datasets combine 2 or 3 collection types. | |
| List ALL applicable types from the controlled vocabulary, e.g., | |
| "Web Scraping | Manual Human Curator | Document analysis", not just the | |
| primary one. | |
| - rai:dataLimitations: capture ALL limitations discussed (temporal, | |
| geographic, quality, scope, generalization, exclusion). | |
| - rai:dataBiases: capture ALL biases discussed (selection, demographic, | |
| contamination, source-medium, stylistic, etc.). | |
| - rai:dataUseCases: list ALL stated use cases (training, testing, validation, | |
| fine-tuning, benchmarking). | |
| Format multi-aspect values with " | " separator OR comprehensive prose. | |
| Do not omit aspects to keep the answer brief. | |
| Return ONLY valid JSON with these keys: %s | |
| PAPER SECTION: | |
| %s""" | |
| GROUP_USER_PROMPT_V2 = """CRITICAL β DEFAULT TO NULL WHEN UNCERTAIN. | |
| Many Croissant RAI fields are rarely documented in dataset papers. If the | |
| section text does NOT explicitly discuss a field, return null. Do not | |
| generate plausible-sounding content based on what a similar paper might say. | |
| Do not infer from related content. Every non-null extraction must have an | |
| evidence quote that is a near-verbatim sentence from the section text. | |
| EXAMPLE A -- section does NOT discuss the field, so the answer is null: | |
| Section: "We crowdsourced 10K questions and verified accuracy..." | |
| Field: rai:dataAnnotationPlatform | |
| CORRECT: {{"value": null, "evidence": null}} | |
| WRONG: extracting "Amazon Mechanical Turk" because the section says "crowdsourced" | |
| EXAMPLE B -- section DOES discuss the field, so extract with evidence: | |
| Section: "Annotations were collected via Amazon Mechanical Turk with 3 workers per item." | |
| Field: rai:dataAnnotationPlatform | |
| CORRECT: {{"value": "Amazon Mechanical Turk", "evidence": | |
| "Annotations were collected via Amazon Mechanical Turk..."}} | |
| EXAMPLE C -- section is silent on imputation, so the field is null: | |
| Section: "Our dataset contains 50K complete examples." | |
| Field: rai:dataImputationProtocol | |
| CORRECT: {{"value": null, "evidence": null}} | |
| WRONG: extracting any imputation protocol (would be hallucinated) | |
| Now extract the fields below. For each: provide "value" and "evidence" (a | |
| brief quote from the text supporting your extraction). For fields not | |
| discussed in the section: {{"value": null, "evidence": null}}. | |
| Return ONLY valid JSON with these keys: %s | |
| PAPER SECTION: | |
| %s""" | |
| # LEV's specialist prompts differ from the canonical SYSTEM_PROMPT. Hash the | |
| # actual strings the LLM sees so _meta.specialist_prompts_sha256_prefix is an | |
| # honest provenance record (PROMPT_HASH from helpers tracks the canonical | |
| # definitions source, unchanged across LEV-prompt revisions). | |
| SPECIALIST_PROMPTS_HASH = hashlib.sha256( | |
| json.dumps( | |
| {"group_system_prompts": GROUP_SYSTEM_PROMPTS, | |
| "group_user_prompt": GROUP_USER_PROMPT}, | |
| sort_keys=True, | |
| ).encode() | |
| ).hexdigest()[:16] | |
| LICENSE_MAP = { | |
| "cc-by-4.0": "CC-BY-4.0", "cc by 4.0": "CC-BY-4.0", | |
| "cc-by-sa-4.0": "CC-BY-SA-4.0", "cc by-sa 4.0": "CC-BY-SA-4.0", | |
| "cc-by-nc-4.0": "CC-BY-NC-4.0", "cc by-nc 4.0": "CC-BY-NC-4.0", | |
| "cc-by-nc-sa-4.0": "CC-BY-NC-SA-4.0", | |
| "cc0-1.0": "CC0-1.0", "cc0": "CC0-1.0", "public domain": "CC0-1.0", | |
| "mit": "MIT", "mit license": "MIT", | |
| "apache-2.0": "Apache-2.0", "apache 2.0": "Apache-2.0", | |
| } | |
| # ββ Phase 0 / Phase 1 loaders βββββββββββββββββββββββββββββββββββββ | |
| def load_paper_text(ds_id: str) -> str | None: | |
| text = get_paper_text(ds_id) | |
| if text is None: | |
| return None | |
| return strip_references(text).strip() | |
| def load_triage(ds_id: str) -> dict: | |
| path = PHASE1_DIR / f"{ds_id}.json" | |
| if not path.exists(): | |
| return {} | |
| with open(path) as f: | |
| return json.load(f).get("triage", {}) | |
| def load_section_text(ds_id: str) -> dict[str, str]: | |
| path = PHASE0_DIR / f"{ds_id}.json" | |
| if not path.exists(): | |
| return {} | |
| with open(path) as f: | |
| data = json.load(f) | |
| sections: defaultdict[str, str] = defaultdict(str) | |
| for chunk in data.get("chunks", []): | |
| name = str(chunk.get("section", "unknown")).strip().upper() | |
| text = chunk.get("text", "") | |
| if name and text: | |
| sections[name] += " " + text | |
| return dict(sections) | |
| # ββ LOCATE ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| LOCATOR_SYSTEM_PROMPT = """You are an expert at navigating academic papers about ML datasets. Given a paper's text and a list of available section names, your job is to identify which sections are most relevant for extracting specific groups of metadata fields. | |
| You will return a JSON object with one entry per field group, listing the section names (from the provided list, exact match) that contain content relevant to that group, plus a presence indicator. | |
| Be precise: only include sections that genuinely discuss the topic. If no section discusses a group, return an empty list and presence="unlikely". If discussion is brief or scattered, mark presence="possible". If a clearly dedicated section exists, mark presence="likely".""" | |
| LOCATOR_USER_TEMPLATE = """Paper text (may be truncated for length): | |
| {paper_text} | |
| Available section names in this paper (use these EXACTLY β they are case-sensitive, do not paraphrase): | |
| {section_list} | |
| Identify relevant sections for each of these field groups: | |
| - G3_collection: How the data was collected β sources, methodology, raw data, timeframe, missing data handling. | |
| - G4_annotation: Annotation procedures β labeling protocols, annotator demographics, annotation platforms (MTurk, Label Studio, etc.), per-item annotation counts, machine annotation tools, annotation analysis / quality metrics. | |
| - G5_rai: Responsible-AI considerations β biases, limitations, ethical issues, social impact, sensitive personal information, intended use cases, release maintenance plan. | |
| - G6_processing: Data processing β preprocessing steps, data manipulation/transformation, imputation of missing values. | |
| Return JSON only, in this exact schema: | |
| {{ | |
| "G3_collection": {{"sections": ["EXACT_SECTION_NAME_1", ...], "presence": "likely"|"possible"|"unlikely"}}, | |
| "G4_annotation": {{"sections": [...], "presence": "..."}}, | |
| "G5_rai": {{"sections": [...], "presence": "..."}}, | |
| "G6_processing": {{"sections": [...], "presence": "..."}} | |
| }}""" | |
| def locate_with_llm(cfg, ds_id: str, full_text: str) -> tuple[dict, dict]: | |
| """LLM-based locator: replaces precomputed Phase 1 Gemini-Flash triage. | |
| Calls the locator LLM with the full paper text + the list of available | |
| section names from PHASE0 chunks; the LLM returns a triage dict in the | |
| same format as load_triage() so the rest of the pipeline is unchanged. | |
| Returns (triage_dict, usage_dict). On failure, returns ({}, usage) and | |
| locate_sections() will fall back to keyword-based section selection. | |
| """ | |
| sections = load_section_text(ds_id) | |
| section_names = sorted(sections.keys()) | |
| if not section_names: | |
| return {}, {"input_tokens": 0, "output_tokens": 0} | |
| section_list = "\n".join(f"- {n}" for n in section_names) | |
| paper_text = full_text[:LOCATOR_CHAR_CAP] | |
| user = LOCATOR_USER_TEMPLATE.format( | |
| paper_text=paper_text, | |
| section_list=section_list, | |
| ) | |
| try: | |
| raw, usage = call_llm(cfg, LOCATOR_SYSTEM_PROMPT, user, MAX_TOKENS_LOCATOR) | |
| parsed = parse_json_response(raw) | |
| except Exception as e: | |
| print(f" {ds_id}/locator FAILED: {str(e)[:80]} β falling back to keyword search") | |
| return {}, {"input_tokens": 0, "output_tokens": 0} | |
| # Normalize: ensure section names match available ones (uppercase compare) | |
| name_to_canonical = {n.upper(): n for n in section_names} | |
| triage: dict = {} | |
| for group_key in ("G3_collection", "G4_annotation", "G5_rai", "G6_processing"): | |
| entry = parsed.get(group_key, {}) | |
| if not isinstance(entry, dict): | |
| continue | |
| raw_sections = entry.get("sections", []) | |
| if not isinstance(raw_sections, list): | |
| raw_sections = [] | |
| # Map each LLM-returned name to a canonical section name; drop unknowns. | |
| canonical_sections: list[str] = [] | |
| for s in raw_sections: | |
| key = str(s).upper().strip() | |
| if key in name_to_canonical: | |
| canonical_sections.append(name_to_canonical[key]) | |
| else: | |
| # Fuzzy match: substring either way | |
| for known_upper, known in name_to_canonical.items(): | |
| if key in known_upper or known_upper in key: | |
| canonical_sections.append(known) | |
| break | |
| presence = entry.get("presence", "possible") | |
| if presence not in ("likely", "possible", "unlikely"): | |
| presence = "possible" | |
| triage[group_key] = {"sections": canonical_sections, "presence": presence} | |
| return triage, usage | |
| def locate_sections_v7(ds_id: str, full_text: str, | |
| triage_override: dict | None = None) -> tuple[dict[str, str], dict]: | |
| """v7 locator output assembly. Fixes 1, 4, 5, 6, 9 from the 2026-05-02 audit: | |
| 1. Stricter section matching β prefer exact upper-case match before | |
| substring fallback. Reduces wrong-section pickups. | |
| 4. Deduplicate matched sections β never include the same paragraph twice. | |
| 5. Preserve paper order β sections appear in document order, not match | |
| loop order. Helps the model reason about earlier/later context. | |
| 6. Section metadata in input β each section is prefixed with | |
| "=== Section: NAME ===" so the extractor can reference positions. | |
| 9. Core-group hack removed β abstract/intro are added via the same | |
| exact/substring match path; no special prepend of full-text fifth. | |
| """ | |
| triage = triage_override if triage_override is not None else load_triage(ds_id) | |
| sections = load_section_text(ds_id) | |
| section_order = list(sections.keys()) # paper order | |
| group_texts: dict[str, str] = {} | |
| for group_name, cfg in FIELD_GROUPS.items(): | |
| triage_key = cfg["triage_key"] | |
| triage_info = triage.get(triage_key, {}) if triage_key else {} | |
| triage_sections = [s.upper() for s in triage_info.get("sections", [])] | |
| fallback = cfg["fallback_sections"] | |
| # Match step. Build a set of section *names* (not text) to dedup | |
| # at the section-id level, then assemble in paper order. | |
| matched_names: set[str] = set() | |
| # Exact match first | |
| for target in triage_sections: | |
| for name in section_order: | |
| if name == target: | |
| matched_names.add(name) | |
| # Substring match second (target β name OR name β target), only for | |
| # targets that didn't already match exactly. | |
| unmatched_targets = [t for t in triage_sections | |
| if not any(t == n for n in section_order)] | |
| for target in unmatched_targets: | |
| for name in section_order: | |
| if target in name or (len(name) >= 4 and name in target): | |
| matched_names.add(name) | |
| # Keyword fallback if nothing matched | |
| if not matched_names: | |
| for name in section_order: | |
| low = name.lower() | |
| if any(kw in low for kw in fallback): | |
| matched_names.add(name) | |
| # Core group: also include abstract / introduction by name | |
| if group_name == "core": | |
| for name in section_order: | |
| if any(kw in name.lower() for kw in ("abstract", "introduction", "1 intro")): | |
| matched_names.add(name) | |
| # Assemble in paper order, prefixed with section metadata. | |
| parts: list[str] = [] | |
| for name in section_order: | |
| if name in matched_names: | |
| txt = sections.get(name, "").strip() | |
| if txt: | |
| parts.append(f"=== Section: {name} ===\n{txt}") | |
| combined = "\n\n".join(parts).strip() | |
| if len(combined) < GROUP_MIN_CONTEXT and full_text: | |
| # Fallback: dump first half of paper rather than nothing | |
| combined = "=== Paper context (locator returned no usable sections) ===\n" | |
| combined += full_text[: len(full_text) // 2] | |
| if len(combined) > GROUP_CHAR_CAP: | |
| combined = combined[:GROUP_CHAR_CAP] | |
| group_texts[group_name] = combined | |
| return group_texts, triage | |
| def locate_sections(ds_id: str, full_text: str, | |
| triage_override: dict | None = None) -> tuple[dict[str, str], dict]: | |
| triage = triage_override if triage_override is not None else load_triage(ds_id) | |
| sections = load_section_text(ds_id) | |
| group_texts: dict[str, str] = {} | |
| for group_name, cfg in FIELD_GROUPS.items(): | |
| triage_key = cfg["triage_key"] | |
| triage_info = triage.get(triage_key, {}) if triage_key else {} | |
| triage_sections = [s.upper() for s in triage_info.get("sections", [])] | |
| fallback = cfg["fallback_sections"] | |
| matched: list[str] = [] | |
| for target in triage_sections: | |
| for name, text in sections.items(): | |
| if target in name or name in target: | |
| matched.append(text) | |
| if not matched: | |
| for name, text in sections.items(): | |
| low = name.lower() | |
| if any(kw in low for kw in fallback): | |
| matched.append(text) | |
| if group_name == "core": | |
| for name, text in sections.items(): | |
| if any(kw in name.lower() for kw in ("abstract", "introduction", "1 intro")): | |
| if text not in matched: | |
| matched.insert(0, text) | |
| if full_text: | |
| matched.insert(0, full_text[: len(full_text) // 5]) | |
| combined = "\n\n".join(matched).strip() | |
| if len(combined) < GROUP_MIN_CONTEXT and full_text: | |
| combined = full_text[: len(full_text) // 2] | |
| if len(combined) > GROUP_CHAR_CAP: | |
| combined = combined[:GROUP_CHAR_CAP] | |
| group_texts[group_name] = combined | |
| return group_texts, triage | |
| # ββ EXTRACT βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _schema_for_group(fields: list[str]) -> str: | |
| parts = [f'"{f}": {{"value": "string or null", "evidence": "quote or null"}}' | |
| for f in fields] | |
| return "{\n " + ",\n ".join(parts) + "\n}" | |
| def extract_group_v7(cfg, group_name: str, section_text: str, fields: list[str]): | |
| """v7 extractor with fixes 7, 8 from the 2026-05-02 audit: | |
| 7. Retry once on JSON parse failure with an explicit schema reminder. | |
| 8. Increased max_tokens from 4096 to 8192 to avoid evidence-quote | |
| truncation that loses fields silently. | |
| Uses the v1 round-0 user prompt template (multi-aspect emphasis applies | |
| via Fix D in v5 only; v7 starts from v1 baseline + locator gating). | |
| """ | |
| system = GROUP_SYSTEM_PROMPTS[group_name] | |
| schema = _schema_for_group(fields) | |
| user = GROUP_USER_PROMPT % (schema, section_text) | |
| raw, usage = call_llm(cfg, system, user, MAX_TOKENS_GROUP_V7) | |
| try: | |
| parsed = parse_json_response(raw) | |
| except Exception: | |
| # Retry once with a schema reminder appended | |
| retry_user = ( | |
| f"Your previous response was not valid JSON. Please try again. " | |
| f"Return ONLY a valid JSON object with these exact keys: " | |
| f"{', '.join(fields)}. Each value must follow the schema " | |
| f"{{\"value\": <string|null>, \"evidence\": <string|null>}}.\n\n" | |
| f"{user}" | |
| ) | |
| raw2, usage2 = call_llm(cfg, system, retry_user, MAX_TOKENS_GROUP_V7) | |
| usage = {"input_tokens": usage["input_tokens"] + usage2["input_tokens"], | |
| "output_tokens": usage["output_tokens"] + usage2["output_tokens"]} | |
| try: | |
| parsed = parse_json_response(raw2) | |
| except Exception: | |
| parsed = {} | |
| extraction: dict = {} | |
| evidence: dict = {} | |
| for field in fields: | |
| entry = parsed.get(field, {}) | |
| if isinstance(entry, dict): | |
| extraction[field] = entry.get("value") | |
| evidence[field] = entry.get("evidence") | |
| else: | |
| extraction[field] = entry | |
| evidence[field] = None | |
| return extraction, evidence, usage | |
| VERIFY_CORRECT_USER = """You previously extracted some fields from a section but returned NULL for these: | |
| {null_fields} | |
| This was your section context (the same locator output as before): | |
| {section_text} | |
| For each null field above, RE-READ the section carefully and decide: | |
| - If the section TRULY does not discuss the field, keep it null. Do NOT hallucinate. | |
| - If the section DOES contain relevant content you missed, extract it now with an evidence quote. | |
| Return JSON: {{"<field_id>": {{"value": "string or null", "evidence": "quote or null"}}, ...}} | |
| ONLY include the fields listed above. No other fields. No prose. JSON only.""" | |
| def verify_correct_group(cfg, group_name: str, section_text: str, fields: list[str], | |
| current_extraction: dict, current_evidence: dict, | |
| triage_presence: str = "possible") -> tuple[dict, dict, dict]: | |
| """v7 verify+correct phase (Fix 2 from audit). | |
| Re-checks fields that came back null IF the locator believed the topic | |
| was 'likely' or 'possible' present in the paper. Skips when locator says | |
| 'unlikely' (Fix B already nulled them anyway). Adds at most 1 extra LLM | |
| call per group, only when there are nulls worth re-checking. | |
| """ | |
| if triage_presence == "unlikely": | |
| return current_extraction, current_evidence, {"input_tokens": 0, "output_tokens": 0} | |
| null_fields = [f for f in fields | |
| if current_extraction.get(f) is None or | |
| (isinstance(current_extraction.get(f), str) and | |
| not current_extraction[f].strip())] | |
| if not null_fields: | |
| return current_extraction, current_evidence, {"input_tokens": 0, "output_tokens": 0} | |
| system = GROUP_SYSTEM_PROMPTS[group_name] | |
| user = VERIFY_CORRECT_USER.format( | |
| null_fields="\n".join(f" - {f}" for f in null_fields), | |
| section_text=section_text[:50000], # cap for verify-prompt | |
| ) | |
| try: | |
| raw, usage = call_llm(cfg, system, user, MAX_TOKENS_GROUP_V7) | |
| parsed = parse_json_response(raw) | |
| except Exception: | |
| return current_extraction, current_evidence, {"input_tokens": 0, "output_tokens": 0} | |
| # Apply only filling (don't overwrite already-filled values) | |
| for f in null_fields: | |
| entry = parsed.get(f, {}) | |
| if not isinstance(entry, dict): | |
| continue | |
| new_val = entry.get("value") | |
| if new_val is not None and isinstance(new_val, str) and new_val.strip(): | |
| current_extraction[f] = new_val | |
| current_evidence[f] = entry.get("evidence") | |
| return current_extraction, current_evidence, usage | |
| def extract_group(cfg, group_name: str, section_text: str, fields: list[str], | |
| prompt_variant: str = "v1"): | |
| system = GROUP_SYSTEM_PROMPTS[group_name] | |
| schema = _schema_for_group(fields) | |
| # v2 and v4 use the anti-null-bias prompt template. | |
| # v5 uses the multi-aspect enumeration template (NO anti-null wrapping). | |
| # v6 = v3 prompt (round-0) + tighter section truncation (8K cap, applied below). | |
| # v3, v4, v5, v6 all apply locator-extractor coupling (in process_paper). | |
| if prompt_variant == "v5": | |
| template = GROUP_USER_PROMPT_V5 | |
| elif prompt_variant in ("v2", "v4"): | |
| template = GROUP_USER_PROMPT_V2 | |
| else: | |
| template = GROUP_USER_PROMPT | |
| # v6: tighter section context to focus the extractor on the most-relevant slice. | |
| if prompt_variant == "v6": | |
| section_text = section_text[:8000] | |
| user = template % (schema, section_text) | |
| raw, usage = call_llm(cfg, system, user, MAX_TOKENS_GROUP) | |
| parsed = parse_json_response(raw) | |
| extraction: dict = {} | |
| evidence: dict = {} | |
| for field in fields: | |
| entry = parsed.get(field, {}) | |
| if isinstance(entry, dict): | |
| extraction[field] = entry.get("value") | |
| evidence[field] = entry.get("evidence") | |
| else: | |
| extraction[field] = entry | |
| evidence[field] = None | |
| return extraction, evidence, usage | |
| # ββ VERIFY: enrichment + validation βββββββββββββββββββββββββββββββ | |
| def _is_null(val) -> bool: | |
| if val is None: | |
| return True | |
| if isinstance(val, str) and val.strip().lower() in ("", "null", "none"): | |
| return True | |
| return False | |
| def enrich(ds_id: str, extraction: dict) -> tuple[dict, dict]: | |
| enrichments: dict = {} | |
| hf_id = ds_id.replace("_", "/", 1) | |
| try: | |
| url = f"https://huggingface.co/api/datasets/{hf_id}" | |
| req = urllib.request.Request(url, headers={"User-Agent": "CroissantMiner/2.0"}) | |
| resp = urllib.request.urlopen(req, timeout=10) | |
| data = json.loads(resp.read()) | |
| card = data.get("cardData", {}) or {} | |
| if _is_null(extraction.get("license")): | |
| lic = card.get("license") | |
| if lic: | |
| enrichments["license"] = lic | |
| if _is_null(extraction.get("inLanguage")): | |
| lang = card.get("language") | |
| if lang: | |
| enrichments["inLanguage"] = ", ".join(lang) if isinstance(lang, list) else str(lang) | |
| if _is_null(extraction.get("url")): | |
| enrichments["url"] = f"https://huggingface.co/datasets/{hf_id}" | |
| except Exception: | |
| pass | |
| if _is_null(extraction.get("citeAs")): | |
| name = extraction.get("name") or ds_id | |
| try: | |
| q = urllib.parse.quote(str(name)) | |
| url = (f"https://api.semanticscholar.org/graph/v1/paper/search" | |
| f"?query={q}&limit=1&fields=title,citationStyles") | |
| req = urllib.request.Request(url, headers={"User-Agent": "CroissantMiner/2.0"}) | |
| resp = urllib.request.urlopen(req, timeout=10) | |
| data = json.loads(resp.read()) | |
| papers = data.get("data", []) | |
| if papers: | |
| bibtex = papers[0].get("citationStyles", {}).get("bibtex", "") | |
| if bibtex: | |
| enrichments["citeAs"] = bibtex | |
| except Exception: | |
| pass | |
| for field, val in enrichments.items(): | |
| extraction[field] = val | |
| return extraction, enrichments | |
| def validate_and_score(extraction, evidence, enrichments, triage) -> dict: | |
| details: dict = {} | |
| for field in ALL_FIELDS: | |
| val = extraction.get(field) | |
| if isinstance(val, str) and val.strip().lower() in ( | |
| "", "null", "none", "n/a", "not applicable", | |
| "[null]", "[null - not found in paper]", | |
| ): | |
| val = None | |
| if isinstance(val, list): | |
| val = ", ".join(str(x) for x in val) | |
| if field == "license" and val: | |
| norm = LICENSE_MAP.get(str(val).strip().lower()) | |
| if norm: | |
| val = norm | |
| if field == "url" and val and not re.match(r"https?://", str(val)): | |
| val = f"https://{val}" | |
| extraction[field] = val | |
| if val is None: | |
| details[field] = {"confidence": 0.0, "status": "NULL", "source": "absent", "evidence": None} | |
| continue | |
| confidence = 0.8 | |
| source = "extraction" | |
| field_ev = evidence.get(field) | |
| has_ev = field_ev is not None and str(field_ev).strip() not in ("", "null", "None") | |
| if has_ev: | |
| confidence = min(confidence + 0.15, 1.0) | |
| if field in enrichments: | |
| source = "enrichment" | |
| confidence = 0.7 | |
| for group_name, gcfg in FIELD_GROUPS.items(): | |
| if field in gcfg["fields"]: | |
| tk = gcfg["triage_key"] | |
| if tk and triage.get(tk, {}).get("presence") == "unlikely": | |
| confidence = min(confidence, 0.5) | |
| break | |
| status = ("VERIFIED" if confidence >= 0.7 | |
| else "UNCERTAIN" if confidence >= 0.4 | |
| else "LOW_CONFIDENCE") | |
| details[field] = { | |
| "confidence": round(confidence, 2), | |
| "status": status, | |
| "source": source, | |
| "evidence": field_ev if has_ev else None, | |
| } | |
| return details | |
| # ββ Save canonical output βββββββββββββββββββββββββββββββββββββββββ | |
| def _paper_stats(details: dict) -> dict: | |
| extracted = sum(1 for d in details.values() if d["status"] != "NULL") | |
| verified = sum(1 for d in details.values() if d["status"] == "VERIFIED") | |
| ev = sum(1 for d in details.values() if d.get("evidence")) | |
| confs = [d["confidence"] for d in details.values() if d["status"] != "NULL"] | |
| avg = sum(confs) / len(confs) if confs else 0.0 | |
| return { | |
| "fields_extracted": extracted, | |
| "fields_null": 30 - extracted, | |
| "fields_verified": verified, | |
| "evidence_count": ev, | |
| "evidence_coverage": round(ev / max(extracted, 1) * 100, 1), | |
| "avg_confidence": round(avg, 3), | |
| } | |
| def save_paper(out_dir: Path, ds_id: str, cfg, extraction, details, usage, | |
| mode: str, extra_meta: dict): | |
| stats = _paper_stats(details) | |
| extra = { | |
| "pipeline": "agentic_lev", | |
| "specialist_prompts_sha256_prefix": SPECIALIST_PROMPTS_HASH, | |
| "extraction_details": details, | |
| "paper_stats": stats, | |
| "cost_usd": estimate_cost(cfg, usage, batch_discount=(mode == "batch")), | |
| "split": "dev" if ds_id in SPLIT["dev"] else "test", | |
| } | |
| extra.update(extra_meta) | |
| meta = meta_block(cfg, SCRIPT_NAME, mode, extra=extra) | |
| write_canonical_output(out_dir, ds_id, extraction, usage, cfg, meta) | |
| return stats | |
| # ββ Sequential processing βββββββββββββββββββββββββββββββββββββββββ | |
| def process_paper(cfg, ds_id: str, out_dir: Path, | |
| locator_cfg: dict | None = None, | |
| prompt_variant: str = "v1") -> dict | None: | |
| t0 = time.time() | |
| full_text = load_paper_text(ds_id) | |
| if not full_text: | |
| print(f" SKIP {ds_id}: no PDF") | |
| return None | |
| locator_usage = {"input_tokens": 0, "output_tokens": 0} | |
| if locator_cfg is not None: | |
| triage_override, locator_usage = locate_with_llm(locator_cfg, ds_id, full_text) | |
| if prompt_variant == "v7": | |
| group_texts, triage = locate_sections_v7(ds_id, full_text, triage_override=triage_override) | |
| else: | |
| group_texts, triage = locate_sections(ds_id, full_text, triage_override=triage_override) | |
| else: | |
| if prompt_variant == "v7": | |
| group_texts, triage = locate_sections_v7(ds_id, full_text) | |
| else: | |
| group_texts, triage = locate_sections(ds_id, full_text) | |
| extraction: dict = {} | |
| evidence: dict = {} | |
| group_stats: dict = {} | |
| total_usage = {"input_tokens": 0, "output_tokens": 0} | |
| skipped_unlikely = 0 | |
| for group_name in GROUP_ORDER: | |
| section_text = group_texts.get(group_name, "") | |
| fields = FIELD_GROUPS[group_name]["fields"] | |
| # Phase 3 Fix B: locator-extractor coupling (variants v3, v4, v5, v6, v7). | |
| # v4 = +Fix A; v5 = +Fix D (multi-aspect); v6 = +tighter section context; | |
| # v7 = full implementation cleanup (audit fixes 1-9). | |
| if prompt_variant in ("v3", "v4", "v5", "v6", "v7"): | |
| triage_key = FIELD_GROUPS[group_name].get("triage_key") | |
| if triage_key: | |
| presence = triage.get(triage_key, {}).get("presence", "possible") | |
| if presence == "unlikely": | |
| for f in fields: | |
| extraction[f] = None | |
| evidence[f] = None | |
| group_stats[group_name] = {"section_chars": len(section_text), | |
| "skipped_unlikely": True} | |
| skipped_unlikely += 1 | |
| continue | |
| if not section_text.strip(): | |
| for f in fields: | |
| extraction[f] = None | |
| evidence[f] = None | |
| group_stats[group_name] = {"section_chars": 0, "skipped": True} | |
| continue | |
| try: | |
| if prompt_variant == "v7": | |
| ext, ev, usage = extract_group_v7(cfg, group_name, section_text, fields) | |
| # v7 verify+correct phase: re-prompt for null fields if locator | |
| # said the topic was 'likely' or 'possible'. | |
| triage_key_v7 = FIELD_GROUPS[group_name].get("triage_key") | |
| presence_v7 = (triage.get(triage_key_v7, {}).get("presence", "possible") | |
| if triage_key_v7 else "possible") | |
| ext, ev, vc_usage = verify_correct_group( | |
| cfg, group_name, section_text, fields, ext, ev, | |
| triage_presence=presence_v7, | |
| ) | |
| usage = {"input_tokens": usage["input_tokens"] + vc_usage["input_tokens"], | |
| "output_tokens": usage["output_tokens"] + vc_usage["output_tokens"]} | |
| else: | |
| ext, ev, usage = extract_group(cfg, group_name, section_text, fields, | |
| prompt_variant=prompt_variant) | |
| except Exception as e: | |
| print(f" {ds_id}/{group_name} FAILED: {str(e)[:60]}") | |
| for f in fields: | |
| extraction[f] = None | |
| evidence[f] = None | |
| group_stats[group_name] = {"error": str(e)[:80]} | |
| continue | |
| extraction.update(ext) | |
| evidence.update(ev) | |
| total_usage["input_tokens"] += usage["input_tokens"] | |
| total_usage["output_tokens"] += usage["output_tokens"] | |
| group_stats[group_name] = { | |
| "section_chars": len(section_text), | |
| "tokens_in": usage["input_tokens"], | |
| "tokens_out": usage["output_tokens"], | |
| } | |
| time.sleep(0.3) | |
| # Fold locator tokens into total so cost reflects full pipeline. | |
| total_usage["input_tokens"] += locator_usage["input_tokens"] | |
| total_usage["output_tokens"] += locator_usage["output_tokens"] | |
| extraction, enrichments = enrich(ds_id, extraction) | |
| details = validate_and_score(extraction, evidence, enrichments, triage) | |
| extra_meta = { | |
| "group_stats": group_stats, | |
| "enrichments": list(enrichments.keys()), | |
| "elapsed_seconds": round(time.time() - t0, 1), | |
| } | |
| if locator_cfg is not None: | |
| extra_meta["locator"] = { | |
| "backbone_key": locator_cfg.get("backbone_key", "unknown"), | |
| "model_id": locator_cfg.get("model_id"), | |
| "provider": locator_cfg.get("provider"), | |
| "input_tokens": locator_usage["input_tokens"], | |
| "output_tokens": locator_usage["output_tokens"], | |
| } | |
| else: | |
| extra_meta["locator"] = {"source": "precomputed_phase1_triage"} | |
| extra_meta["prompt_variant"] = prompt_variant | |
| extra_meta["skipped_unlikely_groups"] = skipped_unlikely | |
| return save_paper(out_dir, ds_id, cfg, extraction, details, total_usage, "realtime", extra_meta) | |
| def run_sequential(cfg, papers, args, out_dir, locator_cfg: dict | None = None, | |
| prompt_variant: str = "v1"): | |
| ok = fail = total_extracted = total_verified = 0 | |
| total_usage = {"input_tokens": 0, "output_tokens": 0} | |
| for i, ds_id in enumerate(papers): | |
| safe = ds_id.replace("/", "__") | |
| out_path = out_dir / f"{safe}.json" | |
| if out_path.exists() and not args.paper: | |
| ok += 1 | |
| if (i + 1) % 25 == 0: | |
| print(f" [{i+1}/{len(papers)}] {ok} done (skipping existing)") | |
| continue | |
| try: | |
| stats = process_paper(cfg, ds_id, out_dir, locator_cfg=locator_cfg, | |
| prompt_variant=prompt_variant) | |
| if stats is None: | |
| fail += 1 | |
| continue | |
| ok += 1 | |
| total_extracted += stats["fields_extracted"] | |
| total_verified += stats["fields_verified"] | |
| with open(out_path) as f: | |
| saved = json.load(f) | |
| total_usage["input_tokens"] += saved["usage"]["input_tokens"] | |
| total_usage["output_tokens"] += saved["usage"]["output_tokens"] | |
| print(f" [{i+1}/{len(papers)}] {ds_id}: " | |
| f"{stats['fields_extracted']}/30, " | |
| f"${saved['_meta']['cost_usd']:.3f}, " | |
| f"ev={stats['evidence_coverage']:.0f}%, " | |
| f"{saved['_meta']['elapsed_seconds']:.0f}s") | |
| except Exception as e: | |
| print(f" [{i+1}/{len(papers)}] {ds_id}: FAILED ({str(e)[:80]})") | |
| fail += 1 | |
| time.sleep(0.3) | |
| total_cost = estimate_cost(cfg, total_usage, batch_discount=False) | |
| total_fields = ok * 30 | |
| print(f"\n{'=' * 70}") | |
| print(f"SUMMARY β LEV sequential ({cfg['name']})") | |
| print(f"{'=' * 70}") | |
| print(f"Papers: {ok} OK, {fail} failed") | |
| print(f"Fields: {total_extracted}/{total_fields} " | |
| f"({total_extracted / max(total_fields, 1) * 100:.1f}%)") | |
| print(f"Verified: {total_verified}/{total_fields}") | |
| print(f"Tokens: {total_usage['input_tokens']:,} in, " | |
| f"{total_usage['output_tokens']:,} out") | |
| print(f"Cost: ${total_cost:.2f}") | |
| print(f"Output: {out_dir}/") | |
| summary = { | |
| "backbone": cfg["model_id"], | |
| "extractor_backbone": cfg["model_id"], | |
| "locator_backbone": (locator_cfg["model_id"] if locator_cfg else | |
| "precomputed_phase1_gemini_flash_triage"), | |
| "papers_ok": ok, "papers_failed": fail, | |
| "total_extracted": total_extracted, | |
| "total_verified": total_verified, | |
| "fill_rate": round(total_extracted / max(total_fields, 1) * 100, 1), | |
| "usage": total_usage, "cost_usd": total_cost, | |
| } | |
| with open(out_dir / "_summary.json", "w") as f: | |
| json.dump(summary, f, indent=2) | |
| # ββ Anthropic-only batch mode βββββββββββββββββββββββββββββββββββββ | |
| def _require_anthropic(cfg): | |
| if cfg["provider"] != "anthropic": | |
| raise SystemExit( | |
| f"Batch mode currently supports provider=anthropic only " | |
| f"(backbone '{cfg['model_id']}' uses provider={cfg['provider']}). " | |
| f"Run without --batch for sequential execution." | |
| ) | |
| def _sanitize_custom_id(text: str, limit: int = 64) -> str: | |
| return re.sub(r"[^a-zA-Z0-9_-]", "_", text.replace("/", "__"))[:limit] | |
| def run_batch(cfg, papers, out_dir): | |
| _require_anthropic(cfg) | |
| import anthropic | |
| from anthropic.types.message_create_params import MessageCreateParamsNonStreaming | |
| from anthropic.types.messages.batch_create_params import Request | |
| client = anthropic.Anthropic() | |
| wave_dir = out_dir / "_batch" | |
| wave_dir.mkdir(parents=True, exist_ok=True) | |
| batch_meta_path = wave_dir / "batch.json" | |
| paper_data: dict[str, tuple[dict, dict, str]] = {} | |
| for ds_id in papers: | |
| safe = ds_id.replace("/", "__") | |
| if (out_dir / f"{safe}.json").exists(): | |
| continue | |
| full_text = load_paper_text(ds_id) | |
| if not full_text: | |
| continue | |
| group_texts, triage = locate_sections(ds_id, full_text) | |
| paper_data[ds_id] = (group_texts, triage, full_text) | |
| if not paper_data: | |
| print(" No papers pending.") | |
| return | |
| if batch_meta_path.exists(): | |
| info = json.loads(batch_meta_path.read_text()) | |
| batch_id = info["batch_id"] | |
| cid_map = info["cid_map"] | |
| else: | |
| requests: list = [] | |
| cid_map: dict = {} | |
| for ds_id, (group_texts, _, _) in paper_data.items(): | |
| safe_ds = _sanitize_custom_id(ds_id, 50) | |
| for group_name in GROUP_ORDER: | |
| section_text = group_texts.get(group_name, "") | |
| if not section_text.strip(): | |
| continue | |
| fields = FIELD_GROUPS[group_name]["fields"] | |
| schema = _schema_for_group(fields) | |
| user = GROUP_USER_PROMPT % (schema, section_text) | |
| cid = f"{safe_ds}___{group_name}"[:64] | |
| cid_map[cid] = {"ds_id": ds_id, "group": group_name} | |
| requests.append(Request( | |
| custom_id=cid, | |
| params=MessageCreateParamsNonStreaming( | |
| model=cfg["model_id"], | |
| max_tokens=MAX_TOKENS_GROUP, | |
| temperature=0.0, | |
| system=GROUP_SYSTEM_PROMPTS[group_name], | |
| messages=[{"role": "user", "content": user}], | |
| ), | |
| )) | |
| batch = client.messages.batches.create(requests=requests) | |
| batch_id = batch.id | |
| batch_meta_path.write_text(json.dumps( | |
| {"batch_id": batch_id, "cid_map": cid_map, | |
| "submitted_at": time.strftime("%Y-%m-%d %H:%M:%S")}, indent=2)) | |
| print(f" Batch submitted: {batch_id} " | |
| f"({len(requests)} requests, {len(paper_data)} papers Γ β€5 groups)") | |
| while True: | |
| info = client.messages.batches.retrieve(batch_id) | |
| counts = info.request_counts | |
| total_done = counts.succeeded + counts.errored + counts.expired | |
| print(f" {info.processing_status}: {total_done} done " | |
| f"(ok={counts.succeeded}, err={counts.errored})") | |
| if info.processing_status == "ended": | |
| break | |
| time.sleep(30) | |
| group_results: dict = defaultdict(dict) | |
| for result in client.messages.batches.results(batch_id): | |
| meta = cid_map.get(result.custom_id) | |
| if not meta: | |
| continue | |
| ds_id = meta["ds_id"] | |
| group_name = meta["group"] | |
| if result.result.type != "succeeded": | |
| print(f" FAILED {ds_id}/{group_name}") | |
| continue | |
| msg = result.result.message | |
| raw = "".join(getattr(b, "text", "") for b in msg.content) | |
| usage = {"input_tokens": msg.usage.input_tokens, | |
| "output_tokens": msg.usage.output_tokens} | |
| try: | |
| parsed = parse_json_response(raw) | |
| except Exception: | |
| print(f" parse failed {ds_id}/{group_name}") | |
| continue | |
| fields = FIELD_GROUPS[group_name]["fields"] | |
| extraction, evidence = {}, {} | |
| for field in fields: | |
| entry = parsed.get(field, {}) | |
| if isinstance(entry, dict): | |
| extraction[field] = entry.get("value") | |
| evidence[field] = entry.get("evidence") | |
| else: | |
| extraction[field] = entry | |
| evidence[field] = None | |
| group_results[ds_id][group_name] = { | |
| "extraction": extraction, "evidence": evidence, "usage": usage, | |
| } | |
| ok = 0 | |
| for ds_id, (group_texts, triage, _) in paper_data.items(): | |
| safe = ds_id.replace("/", "__") | |
| if (out_dir / f"{safe}.json").exists(): | |
| ok += 1 | |
| continue | |
| extraction: dict = {} | |
| evidence: dict = {} | |
| group_stats: dict = {} | |
| total_usage = {"input_tokens": 0, "output_tokens": 0} | |
| groups = group_results.get(ds_id, {}) | |
| for group_name in GROUP_ORDER: | |
| fields = FIELD_GROUPS[group_name]["fields"] | |
| if group_name in groups: | |
| rec = groups[group_name] | |
| extraction.update(rec["extraction"]) | |
| evidence.update(rec["evidence"]) | |
| total_usage["input_tokens"] += rec["usage"]["input_tokens"] | |
| total_usage["output_tokens"] += rec["usage"]["output_tokens"] | |
| group_stats[group_name] = { | |
| "section_chars": len(group_texts.get(group_name, "")), | |
| "tokens_in": rec["usage"]["input_tokens"], | |
| "tokens_out": rec["usage"]["output_tokens"], | |
| } | |
| else: | |
| for f in fields: | |
| extraction[f] = None | |
| evidence[f] = None | |
| group_stats[group_name] = {"skipped": True} | |
| extraction, enrichments = enrich(ds_id, extraction) | |
| details = validate_and_score(extraction, evidence, enrichments, triage) | |
| extra_meta = { | |
| "group_stats": group_stats, | |
| "enrichments": list(enrichments.keys()), | |
| } | |
| save_paper(out_dir, ds_id, cfg, extraction, details, total_usage, "batch", extra_meta) | |
| ok += 1 | |
| print(f" Batch pipeline done: {ok} papers written to {out_dir}") | |
| # ββ CLI βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def main(): | |
| p = argparse.ArgumentParser(description="CroissantMiner LEV (Locate-Extract-Verify) pipeline") | |
| p.add_argument("--backbone", default=None, | |
| help="(Legacy) single backbone for extractor; " | |
| "uses precomputed Phase-1 Gemini-Flash triage as locator. " | |
| "Mutually exclusive with --extractor-backbone.") | |
| p.add_argument("--locator-backbone", default=None, | |
| help="LLM backbone for the locate step (replaces precomputed triage). " | |
| "When set, --extractor-backbone must also be set.") | |
| p.add_argument("--extractor-backbone", default=None, | |
| help="LLM backbone for the 5 extract calls. " | |
| "When set, --locator-backbone must also be set.") | |
| p.add_argument("--dev-only", action="store_true", help="Process only the dev papers") | |
| p.add_argument("--paper", type=str, help="Process a single paper by dataset_id") | |
| p.add_argument("--batch", action="store_true", | |
| help="Use Anthropic batch API (sonnet-4-5 only); " | |
| "currently incompatible with --locator-backbone.") | |
| p.add_argument("--prompt-variant", default="v1", choices=["v1", "v2", "v3", "v4", "v5", "v6", "v7"], | |
| help="v1=round-0; v2=Fix A; v3=Fix B (accepted +0.028); v4=A+B; v5=B+D; v6=B + tighter section; v7=Fix B + 9-item implementation cleanup (audit 2026-05-02): exact-match locator, dedup sections, paper order, section metadata, retry on JSON fail, 8K tokens, verify+correct phase, no core-group hack.") | |
| args = p.parse_args() | |
| # CLI validation: either legacy --backbone alone, or both new flags together. | |
| using_mixed = args.locator_backbone is not None or args.extractor_backbone is not None | |
| if using_mixed: | |
| if args.locator_backbone is None or args.extractor_backbone is None: | |
| raise SystemExit( | |
| "--locator-backbone and --extractor-backbone must be set together; " | |
| "use --backbone alone for the legacy precomputed-triage path." | |
| ) | |
| if args.backbone is not None: | |
| raise SystemExit( | |
| "--backbone is mutually exclusive with --locator-backbone/--extractor-backbone." | |
| ) | |
| if args.batch: | |
| raise SystemExit( | |
| "--batch path not yet supported with mixed-backbone LLM locator." | |
| ) | |
| locator_cfg = dict(resolve_backbone(args.locator_backbone)) | |
| locator_cfg["backbone_key"] = args.locator_backbone | |
| extractor_cfg = dict(resolve_backbone(args.extractor_backbone)) | |
| extractor_cfg["backbone_key"] = args.extractor_backbone | |
| cfg = extractor_cfg # cfg semantics: the model used by extract_group() | |
| out_dir = (ROOT / "data" / "extractions" / | |
| f"agentic_lev_{locator_cfg['output_slug']}_{extractor_cfg['output_slug']}") | |
| backbone_label = (f"locator={args.locator_backbone} ({locator_cfg['name']}) | " | |
| f"extractor={args.extractor_backbone} ({extractor_cfg['name']})") | |
| else: | |
| if args.backbone is None: | |
| args.backbone = "sonnet-4-5" # restore legacy default | |
| cfg = dict(resolve_backbone(args.backbone)) | |
| cfg["backbone_key"] = args.backbone | |
| locator_cfg = None | |
| out_dir = output_dir_for("lev", args.backbone) | |
| backbone_label = f"backbone={args.backbone} ({cfg['name']}) locator=precomputed_phase1_triage" | |
| if not any(PHASE0_DIR.glob("*.json")): | |
| raise SystemExit("Locator-Extractor needs the section index in data/agentic/phase0/, which is not " | |
| "redistributed (it contains the papers' text). Build it with " | |
| "python scripts/agentic_phase0.py after downloading the PDFs.") | |
| if locator_cfg is None and not any(PHASE1_DIR.glob("*.json")): | |
| raise SystemExit("The --backbone path reads the stored triage in data/agentic/phase1/, which is missing.") | |
| if args.prompt_variant != "v1": | |
| out_dir = out_dir.parent / f"{out_dir.name}_{args.prompt_variant}" | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| if args.paper: | |
| papers = [args.paper] | |
| elif args.dev_only: | |
| papers = sorted(SPLIT["dev"]) | |
| else: | |
| papers = sorted(SPLIT["dev"] + SPLIT["test"]) | |
| print(f"{'=' * 70}") | |
| print(f"CROISSANTMINER LEV β {backbone_label}") | |
| print(f"Papers: {len(papers)} Output: {out_dir}") | |
| print(f"Architecture: {'1 locator + 5 specialist' if locator_cfg else '5 specialist'} calls per paper (core/collection/annotation/impact/processing)") | |
| print(f"{'=' * 70}") | |
| if args.batch: | |
| run_batch(cfg, papers, out_dir) | |
| else: | |
| run_sequential(cfg, papers, args, out_dir, locator_cfg=locator_cfg, | |
| prompt_variant=args.prompt_variant) | |
| if __name__ == "__main__": | |
| main() | |