pegas-memory/pegas_memory/verify.py

"""Верификация атомов: self-critique + кросс-подтверждение.

Двухступенчатая верификация знания:
  1. self-critique (LLM) — вердикт по новизне/противоречию против существующих.
  2. кросс-подтверждение (эмбеддинги) — тот же факт из N независимых источников.

Статусы:
  candidate  — новый атом, не проверен
  verified   — подтверждён (self-critique прошёл И/ИЛИ 2+ источника)
  contested  — противоречит существующему знанию
  stale      — устарел (не актуально)

Верификация не удаляет атомы, только меняет статус/confidence.
"""
from __future__ import annotations
import json
import httpx
from . import config, db

_client = httpx.Client(timeout=600)

VERDICTS = ("new", "duplicate", "corroborates", "contradicts")


def self_critique(atom: dict, context_atoms: list[dict], model: str | None = None) -> dict:
    """LLM-вердикт по атому против существующих знаний.

    Возвращает {"verdict", "confidence", "reason"}.
    """
    m = model or config.LLM_MODEL
    ctx = "\n".join(f"- {a.get('claim','')[:120]}" for a in context_atoms[:8])
    prompt = (
        f"Существующие знания:\n{ctx}\n\n"
        f"Новый атом ({atom.get('type')}):\n{atom.get('claim')}\n{atom.get('content','')[:400]}\n\n"
        f"Оцени атом относительно существующих знаний. Верни ТОЛЬКО JSON:\n"
        f'{{"verdict": "new|duplicate|corroborates|contradicts", "confidence": 0.0-1.0, "reason": "кратко"}}\n'
        f"- new: уникальное знание, которого нет в контексте\n"
        f"- duplicate: то же самое, что уже есть\n"
        f"- corroborates: подтверждает/уточняет существующее\n"
        f"- contradicts: противоречит существующему"
    )
    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": "new", "confidence": 0.5, "reason": "parse error"}
    try:
        v = json.loads(out[start:end + 1])
    except json.JSONDecodeError:
        return {"verdict": "new", "confidence": 0.5, "reason": "json error"}
    if v.get("verdict") not in VERDICTS:
        v["verdict"] = "new"
    return v


def corroborate_atom(conn, atom_id: int, threshold: float = 0.90) -> int:
    """Найти близкие атомы из ДРУГИХ источников (кросс-подтверждение).

    Возвращает число найденных подтверждений (не считая сам атом).
    """
    from . import qdrant_store
    with conn.cursor() as cur:
        cur.execute("SELECT id, doc_id, claim, content FROM pm_atom WHERE id = %s", (atom_id,))
        row = cur.fetchone()
        if not row:
            return 0
    text = (row["claim"] or "") + " " + (row["content"] or "")[:800]
    hits = qdrant_store.search(text, top_k=5)
    count = 0
    for h in hits:
        if int(h["id"]) == atom_id:
            continue
        if h["score"] >= threshold and h.get("doc_id") != row["doc_id"]:
            count += 1
    return count


def verify_atom(conn, atom_id: int, *, self_critique_enabled: bool = True,
                model: str | None = None) -> dict:
    """Полная верификация одного атома. Возвращает результат.

    - self-critique: вердикт LLM (contradicts → contested).
    - кросс-подтверждение: N близких из других источников (N>=1 → verified).
    """
    with conn.cursor() as cur:
        cur.execute("SELECT id, type, claim, content, confidence, status, doc_id FROM pm_atom WHERE id = %s", (atom_id,))
        atom = cur.fetchone()
        if not atom:
            return {"atom_id": atom_id, "error": "not found"}

    result = {"atom_id": atom_id, "verdict": None, "corroborations": 0,
              "new_status": atom["status"], "new_confidence": atom["confidence"]}

    # 1) кросс-подтверждение по эмбеддингам (быстро, без LLM)
    corr = corroborate_atom(conn, atom_id)
    result["corroborations"] = corr

    # 2) self-critique (LLM): контекст — ближайшие по смыслу (эмбеддинги), а не по id
    if self_critique_enabled:
        from . import qdrant_store
        qtext = (atom["claim"] or "") + " " + (atom["content"] or "")[:400]
        hits = qdrant_store.search(qtext, top_k=8)
        context = [{"claim": h.get("claim", "")} for h in hits if int(h["id"]) != atom_id]
        v = self_critique(atom, context, model=model)
        result["verdict"] = v["verdict"]
        result["reason"] = v.get("reason")

        if v["verdict"] == "contradicts":
            result["new_status"] = "contested"
        elif v["verdict"] == "duplicate":
            result["new_status"] = "stale"  # дубликат — пометить неактуальным
        elif v["verdict"] in ("corroborates", "new") or corr >= 1:
            result["new_status"] = "verified"
    elif corr >= 1:
        result["new_status"] = "verified"

    # confidence: кросс-подтверждение повышает уверенность
    new_conf = min(1.0, float(atom["confidence"] or 0.5) + 0.1 * corr)
    result["new_confidence"] = round(new_conf, 2)

    with conn.cursor() as cur:
        cur.execute(
            "UPDATE pm_atom SET status = %s, confidence = %s, updated_at = now() WHERE id = %s",
            (result["new_status"], result["new_confidence"], atom_id),
        )
    conn.commit()
    return result


def verify_all(conn, *, limit: int = 50, self_critique_enabled: bool = True,
               model: str | None = None) -> dict:
    """Верифицировать все candidate-атомы (порциями). Возвращает сводку."""
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM pm_atom WHERE status = 'candidate' ORDER BY id LIMIT %s", (limit,))
        ids = [r["id"] for r in cur.fetchall()]
    summary = {"total": len(ids), "verified": 0, "contested": 0, "stale": 0, "unchanged": 0}
    for aid in ids:
        try:
            r = verify_atom(conn, aid, self_critique_enabled=self_critique_enabled, model=model)
        except Exception as e:
            print(f"[verify] atom {aid} failed: {e}")
            continue
        ns = r.get("new_status")
        if ns == "verified":
            summary["verified"] += 1
        elif ns == "contested":
            summary["contested"] += 1
        elif ns == "stale":
            summary["stale"] += 1
        else:
            summary["unchanged"] += 1
    return summary