Spaces:
Sleeping
Sleeping
| """ | |
| 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", "?") | |