| """Tier B: sparse black-box validation/refinement on OpenRouter. |
| |
| Validates the top-K gate-passing candidates against the real target models, |
| scoring by embedding distance-to-truth (aggregated across models). Uses a |
| budget accountant; every query is gated by `budget.can_spend()`. |
| |
| A binary-comparison hill-climb ("which image would you more likely call X?") |
| can further separate near-ties without absolute-confidence prompts, but the |
| default path is direct label + distance scoring which is cheaper and monotonic. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from PIL import Image |
|
|
| from veil_pgd.config import Settings |
| from veil_pgd.fitness.embed import Embedder |
| from veil_pgd.fitness.semantic import aggregate, embedding_distance |
| from veil_pgd.optimizer.budget import Budget |
| from veil_pgd.render import render |
| from veil_pgd.robustness import scraper_sim |
| from veil_pgd.targets.base import LabelPrompt |
| from veil_pgd.types import Candidate |
|
|
|
|
| class TierB: |
| def __init__(self, settings: Settings, blackbox_targets: list, embedder: Embedder): |
| self.s = settings |
| self.targets = blackbox_targets |
| self.embedder = embedder |
| self.prompt = LabelPrompt() |
|
|
| def validate( |
| self, image: Image.Image, truth: str, candidates: list[Candidate] |
| ) -> list[Candidate]: |
| budget = Budget( |
| max_queries=self.s.budget.max_blackbox_queries_per_image, |
| usd_cap=self.s.budget.hard_usd_cap_per_image, |
| ) |
|
|
| for cand in candidates: |
| rendered = render(image, cand.spec) |
| probe = scraper_sim(rendered) |
| dists: list[float] = [] |
| preds: dict[str, str] = {} |
| for t in self.targets: |
| if not budget.can_spend(1): |
| break |
| res = t.label(probe, self.prompt) |
| budget.record(t.name, res.cost_usd, 1) |
| preds[t.name] = res.parsed_label |
| if res.refused or res.error: |
| dists.append(0.0) |
| else: |
| dists.append(embedding_distance(self.embedder, res.parsed_label, truth)) |
| cand.blackbox_fitness = aggregate(dists, self.s.aggregation) |
| cand.per_model_distance.update({k: d for k, d in zip(preds, dists)}) |
| cand.notes["blackbox_preds"] = preds |
| cand.notes["budget"] = { |
| "queries": budget.queries_used, |
| "usd": round(budget.usd_spent, 6), |
| } |
|
|
| candidates.sort(key=lambda c: c.blackbox_fitness, reverse=True) |
| return candidates |
|
|
