Spaces:
Sleeping
Sleeping
File size: 7,580 Bytes
1605cbb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | """
Low-level graph operations for FALSIFY.
This module is the single, verified surface between FALSIFY's belief logic and
Cognee's graph/vector engines. Every engine call used elsewhere in the project goes
through a helper here, so the (few) places that touch Cognee internals are auditable.
Verified Cognee API shapes (grep-confirmed against /workspaces/cognee — see REQUIREMENTS.md §0):
ge = await get_graph_engine() # ASYNC factory
nodes, edges = await ge.get_graph_data() # ([(id, props)], [(src, dst, rel, props)])
nodes, edges = await ge.get_neighborhood(ids, depth=, edge_types=)
await ge.add_edge(from_node, to_node, relationship_name, edge_properties={})
await ge.set_node_truth_state({id: {"truth_alignment": [...], "truth_epoch": N}})
state = await ge.get_node_truth_state([ids]) # {id: {"truth_alignment": [...], ...}}
await ge.set_node_feedback_weights({id: 0.0})
await ge.delete_nodes([ids])
ve = get_vector_engine() # SYNC factory
await ve.delete_data_points(collection_name, [uuids])
hits = await ve.search(collection_name, query_text=, limit=, include_payload=)
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional, Set, Tuple
from cognee.infrastructure.databases.graph import get_graph_engine
from cognee.infrastructure.databases.vector import get_vector_engine
from falsify.models import TruthState
logger = logging.getLogger("falsify.graph_ops")
# A parsed edge: (source_id, target_id, relationship_name, properties)
Edge = Tuple[str, str, str, Dict[str, Any]]
# A parsed node: (node_id, properties)
GraphNode = Tuple[str, Dict[str, Any]]
async def load_graph() -> Tuple[List[GraphNode], List[Edge]]:
"""Return the full graph as ``(nodes, edges)``.
Nodes are ``(id, props)`` tuples; edges are ``(src, dst, rel, props)`` tuples.
IDs are normalized to ``str`` so they can be used as dict keys regardless of
whether the adapter returns ``UUID`` or ``str``.
"""
ge = await get_graph_engine()
raw_nodes, raw_edges = await ge.get_graph_data()
nodes: List[GraphNode] = [(str(nid), props or {}) for nid, props in raw_nodes]
edges: List[Edge] = []
for e in raw_edges:
# Adapters return 4-tuples (src, dst, rel, props); be defensive about arity.
if len(e) >= 4:
src, dst, rel, props = e[0], e[1], e[2], e[3]
elif len(e) == 3:
src, dst, rel, props = e[0], e[1], e[2], {}
else: # pragma: no cover - unexpected adapter shape
continue
edges.append((str(src), str(dst), str(rel), props or {}))
return nodes, edges
async def get_truth(node_ids: List[str]) -> Dict[str, List[str]]:
"""Return ``{node_id: truth_alignment_list}`` for the given ids.
A node with no stored truth state is treated as ``["alive"]`` (the default).
"""
if not node_ids:
return {}
ge = await get_graph_engine()
ids = [str(n) for n in node_ids]
try:
raw = await ge.get_node_truth_state(ids)
except Exception as exc: # adapter may not have state yet
logger.debug("get_node_truth_state failed (%s); defaulting to alive", exc)
raw = {}
out: Dict[str, List[str]] = {}
for nid in ids:
entry = (raw or {}).get(nid) or (raw or {}).get(str(nid))
alignment = (entry or {}).get("truth_alignment") if entry else None
out[nid] = list(alignment) if alignment else [TruthState.ALIVE.value]
return out
async def is_alive(node_id: str) -> bool:
"""True iff the node's truth_alignment currently contains ``alive``."""
state = await get_truth([str(node_id)])
return TruthState.ALIVE.value in state.get(str(node_id), [TruthState.ALIVE.value])
async def set_state(node_id: str, state: TruthState, epoch: int) -> None:
"""Persist ``truth_alignment=[state]`` + ``truth_epoch=epoch`` on a node.
Truth state is stored ON the graph node, so it survives a process restart — the
basis of FALSIFY's cross-session persistence guarantee.
"""
ge = await get_graph_engine()
value = state.value if isinstance(state, TruthState) else str(state)
try:
await ge.set_node_truth_state(
{str(node_id): {"truth_alignment": [value], "truth_epoch": int(epoch)}}
)
except Exception as exc:
logger.error("set_node_truth_state failed for %s -> %s: %s", node_id, value, exc)
raise
async def set_weight(node_id: str, weight: float) -> None:
"""Set a node's feedback weight (confidence/health signal). Best-effort."""
ge = await get_graph_engine()
try:
await ge.set_node_feedback_weights({str(node_id): float(weight)})
except Exception as exc: # non-fatal: weight is a secondary signal
logger.debug("set_node_feedback_weights failed for %s: %s", node_id, exc)
async def add_edge(src: str, dst: str, rel: str, props: Optional[Dict[str, Any]] = None) -> None:
"""Add a directed edge ``src --rel--> dst`` with optional properties."""
ge = await get_graph_engine()
await ge.add_edge(str(src), str(dst), rel, props or {})
async def delete_from_both_stores(node_ids: List[str], collections: List[str]) -> int:
"""Hard-delete nodes from the graph AND their rows from vector collections.
Returns the number of node ids deleted. Vector deletion is attempted per
collection and is best-effort (a node may not live in every collection).
"""
if not node_ids:
return 0
ids = [str(n) for n in node_ids]
ge = await get_graph_engine()
try:
await ge.delete_nodes(ids)
except Exception as exc:
logger.error("delete_nodes failed: %s", exc)
raise
ve = get_vector_engine()
for collection in collections:
try:
await ve.delete_data_points(collection, ids)
except Exception as exc: # collection may not exist / id not present
logger.debug("delete_data_points(%s) best-effort skip: %s", collection, exc)
return len(ids)
# --------------------------------------------------------------------------- #
# Adjacency helpers (built from a single load_graph() snapshot)
# --------------------------------------------------------------------------- #
def dependents_of(evidence_id: str, edges: List[Edge], rel: str) -> List[str]:
"""Return source ids of ``rel`` edges pointing INTO ``evidence_id``.
For ``depends_on`` (Conclusion -> Evidence) this yields the Conclusions that
depend on the given Evidence — i.e. the forward-cascade victims.
"""
tid = str(evidence_id)
return [src for (src, dst, r, _p) in edges if r == rel and str(dst) == tid]
def incoming(node_id: str, edges: List[Edge], rel: str) -> List[Tuple[str, Dict[str, Any]]]:
"""Return ``[(source_id, props)]`` for ``rel`` edges pointing into ``node_id``."""
tid = str(node_id)
return [(src, p) for (src, dst, r, p) in edges if r == rel and str(dst) == tid]
def outgoing(node_id: str, edges: List[Edge], rel: str) -> List[Tuple[str, Dict[str, Any]]]:
"""Return ``[(target_id, props)]`` for ``rel`` edges leaving ``node_id``."""
sid = str(node_id)
return [(dst, p) for (src, dst, r, p) in edges if r == rel and str(src) == sid]
def node_label(props: Dict[str, Any]) -> str:
"""Best-effort human label for a node from its properties."""
for key in ("statement", "claim", "question", "text", "name"):
if props.get(key):
return str(props[key])
return props.get("id", "?")
|