176 lines
7.9 KiB
Python
176 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Constraint probe for the ADAPTIVE-MEMORY research (docs/research/14-adaptive-memory.md).
|
||
|
||
Question it answers empirically (mandate: «проверяй заблокированность методов пробой,
|
||
а не по памяти» — ADAPTIVE_MEMORY_RESEARCH_PROMPT §Жёсткий контур п.1): what LOGIT-LEVEL
|
||
control do our closed flagship providers actually expose in 2026? This bounds which
|
||
"learns-during-the-book" mechanisms are runnable on the flagship TRANSLATOR:
|
||
|
||
logit-level lever needs from the API available on flagship?
|
||
------------------------------------------------------------------------------------
|
||
kNN-MT (Khandelwal 2021) DECODER HIDDEN STATES per step NO API exposes this
|
||
constrained/beam decoding beam access / token masking logit_bias at best
|
||
output-distribution reading per-token logprobs <- THIS we probe live
|
||
hard steering logit_bias honored <- THIS we probe live
|
||
|
||
So even the BEST case (logprobs + logit_bias both honored) does NOT unblock kNN-MT: the
|
||
datastore key in kNN-MT is the decoder's context representation, which none of these APIs
|
||
return. This probe measures the ceiling (logprobs/logit_bias), and the ceiling is still
|
||
below what kNN-MT needs — a doubly-closed door. Local decoders (llama.cpp/vLLM) DO expose
|
||
logprobs/logits by design (n_probs / prompt_logprobs), so local kNN-MT is physically
|
||
possible; that half is asserted from the runtimes' documented APIs, not reprobed here (the
|
||
stand GPU is held by the polygon's resident model — do not evict, exp03/§coordination).
|
||
|
||
Cheap by design: ~1 short call per provider. Reuses eval/.env + eval/providers.json quirks.
|
||
Usage: eval/.venv/bin/python eval/adaptive_probe.py
|
||
Output: stdout table + eval/data/adaptive_probe.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
OUT = ROOT / "data" / "adaptive_probe.json"
|
||
|
||
|
||
def load_env() -> None:
|
||
for cand in (ROOT / ".env", Path("~/.textmachine.env").expanduser()):
|
||
if not cand.exists():
|
||
continue
|
||
for line in cand.read_text().splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
if v.strip():
|
||
os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
|
||
return
|
||
|
||
|
||
load_env()
|
||
|
||
# The models we actually probe. Includes the REAL translator/editor stack models
|
||
# (deepseek-v4-flash draft, grok-4.3 editor, gemini-3.1-pro premium) plus a clean
|
||
# non-reasoning path per family to give logprobs its best chance to appear.
|
||
PROBES = [
|
||
{"name": "deepseek-v4-flash", "base": "https://api.deepseek.com/v1",
|
||
"model": "deepseek-v4-flash", "key": "DEEPSEEK_API_KEY", "max": 200},
|
||
{"name": "grok-non-reasoning", "base": "https://api.x.ai/v1",
|
||
"model": "grok-4.20-0309-non-reasoning", "key": "XAI_API_KEY", "max": 32},
|
||
{"name": "grok-4.3(editor)", "base": "https://api.x.ai/v1",
|
||
"model": "grok-4.3", "key": "XAI_API_KEY", "max": 200},
|
||
{"name": "glm-4.5-air", "base": "https://api.z.ai/api/paas/v4",
|
||
"model": "glm-4.5-air", "key": "ZAI_API_KEY", "max": 64,
|
||
"extra": {"thinking": {"type": "disabled"}}},
|
||
{"name": "gemini-2.5-flash", "base": "https://generativelanguage.googleapis.com/v1beta/openai",
|
||
"model": "gemini-2.5-flash", "key": "GEMINI_API_KEY", "max": 64,
|
||
"extra": {"extra_body": {"google": {"thinking_config": {"thinking_budget": 0}}}}},
|
||
{"name": "gemini-3.1-pro(premium)", "base": "https://generativelanguage.googleapis.com/v1beta/openai",
|
||
"model": "gemini-3.1-pro-preview", "key": "GEMINI_API_KEY", "max": 8000},
|
||
{"name": "gpt-5-mini", "base": "https://api.openai.com/v1",
|
||
"model": "gpt-5-mini", "key": "OPENAI_API_KEY", "max": 8000, "reasoning": True},
|
||
]
|
||
|
||
|
||
def one_call(p: dict, with_logprobs: bool, with_logit_bias: bool):
|
||
"""Return dict describing what the provider did with logprobs / logit_bias."""
|
||
key = os.environ.get(p["key"], "")
|
||
if not key:
|
||
return {"skip": f"no key {p['key']}"}
|
||
body = {
|
||
"model": p["model"],
|
||
"messages": [{"role": "user", "content": "Reply with exactly the single word: ok"}],
|
||
"max_tokens": p["max"],
|
||
}
|
||
if p.get("reasoning"):
|
||
body["max_completion_tokens"] = body.pop("max_tokens")
|
||
body["reasoning_effort"] = "minimal"
|
||
else:
|
||
body["temperature"] = 0.0
|
||
body.update(p.get("extra", {}))
|
||
if with_logprobs:
|
||
body["logprobs"] = True
|
||
body["top_logprobs"] = 5
|
||
if with_logit_bias:
|
||
# bias an arbitrary token id hard; we only check acceptance, not effect.
|
||
body["logit_bias"] = {"1734": -100}
|
||
req = urllib.request.Request(
|
||
p["base"].rstrip("/") + "/chat/completions",
|
||
data=json.dumps(body).encode(),
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
data = json.loads(resp.read())
|
||
except urllib.error.HTTPError as e:
|
||
return {"http_error": e.code, "detail": e.read().decode(errors="replace")[:300].replace("\n", " ")}
|
||
except Exception as e:
|
||
return {"transport_error": f"{type(e).__name__}: {e}"}
|
||
ch = (data.get("choices") or [{}])[0]
|
||
lp = ch.get("logprobs")
|
||
# logprobs "present" = non-null object carrying per-token content entries
|
||
has_lp = bool(lp) and bool((lp or {}).get("content"))
|
||
sample = None
|
||
if has_lp:
|
||
first = lp["content"][0]
|
||
sample = {"token": first.get("token"),
|
||
"logprob": round(first.get("logprob", 0.0), 3),
|
||
"n_top": len(first.get("top_logprobs", []))}
|
||
return {
|
||
"http": 200,
|
||
"content": (ch.get("message", {}).get("content") or "")[:40],
|
||
"finish": ch.get("finish_reason"),
|
||
"logprobs_present": has_lp,
|
||
"logprobs_sample": sample,
|
||
"usage": {k: data.get("usage", {}).get(k) for k in ("prompt_tokens", "completion_tokens")},
|
||
}
|
||
|
||
|
||
def main():
|
||
results = []
|
||
print(f"{'provider':<26} {'logprobs?':<11} {'logit_bias':<12} sample / note")
|
||
print("-" * 88)
|
||
for p in PROBES:
|
||
r_lp = one_call(p, with_logprobs=True, with_logit_bias=False)
|
||
r_lb = one_call(p, with_logprobs=False, with_logit_bias=True)
|
||
rec = {"provider": p["name"], "model": p["model"], "logprobs": r_lp, "logit_bias": r_lb}
|
||
results.append(rec)
|
||
|
||
if "skip" in r_lp:
|
||
print(f"{p['name']:<26} {'SKIP':<11} {'':<12} {r_lp['skip']}")
|
||
continue
|
||
if r_lp.get("http") == 200:
|
||
lp_verdict = "YES" if r_lp["logprobs_present"] else "no(200)"
|
||
elif "http_error" in r_lp:
|
||
lp_verdict = f"HTTP{r_lp['http_error']}"
|
||
else:
|
||
lp_verdict = "ERR"
|
||
# logit_bias: 200 = accepted (not rejected); HTTP4xx = rejected
|
||
if r_lb.get("http") == 200:
|
||
lb_verdict = "accepted"
|
||
elif "http_error" in r_lb:
|
||
lb_verdict = f"HTTP{r_lb['http_error']}"
|
||
else:
|
||
lb_verdict = "ERR"
|
||
note = ""
|
||
if r_lp.get("logprobs_sample"):
|
||
s = r_lp["logprobs_sample"]
|
||
note = f"tok={s['token']!r} lp={s['logprob']} top{s['n_top']}"
|
||
elif "detail" in r_lp:
|
||
note = r_lp["detail"][:60]
|
||
elif r_lp.get("http") == 200 and not r_lp["logprobs_present"]:
|
||
note = f"finish={r_lp.get('finish')} content={r_lp.get('content')!r}"
|
||
print(f"{p['name']:<26} {lp_verdict:<11} {lb_verdict:<12} {note}")
|
||
|
||
OUT.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"\nsaved -> {OUT}")
|
||
print("\nReading: 'logprobs YES' = output distribution readable (still NOT decoder hidden\n"
|
||
"states → kNN-MT still blocked). 'logit_bias accepted' = crude token steering only.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|