Graph-based Label-Embedding Alignment

The problem

It is difficult to measure how logically consistent a web of interconnected concepts or data points actually is. Standard systems often struggle to spot inconsistencies in how different pieces of information relate to one another.

What it does

It maps out relationships between data points and assigns a score based on how well those connections align with each other. It looks at the semantic meaning of the links to see if the structure makes sense.

Why it matters

It provides a way to mathematically measure the logical consistency of complex information networks.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 graph_label_alignment.py
Recursive Reachability Score for node 1: 1.0
Connectivity Density: 0.33
Semantic Consistency Score: 0.62

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 143 lines, one file, standard library only.

# graph_label_alignment.py
import math
from collections import deque

class Node:
    def __init__(self, label, embedding):
        self.label = label
        self.embedding = embedding

class Graph:
    def __init__(self):
        self.nodes = {}
        self.edges = []

    def add_node(self, node_id, label, embedding):
        self.nodes[node_id] = Node(label, embedding)

    def add_edge(self, node_id1, node_id2):
        self.edges.append((node_id1, node_id2))
        self.edges.append((node_id2, node_id1))  # Undirected graph

    def shortest_path_length(self, start, end):
        if start not in self.nodes or end not in self.nodes:
            return None
        visited = set()
        queue = deque([(start, 0)])
        while queue:
            node, length = queue.popleft()
            if node == end:
                return length
            visited.add(node)
            for edge in self.edges:
                neighbor = edge[0] if edge[1] == node else edge[1]
                if neighbor not in visited:
                    queue.append((neighbor, length + 1))
        return None  # No path exists

    def connectivity_density(self):
        n = len(self.nodes)
        max_edges = n * (n - 1)  # Undirected graph
        actual_edges = len(self.edges) // 2
        return actual_edges / max_edges if max_edges > 0 else 0

    def recursive_reachability_score(self, node_id):
        if node_id not in self.nodes:
            return None
        total = 0
        num_nodes = len(self.nodes)
        for other_id in self.nodes:
            if other_id == node_id:
                continue
            path_len = self.shortest_path_length(node_id, other_id)
            if path_len is not None:
                total += path_len
        return total / (num_nodes - 1) if num_nodes > 1 else 0

    def compute_semantic_consistency(self):
        consistency_score = 0
        for edge in self.edges:
            node1_id, node2_id = edge
            node1 = self.nodes[node1_id]
            node2 = self.nodes[node2_id]

            # Cosine similarity
            dot_product = sum(a*b for a, b in zip(node1.embedding, node2.embedding))
            mag1 = math.sqrt(sum(x**2 for x in node1.embedding))
            mag2 = math.sqrt(sum(x**2 for x in node2.embedding))
            similarity = 0 if mag1 == 0 or mag2 == 0 else dot_product / (mag1 * mag2)

            # Reachability component
            path_len = self.shortest_path_length(node1_id, node2_id)
            reachability = 1 / (1 + path_len) if path_len is not None else 0

            # Combine components
            consistency_score += 0.5 * similarity + 0.5 * reachability

        return consistency_score / len(self.edges) if self.edges else 0

    def compute_graph_weighted_cluster(self):
        best_cluster = []
        best_score = -float('inf')

        for node_id in self.nodes:
            # Consider 1-hop neighbors
            cluster_nodes = {node_id}
            for edge in self.edges:
                if edge[0] == node_id or edge[1] == node_id:
                    cluster_nodes.add(edge[0] if edge[1] == node_id else edge[1])

            # Calculate scores
            cluster_edges = [e for e in self.edges if e[0] in cluster_nodes and e[1] in cluster_nodes]
            if not cluster_edges:
                continue

            # Semantic consistency
            total_consistency = 0
            for edge in cluster_edges:
                n1, n2 = edge
                n1_emb = self.nodes[n1].embedding
                n2_emb = self.nodes[n2].embedding
                dot = sum(a*b for a,b in zip(n1_emb, n2_emb))
                mag1 = math.sqrt(sum(x**2 for x in n1_emb))
                mag2 = math.sqrt(sum(x**2 for x in n2_emb))
                similarity = dot / (mag1 * mag2) if mag1 and mag2 else 0
                path_len = self.shortest_path_length(n1, n2)
                reachability = 1 / (1 + path_len) if path_len is not None else 0
                total_consistency += 0.5 * similarity + 0.5 * reachability

            avg_consistency = total_consistency / len(cluster_edges)

            # Connectivity density
            n = len(cluster_nodes)
            max_edges = n * (n-1)
            actual_edges = len(cluster_edges)
            connectivity = actual_edges / max_edges if max_edges else 0

            # Combined score
            combined = avg_consistency + connectivity

            if combined > best_score:
                best_score = combined
                best_cluster = list(cluster_nodes)

        return best_cluster

if __name__ == "__main__":
    g = Graph()
    # Example usage with dummy embeddings
    g.add_node("1", "labelA", [0.1, 0.5, 0.8])
    g.add_node("2", "labelB", [0.3, 0.6, 0.4])
    g.add_node("3", "labelC", [0.9, 0.2, 0.5])
    g.add_edge("1", "2")
    g.add_edge("1", "3")
    
    reachability_score = g.recursive_reachability_score("1")
    connectivity = g.connectivity_density()
    consistency = g.compute_semantic_consistency()
    cluster = g.compute_graph_weighted_cluster()
    
    print(f"Recursive Reachability Score for node 1: {reachability_score:.2f}")
    print(f"Connectivity Density: {connectivity:.2f}")
    print(f"Semantic Consistency Score: {consistency:.2f}")
    print(f"Graph-Weighted Cluster: {cluster}")