569 lines
22 KiB
Python
569 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Run and validate the maintained V16 extraction baseline on a real PDF."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, List
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
PC_REQUIRED_FILES = (
|
|
"scientific_chunks.json",
|
|
"gene_alias.json",
|
|
"all_dic.json",
|
|
"relationships_by_category_final.json",
|
|
)
|
|
|
|
TABLE_READER_MODULES = {
|
|
"pandas": "pandas",
|
|
"xlsx": "openpyxl",
|
|
"xls": "xlrd",
|
|
"docx": "docx",
|
|
}
|
|
|
|
|
|
CPU_MINERU_ENV = {
|
|
"CUDA_VISIBLE_DEVICES": "",
|
|
"MINERU_MODEL_SOURCE": "modelscope",
|
|
"TORCH_CUDNN_V8_API_DISABLED": "1",
|
|
"V15_MINERU_BACKEND": "pipeline",
|
|
"V15_MINERU_METHOD": "txt",
|
|
"V15_MINERU_LANG": "en",
|
|
"V15_MINERU_USE_API": "0",
|
|
"V15_MINERU_FORMULA": "0",
|
|
"V15_MINERU_TABLE": "0",
|
|
}
|
|
|
|
|
|
def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Run V16Extractor on a real PDF and validate baseline invariants."
|
|
)
|
|
parser.add_argument("--pdf", required=True, type=Path, help="Input PDF path.")
|
|
parser.add_argument("--pmid", default=None, help="PMID passed to the extractor.")
|
|
parser.add_argument("--doi", default=None, help="Optional DOI override.")
|
|
parser.add_argument("--output-dir", required=True, type=Path, help="Extraction output directory.")
|
|
parser.add_argument(
|
|
"--cpu-mineru",
|
|
action="store_true",
|
|
help="Force low-memory MinerU pipeline settings and disable CUDA visibility.",
|
|
)
|
|
parser.add_argument("--no-llm", action="store_true", help="Keep LLM relation extraction off.")
|
|
parser.add_argument("--no-vlm", action="store_true", help="Keep VLM image extraction off.")
|
|
parser.add_argument("--with-llm", action="store_true", help="Enable LLM relation extraction when keys are configured.")
|
|
parser.add_argument(
|
|
"--with-llm-definitions",
|
|
action="store_true",
|
|
help="Also generate per-entity LLM definitions. Off by default for regression stability.",
|
|
)
|
|
parser.add_argument("--with-vlm", action="store_true", help="Enable VLM image extraction.")
|
|
parser.add_argument("--no-pc-data", action="store_true", help="Disable PC data layer.")
|
|
parser.add_argument(
|
|
"--extractor-version",
|
|
choices=("v16_1_2", "v16_1_1"),
|
|
default="v16_1_2",
|
|
help="Extractor package to validate. Defaults to the latest maintained version.",
|
|
)
|
|
parser.add_argument(
|
|
"--pc-data-root",
|
|
type=Path,
|
|
default=None,
|
|
help="Override PC_DATA_ROOT for this validation run without editing .env.",
|
|
)
|
|
parser.add_argument(
|
|
"--split-data-root",
|
|
type=Path,
|
|
default=None,
|
|
help="Record/check the user-managed split-data root for this run. No data processing is performed.",
|
|
)
|
|
parser.add_argument(
|
|
"--require-pc-data",
|
|
action="store_true",
|
|
help="Fail if PC required files are missing or PC loader counts are zero.",
|
|
)
|
|
return parser.parse_args(list(argv) if argv is not None else None)
|
|
|
|
|
|
def load_env_file(env_path: Path = PROJECT_ROOT / ".env") -> None:
|
|
if not env_path.exists():
|
|
return
|
|
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|
|
|
|
|
|
def apply_runtime_environment(cpu_mineru: bool = False) -> Dict[str, str]:
|
|
applied: Dict[str, str] = {}
|
|
if cpu_mineru:
|
|
for key, value in CPU_MINERU_ENV.items():
|
|
os.environ[key] = value
|
|
applied[key] = value
|
|
|
|
mineru_cmd = os.environ.get("MINERU_CMD", "").strip()
|
|
if mineru_cmd and not os.environ.get("V15_MINERU_CMD"):
|
|
os.environ["V15_MINERU_CMD"] = mineru_cmd
|
|
applied["V15_MINERU_CMD"] = mineru_cmd
|
|
|
|
return applied
|
|
|
|
|
|
def _resolve_path(value: str | Path | None, default: str | Path | None = None) -> Path | None:
|
|
raw_value = str(value).strip() if value is not None else ""
|
|
raw_default = str(default).strip() if default is not None else ""
|
|
chosen = raw_value or raw_default
|
|
if not chosen:
|
|
return None
|
|
path = Path(chosen)
|
|
if not path.is_absolute():
|
|
path = PROJECT_ROOT / path
|
|
return path.resolve()
|
|
|
|
|
|
def apply_external_data_overrides(args: argparse.Namespace) -> Dict[str, str]:
|
|
applied: Dict[str, str] = {}
|
|
if args.pc_data_root is not None:
|
|
pc_root = _resolve_path(args.pc_data_root)
|
|
if pc_root is not None:
|
|
os.environ["PC_DATA_ROOT"] = str(pc_root)
|
|
applied["PC_DATA_ROOT"] = str(pc_root)
|
|
if args.split_data_root is not None:
|
|
split_root = _resolve_path(args.split_data_root)
|
|
if split_root is not None:
|
|
os.environ["PLANTAI_SPLIT_DATA_ROOT"] = str(split_root)
|
|
applied["PLANTAI_SPLIT_DATA_ROOT"] = str(split_root)
|
|
return applied
|
|
|
|
|
|
def inspect_data_root(root: Path | None, required_files: Iterable[str] = ()) -> Dict[str, Any]:
|
|
required = list(required_files)
|
|
if root is None:
|
|
return {
|
|
"configured": False,
|
|
"root": "",
|
|
"exists": False,
|
|
"is_dir": False,
|
|
"readable": False,
|
|
"required_files": {name: False for name in required},
|
|
"missing_required_files": required,
|
|
"available": False,
|
|
}
|
|
|
|
file_status = {name: (root / name).is_file() for name in required}
|
|
missing = [name for name, exists in file_status.items() if not exists]
|
|
exists = root.exists()
|
|
is_dir = root.is_dir()
|
|
readable = os.access(root, os.R_OK) if exists else False
|
|
return {
|
|
"configured": True,
|
|
"root": str(root),
|
|
"exists": exists,
|
|
"is_dir": is_dir,
|
|
"readable": readable,
|
|
"required_files": file_status,
|
|
"missing_required_files": missing,
|
|
"available": exists and is_dir and readable and not missing,
|
|
}
|
|
|
|
|
|
def inspect_external_data(args: argparse.Namespace) -> Dict[str, Any]:
|
|
pc_root = _resolve_path(args.pc_data_root, os.environ.get("PC_DATA_ROOT") or "data/pc")
|
|
split_root = _resolve_path(args.split_data_root, os.environ.get("PLANTAI_SPLIT_DATA_ROOT"))
|
|
return {
|
|
"pc": inspect_data_root(pc_root, PC_REQUIRED_FILES),
|
|
"split": inspect_data_root(split_root),
|
|
}
|
|
|
|
|
|
def inspect_table_reader_dependencies() -> Dict[str, Any]:
|
|
modules = {
|
|
label: importlib.util.find_spec(module_name) is not None
|
|
for label, module_name in TABLE_READER_MODULES.items()
|
|
}
|
|
return {
|
|
"modules": modules,
|
|
"csv_available": modules["pandas"],
|
|
"tsv_available": modules["pandas"],
|
|
"xlsx_available": modules["pandas"] and modules["xlsx"],
|
|
"xls_available": modules["pandas"] and modules["xls"],
|
|
"docx_available": modules["pandas"] and modules["docx"],
|
|
"missing_modules": [label for label, available in modules.items() if not available],
|
|
}
|
|
|
|
|
|
def preflight_configuration(args: argparse.Namespace, external_data: Dict[str, Any]) -> List[str]:
|
|
failures: List[str] = []
|
|
if args.with_llm and not args.no_llm and not os.environ.get("DEEPSEEK_API_KEY", "").strip():
|
|
failures.append("DEEPSEEK_API_KEY is required when --with-llm is used")
|
|
if args.with_vlm and not args.no_vlm and not os.environ.get("DASHSCOPE_API_KEY", "").strip():
|
|
failures.append("DASHSCOPE_API_KEY is required when --with-vlm is used")
|
|
if args.require_pc_data:
|
|
pc_info = external_data.get("pc") or {}
|
|
if not pc_info.get("available"):
|
|
failures.append(f"PC data is required but unavailable: {pc_info.get('root', '')}")
|
|
for file_name in pc_info.get("missing_required_files") or []:
|
|
failures.append(f"PC required file missing: {file_name}")
|
|
return failures
|
|
|
|
|
|
def _count_items(result: Dict[str, Any], key: str) -> int:
|
|
value = result.get(key) or []
|
|
return len(value) if isinstance(value, list) else 0
|
|
|
|
|
|
def _resolve_audit_path(path_value: str | Path | None, output_dir: Path) -> Path | None:
|
|
if not path_value:
|
|
return None
|
|
path = Path(str(path_value))
|
|
if not path.is_absolute():
|
|
path = output_dir / path
|
|
return path
|
|
|
|
|
|
def _count_jsonl_rows(path: Path) -> int:
|
|
if not path.is_file():
|
|
return 0
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
return sum(1 for line in handle if line.strip())
|
|
|
|
|
|
def _count_csv_data_rows(path: Path) -> int:
|
|
if not path.is_file():
|
|
return 0
|
|
with path.open("r", encoding="utf-8", newline="") as handle:
|
|
return sum(1 for _row in csv.DictReader(handle))
|
|
|
|
|
|
def summarize_run_request(args: argparse.Namespace) -> Dict[str, Any]:
|
|
pdf_path = args.pdf.resolve()
|
|
output_dir = args.output_dir.resolve()
|
|
use_llm = bool(args.with_llm and not args.no_llm)
|
|
use_vlm = bool(args.with_vlm and not args.no_vlm)
|
|
return {
|
|
"scope": "single_real_pdf",
|
|
"pdf_path": str(pdf_path),
|
|
"pdf_exists": pdf_path.is_file(),
|
|
"pdf_size_bytes": pdf_path.stat().st_size if pdf_path.is_file() else 0,
|
|
"pmid": args.pmid or "",
|
|
"doi": args.doi or "",
|
|
"output_dir": str(output_dir),
|
|
"extractor_version": args.extractor_version,
|
|
"cpu_mineru": bool(args.cpu_mineru),
|
|
"use_llm_relations": use_llm,
|
|
"use_llm_definitions": bool(args.with_llm_definitions and use_llm),
|
|
"use_vlm": use_vlm,
|
|
"use_pc_data": not args.no_pc_data,
|
|
"require_pc_data": bool(args.require_pc_data),
|
|
}
|
|
|
|
|
|
def _read_llm_raw_response_rows(path: Path) -> List[Dict[str, Any]]:
|
|
if not path.is_file():
|
|
return []
|
|
rows: List[Dict[str, Any]] = []
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
raw = line.strip()
|
|
if not raw:
|
|
continue
|
|
try:
|
|
row = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
rows.append(
|
|
{
|
|
"window_id": "",
|
|
"section": "",
|
|
"status": "parse_error",
|
|
"error": "raw response audit row is not valid JSON",
|
|
}
|
|
)
|
|
continue
|
|
if isinstance(row, dict):
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def collect_llm_runtime_diagnostics(
|
|
output_dir: Path,
|
|
llm_metadata: Dict[str, Any] | None = None,
|
|
) -> Dict[str, Any]:
|
|
"""Summarize LLM runtime stability without copying prompt/response text."""
|
|
llm_metadata = llm_metadata or {}
|
|
audit_files = llm_metadata.get("audit_files") or {}
|
|
raw_candidates = [
|
|
_resolve_audit_path(audit_files.get("llm1_raw_response"), output_dir),
|
|
_resolve_audit_path(audit_files.get("llm_relation_raw_responses"), output_dir),
|
|
output_dir / "llm1_raw_response.jsonl",
|
|
output_dir / "llm_relation_raw_responses.jsonl",
|
|
]
|
|
raw_response_path = next((path for path in raw_candidates if path and path.is_file()), None)
|
|
raw_rows = _read_llm_raw_response_rows(raw_response_path) if raw_response_path else []
|
|
|
|
status_counts = Counter(str(row.get("status") or "unknown") for row in raw_rows)
|
|
error_rows = [row for row in raw_rows if str(row.get("status") or "").lower() != "ok"]
|
|
timeout_rows = [
|
|
row
|
|
for row in error_rows
|
|
if "timed out" in str(row.get("error") or "").lower()
|
|
or "timeout" in str(row.get("error") or "").lower()
|
|
]
|
|
error_windows = [
|
|
{
|
|
"window_id": str(row.get("window_id") or ""),
|
|
"section": str(row.get("section") or ""),
|
|
"error": str(row.get("error") or "")[:240],
|
|
}
|
|
for row in error_rows[:50]
|
|
]
|
|
|
|
candidate_files = sorted(output_dir.glob("*.llm1_candidate_relations.jsonl"))
|
|
rejected_files = sorted(output_dir.glob("*.llm2_rejected_relations.csv"))
|
|
|
|
return {
|
|
"enabled": bool(llm_metadata.get("enabled", False)),
|
|
"client_available": bool(llm_metadata.get("client_available", False)),
|
|
"metadata_llm_requests": int(llm_metadata.get("llm_requests") or 0),
|
|
"metadata_llm_failures": int(llm_metadata.get("llm_failures") or 0),
|
|
"raw_response_file": str(raw_response_path) if raw_response_path else "",
|
|
"raw_response_rows": len(raw_rows),
|
|
"raw_response_status_counts": dict(status_counts),
|
|
"raw_response_error_count": len(error_rows),
|
|
"timeout_error_count": len(timeout_rows),
|
|
"error_windows": error_windows,
|
|
"error_windows_truncated": len(error_rows) > len(error_windows),
|
|
"candidate_relation_files": [str(path) for path in candidate_files],
|
|
"candidate_relation_rows": sum(_count_jsonl_rows(path) for path in candidate_files),
|
|
"rejected_candidate_files": [str(path) for path in rejected_files],
|
|
"rejected_candidate_rows": sum(_count_csv_data_rows(path) for path in rejected_files),
|
|
}
|
|
|
|
|
|
def summarize_result(
|
|
result: Dict[str, Any],
|
|
output_dir: Path,
|
|
external_data: Dict[str, Any] | None = None,
|
|
table_readers: Dict[str, Any] | None = None,
|
|
) -> Dict[str, Any]:
|
|
relations = result.get("relations") or []
|
|
entities = result.get("entities") or []
|
|
sections = result.get("sections") or []
|
|
chunks = result.get("chunks") or []
|
|
metadata = result.get("metadata") or {}
|
|
|
|
missing_source = sum(1 for rel in relations if not str(rel.get("source", "")).strip())
|
|
missing_target = sum(1 for rel in relations if not str(rel.get("target", "")).strip())
|
|
missing_evidence = sum(1 for rel in relations if not str(rel.get("evidence", "")).strip())
|
|
|
|
relation_types = Counter(
|
|
str(rel.get("relationship_resolved") or rel.get("relation") or rel.get("relation_type") or "unknown")
|
|
for rel in relations
|
|
)
|
|
output_layers = Counter(str(rel.get("output_layer") or "") for rel in relations)
|
|
|
|
external_data = external_data or {}
|
|
table_readers = table_readers or inspect_table_reader_dependencies()
|
|
pc_data = external_data.get("pc") or {}
|
|
split_data = external_data.get("split") or {}
|
|
|
|
llm_relation_extraction = metadata.get("llm_relation_extraction", {})
|
|
llm_runtime_diagnostics = collect_llm_runtime_diagnostics(
|
|
output_dir,
|
|
llm_relation_extraction,
|
|
)
|
|
|
|
return {
|
|
"output_dir": str(output_dir),
|
|
"result_json": str(output_dir / "result.json"),
|
|
"entities": len(entities),
|
|
"relations": len(relations),
|
|
"sections": len(sections),
|
|
"chunks": len(chunks),
|
|
"missing_source": missing_source,
|
|
"missing_target": missing_target,
|
|
"missing_evidence": missing_evidence,
|
|
"relation_types_top": relation_types.most_common(20),
|
|
"output_layers": dict(output_layers),
|
|
"pc_data_stats": metadata.get("pc_data_stats", {}),
|
|
"relation_block_extraction": metadata.get("relation_block_extraction", {}),
|
|
"llm_relation_extraction": llm_relation_extraction,
|
|
"llm_runtime_diagnostics": llm_runtime_diagnostics,
|
|
"pc_data_root": pc_data.get("root", ""),
|
|
"pc_data_available": bool(pc_data.get("available", False)),
|
|
"split_data_root": split_data.get("root", ""),
|
|
"split_data_available": bool(split_data.get("available", False)),
|
|
"external_data": external_data,
|
|
"supplementary_table_readers": table_readers,
|
|
"version": metadata.get("version"),
|
|
"extractor_version": metadata.get("extractor_version"),
|
|
}
|
|
|
|
|
|
def validate_summary(
|
|
summary: Dict[str, Any],
|
|
require_pc_data: bool = False,
|
|
require_llm_relations: bool = False,
|
|
require_run_request: bool = False,
|
|
) -> List[str]:
|
|
failures: List[str] = []
|
|
for key in ("sections", "entities", "relations", "chunks"):
|
|
if int(summary.get(key) or 0) <= 0:
|
|
failures.append(f"{key} must be > 0")
|
|
for key in ("missing_source", "missing_target", "missing_evidence"):
|
|
if int(summary.get(key) or 0) != 0:
|
|
failures.append(f"{key} must be 0")
|
|
result_json = Path(str(summary.get("result_json") or ""))
|
|
if not result_json.exists():
|
|
failures.append(f"result.json not found: {result_json}")
|
|
if require_run_request:
|
|
request = summary.get("run_request") or {}
|
|
if not isinstance(request, dict) or not request:
|
|
failures.append("run_request must be recorded for real validation runs")
|
|
else:
|
|
if request.get("scope") != "single_real_pdf":
|
|
failures.append("run_request.scope must be single_real_pdf")
|
|
if not str(request.get("pdf_path") or "").strip():
|
|
failures.append("run_request.pdf_path must be recorded")
|
|
if not request.get("pdf_exists"):
|
|
failures.append("run_request.pdf_exists must be true")
|
|
if int(request.get("pdf_size_bytes") or 0) <= 0:
|
|
failures.append("run_request.pdf_size_bytes must be > 0")
|
|
if request.get("extractor_version") not in {"v16_1_1", "v16_1_2"}:
|
|
failures.append("run_request.extractor_version must be a maintained V16 version")
|
|
if not str(request.get("output_dir") or "").strip():
|
|
failures.append("run_request.output_dir must be recorded")
|
|
if require_pc_data:
|
|
pc_info = (summary.get("external_data") or {}).get("pc") or {}
|
|
missing_files = pc_info.get("missing_required_files") or []
|
|
if not pc_info.get("available"):
|
|
failures.append(f"PC data is required but unavailable: {pc_info.get('root', '')}")
|
|
for file_name in missing_files:
|
|
failures.append(f"PC required file missing: {file_name}")
|
|
|
|
pc_stats = summary.get("pc_data_stats") or {}
|
|
for key in (
|
|
"alias_map_loaded",
|
|
"gene_aliases_loaded",
|
|
"relationship_patterns_loaded",
|
|
"entity_dict_loaded",
|
|
):
|
|
if int(pc_stats.get(key) or 0) <= 0:
|
|
failures.append(f"pc_data_stats.{key} must be > 0 when --require-pc-data is used")
|
|
if require_llm_relations:
|
|
block = summary.get("relation_block_extraction") or {}
|
|
if not block.get("enabled"):
|
|
failures.append("LLM relation block extraction must be enabled when --with-llm is used")
|
|
if int(block.get("chunk_count") or 0) <= 0:
|
|
failures.append("relation_block_extraction.chunk_count must be > 0 when --with-llm is used")
|
|
if int(block.get("relation_count_before_gate") or 0) <= 0:
|
|
failures.append(
|
|
"relation_block_extraction.relation_count_before_gate must be > 0 when --with-llm is used"
|
|
)
|
|
return failures
|
|
|
|
|
|
def run_extraction(args: argparse.Namespace) -> Dict[str, Any]:
|
|
if args.extractor_version == "v16_1_1":
|
|
from src.rag.extractors.v16_1_1 import V16Extractor
|
|
else:
|
|
from src.rag.extractors.v16_1_2 import V16Extractor
|
|
|
|
output_dir = args.output_dir.resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
use_llm = bool(args.with_llm and not args.no_llm)
|
|
use_llm_definitions = bool(args.with_llm_definitions and use_llm)
|
|
use_vlm = bool(args.with_vlm and not args.no_vlm)
|
|
|
|
extractor = V16Extractor(
|
|
use_vlm=use_vlm,
|
|
use_mineru=True,
|
|
output_dir=output_dir,
|
|
use_pc_data=not args.no_pc_data,
|
|
use_disambiguation=True,
|
|
use_llm_definitions=use_llm_definitions,
|
|
use_llm_relations=use_llm,
|
|
use_rule_relation_fallback=args.extractor_version == "v16_1_1",
|
|
use_neo4j=False,
|
|
use_unified_storage=False,
|
|
)
|
|
return extractor.extract(args.pdf.resolve(), pmid=args.pmid, doi=args.doi)
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
load_env_file()
|
|
external_overrides = apply_external_data_overrides(args)
|
|
applied_env = apply_runtime_environment(cpu_mineru=args.cpu_mineru)
|
|
applied_env.update(external_overrides)
|
|
|
|
if not args.pdf.exists():
|
|
print(f"ERROR: input PDF does not exist: {args.pdf}", file=sys.stderr)
|
|
return 2
|
|
if args.no_pc_data and args.require_pc_data:
|
|
print("ERROR: --no-pc-data conflicts with --require-pc-data", file=sys.stderr)
|
|
return 2
|
|
|
|
external_data = inspect_external_data(args)
|
|
table_readers = inspect_table_reader_dependencies()
|
|
preflight_failures = preflight_configuration(args, external_data)
|
|
if preflight_failures:
|
|
print("VALIDATION_PREFLIGHT_FAILED")
|
|
for failure in preflight_failures:
|
|
print(f"- {failure}")
|
|
return 2
|
|
|
|
result = run_extraction(args)
|
|
summary = summarize_result(
|
|
result,
|
|
args.output_dir.resolve(),
|
|
external_data=external_data,
|
|
table_readers=table_readers,
|
|
)
|
|
summary["run_request"] = summarize_run_request(args)
|
|
summary["applied_env"] = {
|
|
key: ("<empty>" if value == "" else value)
|
|
for key, value in applied_env.items()
|
|
if "KEY" not in key and "PASSWORD" not in key and "TOKEN" not in key
|
|
}
|
|
summary_path = args.output_dir.resolve() / "validation_summary.json"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
require_llm_relations = bool(
|
|
args.with_llm and not args.no_llm and args.extractor_version == "v16_1_2"
|
|
)
|
|
failures = validate_summary(
|
|
summary,
|
|
require_pc_data=args.require_pc_data,
|
|
require_llm_relations=require_llm_relations,
|
|
require_run_request=True,
|
|
)
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
print(f"VALIDATION_SUMMARY_JSON {summary_path}")
|
|
if failures:
|
|
print("VALIDATION_FAILED")
|
|
for failure in failures:
|
|
print(f"- {failure}")
|
|
return 2
|
|
print("VALIDATION_PASSED")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|