Pegas
13e0b87446
- ingest.parse_web: direct fetch; on block/timeout retry via socks5h://127.0.0.1:1080 (SOCKS_FALLBACK_PROXY); raises ValueError if both fail. - config: add SOCKS_FALLBACK_PROXY default (socks5h://127.0.0.1:1080). - self_improve.stale_guard: supported LLM verdict now sets status_change='supported' (authoritative), no fallthrough to age fallback -> no double-checked/skipped_no_source. - README: document auto-fallback.
230 lines
10 KiB
Python
230 lines
10 KiB
Python
"""Самообучение (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
|