mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-07-16 08:57:21 +02:00
pliny
711 lines
26 KiB
Python
711 lines
26 KiB
Python
"""Telemetry-driven adaptive defaults for OBLITERATUS.
|
|
|
|
Fetches community telemetry from the HuggingFace Hub dataset and analyzes
|
|
historical runs to recommend the best abliteration method and hyperparameters
|
|
for a given model architecture.
|
|
|
|
Architecture bucketing:
|
|
Records are grouped by (arch_class, reasoning_class, param_bucket) where
|
|
param_bucket is a coarse size tier (tiny/small/medium/large/frontier).
|
|
Within each bucket, methods are ranked by composite score and the
|
|
best-performing hyperparameter ranges are extracted.
|
|
|
|
The ``get_adaptive_recommendation()`` function returns an
|
|
``AdaptiveRecommendation`` that the pipeline/UI can apply on top of
|
|
(or instead of) the static research-grounded defaults in
|
|
``architecture_profiles.py``.
|
|
|
|
Data flow:
|
|
HF Hub (OBLITERATUS-TELEMETRY) ──► fetch_hub_records()
|
|
│ │
|
|
▼ ▼
|
|
Local JSONL cache ──────────► build_knowledge_base()
|
|
│
|
|
▼
|
|
get_adaptive_recommendation()
|
|
│
|
|
▼
|
|
AdaptiveRecommendation
|
|
(best method, overrides, confidence)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import statistics
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Cache config ──────────────────────────────────────────────────────────
|
|
|
|
_CACHE_TTL_S = 600 # 10 minutes — telemetry doesn't change that fast
|
|
_cache: dict[str, Any] = {}
|
|
_cache_ts: float = 0.0
|
|
|
|
# Minimum records per bucket to trust the recommendation
|
|
_MIN_RECORDS_FOR_CONFIDENCE = 5
|
|
_HIGH_CONFIDENCE_RECORDS = 20
|
|
|
|
|
|
# ── Size bucketing ────────────────────────────────────────────────────────
|
|
|
|
def _param_bucket(total_params_b: float) -> str:
|
|
"""Coarse size tier matching presets.py tiers."""
|
|
if total_params_b <= 0.5:
|
|
return "tiny"
|
|
if total_params_b <= 4:
|
|
return "small"
|
|
if total_params_b <= 16:
|
|
return "medium"
|
|
if total_params_b <= 80:
|
|
return "large"
|
|
return "frontier"
|
|
|
|
|
|
def _extract_arch_key(record: dict) -> tuple[str, str, str] | None:
|
|
"""Extract (arch_class, reasoning_class, param_bucket) from a telemetry record.
|
|
|
|
Returns None if the record lacks enough information to classify.
|
|
"""
|
|
model = record.get("model", {})
|
|
if isinstance(model, str):
|
|
# Schema v1 — just model name, can't reliably bucket
|
|
return None
|
|
|
|
arch_str = model.get("architecture", "")
|
|
num_layers = model.get("num_layers", 0)
|
|
hidden_size = model.get("hidden_size", 0)
|
|
total_params = model.get("total_params", 0)
|
|
|
|
# Estimate params in billions
|
|
if total_params > 0:
|
|
params_b = total_params / 1e9
|
|
elif num_layers > 0 and hidden_size > 0:
|
|
# Rough estimate: 12 * hidden² * num_layers (transformer scaling)
|
|
params_b = (12 * hidden_size**2 * num_layers) / 1e9
|
|
else:
|
|
return None
|
|
|
|
# Detect architecture class from the architecture string or model config
|
|
arch_lower = arch_str.lower()
|
|
moe_keywords = {"moe", "mixtral", "qwen2_moe", "qwen3_moe", "deepseek_v2",
|
|
"deepseek_v3", "dbrx", "grok", "jamba", "arctic", "olmoe",
|
|
"switch", "llama4"}
|
|
is_moe = any(kw in arch_lower for kw in moe_keywords)
|
|
|
|
# Check method_config for per_expert_directions as MoE signal
|
|
mc = record.get("method_config", {})
|
|
if mc.get("per_expert_directions"):
|
|
is_moe = True
|
|
|
|
if is_moe:
|
|
arch_class = "large_moe" if params_b > 100 else "small_moe"
|
|
else:
|
|
arch_class = "dense"
|
|
|
|
# Detect reasoning from analysis insights or architecture name
|
|
analysis = record.get("analysis_insights", {})
|
|
reasoning_class = "standard"
|
|
reasoning_keywords = {"reason", "think", "cot", "r1", "qwq", "o1", "o3"}
|
|
if any(kw in arch_lower for kw in reasoning_keywords):
|
|
reasoning_class = "reasoning"
|
|
if analysis.get("cot_aware") or mc.get("cot_aware"):
|
|
reasoning_class = "reasoning"
|
|
|
|
return (arch_class, reasoning_class, _param_bucket(params_b))
|
|
|
|
|
|
# ── Composite scoring (same as tourney.py) ────────────────────────────────
|
|
|
|
def _composite_score(qm: dict[str, Any]) -> float:
|
|
"""Score a run on [0, 1]. Higher is better."""
|
|
rr = qm.get("refusal_rate")
|
|
co = qm.get("coherence")
|
|
kl = qm.get("kl_divergence")
|
|
pp = qm.get("perplexity")
|
|
|
|
refusal_score = (1.0 - rr) if rr is not None else 0.0
|
|
coherence_score = co if co is not None else 0.0
|
|
kl_score = 1.0 / (1.0 + kl) if kl is not None else 0.5
|
|
ppl_score = 1.0 / (1.0 + pp / 100.0) if pp is not None else 0.5
|
|
|
|
return (
|
|
refusal_score * 0.4
|
|
+ coherence_score * 0.3
|
|
+ kl_score * 0.2
|
|
+ ppl_score * 0.1
|
|
)
|
|
|
|
|
|
# ── Data structures ──────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class MethodStats:
|
|
"""Aggregated statistics for one method within an architecture bucket."""
|
|
|
|
method: str
|
|
n_runs: int = 0
|
|
scores: list[float] = field(default_factory=list)
|
|
refusal_rates: list[float] = field(default_factory=list)
|
|
coherences: list[float] = field(default_factory=list)
|
|
kl_divergences: list[float] = field(default_factory=list)
|
|
perplexities: list[float] = field(default_factory=list)
|
|
configs: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
@property
|
|
def mean_score(self) -> float:
|
|
return statistics.mean(self.scores) if self.scores else 0.0
|
|
|
|
@property
|
|
def best_score(self) -> float:
|
|
return max(self.scores) if self.scores else 0.0
|
|
|
|
@property
|
|
def median_score(self) -> float:
|
|
return statistics.median(self.scores) if self.scores else 0.0
|
|
|
|
def best_config_ranges(self) -> dict[str, Any]:
|
|
"""Extract the hyperparameter ranges from top-performing runs.
|
|
|
|
Takes the top 25% of runs by composite score and returns the median
|
|
value for each numeric config key, or the mode for booleans.
|
|
"""
|
|
if not self.configs or not self.scores:
|
|
return {}
|
|
|
|
# Pair scores with configs and take top 25%
|
|
paired = sorted(zip(self.scores, self.configs), key=lambda x: x[0], reverse=True)
|
|
top_n = max(1, len(paired) // 4)
|
|
top_configs = [c for _, c in paired[:top_n]]
|
|
|
|
ranges: dict[str, Any] = {}
|
|
all_keys = set()
|
|
for c in top_configs:
|
|
all_keys.update(c.keys())
|
|
|
|
for key in all_keys:
|
|
values = [c[key] for c in top_configs if key in c and c[key] is not None]
|
|
if not values:
|
|
continue
|
|
|
|
if all(isinstance(v, bool) for v in values):
|
|
# Mode for booleans
|
|
true_count = sum(1 for v in values if v)
|
|
ranges[key] = true_count > len(values) / 2
|
|
elif all(isinstance(v, (int, float)) for v in values):
|
|
# Median for numerics
|
|
ranges[key] = statistics.median(values)
|
|
# Round ints back to ints
|
|
if all(isinstance(v, int) for v in values):
|
|
ranges[key] = int(round(ranges[key]))
|
|
# Skip strings and other types
|
|
|
|
return ranges
|
|
|
|
|
|
@dataclass
|
|
class BucketKnowledge:
|
|
"""Everything we know about one architecture bucket from telemetry."""
|
|
|
|
arch_key: tuple[str, str, str] # (arch_class, reasoning_class, param_bucket)
|
|
methods: dict[str, MethodStats] = field(default_factory=dict)
|
|
total_runs: int = 0
|
|
|
|
@property
|
|
def best_method(self) -> str | None:
|
|
"""Method with highest mean composite score (min 3 runs)."""
|
|
candidates = [
|
|
(name, ms) for name, ms in self.methods.items()
|
|
if ms.n_runs >= 3
|
|
]
|
|
if not candidates:
|
|
# Fall back to any method with runs
|
|
candidates = [(name, ms) for name, ms in self.methods.items() if ms.n_runs > 0]
|
|
if not candidates:
|
|
return None
|
|
return max(candidates, key=lambda x: x[1].mean_score)[0]
|
|
|
|
@property
|
|
def ranked_methods(self) -> list[tuple[str, MethodStats]]:
|
|
"""All methods ranked by mean score, descending."""
|
|
return sorted(
|
|
self.methods.items(),
|
|
key=lambda x: x[1].mean_score,
|
|
reverse=True,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class AdaptiveRecommendation:
|
|
"""A telemetry-driven recommendation for a specific model."""
|
|
|
|
# What we recommend
|
|
recommended_method: str
|
|
method_overrides: dict[str, Any]
|
|
|
|
# How confident we are
|
|
confidence: str # "high", "medium", "low", "none"
|
|
n_records: int # total records in bucket
|
|
n_method_records: int # records for this specific method
|
|
|
|
# Context
|
|
arch_key: tuple[str, str, str]
|
|
bucket_label: str # human-readable e.g. "Dense Standard Medium"
|
|
method_ranking: list[tuple[str, float]] # [(method, mean_score), ...]
|
|
|
|
# Best metrics seen in this bucket
|
|
best_refusal_rate: float | None = None
|
|
best_coherence: float | None = None
|
|
|
|
# Explanation
|
|
reason: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"recommended_method": self.recommended_method,
|
|
"method_overrides": self.method_overrides,
|
|
"confidence": self.confidence,
|
|
"n_records": self.n_records,
|
|
"n_method_records": self.n_method_records,
|
|
"arch_key": list(self.arch_key),
|
|
"bucket_label": self.bucket_label,
|
|
"method_ranking": self.method_ranking,
|
|
"best_refusal_rate": self.best_refusal_rate,
|
|
"best_coherence": self.best_coherence,
|
|
"reason": self.reason,
|
|
}
|
|
|
|
|
|
# ── Knowledge base construction ──────────────────────────────────────────
|
|
|
|
|
|
def build_knowledge_base(
|
|
records: list[dict[str, Any]] | None = None,
|
|
) -> dict[tuple[str, str, str], BucketKnowledge]:
|
|
"""Build per-bucket knowledge from telemetry records.
|
|
|
|
If *records* is None, fetches from local + Hub automatically.
|
|
"""
|
|
if records is None:
|
|
records = _fetch_all_records()
|
|
|
|
buckets: dict[tuple[str, str, str], BucketKnowledge] = {}
|
|
|
|
for record in records:
|
|
# Skip errored runs
|
|
if record.get("error"):
|
|
continue
|
|
|
|
arch_key = _extract_arch_key(record)
|
|
if arch_key is None:
|
|
continue
|
|
|
|
method = record.get("method", "")
|
|
if not method:
|
|
continue
|
|
|
|
qm = record.get("quality_metrics", {})
|
|
if not qm:
|
|
continue
|
|
|
|
score = _composite_score(qm)
|
|
|
|
if arch_key not in buckets:
|
|
buckets[arch_key] = BucketKnowledge(arch_key=arch_key)
|
|
|
|
bucket = buckets[arch_key]
|
|
bucket.total_runs += 1
|
|
|
|
if method not in bucket.methods:
|
|
bucket.methods[method] = MethodStats(method=method)
|
|
|
|
ms = bucket.methods[method]
|
|
ms.n_runs += 1
|
|
ms.scores.append(score)
|
|
|
|
rr = qm.get("refusal_rate")
|
|
if rr is not None:
|
|
ms.refusal_rates.append(rr)
|
|
co = qm.get("coherence")
|
|
if co is not None:
|
|
ms.coherences.append(co)
|
|
kl = qm.get("kl_divergence")
|
|
if kl is not None:
|
|
ms.kl_divergences.append(kl)
|
|
pp = qm.get("perplexity")
|
|
if pp is not None:
|
|
ms.perplexities.append(pp)
|
|
|
|
mc = record.get("method_config", {})
|
|
if mc:
|
|
ms.configs.append(mc)
|
|
|
|
return buckets
|
|
|
|
|
|
def _fetch_all_records() -> list[dict[str, Any]]:
|
|
"""Fetch telemetry from local file + Hub, with caching."""
|
|
global _cache, _cache_ts
|
|
|
|
now = time.time()
|
|
if _cache.get("records") is not None and (now - _cache_ts) < _CACHE_TTL_S:
|
|
return _cache["records"]
|
|
|
|
records: list[dict[str, Any]] = []
|
|
|
|
# Local records
|
|
try:
|
|
from obliteratus.telemetry import read_telemetry
|
|
records.extend(read_telemetry())
|
|
except Exception as e:
|
|
logger.debug("Failed to read local telemetry: %s", e)
|
|
|
|
# Hub records
|
|
try:
|
|
from obliteratus.telemetry import fetch_hub_records
|
|
hub = fetch_hub_records()
|
|
records.extend(hub)
|
|
except Exception as e:
|
|
logger.debug("Failed to fetch Hub telemetry: %s", e)
|
|
|
|
# Deduplicate by (session_id, timestamp)
|
|
seen: set[tuple[str, str]] = set()
|
|
deduped = []
|
|
for r in records:
|
|
key = (r.get("session_id", ""), r.get("timestamp", ""))
|
|
if key not in seen:
|
|
seen.add(key)
|
|
deduped.append(r)
|
|
|
|
_cache["records"] = deduped
|
|
_cache_ts = now
|
|
return deduped
|
|
|
|
|
|
# ── Recommendation engine ────────────────────────────────────────────────
|
|
|
|
|
|
def get_adaptive_recommendation(
|
|
arch_class: str,
|
|
reasoning_class: str,
|
|
total_params_b: float,
|
|
model_name: str = "",
|
|
knowledge: dict[tuple[str, str, str], BucketKnowledge] | None = None,
|
|
) -> AdaptiveRecommendation:
|
|
"""Get a telemetry-based recommendation for the given architecture.
|
|
|
|
Looks up the closest bucket in the knowledge base and returns the
|
|
best-performing method + hyperparameter overrides.
|
|
|
|
Falls through to broader buckets if the exact match has too few records:
|
|
1. Exact match: (arch_class, reasoning_class, param_bucket)
|
|
2. Size-agnostic: (arch_class, reasoning_class, "*")
|
|
3. Arch-only: (arch_class, "*", "*")
|
|
|
|
Args:
|
|
arch_class: "dense", "small_moe", or "large_moe"
|
|
reasoning_class: "standard" or "reasoning"
|
|
total_params_b: Total params in billions
|
|
model_name: Optional, for model-specific matching
|
|
knowledge: Pre-built knowledge base (fetches if None)
|
|
"""
|
|
if knowledge is None:
|
|
knowledge = build_knowledge_base()
|
|
|
|
param_bucket = _param_bucket(total_params_b)
|
|
bucket_label = f"{arch_class.replace('_', ' ').title()} {reasoning_class.title()} {param_bucket.title()}"
|
|
|
|
# Try exact match first, then broaden
|
|
candidates = [
|
|
(arch_class, reasoning_class, param_bucket),
|
|
]
|
|
|
|
# Also check model-specific records (exact model name match)
|
|
# This is for the future when we have enough data per-model
|
|
model_short = model_name.split("/")[-1].lower() if model_name else ""
|
|
|
|
bucket = None
|
|
used_key = None
|
|
for key in candidates:
|
|
if key in knowledge and knowledge[key].total_runs >= _MIN_RECORDS_FOR_CONFIDENCE:
|
|
bucket = knowledge[key]
|
|
used_key = key
|
|
break
|
|
|
|
# Fall back: merge all buckets that share (arch_class, reasoning_class)
|
|
if bucket is None:
|
|
merged = BucketKnowledge(arch_key=(arch_class, reasoning_class, "*"))
|
|
for key, bkt in knowledge.items():
|
|
if key[0] == arch_class and key[1] == reasoning_class:
|
|
for method_name, ms in bkt.methods.items():
|
|
if method_name not in merged.methods:
|
|
merged.methods[method_name] = MethodStats(method=method_name)
|
|
target = merged.methods[method_name]
|
|
target.n_runs += ms.n_runs
|
|
target.scores.extend(ms.scores)
|
|
target.refusal_rates.extend(ms.refusal_rates)
|
|
target.coherences.extend(ms.coherences)
|
|
target.kl_divergences.extend(ms.kl_divergences)
|
|
target.perplexities.extend(ms.perplexities)
|
|
target.configs.extend(ms.configs)
|
|
merged.total_runs += bkt.total_runs
|
|
if merged.total_runs >= _MIN_RECORDS_FOR_CONFIDENCE:
|
|
bucket = merged
|
|
used_key = merged.arch_key
|
|
bucket_label = f"{arch_class.replace('_', ' ').title()} {reasoning_class.title()} (all sizes)"
|
|
|
|
# Last resort: merge all buckets that share arch_class
|
|
if bucket is None:
|
|
merged = BucketKnowledge(arch_key=(arch_class, "*", "*"))
|
|
for key, bkt in knowledge.items():
|
|
if key[0] == arch_class:
|
|
for method_name, ms in bkt.methods.items():
|
|
if method_name not in merged.methods:
|
|
merged.methods[method_name] = MethodStats(method=method_name)
|
|
target = merged.methods[method_name]
|
|
target.n_runs += ms.n_runs
|
|
target.scores.extend(ms.scores)
|
|
target.refusal_rates.extend(ms.refusal_rates)
|
|
target.coherences.extend(ms.coherences)
|
|
target.kl_divergences.extend(ms.kl_divergences)
|
|
target.perplexities.extend(ms.perplexities)
|
|
target.configs.extend(ms.configs)
|
|
merged.total_runs += bkt.total_runs
|
|
if merged.total_runs > 0:
|
|
bucket = merged
|
|
used_key = merged.arch_key
|
|
bucket_label = f"{arch_class.replace('_', ' ').title()} (all)"
|
|
|
|
# No data at all
|
|
if bucket is None or not bucket.methods:
|
|
return AdaptiveRecommendation(
|
|
recommended_method="",
|
|
method_overrides={},
|
|
confidence="none",
|
|
n_records=0,
|
|
n_method_records=0,
|
|
arch_key=(arch_class, reasoning_class, param_bucket),
|
|
bucket_label=bucket_label,
|
|
method_ranking=[],
|
|
reason="No telemetry data available for this architecture.",
|
|
)
|
|
|
|
# Get best method
|
|
best_method = bucket.best_method
|
|
if not best_method:
|
|
return AdaptiveRecommendation(
|
|
recommended_method="",
|
|
method_overrides={},
|
|
confidence="none",
|
|
n_records=bucket.total_runs,
|
|
n_method_records=0,
|
|
arch_key=used_key or (arch_class, reasoning_class, param_bucket),
|
|
bucket_label=bucket_label,
|
|
method_ranking=[],
|
|
reason="Telemetry records found but no method has enough runs.",
|
|
)
|
|
|
|
ms = bucket.methods[best_method]
|
|
|
|
# Extract best hyperparams from top runs
|
|
overrides = ms.best_config_ranges()
|
|
|
|
# Confidence level
|
|
if ms.n_runs >= _HIGH_CONFIDENCE_RECORDS:
|
|
confidence = "high"
|
|
elif ms.n_runs >= _MIN_RECORDS_FOR_CONFIDENCE:
|
|
confidence = "medium"
|
|
else:
|
|
confidence = "low"
|
|
|
|
# Method ranking
|
|
ranking = [
|
|
(name, stats.mean_score)
|
|
for name, stats in bucket.ranked_methods
|
|
]
|
|
|
|
# Best metrics seen
|
|
best_rr = min(ms.refusal_rates) if ms.refusal_rates else None
|
|
best_co = max(ms.coherences) if ms.coherences else None
|
|
|
|
# Build explanation
|
|
runner_up = ranking[1] if len(ranking) > 1 else None
|
|
reason_parts = [
|
|
f"Based on {bucket.total_runs} community runs for {bucket_label}.",
|
|
f"`{best_method}` achieves a mean composite score of {ms.mean_score:.4f} "
|
|
f"across {ms.n_runs} runs.",
|
|
]
|
|
if runner_up:
|
|
reason_parts.append(
|
|
f"Runner-up: `{runner_up[0]}` ({runner_up[1]:.4f})."
|
|
)
|
|
if best_rr is not None:
|
|
reason_parts.append(f"Best refusal rate seen: {best_rr:.1%}.")
|
|
if overrides:
|
|
override_strs = [f"{k}={v}" for k, v in sorted(overrides.items())]
|
|
reason_parts.append(f"Optimal hyperparams from top runs: {', '.join(override_strs[:6])}")
|
|
|
|
return AdaptiveRecommendation(
|
|
recommended_method=best_method,
|
|
method_overrides=overrides,
|
|
confidence=confidence,
|
|
n_records=bucket.total_runs,
|
|
n_method_records=ms.n_runs,
|
|
arch_key=used_key or (arch_class, reasoning_class, param_bucket),
|
|
bucket_label=bucket_label,
|
|
method_ranking=ranking,
|
|
best_refusal_rate=best_rr,
|
|
best_coherence=best_co,
|
|
reason=" ".join(reason_parts),
|
|
)
|
|
|
|
|
|
# ── Cross-architecture insights ──────────────────────────────────────────
|
|
|
|
|
|
def get_global_insights(
|
|
knowledge: dict[tuple[str, str, str], BucketKnowledge] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Compute cross-architecture insights from all telemetry.
|
|
|
|
Returns a summary dict with:
|
|
- overall_best_methods: top methods across all architectures
|
|
- architecture_breakdown: per-bucket summaries
|
|
- total_records: total telemetry records analyzed
|
|
- hyperparameter_trends: keys that consistently appear in top configs
|
|
"""
|
|
if knowledge is None:
|
|
knowledge = build_knowledge_base()
|
|
|
|
total_records = sum(b.total_runs for b in knowledge.values())
|
|
|
|
# Global method scores (weighted by bucket size)
|
|
global_method_scores: dict[str, list[float]] = {}
|
|
for bucket in knowledge.values():
|
|
for name, ms in bucket.methods.items():
|
|
if name not in global_method_scores:
|
|
global_method_scores[name] = []
|
|
global_method_scores[name].extend(ms.scores)
|
|
|
|
overall_ranking = sorted(
|
|
[
|
|
(name, statistics.mean(scores), len(scores))
|
|
for name, scores in global_method_scores.items()
|
|
if scores
|
|
],
|
|
key=lambda x: x[1],
|
|
reverse=True,
|
|
)
|
|
|
|
# Per-bucket summaries
|
|
arch_breakdown = {}
|
|
for key, bucket in sorted(knowledge.items()):
|
|
label = f"{key[0]} / {key[1]} / {key[2]}"
|
|
best = bucket.best_method
|
|
arch_breakdown[label] = {
|
|
"total_runs": bucket.total_runs,
|
|
"best_method": best,
|
|
"best_score": bucket.methods[best].mean_score if best and best in bucket.methods else 0,
|
|
"n_methods_tested": len(bucket.methods),
|
|
}
|
|
|
|
# Hyperparameter trends across top runs
|
|
all_top_configs: list[dict] = []
|
|
for bucket in knowledge.values():
|
|
for ms in bucket.methods.values():
|
|
if ms.configs and ms.scores:
|
|
paired = sorted(zip(ms.scores, ms.configs), key=lambda x: x[0], reverse=True)
|
|
top_n = max(1, len(paired) // 4)
|
|
all_top_configs.extend(c for _, c in paired[:top_n])
|
|
|
|
hp_trends: dict[str, Any] = {}
|
|
if all_top_configs:
|
|
all_keys = set()
|
|
for c in all_top_configs:
|
|
all_keys.update(c.keys())
|
|
for key in sorted(all_keys):
|
|
values = [c[key] for c in all_top_configs if key in c and c[key] is not None]
|
|
if not values:
|
|
continue
|
|
if all(isinstance(v, bool) for v in values):
|
|
true_pct = sum(1 for v in values if v) / len(values)
|
|
hp_trends[key] = {"type": "bool", "true_pct": round(true_pct, 2), "n": len(values)}
|
|
elif all(isinstance(v, (int, float)) for v in values):
|
|
hp_trends[key] = {
|
|
"type": "numeric",
|
|
"median": round(statistics.median(values), 4),
|
|
"mean": round(statistics.mean(values), 4),
|
|
"min": min(values),
|
|
"max": max(values),
|
|
"n": len(values),
|
|
}
|
|
|
|
return {
|
|
"total_records": total_records,
|
|
"overall_best_methods": [
|
|
{"method": name, "mean_score": round(score, 4), "n_runs": n}
|
|
for name, score, n in overall_ranking
|
|
],
|
|
"architecture_breakdown": arch_breakdown,
|
|
"hyperparameter_trends": hp_trends,
|
|
}
|
|
|
|
|
|
# ── Format helpers ────────────────────────────────────────────────────────
|
|
|
|
|
|
def format_recommendation(rec: AdaptiveRecommendation) -> str:
|
|
"""Format a recommendation as a human-readable markdown string."""
|
|
if rec.confidence == "none":
|
|
return (
|
|
f"**No telemetry data** for {rec.bucket_label}.\n\n"
|
|
"Using research-grounded defaults from `architecture_profiles.py`.\n"
|
|
"Run some abliterations and the adaptive system will learn!"
|
|
)
|
|
|
|
confidence_emoji = {"high": "HIGH", "medium": "MEDIUM", "low": "LOW"}
|
|
conf = confidence_emoji.get(rec.confidence, rec.confidence.upper())
|
|
|
|
lines = [
|
|
f"### Adaptive Recommendation [{conf} confidence]",
|
|
f"**Architecture bucket:** {rec.bucket_label}",
|
|
f"**Based on:** {rec.n_records} community runs",
|
|
"",
|
|
f"**Recommended method:** `{rec.recommended_method}` "
|
|
f"(score: {rec.method_ranking[0][1]:.4f}, {rec.n_method_records} runs)",
|
|
"",
|
|
]
|
|
|
|
if len(rec.method_ranking) > 1:
|
|
lines.append("**Method ranking:**")
|
|
lines.append("| Rank | Method | Mean Score | Runs |")
|
|
lines.append("|------|--------|------------|------|")
|
|
for i, (name, score) in enumerate(rec.method_ranking[:8], 1):
|
|
ms_runs = 0
|
|
# Get run count from the knowledge (not stored directly, but we have n_method_records for winner)
|
|
lines.append(f"| {i} | `{name}` | {score:.4f} | — |")
|
|
lines.append("")
|
|
|
|
if rec.method_overrides:
|
|
lines.append("**Optimal hyperparameters** (from top 25% of runs):")
|
|
for k, v in sorted(rec.method_overrides.items()):
|
|
lines.append(f" - `{k}`: {v}")
|
|
lines.append("")
|
|
|
|
if rec.best_refusal_rate is not None:
|
|
lines.append(f"**Best refusal rate achieved:** {rec.best_refusal_rate:.1%}")
|
|
if rec.best_coherence is not None:
|
|
lines.append(f"**Best coherence achieved:** {rec.best_coherence:.3f}")
|
|
|
|
lines.append("")
|
|
lines.append(f"*{rec.reason}*")
|
|
|
|
return "\n".join(lines)
|