Celes Renata
a72f336ad1
Multi-stage evidence-grounded inference architecture replacing the monolithic 9B model extraction pipeline. CPU-first specialist services handle routine extraction while the 9B vLLM model is preserved for semantic adjudication of ambiguous cases. Key components: - Capability-aware inference gateway (OpenAI-compatible + Ollama) - Endpoint registry with DB migrations and REST API - Sentence-aware document segmenter (property tests) - Deterministic financial parsing with offset integrity - Symbol resolution with ambiguity detection - Specialist service (GLiNER2, dynamic batching, K8s deployment) - Company-specific sentiment (FinBERT, calibration) - Retrieval-based novelty and duplicate detection - Confidence calibration pipeline - Deterministic routing engine (property tests) - 9B adjudication layer with VRAM gating - Stock-specific impact model (features, labels, baseline, trained) - Pipeline orchestrator (state machine, queues, leases, feature flags) - Bounded parallelism (async workers, semaphore, load shedding) - Observability (tracing, metrics, alerts) - Compatibility adapter (v3→v2 golden mapping tests) - Shadow/canary promotion framework - Active learning and fine-tuning pipeline Test results: 1,161 tests pass, ruff lint clean. All 282 spec tasks completed.
295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""Comparison and attribution for benchmark configurations.
|
|
|
|
Produces delta tables and attribution reports to quantify how much of
|
|
the apparent architecture gain comes from fixing the current request alone
|
|
(temperature, schema constraints) versus the full v3 architecture.
|
|
|
|
Validates: Requirements 16.2, 16.3, 16.5
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from services.intelligence_pipeline_v3.benchmark.runner import BenchmarkRun
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Result Models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class FieldDelta(BaseModel):
|
|
"""Per-field improvement between two configurations."""
|
|
|
|
field_name: str = Field(description="Name of the compared field/metric")
|
|
baseline_value: float = Field(description="Value in the baseline configuration")
|
|
comparison_value: float = Field(description="Value in the compared configuration")
|
|
absolute_delta: float = Field(description="comparison - baseline")
|
|
relative_delta_percent: float = Field(
|
|
description="Percentage change from baseline ((comp - base) / base * 100)",
|
|
)
|
|
improved: bool = Field(
|
|
description="Whether the delta represents improvement (higher is better assumed unless inverted)",
|
|
)
|
|
|
|
|
|
class ResourceDelta(BaseModel):
|
|
"""Resource usage comparison between configurations."""
|
|
|
|
metric_name: str = Field(description="Resource metric name")
|
|
baseline_value: float = Field(description="Baseline resource usage")
|
|
comparison_value: float = Field(description="Compared configuration resource usage")
|
|
absolute_delta: float = Field(description="comparison - baseline")
|
|
relative_delta_percent: float = Field(description="Percentage change")
|
|
improved: bool = Field(
|
|
description="Whether the delta represents improvement (lower is better for resources)",
|
|
)
|
|
|
|
|
|
class ConfigDelta(BaseModel):
|
|
"""Comparison results between a baseline and one other configuration."""
|
|
|
|
baseline_config: str = Field(description="Baseline configuration name")
|
|
comparison_config: str = Field(description="Compared configuration name")
|
|
field_deltas: list[FieldDelta] = Field(default_factory=list)
|
|
resource_deltas: list[ResourceDelta] = Field(default_factory=list)
|
|
|
|
|
|
class ComparisonReport(BaseModel):
|
|
"""Full comparison report across multiple configurations.
|
|
|
|
Attributes:
|
|
configs_compared: Names of all configurations in this comparison.
|
|
deltas: Per-configuration comparison against the baseline.
|
|
attribution_summary: Human-readable attribution of improvement sources.
|
|
"""
|
|
|
|
configs_compared: list[str] = Field(
|
|
description="All configuration names included in this comparison",
|
|
)
|
|
deltas: list[ConfigDelta] = Field(
|
|
default_factory=list,
|
|
description="Delta tables for each non-baseline config vs baseline",
|
|
)
|
|
attribution_summary: dict[str, float] = Field(
|
|
default_factory=dict,
|
|
description=(
|
|
"Attribution percentages: maps source (e.g. 'temperature_fix', "
|
|
"'schema_constraint', 'architecture') to fraction of total improvement"
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _compute_field_delta(
|
|
field_name: str,
|
|
baseline_val: float,
|
|
comparison_val: float,
|
|
*,
|
|
higher_is_better: bool = True,
|
|
) -> FieldDelta:
|
|
"""Compute a single field delta with direction awareness."""
|
|
absolute = comparison_val - baseline_val
|
|
relative = (
|
|
(absolute / baseline_val * 100.0) if baseline_val != 0.0 else 0.0
|
|
)
|
|
improved = absolute > 0.0 if higher_is_better else absolute < 0.0
|
|
|
|
return FieldDelta(
|
|
field_name=field_name,
|
|
baseline_value=baseline_val,
|
|
comparison_value=comparison_val,
|
|
absolute_delta=absolute,
|
|
relative_delta_percent=relative,
|
|
improved=improved,
|
|
)
|
|
|
|
|
|
def _compute_resource_delta(
|
|
metric_name: str,
|
|
baseline_val: float,
|
|
comparison_val: float,
|
|
) -> ResourceDelta:
|
|
"""Compute a resource delta (lower is better)."""
|
|
absolute = comparison_val - baseline_val
|
|
relative = (
|
|
(absolute / baseline_val * 100.0) if baseline_val != 0.0 else 0.0
|
|
)
|
|
improved = absolute < 0.0 # Lower resource usage is better
|
|
|
|
return ResourceDelta(
|
|
metric_name=metric_name,
|
|
baseline_value=baseline_val,
|
|
comparison_value=comparison_val,
|
|
absolute_delta=absolute,
|
|
relative_delta_percent=relative,
|
|
improved=improved,
|
|
)
|
|
|
|
|
|
def _run_metrics(run: BenchmarkRun) -> dict[str, float]:
|
|
"""Extract summary metrics from a benchmark run."""
|
|
n = len(run.results) or 1 # Avoid division by zero
|
|
|
|
return {
|
|
"schema_validity_rate": run.schema_validity_rate,
|
|
"success_count": float(run.success_count),
|
|
"failure_count": float(run.failure_count),
|
|
"mean_duration_ms": run.mean_duration_ms,
|
|
"total_input_tokens": float(run.total_input_tokens),
|
|
"total_output_tokens": float(run.total_output_tokens),
|
|
"mean_retries": sum(r.retries for r in run.results) / n,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def compare_configurations(
|
|
baseline_run: BenchmarkRun,
|
|
comparison_runs: list[BenchmarkRun],
|
|
) -> ComparisonReport:
|
|
"""Compare benchmark runs to produce delta tables and attribution.
|
|
|
|
Computes per-field and per-resource deltas between the baseline and
|
|
each comparison configuration, then attributes improvement sources.
|
|
|
|
Args:
|
|
baseline_run: The baseline (typically BASELINE_CURRENT) run results.
|
|
comparison_runs: One or more comparison configuration runs.
|
|
|
|
Returns:
|
|
ComparisonReport with deltas and attribution percentages.
|
|
"""
|
|
configs_compared = [baseline_run.config_name] + [
|
|
r.config_name for r in comparison_runs
|
|
]
|
|
|
|
baseline_metrics = _run_metrics(baseline_run)
|
|
deltas: list[ConfigDelta] = []
|
|
|
|
# Fields where higher is better
|
|
_higher_is_better = {"schema_validity_rate", "success_count"}
|
|
# Fields where lower is better (resource-like)
|
|
_resource_fields = {
|
|
"mean_duration_ms",
|
|
"total_input_tokens",
|
|
"total_output_tokens",
|
|
"mean_retries",
|
|
"failure_count",
|
|
}
|
|
|
|
for comp_run in comparison_runs:
|
|
comp_metrics = _run_metrics(comp_run)
|
|
field_deltas: list[FieldDelta] = []
|
|
resource_deltas: list[ResourceDelta] = []
|
|
|
|
for metric_name, baseline_val in baseline_metrics.items():
|
|
comp_val = comp_metrics[metric_name]
|
|
|
|
if metric_name in _resource_fields:
|
|
resource_deltas.append(
|
|
_compute_resource_delta(metric_name, baseline_val, comp_val)
|
|
)
|
|
else:
|
|
field_deltas.append(
|
|
_compute_field_delta(
|
|
metric_name,
|
|
baseline_val,
|
|
comp_val,
|
|
higher_is_better=(metric_name in _higher_is_better),
|
|
)
|
|
)
|
|
|
|
deltas.append(
|
|
ConfigDelta(
|
|
baseline_config=baseline_run.config_name,
|
|
comparison_config=comp_run.config_name,
|
|
field_deltas=field_deltas,
|
|
resource_deltas=resource_deltas,
|
|
)
|
|
)
|
|
|
|
# Attribution: estimate how much improvement comes from each fix
|
|
attribution = _compute_attribution(baseline_metrics, comparison_runs)
|
|
|
|
return ComparisonReport(
|
|
configs_compared=configs_compared,
|
|
deltas=deltas,
|
|
attribution_summary=attribution,
|
|
)
|
|
|
|
|
|
def _compute_attribution(
|
|
baseline_metrics: dict[str, float],
|
|
comparison_runs: list[BenchmarkRun],
|
|
) -> dict[str, float]:
|
|
"""Compute attribution percentages for improvement sources.
|
|
|
|
Uses schema_validity_rate as the primary improvement signal.
|
|
Attribution is computed as the fraction of total improvement each
|
|
configuration step contributes.
|
|
|
|
Returns a dict mapping source labels to fraction (0.0-1.0).
|
|
"""
|
|
attribution: dict[str, float] = {}
|
|
|
|
if not comparison_runs:
|
|
return attribution
|
|
|
|
baseline_validity = baseline_metrics["schema_validity_rate"]
|
|
|
|
# Find temp_zero and strict_schema runs by config name
|
|
temp_zero_validity: float | None = None
|
|
strict_schema_validity: float | None = None
|
|
|
|
for run in comparison_runs:
|
|
run_metrics = _run_metrics(run)
|
|
if "temp_zero" in run.config_name:
|
|
temp_zero_validity = run_metrics["schema_validity_rate"]
|
|
elif "strict_schema" in run.config_name:
|
|
strict_schema_validity = run_metrics["schema_validity_rate"]
|
|
|
|
# Compute incremental gains
|
|
# Total improvement = strict_schema - baseline (or best comparison - baseline)
|
|
best_validity = max(
|
|
_run_metrics(r)["schema_validity_rate"] for r in comparison_runs
|
|
)
|
|
total_improvement = best_validity - baseline_validity
|
|
|
|
if total_improvement <= 0.0:
|
|
# No improvement detected; equal attribution
|
|
attribution["temperature_fix"] = 0.0
|
|
attribution["schema_constraint"] = 0.0
|
|
return attribution
|
|
|
|
# Temperature fix contribution
|
|
if temp_zero_validity is not None:
|
|
temp_gain = temp_zero_validity - baseline_validity
|
|
attribution["temperature_fix"] = max(0.0, temp_gain / total_improvement)
|
|
else:
|
|
attribution["temperature_fix"] = 0.0
|
|
|
|
# Schema constraint contribution (incremental over temp fix)
|
|
if strict_schema_validity is not None and temp_zero_validity is not None:
|
|
schema_gain = strict_schema_validity - temp_zero_validity
|
|
attribution["schema_constraint"] = max(0.0, schema_gain / total_improvement)
|
|
elif strict_schema_validity is not None:
|
|
schema_gain = strict_schema_validity - baseline_validity
|
|
attribution["schema_constraint"] = max(0.0, schema_gain / total_improvement)
|
|
else:
|
|
attribution["schema_constraint"] = 0.0
|
|
|
|
# Remaining is attributed to other factors
|
|
accounted = attribution.get("temperature_fix", 0.0) + attribution.get(
|
|
"schema_constraint", 0.0
|
|
)
|
|
attribution["other"] = max(0.0, 1.0 - accounted)
|
|
|
|
return attribution
|