75 lines
4.7 KiB
Python
75 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Protocol
|
|
|
|
|
|
@dataclass
|
|
class ArchaeologyFinding:
|
|
subject: str
|
|
confidence: str
|
|
evidence: list[str]
|
|
details: dict[str, Any]
|
|
|
|
|
|
class ModelProjectProfile(Protocol):
|
|
name: str
|
|
def archaeology(self, repository_path: str) -> dict[str, Any]: ...
|
|
def validate_evaluation_suite(self, repository_path: str, suite_reference: str) -> dict[str, Any]: ...
|
|
def training_command(self, recipe: dict[str, Any], output_directory: str) -> list[str]: ...
|
|
|
|
|
|
class GuardModelProfile:
|
|
name = "guard"
|
|
base_model = "Qwen/Qwen2.5-Coder-3B-Instruct"
|
|
|
|
def archaeology(self, repository_path: str) -> dict[str, Any]:
|
|
root = Path(repository_path)
|
|
findings: list[dict[str, Any]] = []
|
|
def add(subject: str, confidence: str, evidence: list[Path], **details: Any) -> None:
|
|
findings.append({"subject": subject, "confidence": confidence, "evidence": [str(item) for item in evidence], "details": details})
|
|
|
|
trainer = root / "scripts" / "05_finetune.py"
|
|
evaluator = root / "scripts" / "run_evmbench.py"
|
|
strict_scorer = root / "scripts" / "score_evmbench_strict.py"
|
|
add("base_model", "CONFIRMED" if trainer.exists() else "UNKNOWN", [trainer] if trainer.exists() else [], value=self.base_model)
|
|
add("training_entrypoint", "CONFIRMED" if trainer.exists() else "UNKNOWN", [trainer] if trainer.exists() else [], command="python scripts/05_finetune.py --model ... --training-data ... --output ...")
|
|
add("evaluation_entrypoint", "CONFIRMED" if evaluator.exists() else "UNKNOWN", [item for item in [evaluator, strict_scorer] if item.exists()], command="python scripts/run_evmbench.py ...", strict_scoring=bool(strict_scorer.exists()))
|
|
checkpoints = []
|
|
for adapter in sorted((root / "models").glob("**/adapter_model.*")) if (root / "models").exists() else []:
|
|
checkpoints.append({"name": adapter.parent.name, "reference": str(adapter.parent), "hash": self._hash_file(adapter), "adapter_type": "LORA", "confidence": "CONFIRMED"})
|
|
datasets = []
|
|
for manifest in sorted((root / "data" / "production_authorized").glob("*.json")) if (root / "data" / "production_authorized").exists() else []:
|
|
try:
|
|
payload = json.loads(manifest.read_text(encoding="utf-8"))
|
|
count = len(payload) if isinstance(payload, list) else payload.get("record_count")
|
|
except (OSError, json.JSONDecodeError):
|
|
count = None
|
|
datasets.append({"name": manifest.stem, "reference": str(manifest), "hash": self._hash_file(manifest), "record_count": count, "tags": ["imported", "training"]})
|
|
reports = [str(path) for path in sorted((root / "results").glob("qwen25_coder_3b_*") if (root / "results").exists() else [])]
|
|
benchmark = root / "evmbench"
|
|
add("benchmark_checkout", "CONFIRMED" if benchmark.exists() else "UNKNOWN", [benchmark] if benchmark.exists() else [], strict_scorer_exists=strict_scorer.exists())
|
|
return {"root_exists": root.exists(), "findings": findings, "checkpoints": checkpoints, "datasets": datasets, "historical_reports": reports, "benchmark_reference": str(benchmark), "trainer_reference": str(trainer), "evaluator_reference": str(evaluator), "strict_scorer_reference": str(strict_scorer)}
|
|
|
|
def validate_evaluation_suite(self, repository_path: str, suite_reference: str) -> dict[str, Any]:
|
|
root = Path(repository_path)
|
|
evaluator = root / "scripts" / "run_evmbench.py"
|
|
scorer = root / "scripts" / "score_evmbench_strict.py"
|
|
benchmark = Path(suite_reference)
|
|
available = evaluator.exists() and scorer.exists() and benchmark.exists()
|
|
return {"status": "WARNING" if available else "BLOCKED", "evidence": {"evaluator": str(evaluator), "strict_scorer": str(scorer), "benchmark": str(benchmark), "reason": "Scripts and checkout are present, but no fresh Spark baseline has yet proven model loading, output parsing, reproducibility, or contamination checks." if available else "Guard evaluation harness or strict scorer missing."}}
|
|
|
|
def training_command(self, recipe: dict[str, Any], output_directory: str) -> list[str]:
|
|
dataset = str(recipe["training_data"])
|
|
return ["python", "scripts/05_finetune.py", "--model", str(recipe.get("base_model", self.base_model)), "--training-data", dataset, "--output", output_directory]
|
|
|
|
@staticmethod
|
|
def _hash_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|