mirror of
https://github.com/wassname/moral-maps.git
synced 2026-09-22 13:20:36 +08:00
Add versioned DeepSeek WVS reliability pilot
Co-Authored-By: PI[openai-codex] <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
wassname
co-authored by
PI[openai-codex]
PI[openai-codex]
parent
f2b321a0b2
commit
1b6f2d7ef1
@@ -0,0 +1,261 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replicate seven complete DeepSeek WVS v1 panels without touching canonical cache or Pages data."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
from collections import Counter
|
||||
from datetime import UTC, datetime
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from moralmaps.iw_axes import X_AXIS, Y_AXIS, resolve_items
|
||||
from moralmaps.read_api import rated_protocol_identity, read_items_rated
|
||||
from wvs_map import _sample_only_coord_se, load_wvs_all, model_coord_ci
|
||||
from wvs_score_all_options_refresh import OSS_PROVIDER, release, reserve
|
||||
|
||||
EVAL_VERSION = "wvs-score-all-options-v1"
|
||||
MODELS = (
|
||||
"deepseek/deepseek-chat-v3-0324",
|
||||
"deepseek/deepseek-chat-v3.1",
|
||||
"deepseek/deepseek-v3.2",
|
||||
"deepseek/deepseek-v3.2-exp",
|
||||
"deepseek/deepseek-v4-flash",
|
||||
"deepseek/deepseek-v4-flash-0731",
|
||||
"deepseek/deepseek-v4.1-flash",
|
||||
)
|
||||
REPLICATES = 3
|
||||
N_SAMPLES = 24
|
||||
OUT = Path("slop/research/wvs/20260917_deepseek_reliability")
|
||||
MANIFEST = OUT / "manifest.json"
|
||||
RESULTS = OUT / "results.json"
|
||||
STATE = OUT / "budget.json"
|
||||
LOCK = OUT / "budget.lock"
|
||||
CACHE = Path("slop/research/wvs/20260916_openrouter/wvs_iw_rated.json")
|
||||
LEDGER = Path("slop/research/wvs/20260916_openrouter/wvs_iw_requests.jsonl")
|
||||
PILOT_RESERVATION_ID = "pilot/deepseek-wvs-v1-reliability"
|
||||
PILOT_CAP_USD = Decimal("1")
|
||||
|
||||
|
||||
def atomic_json(path: Path, value: object) -> None:
|
||||
temp = path.with_suffix(path.suffix + ".tmp")
|
||||
temp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
|
||||
temp.replace(path)
|
||||
|
||||
|
||||
def pilot_state(update) -> dict:
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
with LOCK.open("w") as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX)
|
||||
state = json.loads(STATE.read_text()) if STATE.exists() else {
|
||||
"schema": 1, "pilot_cap_usd": str(PILOT_CAP_USD), "spent_usd": "0", "reserved_usd": "0",
|
||||
}
|
||||
update(state)
|
||||
atomic_json(STATE, state)
|
||||
fcntl.flock(lock, fcntl.LOCK_UN)
|
||||
return state
|
||||
|
||||
|
||||
def deterministic_seeds(model: str, replicate: int, count: int) -> list[int]:
|
||||
seeds = []
|
||||
for sequence in range(count):
|
||||
digest = hashlib.sha256(f"{EVAL_VERSION}|{model}|{replicate}|{sequence}".encode()).digest()
|
||||
seeds.append(int.from_bytes(digest[:4], "big") & 0x7FFF_FFFF)
|
||||
if len(set(seeds)) != len(seeds):
|
||||
raise RuntimeError(f"seed collision for {model} replicate {replicate}")
|
||||
return seeds
|
||||
|
||||
|
||||
def completed_v1_settings() -> dict[str, dict]:
|
||||
starts = {}
|
||||
finished = set()
|
||||
for line in LEDGER.read_text().splitlines():
|
||||
record = json.loads(line)
|
||||
if record.get("model") not in MODELS:
|
||||
continue
|
||||
if record.get("event") in {"run_started", "request_started"}:
|
||||
starts[record["run_id"]] = record
|
||||
if record.get("event") == "run_finished" and record["valid_samples"] == 144 and record["failed_samples"] == 0:
|
||||
finished.add(record["run_id"])
|
||||
settings = {}
|
||||
for run_id in sorted(finished):
|
||||
start = starts[run_id]
|
||||
settings[start["model"]] = start["settings"]
|
||||
if set(settings) != set(MODELS):
|
||||
raise RuntimeError(f"missing complete v1 DeepSeek settings: {sorted(set(MODELS) - set(settings))}")
|
||||
return settings
|
||||
|
||||
|
||||
def rated_items() -> tuple[list[dict], dict]:
|
||||
resolved = resolve_items(load_wvs_all())
|
||||
items, seen = [], set()
|
||||
for axis in (X_AXIS, Y_AXIS):
|
||||
for item in resolved[axis]:
|
||||
if item["suffix"] in seen:
|
||||
continue
|
||||
seen.add(item["suffix"])
|
||||
items.append({"id": item["suffix"], "question": item["rec"]["q"], "options": item["rec"]["opts"], "n": item["n"]})
|
||||
return items, resolved
|
||||
|
||||
|
||||
def catalog_seed_support() -> dict[str, bool]:
|
||||
catalog = json.loads(Path("slop/research/wvs/20260917_openrouter_models.json").read_text())["data"]
|
||||
by_id = {row["id"]: row for row in catalog}
|
||||
return {model: "seed" in by_id[model].get("supported_parameters", []) for model in MODELS}
|
||||
|
||||
|
||||
def protocol_settings(model: str, v1: dict, seeds: list[int]) -> dict:
|
||||
return {
|
||||
"model": model, "n_samples": N_SAMPLES, "temperature": v1["temperature"],
|
||||
"max_tokens": v1["max_tokens"], "concurrency": 1, "req_timeout": v1["req_timeout"],
|
||||
"reasoning": v1["reasoning"], "structured_output": v1["structured_output"],
|
||||
"provider": v1.get("provider", OSS_PROVIDER), "eval_version": EVAL_VERSION, "seed_schedule": seeds,
|
||||
}
|
||||
|
||||
|
||||
def manifest() -> dict:
|
||||
items, _ = rated_items()
|
||||
settings = completed_v1_settings()
|
||||
seed_support = catalog_seed_support()
|
||||
rows = []
|
||||
for model in MODELS:
|
||||
replicates = []
|
||||
for replicate in range(REPLICATES):
|
||||
seeds = deterministic_seeds(model, replicate, len(items) * N_SAMPLES)
|
||||
cfg = protocol_settings(model, settings[model], seeds)
|
||||
protocol_id = rated_protocol_identity(model, items, **{key: cfg[key] for key in (
|
||||
"n_samples", "temperature", "max_tokens", "concurrency", "req_timeout", "reasoning",
|
||||
"structured_output", "provider", "eval_version", "seed_schedule")})
|
||||
replicates.append({"replicate": replicate, "protocol_id": protocol_id, "seed_schedule": seeds,
|
||||
"records": str(OUT / "records" / model.replace("/", "__") / f"replicate_{replicate}.jsonl")})
|
||||
rows.append({"id": model, "v1_settings": settings[model], "provider_policy": OSS_PROVIDER,
|
||||
"endpoint_advertises_seed": seed_support[model], "replicates": replicates})
|
||||
return {
|
||||
"schema": 1, "eval_version": EVAL_VERSION, "purpose": "reliability replication, not a new evaluator",
|
||||
"models": rows, "items": len(items), "replicates": REPLICATES, "samples_per_item": N_SAMPLES,
|
||||
"expected_calls": len(MODELS) * REPLICATES * len(items) * N_SAMPLES,
|
||||
"pilot_cap_usd": str(PILOT_CAP_USD), "expected_cost_usd": "0.318363267912",
|
||||
"global_cap_reservation_id": PILOT_RESERVATION_ID,
|
||||
"hidden_reasoning": "unavailable when a provider does not return it; returned response fields are preserved verbatim",
|
||||
"created_utc": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def usage(records: Path) -> tuple[Decimal, Counter, int, int]:
|
||||
cost, providers, rescues, failures = Decimal(), Counter(), 0, 0
|
||||
for line in records.read_text().splitlines():
|
||||
record = json.loads(line)
|
||||
if record["event"] == "request_completed":
|
||||
cost += Decimal(str(record.get("usage", {}).get("cost", 0)))
|
||||
providers[record.get("provider")] += 1
|
||||
if record["event"] == "item_result":
|
||||
rescues += record["rescued_samples"]
|
||||
failures += record["failed_samples"]
|
||||
return cost, providers, rescues, failures
|
||||
|
||||
|
||||
def v1_coords(model: str) -> list[float]:
|
||||
cache = json.loads(CACHE.read_text())["completed"]
|
||||
entries = [entry for entry in cache.values() if entry["model"] == model and entry.get("n_samples") == 12]
|
||||
if not entries:
|
||||
raise RuntimeError(f"missing canonical v1 cache entry for {model}")
|
||||
return entries[-1]["coords"]
|
||||
|
||||
|
||||
def run_replicate(model: str, replicate: dict, v1: dict, items: list[dict], resolved: dict) -> dict:
|
||||
records = Path(replicate["records"])
|
||||
records.parent.mkdir(parents=True, exist_ok=True)
|
||||
seeds = replicate["seed_schedule"]
|
||||
rows = read_items_rated(model, items, n_samples=N_SAMPLES, temperature=v1["temperature"],
|
||||
max_tokens=v1["max_tokens"], concurrency=1, req_timeout=v1["req_timeout"],
|
||||
reasoning=v1["reasoning"], structured_output=v1["structured_output"], provider=v1.get("provider", OSS_PROVIDER),
|
||||
records_path=records, verbose_first=True, probe_first=True, eval_version=EVAL_VERSION,
|
||||
identity_eval_version=EVAL_VERSION, seed_schedule=seeds)
|
||||
if any(row["valid_samples"] != N_SAMPLES for row in rows):
|
||||
raise RuntimeError(f"incomplete replicate {model} {replicate['replicate']}")
|
||||
psamples = {row["id"]: np.asarray(row["p_samples"]) for row in rows}
|
||||
coords = model_coord_ci(psamples, resolved, np.random.default_rng(0))
|
||||
response_se = _sample_only_coord_se(psamples, resolved, np.random.default_rng(1), n_draws=N_SAMPLES)
|
||||
cost, providers, rescues, failures = usage(records)
|
||||
return {"replicate": replicate["replicate"], "protocol_id": replicate["protocol_id"], "records": str(records),
|
||||
"coords": list(coords), "response_mean_se": list(response_se), "provider_requests": dict(providers),
|
||||
"rescues": rescues, "failures": failures, "usage_cost_usd": str(cost)}
|
||||
|
||||
|
||||
def run() -> None:
|
||||
data = json.loads(MANIFEST.read_text())
|
||||
items, resolved = rated_items()
|
||||
if not reserve({"id": PILOT_RESERVATION_ID, "lane": "deepseek", "reserve_usd": str(PILOT_CAP_USD)}):
|
||||
raise RuntimeError("global USD 80 cap would be exceeded by the USD 1 pilot reservation")
|
||||
pilot_state(lambda state: state.update({"reserved_usd": str(PILOT_CAP_USD), "started_utc": datetime.now(UTC).isoformat()}))
|
||||
results = {"eval_version": EVAL_VERSION, "models": [], "not_published": True}
|
||||
try:
|
||||
for row in data["models"]:
|
||||
reps = []
|
||||
for replicate in row["replicates"]:
|
||||
result = run_replicate(row["id"], replicate, row["v1_settings"], items, resolved)
|
||||
spent = pilot_state(lambda state: state.update({"spent_usd": str(Decimal(state["spent_usd"]) + Decimal(result["usage_cost_usd"]))}))
|
||||
if Decimal(spent["spent_usd"]) >= PILOT_CAP_USD:
|
||||
raise RuntimeError(f"pilot cap reached: {spent['spent_usd']} >= {PILOT_CAP_USD}")
|
||||
reps.append(result)
|
||||
atomic_json(RESULTS, results | {"models": results["models"] + [{"id": row["id"], "replicates": reps}]})
|
||||
values = np.asarray([rep["coords"][:2] for rep in reps])
|
||||
aggregate_samples = {}
|
||||
for rep in reps:
|
||||
rows = [json.loads(line) for line in Path(rep["records"]).read_text().splitlines()]
|
||||
for record in rows:
|
||||
if record["event"] == "item_result":
|
||||
aggregate_samples.setdefault(record["id"], []).extend(record["p_samples"])
|
||||
aggregate = model_coord_ci({key: np.asarray(value) for key, value in aggregate_samples.items()}, resolved, np.random.default_rng(2))
|
||||
v1 = v1_coords(row["id"])
|
||||
results["models"].append({"id": row["id"], "v1_coords": v1, "replicates": reps,
|
||||
"between_replicate_coordinate_sd": np.std(values, axis=0, ddof=1).tolist(),
|
||||
"aggregate_n72_coords": list(aggregate),
|
||||
"delta_aggregate_minus_v1": (np.asarray(aggregate[:2]) - np.asarray(v1[:2])).tolist()})
|
||||
atomic_json(RESULTS, results)
|
||||
finally:
|
||||
release(PILOT_RESERVATION_ID)
|
||||
pilot_state(lambda state: state.update({"reserved_usd": "0", "finished_utc": datetime.now(UTC).isoformat()}))
|
||||
|
||||
|
||||
def smoke() -> None:
|
||||
data = json.loads(MANIFEST.read_text())
|
||||
row, replicate = data["models"][0], data["models"][0]["replicates"][0]
|
||||
items, _ = rated_items()
|
||||
full_identity = replicate["protocol_id"]
|
||||
record = OUT / "smoke.jsonl"
|
||||
rows = read_items_rated(row["id"], [items[0]], n_samples=1, temperature=row["v1_settings"]["temperature"],
|
||||
max_tokens=row["v1_settings"]["max_tokens"], concurrency=1, req_timeout=row["v1_settings"]["req_timeout"],
|
||||
reasoning=row["v1_settings"]["reasoning"], structured_output=row["v1_settings"]["structured_output"],
|
||||
provider=row["v1_settings"].get("provider", OSS_PROVIDER), records_path=record, probe_first=True,
|
||||
eval_version=EVAL_VERSION, identity_eval_version=EVAL_VERSION,
|
||||
seed_schedule=[replicate["seed_schedule"][0]])
|
||||
if rows[0]["valid_samples"] != 1:
|
||||
raise RuntimeError("one-request smoke was not parse-valid")
|
||||
atomic_json(OUT / "smoke.json", {"full_panel_protocol_id": full_identity, "smoke_model": row["id"],
|
||||
"smoke_item": items[0]["id"], "seed": replicate["seed_schedule"][0],
|
||||
"settings": row["v1_settings"], "endpoint_advertises_seed": row["endpoint_advertises_seed"],
|
||||
"result": rows[0]})
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--write-manifest", action="store_true")
|
||||
parser.add_argument("--smoke", action="store_true")
|
||||
parser.add_argument("--run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
if args.write_manifest:
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
atomic_json(MANIFEST, manifest())
|
||||
if args.smoke:
|
||||
smoke()
|
||||
if args.run:
|
||||
run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -208,7 +208,8 @@ def _rate_plan(items: list[dict], n_samples: int, per_call: int = 1) -> list[dic
|
||||
def rated_protocol_identity(model: str, items: list[dict], *, n_samples: int, temperature: float,
|
||||
max_tokens: int, concurrency: int, req_timeout: float,
|
||||
reasoning: dict | None, structured_output: bool,
|
||||
provider: dict | None = None) -> str:
|
||||
provider: dict | None = None, eval_version: str | None = None,
|
||||
seed_schedule: list[int] | None = None) -> str:
|
||||
"""Hash the exact model, rendered prompts, and request settings that define a cacheable panel."""
|
||||
plan = _rate_plan(items, n_samples)
|
||||
protocol = {
|
||||
@@ -226,6 +227,10 @@ def rated_protocol_identity(model: str, items: list[dict], *, n_samples: int, te
|
||||
"requests": [{key: req[key] for key in ("i", "perm", "prompt", "cnt", "sample", "presented_options")}
|
||||
for req in plan],
|
||||
}
|
||||
if eval_version is not None:
|
||||
protocol["eval_version"] = eval_version
|
||||
if seed_schedule is not None:
|
||||
protocol["seed_schedule"] = seed_schedule
|
||||
encoded = json.dumps(protocol, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
|
||||
return hashlib.sha256(encoded).hexdigest()
|
||||
|
||||
@@ -242,7 +247,8 @@ def read_items_rated(model: str, items: list[dict], *, n_samples: int = 12, temp
|
||||
max_tokens: int = 512, concurrency: int = 8, req_timeout: float = 90.0,
|
||||
reasoning: dict | None = None, structured_output: bool = False, records_path: str | Path,
|
||||
verbose_first: bool = False, provider: dict | None = None,
|
||||
probe_first: bool = False) -> list[dict]:
|
||||
probe_first: bool = False, eval_version: str = "wvs-score-all-options-v1",
|
||||
identity_eval_version: str | None = None, seed_schedule: list[int] | None = None) -> list[dict]:
|
||||
"""Run one score-all-options panel and write an fsynced JSONL event for every paid request phase.
|
||||
|
||||
The record is the source of truth. It preserves dispatches, responses, rescues, provider usage,
|
||||
@@ -251,17 +257,21 @@ def read_items_rated(model: str, items: list[dict], *, n_samples: int = 12, temp
|
||||
"""
|
||||
assert temperature > 0, "sampling readout needs temperature > 0"
|
||||
plan = _rate_plan(items, n_samples)
|
||||
if seed_schedule is not None and len(seed_schedule) != len(plan):
|
||||
raise ValueError(f"seed schedule has {len(seed_schedule)} entries, expected {len(plan)}")
|
||||
protocol_id = rated_protocol_identity(model, items, n_samples=n_samples, temperature=temperature,
|
||||
max_tokens=max_tokens, concurrency=concurrency,
|
||||
req_timeout=req_timeout, reasoning=reasoning,
|
||||
structured_output=structured_output, provider=provider)
|
||||
structured_output=structured_output, provider=provider,
|
||||
eval_version=identity_eval_version, seed_schedule=seed_schedule)
|
||||
run_id = f"{datetime.now(UTC).strftime('%Y%m%dT%H%M%SZ')}_{protocol_id[:12]}"
|
||||
rpath = Path(records_path)
|
||||
rpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
settings = {"model": model, "n_samples": n_samples, "temperature": temperature,
|
||||
"max_tokens": max_tokens, "concurrency": concurrency, "req_timeout": req_timeout,
|
||||
"reasoning": reasoning, "structured_output": structured_output, "provider": provider,
|
||||
"probe_first": probe_first}
|
||||
"probe_first": probe_first, "eval_version": eval_version,
|
||||
"identity_eval_version": identity_eval_version, "seed_schedule": seed_schedule}
|
||||
_append_record(rpath, {"event": "run_started", "run_id": run_id, "protocol_id": protocol_id,
|
||||
"settings": settings, "items": items, "planned_requests": len(plan)})
|
||||
|
||||
@@ -271,12 +281,16 @@ def read_items_rated(model: str, items: list[dict], *, n_samples: int = 12, temp
|
||||
async def call(seq: int, req: dict) -> dict:
|
||||
item = items[req["i"]]
|
||||
request_id = f"{run_id}_{seq:03d}"
|
||||
seed = seed_schedule[seq] if seed_schedule is not None else None
|
||||
request_meta = {"request_id": request_id, "run_id": run_id, "protocol_id": protocol_id,
|
||||
"model": model, "item_id": item["id"], "canonical_options": item["options"],
|
||||
"presented_options": req["presented_options"], "presented_order": req["perm"],
|
||||
"sample": req["sample"], "prompt": req["prompt"], "settings": settings}
|
||||
"sample": req["sample"], "prompt": req["prompt"], "settings": settings,
|
||||
"eval_version": eval_version, "seed": seed}
|
||||
payload = {"model": model, "messages": [{"role": "user", "content": req["prompt"]}],
|
||||
"temperature": temperature, "n": req["cnt"], "max_tokens": max_tokens}
|
||||
if seed is not None:
|
||||
payload["seed"] = seed
|
||||
if reasoning is not None:
|
||||
payload["reasoning"] = reasoning
|
||||
if provider is not None:
|
||||
|
||||
Reference in New Issue
Block a user