"""Самообучение (Tier 1): закрытие «дыры commit» + гвард сталкнутости. Два механизма, закрывающих петлю commit/maintain (см. ROADMAP.md): 1. commit_gaps — после закрытия gap пишет верифицированный answer обратно в атомы, чтобы знание не слетивало и не пере-исследовалось. Atom-efficiency: обязательный dedup. 2. stale_guard — пере-проверяет атомы по свежу источнику; противоречит/ устарел → status='stale'. Два сигнала: - content-staleness: рефетч источника + LLM-вердикт (нужен proxy для зарубежных сайтов); - age-staleness: fallback по `fetched_at` (proxy-free). Лимит LLM-вызовов (limit/sample). Лимиты: LLM-вызовы дорогие — каждый вызов инструмента лимитирован. """ from __future__ import annotations import json import random from datetime import datetime from . import config, db, research, qdrant_store from .ingest import parse_web def _fresh_source(uri: str) -> str | None: """Свежный контент источника (прокси/UA/таймаут через ingest.parse_web).""" try: doc = parse_web(uri) except Exception: return None return "\n\n".join(c.content for c in doc.chunks) def _stale_verdict(atom: dict, fresh_text: str, model: str | None) -> dict: """LLM: поддерживает ли свежный источник утверждение атома? Вернит {"verdict": "supported|contradicts|outdated", "confidence", "reason"}. """ import httpx m = model or config.LLM_MODEL ctx = (atom.get("claim", "") or "") + "\n" + (atom.get("content", "") or "")[:600] prompt = ( f"Свежный источник:\n{fresh_text[:4000]}\n\n" f"Утверждение атома ({atom.get('type')}):\n{ctx[:500]}\n\n" f"Оцени, свежный источник всё ещё ПОДДЕРЖИВАЕТ утверждение. " f"Верни ТОЛЬКО JSON:\n" f'{{"verdict": "supported|contradicts|outdated", "confidence": 0.0-1.0, ' f'"reason": "кратко"}}\n' f"- supported: источник подтверждает/сохраняет утверждение\n" f"- contradicts: источник теперь противоречит\n" f"- outdated: информация устарела в источнике" ) try: with httpx.Client(timeout=120) as client: r = client.post(f"{config.OLLAMA_URL}/api/chat", json={ "model": m, "messages": [{"role": "user", "content": prompt}], "stream": False, "options": {"temperature": 0.0}, }) r.raise_for_status() out = r.json()["message"]["content"].strip() start, end = out.find("{"), out.rfind("}") if start == -1 or end == -1: return {"verdict": "supported", "confidence": 0.5, "reason": "parse error"} v = json.loads(out[start:end + 1]) except Exception: return {"verdict": "supported", "confidence": 0.5, "reason": "llm error"} if v.get("verdict") not in ("supported", "contradicts", "outdated"): v["verdict"] = "supported" return v def commit_gaps(conn, *, gap_ids=None, dedup=True, model: str | None = None) -> dict: """Закрывает дыра «commit»: answer закрытого gap пишется в атомы. Для каждого gap из gap_ids: ищет verified-атомы, отвечающие на вопрос, дистиллирует канонический атом-ответ, сохраняет с dedup, resolves gap. Возвращает статистику. dedup — atom-efficiency guardrail (по умолчанию True). """ from . import distill, dedup as _dedup out = {"committed": 0, "skipped_duplicate": 0, "resolved": 0, "skipped_empty": 0, "atoms": []} if not gap_ids: raise ValueError("нужен gap_ids (список id)") for gid in gap_ids: gid = int(gid) gap = research.get_gap(conn, gid) if not gap: continue q = gap["question"] # релевантность — из Qdrant (рейтинг), статус — из Postgres (истина): # verify меняет статус в PG, но не пушит projection, поэтому читаем status оттуда. hits = qdrant_store.search(q, top_k=8) if not hits: out["skipped_empty"] += 1 continue hit_ids = [int(h["id"]) for h in hits] with conn.cursor() as cur: cur.execute( "SELECT id, claim, content, status FROM pm_atom WHERE id = ANY(%s)", (hit_ids,), ) rows = {r["id"]: r for r in cur.fetchall()} verified = [rows[hid] for hid in hit_ids if rows[hid]["status"] == "verified"] if not verified: # нет верифицированного ответа — не резать gap, агент должен его получить out["skipped_empty"] += 1 continue text = f"ВОПРОС: {q}\n\nОТВЕТ (верифицированные знания):\n" + "\n".join( f"- {v.get('claim', '')}\n {str(v.get('content', ''))[:300]}" for v in verified ) atoms = distill.distill_chunk(text, model=model) for a in atoms: if a.get("confidence", 0.5) < 0.4: continue if dedup: dup_id = _dedup.find_duplicate( (a.get("claim") or "") + " " + str(a["content"])[:800], threshold=config.DEDUP_THRESHOLD, ) if dup_id is not None: out["skipped_duplicate"] += 1 continue atom_id, created = db.upsert_atom( conn, type=a["type"], claim=a.get("claim"), content=a["content"], confidence=a.get("confidence", 0.5), status="verified", doc_id=None, metadata={"committed_from_gap": gid, "source_question": q}, ) if created: out["committed"] += 1 out["atoms"].append(atom_id) db.enqueue_outbox(conn, entity_type="atom", entity_id=atom_id) research.update_gap(conn, gid, status="resolved") out["resolved"] += 1 conn.commit() return out def stale_guard(conn, *, atom_id=None, all: bool = False, limit: int = 20, sample: bool = False, min_confidence: float = 0.0, age_only: bool = False, max_age_days: int = 90, model: str | None = None) -> dict: """Гвард сталкнутости: пере-проверка атомов по свежему источнику. atom_id — один атом (1 fetch + 1 LLM-вызов). all — пачка атомов; limit + sample лимитируют число LLM-вызовов. Противоречит/устарел → status='stale'. Возвращает сводку. """ ids = [] if atom_id is not None: with conn.cursor() as cur: cur.execute( "SELECT id, type, claim, content, confidence, status, doc_id " "FROM pm_atom WHERE id = %s", (int(atom_id),), ) row = cur.fetchone() if row: ids = [row] elif all: with conn.cursor() as cur: cur.execute( "SELECT id, type, claim, content, confidence, status, doc_id " "FROM pm_atom WHERE confidence >= %s", (min_confidence,), ) rows = cur.fetchall() if sample and rows: n = min(int(limit), len(rows)) ids = random.sample(rows, n) else: ids = rows[:int(limit)] else: raise ValueError("укажи atom_id или all=true") summary = {"checked": 0, "stale": 0, "supported": 0, "skipped_no_source": 0, "atom_ids": [], "reasons": {}} for atom in ids: with conn.cursor() as cur: cur.execute( "SELECT d.uri, d.fetched_at FROM pm_document d WHERE id = %s", (atom["doc_id"],), ) doc = cur.fetchone() if not doc or not doc.get("uri"): summary["skipped_no_source"] += 1 continue status_change = None reason = None if not age_only: fresh = _fresh_source(doc["uri"]) if fresh: v = _stale_verdict(atom, fresh, model) summary["checked"] += 1 if v["verdict"] in ("contradicts", "outdated"): status_change = "stale" reason = f"source_{v['verdict']}: {v.get('reason', '')}" else: # content-staleness re-fetch succeeded -> verdict is authoritative status_change = "supported" reason = "supported" if status_change is None: # proxy-free fallback: возраст контента fa = doc.get("fetched_at") if fa is None: summary["skipped_no_source"] += 1 continue tz = fa.tzinfo or datetime.timezone.utc age_days = max(0, (datetime.now(tz=tz) - fa).days) if age_days >= max_age_days: status_change = "stale" reason = f"content_age {age_days}d" else: reason = f"fresh {age_days}d" if status_change == "stale": with conn.cursor() as cur: cur.execute( "UPDATE pm_atom SET status = 'stale', updated_at = now() WHERE id = %s", (atom["id"],), ) summary["stale"] += 1 summary["atom_ids"].append(atom["id"]) else: summary["supported"] += 1 summary["reasons"][atom["id"]] = reason conn.commit() return summary