105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
"""Логическая инференция: forward-chaining правил в Memgraph.
|
|
|
|
Правило — атом типа logical_rule, content = JSON:
|
|
{"if": "imports", "then": "depends_on"} # одношаговое
|
|
{"if": ["depends_on", "depends_on"], "then": "depends_on"} # транзитивное (композиция)
|
|
|
|
Инференция итерирует до фиксированной точки: пока правила порождают новые рёбра.
|
|
Выведенные рёбра помечаются inferred=true, source='inference'.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
|
|
from . import memgraph_store
|
|
|
|
|
|
def _is_kind(token) -> bool:
|
|
"""kind ребра — одиночный токен без пробелов (imports, depends_on, calls...).
|
|
|
|
Отсекает человекочитаемые правила от LLM-дистилляции вида
|
|
if='Memgraph хранит граф в RAM' — это не kind ребра.
|
|
"""
|
|
return isinstance(token, str) and " " not in token and len(token) > 0 and len(token) < 64
|
|
|
|
|
|
def load_rules(conn) -> list[dict]:
|
|
"""Загрузить машинные правила из pm_atom (type='logical_rule').
|
|
|
|
Правило валидно для инференции, только если if/then — kind-рёбра
|
|
(одиночные токены без пробелов). Человекочитаемые правила от
|
|
дистилляции отфильтровываются.
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT id, claim, content FROM pm_atom WHERE type = 'logical_rule' ORDER BY id"
|
|
)
|
|
rows = cur.fetchall()
|
|
rules = []
|
|
for r in rows:
|
|
try:
|
|
spec = json.loads(r["content"]) if isinstance(r["content"], str) else r["content"]
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
if "if" not in spec or "then" not in spec:
|
|
continue
|
|
if_ = spec["if"]
|
|
then = spec["then"]
|
|
# валидация: then — всегда kind; if — kind или список kind
|
|
if not _is_kind(then):
|
|
continue
|
|
if isinstance(if_, list):
|
|
if len(if_) != 2 or not all(_is_kind(k) for k in if_):
|
|
continue
|
|
elif not _is_kind(if_):
|
|
continue
|
|
rules.append({"atom_id": r["id"], "claim": r["claim"], "spec": spec})
|
|
return rules
|
|
|
|
|
|
def _apply_rule(spec: dict) -> int:
|
|
"""Применить одно правило. Возвращает число созданных рёбер."""
|
|
if_ = spec["if"]
|
|
then = spec["then"]
|
|
if isinstance(if_, list):
|
|
# композиция: [k1, k2] -> (a)-[k1]->(b)-[k2]->(c) ⇒ (a)-[then]->(c)
|
|
if len(if_) == 2:
|
|
return memgraph_store.apply_rule_chain(if_[0], if_[1], then)
|
|
return 0
|
|
# одношаговое: (a)-[if]->(b) ⇒ (a)-[then]->(b)
|
|
return memgraph_store.apply_rule_one_step(if_, then)
|
|
|
|
|
|
def infer(conn, max_rounds: int = 10) -> dict:
|
|
"""Запустить инференцию до фиксированной точки.
|
|
|
|
Возвращает {rules, rounds, edges_created}.
|
|
"""
|
|
rules = load_rules(conn)
|
|
total = 0
|
|
rounds = 0
|
|
for _ in range(max_rounds):
|
|
created = 0
|
|
for rule in rules:
|
|
created += _apply_rule(rule["spec"])
|
|
total += created
|
|
rounds += 1
|
|
if created == 0:
|
|
break
|
|
return {"rules": len(rules), "rounds": rounds, "edges_created": total}
|
|
|
|
|
|
def store_rule(conn, *, claim: str, if_: str | list, then: str) -> int:
|
|
"""Сохранить логическое правило (атом logical_rule) в PG.
|
|
|
|
Возвращает id атома. Если правило с тем же content уже есть — вернёт существующий.
|
|
"""
|
|
from . import db
|
|
content = json.dumps({"if": if_, "then": then}, ensure_ascii=False)
|
|
atom_id, created = db.upsert_atom(
|
|
conn, type="logical_rule", claim=claim, content=content,
|
|
confidence=0.9, status="verified",
|
|
)
|
|
if created:
|
|
db.enqueue_outbox(conn, entity_type="atom", entity_id=atom_id)
|
|
return atom_id
|