feat(pegas-memory): Tier-1 self-improvement tools (commit + staleness)
- memory_commit: closes the 'commit hole' — writes verified answers from closed gaps back into atoms (mandatory dedup); leaves gap open if no verified answer exists. Reads verified status from PostgreSQL (source of truth), not the Qdrant projection. - memory_stale: staleness guard with two signals — content-staleness (re-fetch + LLM verdict, needs proxy for foreign sites) and age-staleness (proxy-free fallback by pm_document.fetched_at, max_age_days). - db: populate pm_document.fetched_at on upsert (freshness signal). - docs: README/AGENTS/ARCHITECTURE/EXAMPLES updated (18 tools). - self_improve.py: new module (227 lines).
This commit is contained in:
n1kt4r committed
1 parent
a4c4a766c1
commit
eca8a11e23
8 files changed
+439
-6
No files matched your search
@@ -0,0 +1,227 @@
|
||||
"""Самообучение (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:
|
||||
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
|
||||
Reference in new issue
Block a user