andrew/physicalCombinatorics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

drop safety/availability from scoring, holistic p4 rating, phase-parallel pipeline

Browse Source 
safety and availability don't reduce to physics formulas the way
power_density/range_fuel/cost_efficiency do -- they're judgment calls
(risk assessment, infrastructure prevalence), and running them through the
same log-normalize() built for physical quantities produced incoherent
results: safety's raw value is already a "0-1" score, and normalizing it
again turned 0.6 into an unexplainable 0.678 that even the LLM reviewing
it could only cite, never justify (see combo 1540). Removed both from
domain_metric_weights (safety from 4 domains, availability from
urban_commuting) and renormalized the remaining weights to sum to 1.0.

Pass 4 now produces one holistic RATING (LOW/MEDIUM/HIGH) alongside the
existing VERDICT, with safety and accessibility folded in as qualitative
considerations feeding that single judgment rather than scored
separately -- not a checklist of independent numbers. New
qualitative_rating column, filterable in the results UI. Also added
domain name/description to the review prompt so the LLM judges a metric
like range against what the domain actually needs (urban_commuting:
1-50km) instead of generic real-world expectations for the platform
category -- confirmed live on a combo where phi4 had called a 396km range
"limited" by comparing to typical aircraft rather than a domain that
needs 1-50km.

Pass 2 is estimator-only now -- self.llm is never consulted there,
reserved entirely for pass 4. Restructured Pipeline.run() from combo-first
to phase-parallel: each pass now runs to completion across every combo
before the next pass starts, rather than walking each combo through all
four passes before the next combo. This surfaced a real bug: domain-
blocked combos (status stays "valid" by design, not "_fail") were
slipping past a naive status-based skip guard and getting silently
re-processed by pass 2. Fixed with a shared dead-combo check that catches
both generic failures and domain blocks correctly.

Also fixes a results-page display bug found while reviewing a live combo:
the per-metric "position" bar showed raw distance from norm_min without
inverting for lower_is_better metrics, so an excellent cost score (near
the good end) rendered as a ~0%, near-empty bar -- looked bad next to its
own 0.99 normalized score.

Validated live against phi4 (real Ollama calls, not mocked): full-domain
phase-parallel run (2,970 combos, estimator-only p2, 1.6s) followed by a
real pass-4 run (111 reviewed, 11m, 0 crashes, 0 null ratings). Two tests
that relied on the old LLM-driven pass 2 to force deterministic outcomes
were updated to test pass 4's verdict-wiring directly instead. All 100
tests pass.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Andrew Simonson
2026-08-15 16:52:26 -05:00
parent 730a23bac3
commit 76f460499a
16 changed files with 569 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files

607
src/physcom/engine/pipeline.py
View File

@@ -12,6 +12,7 @@ from physcom.engine.combinator import generate_combinations
from physcom.engine.constraint_resolver import ConstraintResolver, ConstraintResult
from physcom.engine.scorer import Scorer
from physcom.llm.base import LLMProvider, LLMRateLimitError
from physcom.llm.parsing import parse_rating
from physcom.models.combination import Combination, ScoredResult
from physcom.models.domain import Domain, MetricBound
@@ -385,308 +386,57 @@ class Pipeline:
# Prepare metric lookup
bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
# ── Combo-first loop ─────────────────────────────────────
# Deterministic passes (1, 3, and 2 without an LLM) defer commits and
# get flushed periodically + in `finally` below -- a crash there costs
# a cheap recompute, not lost work worth committing per write. Pass 4
# (and pass 2 with an LLM) commit immediately after each call: those
# are slow and crash-prone (see the QwQ timeout saga), so that result
# is worth protecting the moment it lands.
# ── Phase-parallel: each pass runs to completion across every combo
# before the next pass starts, instead of walking each combo through
# every pass before moving to the next combo. This maximizes
# progress before the expensive/slow phase (pass 4's LLM calls) and
# keeps that phase's cost visible on its own, separate from the
# deterministic passes. It also removes any need for the two options
# to reconcile: pass 2 is estimator-only now (no LLM call in it at
# all -- self.llm is reserved for pass 4), so there's no combo that
# touches an LLM in both pass 2 and pass 4, and nothing here needs a
# live/resumed conversation across passes.
#
# Deterministic passes (1, 2, 3) defer commits and get flushed
# periodically + in `finally` below -- a crash there costs a cheap
# recompute, not lost work worth committing per write. Pass 4
# commits immediately after each call: those are slow and
# crash-prone (see the QwQ timeout saga), so that result is worth
# protecting the moment it lands.
combos_since_commit = 0
def _tick_commit() -> None:
nonlocal combos_since_commit
combos_since_commit += 1
if combos_since_commit >= 200:
self.repo.commit()
combos_since_commit = 0
try:
for combo in combos:
self._check_cancelled(run_id)
combos_since_commit += 1
if combos_since_commit >= 200:
self.repo.commit()
combos_since_commit = 0
if 1 in passes:
for combo in combos:
self._check_cancelled(run_id)
_tick_commit()
self._process_pass1(combo, domain, result, run_id)
# Check existing progress for this combo in this domain
existing_pass = self.repo.get_combo_pass_reached(
combo.id, domain.id
) or 0
if 2 in passes:
for combo in combos:
self._check_cancelled(run_id)
_tick_commit()
self._process_pass2(combo, domain, bounds_by_name, result, run_id)
# Load existing result to preserve human review data
existing_result = self.repo.get_existing_result(
combo.id, domain.id
)
# ── Pass 1: Constraint Resolution ────────────────
if 1 in passes and existing_pass < 1:
cr: ConstraintResult = self.resolver.resolve(combo)
if cr.status == "p1_fail":
combo.status = "p1_fail"
combo.block_reason = "; ".join(cr.violations)
self.repo.update_combination_status(
combo.id, "p1_fail", combo.block_reason, commit=False
)
# Save a result row so failed combos appear in results
self.repo.save_result(
combo.id,
domain.id,
composite_score=0.0,
pass_reached=1,
commit=False,
)
result.pass1_failed += 1
self._update_run_counters(run_id, result, current_pass=1)
continue # p1_fail — skip remaining passes
else:
combo.status = "valid"
self.repo.update_combination_status(combo.id, "valid", commit=False)
# Domain constraint check (per-domain block only). combo.status
# stays "valid" here on purpose: it's domain-agnostic and the
# same combo can be blocked in this domain but valid in another.
# The per-domain block lives on combination_results.domain_block_reason
# (see count_combinations_by_status / get_all_results, which bucket on it).
if domain.constraints:
dc_result = self.resolver.check_domain_constraints(
combo, domain.constraints
)
if dc_result.status == "p1_fail":
self.repo.save_result(
combo.id, domain.id,
composite_score=0.0, pass_reached=1,
domain_block_reason="; ".join(
dc_result.violations
),
commit=False,
)
result.pass1_failed += 1
self._update_run_counters(
run_id, result, current_pass=1
)
continue
if cr.status == "conditional":
result.pass1_conditional += 1
else:
result.pass1_valid += 1
self._update_run_counters(run_id, result, current_pass=1)
elif 1 in passes:
# Already pass1'd — check if it failed
if combo.status.endswith("_fail"):
result.pass1_failed += 1
continue
# Check if domain-blocked from a prior run
if existing_result and existing_result["pass_reached"] == 1:
result.pass1_failed += 1
continue
result.pass1_valid += 1
else:
# Pass 1 not requested; check if failed from a prior run
if combo.status.endswith("_fail"):
result.pass1_failed += 1
continue
# ── Pass 2: Physics Estimation ───────────────────
raw_metrics: dict[str, float] = {}
if 2 in passes and existing_pass < 2:
description = _describe_combination(combo)
if self.llm:
raw_metrics = self.llm.estimate_physics(
description, domain.metric_bounds
)
else:
raw_metrics = self._stub_estimate(combo, domain.metric_bounds)
# Save raw estimates immediately (crash-safe)
estimate_dicts = []
for mname, rval in raw_metrics.items():
mb = bounds_by_name.get(mname)
if mb and mb.metric_id:
estimate_dicts.append({
"metric_id": mb.metric_id,
"raw_value": rval,
"estimation_method": "llm" if self.llm else "stub",
"confidence": 1.0,
})
# LLM-produced estimates commit immediately (slow/crash-
# prone, worth protecting); stub estimates are instant
# and defer, same as the rest of the deterministic passes.
used_llm = self.llm is not None
if estimate_dicts:
self.repo.save_raw_estimates(
combo.id, domain.id, estimate_dicts, commit=used_llm
)
# Check for all-zero estimates → p2_fail
if raw_metrics and all(v == 0.0 for v in raw_metrics.values()):
combo.status = "p2_fail"
combo.block_reason = "All metric estimates are zero"
self.repo.update_combination_status(
combo.id, "p2_fail", combo.block_reason, commit=used_llm
)
self.repo.save_result(
combo.id, domain.id,
composite_score=0.0, pass_reached=2,
commit=used_llm,
)
result.pass2_failed += 1
self._update_run_counters(run_id, result, current_pass=2)
continue
result.pass2_estimated += 1
self._update_run_counters(run_id, result, current_pass=2)
elif 2 in passes:
# Already estimated — reload raw values from DB
existing_scores = self.repo.get_combination_scores(
combo.id, domain.id
)
raw_metrics = {
s["metric_name"]: s["raw_value"] for s in existing_scores
}
result.pass2_estimated += 1
else:
# Pass 2 not requested, use empty metrics
raw_metrics = {}
# ── Pass 3: Scoring & Ranking ────────────────────
if 3 in passes and existing_pass < 3:
sr = self.scorer.score_combination(combo, raw_metrics)
# Persist per-metric scores with normalized values
score_dicts = []
for s in sr.scores:
mb = bounds_by_name.get(s.metric_name)
if mb and mb.metric_id:
score_dicts.append({
"metric_id": mb.metric_id,
"raw_value": s.raw_value,
"normalized_score": s.normalized_score,
"estimation_method": s.estimation_method,
"confidence": s.confidence,
})
if score_dicts:
self.repo.save_scores(combo.id, domain.id, score_dicts, commit=False)
# Preserve existing human data
novelty_flag = (
existing_result["novelty_flag"] if existing_result else None
)
human_notes = (
existing_result["human_notes"] if existing_result else None
if 3 in passes:
for combo in combos:
self._check_cancelled(run_id)
_tick_commit()
self._process_pass3(
combo, domain, bounds_by_name, result, score_threshold, run_id
)
if sr.composite_score < score_threshold:
self.repo.save_result(
combo.id, domain.id,
sr.composite_score, pass_reached=3,
novelty_flag=novelty_flag,
human_notes=human_notes,
commit=False,
)
combo.status = "p3_fail"
combo.block_reason = (
f"Composite score {sr.composite_score:.4f} "
f"below threshold {score_threshold}"
)
self.repo.update_combination_status(
combo.id, "p3_fail", combo.block_reason, commit=False
)
result.pass3_failed += 1
result.pass3_scored += 1
self._update_run_counters(run_id, result, current_pass=3)
continue
self.repo.save_result(
combo.id,
domain.id,
sr.composite_score,
pass_reached=3,
novelty_flag=novelty_flag,
human_notes=human_notes,
commit=False,
)
self.repo.update_combination_status(combo.id, "scored", commit=False)
result.pass3_scored += 1
result.pass3_above_threshold += 1
self._update_run_counters(run_id, result, current_pass=3)
elif 3 in passes and existing_pass >= 3:
# Already scored — count it
result.pass3_scored += 1
if existing_result and existing_result["composite_score"] is not None:
if existing_result["composite_score"] >= score_threshold:
result.pass3_above_threshold += 1
# ── Pass 4: LLM Review ───────────────────────────
if 4 in passes and self.llm:
cur_pass = self.repo.get_combo_pass_reached(
combo.id, domain.id
) or 0
if cur_pass < 4:
cur_result = self.repo.get_existing_result(
combo.id, domain.id
)
if (
cur_result
and cur_result["composite_score"] is not None
and cur_result["composite_score"] >= score_threshold
):
description = _describe_combination(combo)
db_scores = self.repo.get_combination_scores(
combo.id, domain.id
)
score_dict = {
s["metric_name"]: s["normalized_score"]
for s in db_scores
if s["normalized_score"] is not None
}
raw_dict = {
s["metric_name"]: s["raw_value"]
for s in db_scores
if s["raw_value"] is not None
}
review_result: tuple[str, bool] | None = None
try:
review_result = self.llm.review_plausibility(
description, raw_dict, score_dict, domain.metric_bounds
)
except LLMRateLimitError as exc:
self._wait_for_rate_limit(run_id, exc.retry_after)
try:
review_result = self.llm.review_plausibility(
description, raw_dict, score_dict, domain.metric_bounds
)
except LLMRateLimitError:
pass # still limited; skip, retry next run
if review_result is not None:
review_text, plausible = review_result
if not plausible:
self.repo.save_result(
combo.id, domain.id,
cur_result["composite_score"],
pass_reached=4,
novelty_flag=cur_result.get("novelty_flag"),
llm_review=review_text,
human_notes=cur_result.get("human_notes"),
)
combo.status = "p4_fail"
combo.block_reason = "LLM deemed implausible"
self.repo.update_combination_status(
combo.id, "p4_fail", combo.block_reason
)
result.pass4_failed += 1
else:
self.repo.save_result(
combo.id, domain.id,
cur_result["composite_score"],
pass_reached=4,
novelty_flag=cur_result.get("novelty_flag"),
llm_review=review_text,
human_notes=cur_result.get("human_notes"),
)
self.repo.update_combination_status(
combo.id, "llm_reviewed"
)
result.pass4_reviewed += 1
self._update_run_counters(
run_id, result, current_pass=4
)
if 4 in passes and self.llm:
for combo in combos:
self._check_cancelled(run_id)
self._process_pass4(combo, domain, result, score_threshold, run_id)
except CancelledError:
if run_id is not None:
@@ -716,6 +466,271 @@ class Pipeline:
result.top_results = self.repo.get_top_results(domain.name, limit=20)
return result
@staticmethod
def _already_dead(combo: Combination, existing_result: dict | None) -> bool:
"""True if this combo is dead for every pass after 1 -- either a
generic failure (status ends in _fail) or a domain-specific block.
The domain-block case needs the extra existing_result check:
combo.status stays "valid" on purpose for it (domain-agnostic,
see _process_pass1's own comment on this), so pass_reached==1 with
the block already recorded is what actually marks it dead --
status alone isn't enough to catch it."""
if combo.status.endswith("_fail"):
return True
return bool(existing_result and existing_result["pass_reached"] == 1)
def _process_pass1(
self, combo: Combination, domain: Domain, result: PipelineResult, run_id: int | None
) -> None:
"""Constraint resolution for one combo. All writes deferred (commit=False)."""
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
if existing_pass >= 1:
if combo.status.endswith("_fail"):
result.pass1_failed += 1
return
existing_result = self.repo.get_existing_result(combo.id, domain.id)
if existing_result and existing_result["pass_reached"] == 1:
result.pass1_failed += 1
return
result.pass1_valid += 1
return
cr: ConstraintResult = self.resolver.resolve(combo)
if cr.status == "p1_fail":
combo.status = "p1_fail"
combo.block_reason = "; ".join(cr.violations)
self.repo.update_combination_status(
combo.id, "p1_fail", combo.block_reason, commit=False
)
# Save a result row so failed combos appear in results
self.repo.save_result(
combo.id, domain.id, composite_score=0.0, pass_reached=1, commit=False
)
result.pass1_failed += 1
self._update_run_counters(run_id, result, current_pass=1)
return
combo.status = "valid"
self.repo.update_combination_status(combo.id, "valid", commit=False)
# Domain constraint check (per-domain block only). combo.status stays
# "valid" here on purpose: it's domain-agnostic and the same combo can
# be blocked in this domain but valid in another. The per-domain
# block lives on combination_results.domain_block_reason (see
# count_combinations_by_status / get_all_results, which bucket on it).
if domain.constraints:
dc_result = self.resolver.check_domain_constraints(combo, domain.constraints)
if dc_result.status == "p1_fail":
self.repo.save_result(
combo.id, domain.id,
composite_score=0.0, pass_reached=1,
domain_block_reason="; ".join(dc_result.violations),
commit=False,
)
result.pass1_failed += 1
self._update_run_counters(run_id, result, current_pass=1)
return
if cr.status == "conditional":
result.pass1_conditional += 1
else:
result.pass1_valid += 1
self._update_run_counters(run_id, result, current_pass=1)
def _process_pass2(
self,
combo: Combination,
domain: Domain,
bounds_by_name: dict[str, MetricBound],
result: PipelineResult,
run_id: int | None,
) -> None:
"""Physics estimation for one combo. Estimator-only -- self.llm is
reserved for pass 4, never consulted here. All writes deferred."""
existing_result = self.repo.get_existing_result(combo.id, domain.id)
if self._already_dead(combo, existing_result):
return
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
if existing_pass >= 2:
result.pass2_estimated += 1
return
raw_metrics = self._stub_estimate(combo, domain.metric_bounds)
estimate_dicts = []
for mname, rval in raw_metrics.items():
mb = bounds_by_name.get(mname)
if mb and mb.metric_id:
estimate_dicts.append({
"metric_id": mb.metric_id,
"raw_value": rval,
"estimation_method": "stub",
"confidence": 1.0,
})
if estimate_dicts:
self.repo.save_raw_estimates(combo.id, domain.id, estimate_dicts, commit=False)
# Check for all-zero estimates → p2_fail
if raw_metrics and all(v == 0.0 for v in raw_metrics.values()):
combo.status = "p2_fail"
combo.block_reason = "All metric estimates are zero"
self.repo.update_combination_status(
combo.id, "p2_fail", combo.block_reason, commit=False
)
self.repo.save_result(
combo.id, domain.id, composite_score=0.0, pass_reached=2, commit=False
)
result.pass2_failed += 1
self._update_run_counters(run_id, result, current_pass=2)
return
result.pass2_estimated += 1
self._update_run_counters(run_id, result, current_pass=2)
def _process_pass3(
self,
combo: Combination,
domain: Domain,
bounds_by_name: dict[str, MetricBound],
result: PipelineResult,
score_threshold: float,
run_id: int | None,
) -> None:
"""Scoring for one combo. Reloads raw estimates from the DB (pass 2
ran as its own separate phase, not in-memory from this iteration).
All writes deferred."""
existing_result = self.repo.get_existing_result(combo.id, domain.id)
if self._already_dead(combo, existing_result):
return
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
if existing_pass >= 3:
result.pass3_scored += 1
if existing_result and existing_result["composite_score"] is not None:
if existing_result["composite_score"] >= score_threshold:
result.pass3_above_threshold += 1
return
existing_scores = self.repo.get_combination_scores(combo.id, domain.id)
raw_metrics = {s["metric_name"]: s["raw_value"] for s in existing_scores}
sr = self.scorer.score_combination(combo, raw_metrics)
score_dicts = []
for s in sr.scores:
mb = bounds_by_name.get(s.metric_name)
if mb and mb.metric_id:
score_dicts.append({
"metric_id": mb.metric_id,
"raw_value": s.raw_value,
"normalized_score": s.normalized_score,
"estimation_method": s.estimation_method,
"confidence": s.confidence,
})
if score_dicts:
self.repo.save_scores(combo.id, domain.id, score_dicts, commit=False)
# Preserve existing human data
novelty_flag = existing_result["novelty_flag"] if existing_result else None
human_notes = existing_result["human_notes"] if existing_result else None
if sr.composite_score < score_threshold:
self.repo.save_result(
combo.id, domain.id, sr.composite_score, pass_reached=3,
novelty_flag=novelty_flag, human_notes=human_notes, commit=False,
)
combo.status = "p3_fail"
combo.block_reason = (
f"Composite score {sr.composite_score:.4f} below threshold {score_threshold}"
)
self.repo.update_combination_status(
combo.id, "p3_fail", combo.block_reason, commit=False
)
result.pass3_failed += 1
result.pass3_scored += 1
self._update_run_counters(run_id, result, current_pass=3)
return
self.repo.save_result(
combo.id, domain.id, sr.composite_score, pass_reached=3,
novelty_flag=novelty_flag, human_notes=human_notes, commit=False,
)
self.repo.update_combination_status(combo.id, "scored", commit=False)
result.pass3_scored += 1
result.pass3_above_threshold += 1
self._update_run_counters(run_id, result, current_pass=3)
def _process_pass4(
self,
combo: Combination,
domain: Domain,
result: PipelineResult,
score_threshold: float,
run_id: int | None,
) -> None:
"""LLM plausibility review for one combo. Writes commit immediately
(default commit=True) -- slow, crash-prone calls worth protecting
the moment a result lands."""
cur_result = self.repo.get_existing_result(combo.id, domain.id)
if self._already_dead(combo, cur_result):
return
cur_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
if cur_pass >= 4:
return
if not (
cur_result
and cur_result["composite_score"] is not None
and cur_result["composite_score"] >= score_threshold
):
return
description = _describe_combination(combo)
db_scores = self.repo.get_combination_scores(combo.id, domain.id)
score_dict = {
s["metric_name"]: s["normalized_score"]
for s in db_scores if s["normalized_score"] is not None
}
raw_dict = {
s["metric_name"]: s["raw_value"]
for s in db_scores if s["raw_value"] is not None
}
review_result: tuple[str, bool] | None = None
try:
review_result = self.llm.review_plausibility(
description, raw_dict, score_dict, domain
)
except LLMRateLimitError as exc:
self._wait_for_rate_limit(run_id, exc.retry_after)
try:
review_result = self.llm.review_plausibility(
description, raw_dict, score_dict, domain.metric_bounds
)
except LLMRateLimitError:
return # still limited; skip, retry next run
if review_result is None:
return
review_text, plausible = review_result
rating = parse_rating(review_text)
if not plausible:
self.repo.save_result(
combo.id, domain.id, cur_result["composite_score"], pass_reached=4,
novelty_flag=cur_result.get("novelty_flag"), llm_review=review_text,
human_notes=cur_result.get("human_notes"), qualitative_rating=rating,
)
combo.status = "p4_fail"
combo.block_reason = "LLM deemed implausible"
self.repo.update_combination_status(combo.id, "p4_fail", combo.block_reason)
result.pass4_failed += 1
else:
self.repo.save_result(
combo.id, domain.id, cur_result["composite_score"], pass_reached=4,
novelty_flag=cur_result.get("novelty_flag"), llm_review=review_text,
human_notes=cur_result.get("human_notes"), qualitative_rating=rating,
)
self.repo.update_combination_status(combo.id, "llm_reviewed")
result.pass4_reviewed += 1
self._update_run_counters(run_id, result, current_pass=4)
def _wait_for_rate_limit(self, run_id: int | None, retry_after: int) -> None:
"""Mark run rate_limited, sleep with cancel checks, then resume."""
if run_id is not None: