3ff504143a
- старые записи инбокса без chat_type (до t-261013113629) считались личкой и обходили фильтр адресации — группа теперь и по chat_id < 0; - утренний холодный старт clef:27b в облаке (> 15 с) взводил предохранитель на 5 минут → ответы от CPU-nimble по 10 с: облако ждём до 30 с, его кулдаун 60 с (локальной — по-прежнему 5 минут). Реплей реальной истории «Правления» (62 сообщения, бот отвечает только на адресованное, облако ~0.66 с): приветствия и переписка людей между собой отфильтрованы; разговор Alex с ботом без упоминания распознан по контексту; спорные — 2–3 в каждую сторону, «Коди»/@/reply — всегда. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
394 lines
19 KiB
Python
394 lines
19 KiB
Python
"""Simbiot Harness — decision-модель (Ollama /v1/systemone: облачная clef → локальная nimble).
|
||
|
||
Схема из systemone-bench (ga/router42.py, последняя версия): детерминированная
|
||
механика остаётся быстрым путём, decision-модель — арбитр там, где механика
|
||
слепа (FIFO по задачам, round-robin по темам, порог гейта). Недоступна
|
||
модель / таймаут / мусор в ответе → None, вызывающий берёт свой фолбэк.
|
||
Тик никогда не падает из-за модели.
|
||
|
||
Контракт эндпоинта (README бенча, пробы 2026-10-04):
|
||
- choice: criteria — объект {ключ: описание}, 2–26 опций; ответ — ключ +
|
||
probabilities. Точность 100% на шортлисте 6 → 50% на 10, поэтому здесь
|
||
жёсткий кап MAX_OPTIONS = 9 (вызывающий режет шортлист сам);
|
||
- noul: P(«да») в [0, 1];
|
||
- confidence — концентрация распределения, НЕ вероятность правильности.
|
||
|
||
Бэкенды по порядку (у каждого свой предохранитель):
|
||
1) основной — SIMBIOT_DECISION_URL/MODEL/API_KEY. С ключом Ollama Cloud:
|
||
https://ollama.com/v1/systemone, clef:27b — ~0.6–0.8 с на запрос;
|
||
2) запасной — локальная nimble (SIMBIOT_DECISION_FALLBACK_URL/MODEL,
|
||
SIMBIOT_DECISION_FALLBACK=off — выключить): CPU, 10–30 с, холодный
|
||
старт ~20 с, зато без сети и квот.
|
||
Без API_KEY и URL основной = локальная nimble, запасного нет (как раньше).
|
||
|
||
Ответы детерминированы (один и тот же запрос → те же вероятности), поэтому
|
||
кеш по хешу вопроса+состояния валиден: одинаковая очередь в соседних тиках
|
||
не стоит повторного прогона.
|
||
|
||
Режим — SIMBIOT_DECISION (env процесса сильнее .env инстанса):
|
||
off (дефолт) — модель не зовётся; on — решения применяются;
|
||
shadow — модель зовётся и решение пишется событием, но не применяется.
|
||
Точечный оверрайд потребителя: SIMBIOT_DECISION_<PURPOSE>=off|shadow|on
|
||
(например SIMBIOT_DECISION_GATE=on). Гейт по умолчанию — shadow: пропуск
|
||
LLM-тика по мнению модели включается только явно.
|
||
|
||
Каждый реальный вызов пишет DecisionMade (purpose, опции, выбор,
|
||
вероятности, латентность, applied) — policy trace для будущей GA-сшивки
|
||
с исходом (skills/systemone-bench, §4 «Интеграция в тик харнеса»).
|
||
"""
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
from . import paths as _paths
|
||
|
||
MAX_OPTIONS = 9 # шортлист stage2 строго ≤ 9 (бенч: 100% на 6 → 50% на 10)
|
||
CACHE_TTL = 3600.0 # ответы детерминированы — час держим
|
||
CACHE_MAX = 256
|
||
BREAKER_COOLDOWN = 300.0 # после сбоя молчим 5 минут, не тормозим тики таймаутами
|
||
# облако: холодный старт clef после простоя бывает > 15 с (утро 2026-10-17:
|
||
# один таймаут → 5 минут ответов от CPU-nimble по 10 с) — короткий кулдаун
|
||
CLOUD_BREAKER_COOLDOWN = 60.0
|
||
|
||
_LOCK = threading.Lock()
|
||
_CACHE = {} # sha256(вопросы+состояние) → (ts, answers, model)
|
||
_BREAKERS = {} # имя бэкенда → {"until", "error"}
|
||
|
||
LOCAL_URL = "http://127.0.0.1:11434/v1/systemone"
|
||
CLOUD_URL = "https://ollama.com/v1/systemone"
|
||
|
||
# тексты ниже уходят в модель — на английском: nimble (qwen3.5) калибровался
|
||
# в бенче на английских инструкциях
|
||
AUM_GOAL = ("Goal of the harness: maximize long-term AUM of the cooperative "
|
||
"(income and resources of all projects over a year+ horizon). "
|
||
"Prefer work that adds >=1% to AUM in a year over short wins that "
|
||
"eat the long-term base.")
|
||
|
||
|
||
# ────────────────────────── конфиг ──────────────────────────
|
||
|
||
def _cfg(key: str, default: str = "") -> str:
|
||
"""env процесса > .env инстанса (как runtime._env_or_file, без импорта runtime)."""
|
||
val = os.environ.get(key, "").strip()
|
||
if val:
|
||
return val
|
||
try:
|
||
with open(_paths.env_path(), encoding="utf-8") as f:
|
||
for line in f:
|
||
if line.startswith(key + "="):
|
||
return line.split("=", 1)[1].strip().strip('"').strip("'") or default
|
||
except OSError:
|
||
pass
|
||
return default
|
||
|
||
|
||
def _float(key: str, default: float) -> float:
|
||
try:
|
||
return float(_cfg(key, str(default)))
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def backends() -> list:
|
||
"""[{name, url, model, key, timeout}] в порядке опроса."""
|
||
key = _cfg("SIMBIOT_DECISION_API_KEY")
|
||
if key:
|
||
primary = {"name": "primary", "key": key,
|
||
"url": _cfg("SIMBIOT_DECISION_URL", CLOUD_URL),
|
||
"model": _cfg("SIMBIOT_DECISION_MODEL", "clef:27b"),
|
||
"timeout": _float("SIMBIOT_DECISION_TIMEOUT", 30.0)}
|
||
else:
|
||
primary = {"name": "primary", "key": "",
|
||
"url": _cfg("SIMBIOT_DECISION_URL", LOCAL_URL),
|
||
"model": _cfg("SIMBIOT_DECISION_MODEL", "nimble"),
|
||
"timeout": _float("SIMBIOT_DECISION_TIMEOUT", 90.0)}
|
||
out = [primary]
|
||
fb_default = "on" if primary["key"] else "off" # запасной — только за облаком
|
||
if _cfg("SIMBIOT_DECISION_FALLBACK", fb_default).lower() == "on":
|
||
out.append({"name": "fallback", "key": "",
|
||
"url": _cfg("SIMBIOT_DECISION_FALLBACK_URL", LOCAL_URL),
|
||
"model": _cfg("SIMBIOT_DECISION_FALLBACK_MODEL", "nimble"),
|
||
"timeout": _float("SIMBIOT_DECISION_FALLBACK_TIMEOUT", 90.0)})
|
||
return out
|
||
|
||
|
||
def url() -> str:
|
||
return backends()[0]["url"]
|
||
|
||
|
||
def model_name() -> str:
|
||
return backends()[0]["model"]
|
||
|
||
|
||
_PURPOSE_DEFAULTS = {"gate": "shadow"}
|
||
|
||
|
||
def mode(purpose: str = "") -> str:
|
||
"""off | shadow | on для потребителя purpose (глобальный режим — потолок)."""
|
||
glob = _cfg("SIMBIOT_DECISION", "off").lower()
|
||
if glob not in ("on", "shadow"):
|
||
return "off"
|
||
if purpose:
|
||
local = _cfg("SIMBIOT_DECISION_" + purpose.upper(),
|
||
_PURPOSE_DEFAULTS.get(purpose, glob)).lower()
|
||
if local not in ("on", "shadow"):
|
||
return "off"
|
||
if glob == "shadow":
|
||
return "shadow"
|
||
return local
|
||
return glob
|
||
|
||
|
||
# ────────────────────────── транспорт ──────────────────────────
|
||
|
||
def _post(be: dict, body: dict) -> dict:
|
||
raw = json.dumps(body, ensure_ascii=False).encode()
|
||
headers = {"Content-Type": "application/json"}
|
||
if be["key"]:
|
||
headers["Authorization"] = "Bearer " + be["key"]
|
||
req = urllib.request.Request(be["url"], data=raw, method="POST", headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=be["timeout"]) as r:
|
||
return json.loads(r.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
# тело ошибки systemone информативно («model not found», «not a decision model»)
|
||
raise ValueError(f"HTTP {e.code}: {e.read().decode(errors='replace')[:160]}")
|
||
|
||
|
||
def ask(questions: dict, state: str):
|
||
"""Сырой вызов: {qid: question} → (answers, meta) | (None, meta).
|
||
|
||
Бэкенды по порядку (backends()); сбой бэкенда взводит ЕГО предохранитель
|
||
на BREAKER_COOLDOWN и сразу пробует следующий. meta: {"cached", "ms",
|
||
"error", "model", "backend"}; error — сводка сбоев, если все упали.
|
||
"""
|
||
state = state[:24000]
|
||
key = hashlib.sha256(json.dumps({"s": state, "q": questions}, sort_keys=True,
|
||
ensure_ascii=False).encode()).hexdigest()
|
||
now = time.time()
|
||
with _LOCK:
|
||
hit = _CACHE.get(key)
|
||
if hit and now - hit[0] < CACHE_TTL:
|
||
return hit[1], {"cached": True, "ms": 0, "error": "", "model": hit[2],
|
||
"backend": "cache"}
|
||
errors = []
|
||
t0 = time.time()
|
||
for be in backends():
|
||
with _LOCK:
|
||
br = _BREAKERS.get(be["name"]) or {}
|
||
if time.time() < br.get("until", 0.0):
|
||
errors.append(f"{be['name']}: breaker ({br.get('error')})")
|
||
continue
|
||
body = {"model": be["model"], "state": state, "questions": questions}
|
||
if not be["key"]:
|
||
body["keep_alive"] = _cfg("SIMBIOT_DECISION_KEEP_ALIVE", "2h")
|
||
t1 = time.time()
|
||
try:
|
||
d = _post(be, body)
|
||
answers = d.get("answers")
|
||
if not isinstance(answers, dict):
|
||
raise ValueError(f"нет answers: {str(d)[:120]}")
|
||
except (urllib.error.URLError, OSError, ValueError, TimeoutError) as e:
|
||
err = f"{e.__class__.__name__}: {str(e)[:160]}"
|
||
with _LOCK:
|
||
cool = CLOUD_BREAKER_COOLDOWN if be["key"] else BREAKER_COOLDOWN
|
||
_BREAKERS[be["name"]] = {"until": time.time() + cool, "error": err}
|
||
errors.append(f"{be['name']} {be['model']}: {err} "
|
||
f"({int((time.time() - t1) * 1000)} мс)")
|
||
continue
|
||
with _LOCK:
|
||
if len(_CACHE) >= CACHE_MAX:
|
||
for k in sorted(_CACHE, key=lambda k: _CACHE[k][0])[:CACHE_MAX // 4]:
|
||
_CACHE.pop(k, None)
|
||
_CACHE[key] = (time.time(), answers, be["model"])
|
||
_BREAKERS.pop(be["name"], None)
|
||
return answers, {"cached": False, "ms": int((time.time() - t1) * 1000),
|
||
"error": "; ".join(errors), "model": be["model"],
|
||
"backend": be["name"]}
|
||
return None, {"cached": False, "ms": int((time.time() - t0) * 1000),
|
||
"error": "; ".join(errors) or "нет бэкендов", "model": None,
|
||
"backend": None}
|
||
|
||
|
||
def _record(purpose: str, kind: str, payload: dict, meta: dict, applied: bool):
|
||
"""DecisionMade в стор — только на реальный вызов (кеш-хиты не спамят стор)."""
|
||
if meta.get("cached"):
|
||
return
|
||
try:
|
||
from . import es
|
||
rid = hashlib.sha1(json.dumps(payload, sort_keys=True, ensure_ascii=False)
|
||
.encode()).hexdigest()[:12]
|
||
es.append_event("decision", f"{purpose}:{int(time.time())}:{rid}", "DecisionMade", {
|
||
"purpose": purpose, "kind": kind, "model": meta.get("model"),
|
||
"backend": meta.get("backend"),
|
||
"ms": meta.get("ms"), "error": meta.get("error") or None,
|
||
"applied": applied, **payload,
|
||
})
|
||
except Exception:
|
||
pass # трасса — best effort, решение важнее записи
|
||
|
||
|
||
# ────────────────────────── примитивы ──────────────────────────
|
||
|
||
def choose(purpose: str, options: dict, state: str, instructions: str):
|
||
"""Выбор одного ключа из options {ключ: описание} (2..MAX_OPTIONS).
|
||
|
||
Возвращает {"choice", "probabilities", "confidence", "applied"} или None
|
||
(режим off, <2 опций, модель недоступна, ответ вне опций). applied=False
|
||
в shadow — вызывающий обязан тогда остаться на своём фолбэке.
|
||
"""
|
||
m = mode(purpose)
|
||
if m == "off" or len(options) < 2:
|
||
return None
|
||
opts = dict(list(options.items())[:MAX_OPTIONS])
|
||
crit = {k: (v or k)[:400] for k, v in opts.items()}
|
||
answers, meta = ask({"c": {"type": "choice", "instructions": instructions,
|
||
"criteria": crit}}, state)
|
||
a = (answers or {}).get("c") or {}
|
||
pick = a.get("choice")
|
||
if pick not in opts:
|
||
_record(purpose, "choice", {"options": list(opts), "choice": None}, meta, False)
|
||
return None
|
||
probs = {k: round(float(v), 4) for k, v in (a.get("probabilities") or {}).items()}
|
||
applied = m == "on"
|
||
_record(purpose, "choice", {"options": list(opts), "choice": pick,
|
||
"probabilities": probs}, meta, applied)
|
||
return {"choice": pick, "probabilities": probs,
|
||
"confidence": round(float(a.get("confidence") or 0), 4),
|
||
"applied": applied}
|
||
|
||
|
||
def yes_prob(purpose: str, question: str, state: str):
|
||
"""noul: P(«да») на вопрос. {"p", "applied"} или None."""
|
||
m = mode(purpose)
|
||
if m == "off":
|
||
return None
|
||
answers, meta = ask({"n": {"type": "noul", "instructions": question}}, state)
|
||
p = ((answers or {}).get("n") or {}).get("noul")
|
||
if not isinstance(p, (int, float)):
|
||
_record(purpose, "noul", {"question": question[:300], "p": None}, meta, False)
|
||
return None
|
||
applied = m == "on"
|
||
_record(purpose, "noul", {"question": question[:300], "p": round(float(p), 4)},
|
||
meta, applied)
|
||
return {"p": float(p), "applied": applied}
|
||
|
||
|
||
# ────────────────────────── потребители харнеса ──────────────────────────
|
||
|
||
def _policy_line(model: dict) -> str:
|
||
pol = (model or {}).get("policy") or {}
|
||
aum = pol.get("aum_target")
|
||
if isinstance(aum, dict):
|
||
aum = json.dumps(aum, ensure_ascii=False)
|
||
return f"Current AUM policy: {str(aum)[:300]}" if aum else ""
|
||
|
||
|
||
def prioritize_tasks(level_name: str, tasks: list, model: dict):
|
||
"""Какая из ready-задач (шортлист ≤ 9, старейшие) даст больше AUM — её первой.
|
||
|
||
tasks — уже отсортированы FIFO. Возвращает id выбранной или None
|
||
(тогда остаётся FIFO). Состояние без времени и счётчиков — чтобы
|
||
одинаковая очередь попадала в кеш между тиками.
|
||
"""
|
||
short = tasks[:MAX_OPTIONS]
|
||
if len(short) < 2:
|
||
return None
|
||
opts = {}
|
||
for t in short:
|
||
desc = f"[{t.get('size')}] {t.get('title') or ''}"
|
||
if t.get("details"):
|
||
desc += " — " + t["details"][:120]
|
||
opts[t["id"]] = desc
|
||
state = "\n".join(x for x in (
|
||
AUM_GOAL, _policy_line(model),
|
||
f"Harness level '{level_name}' picks the task to execute FIRST this tick. "
|
||
"Older tasks are listed first; a task blocking money, clients or "
|
||
"other people outweighs internal polish.") if x)
|
||
r = choose("tasks", opts, state,
|
||
"Which task should be executed first to maximize long-term AUM?")
|
||
return r["choice"] if r and r["applied"] else None
|
||
|
||
|
||
def gate_wants_llm(level_name: str, signals: list):
|
||
"""Серая зона гейта (только warn/info-сигналы): стоит ли тратить кредиты роутера.
|
||
|
||
Возвращает (allow: bool | None, p). None — решения нет/shadow: гейт
|
||
остаётся на пороговой механике. Порог SIMBIOT_DECISION_GATE_MIN_P (0.35).
|
||
"""
|
||
if not signals:
|
||
return None, None
|
||
lines = [f"- {s[0]}: {s[1]} — {s[2]}" for s in signals[:10]]
|
||
state = (AUM_GOAL + "\nA full LLM tick costs router credits (direct minus to AUM) "
|
||
f"and takes minutes. Level '{level_name}' sees these non-critical "
|
||
"signals:\n" + "\n".join(lines))
|
||
r = yes_prob("gate", "Do these signals need attention of an expensive LLM "
|
||
"agent right now, rather than waiting for the next scheduled tick?",
|
||
state)
|
||
if not r:
|
||
return None, None
|
||
try:
|
||
thr = float(_cfg("SIMBIOT_DECISION_GATE_MIN_P", "0.35"))
|
||
except ValueError:
|
||
thr = 0.35
|
||
if not r["applied"]:
|
||
return None, r["p"]
|
||
return r["p"] >= thr, r["p"]
|
||
|
||
|
||
def pick_topic(topics: list, counts: dict, model: dict):
|
||
"""Тема инициативной реплики: шортлист ≤ 9 наименее использованных → choice.
|
||
|
||
Возвращает id темы или None (фолбэк — round-robin).
|
||
"""
|
||
ranked = sorted(topics, key=lambda t: counts.get(t["id"], 0))[:MAX_OPTIONS]
|
||
if len(ranked) < 2:
|
||
return None
|
||
opts = {t["id"]: t["text"][:300] for t in ranked}
|
||
tasks = (model or {}).get("tasks") or {}
|
||
active = [t.get("title") or "" for t in tasks.values()
|
||
if t.get("status") in ("ready", "running")][:8]
|
||
state = "\n".join(x for x in (
|
||
AUM_GOAL, _policy_line(model),
|
||
"The harness proactively writes one thought to the founder's chat. "
|
||
"Topics used less often are listed first.",
|
||
("Work in progress: " + "; ".join(a[:80] for a in active)) if active else "",
|
||
) if x)
|
||
r = choose("topic", opts, state,
|
||
"Which topic is the most useful to raise with the founder now "
|
||
"to grow long-term AUM?")
|
||
return r["choice"] if r and r["applied"] else None
|
||
|
||
|
||
def status() -> dict:
|
||
"""Срез для CLI/дашборда: режимы и по каждому бэкенду — доступность,
|
||
модель установлена/в памяти (облаку /api/ps не нужен), предохранитель."""
|
||
out = {"mode": mode(), "modes": {p: mode(p) for p in ("tasks", "gate", "topic")},
|
||
"cache": len(_CACHE), "backends": []}
|
||
for be in backends():
|
||
b = {"name": be["name"], "url": be["url"], "model": be["model"],
|
||
"cloud": bool(be["key"]), "breaker": dict(_BREAKERS.get(be["name"]) or {})}
|
||
base = be["url"].split("/v1/")[0]
|
||
hdr = {"Authorization": "Bearer " + be["key"]} if be["key"] else {}
|
||
want = be["model"] if ":" in be["model"] else be["model"] + ":latest"
|
||
try:
|
||
req = urllib.request.Request(base + "/api/tags", headers=hdr)
|
||
with urllib.request.urlopen(req, timeout=8) as r:
|
||
tags = json.loads(r.read().decode())
|
||
names = {m.get("name", "") for m in tags.get("models") or []}
|
||
b["installed"] = be["model"] in names or want in names
|
||
if not be["key"]:
|
||
with urllib.request.urlopen(base + "/api/ps", timeout=5) as r:
|
||
ps = json.loads(r.read().decode())
|
||
b["loaded"] = any(m.get("name") in (be["model"], want)
|
||
for m in ps.get("models") or [])
|
||
b["reachable"] = True
|
||
except Exception as e:
|
||
b["reachable"] = False
|
||
b["error"] = f"{e.__class__.__name__}: {str(e)[:120]}"
|
||
out["backends"].append(b)
|
||
return out
|