Multi-Hop Knowledge Path Ranking

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 mhkpr.py
usage: mhkpr [-h] {rank,stats,sample} ...

Multi-Hop Knowledge Path Ranking — rank paths in a KG by logical depth &
connectivity

positional arguments:
  {rank,stats,sample}  subcommand
    rank               Discover and rank multi-hop paths
    stats              Print KG statistics
    sample             Output a sample knowledge graph in TSV

options:
  -h, --help           show this help message and exit

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 — 383 lines, one file, standard library only.

#!/usr/bin/env python3
"""
Multi-Hop Knowledge Path Ranking (MHKPR)

Combines SQUIRE-style encoder-decoder multi-hop reasoning with a command-line
interface for ranking paths in a knowledge graph. Paths are scored by:
  - Logical depth (hop count / path length)
  - Connectivity density (edge weight aggregate)
  - Recursive reachability from query entities

No external dependencies — stdlib only.
"""

import argparse
import json
import sys
from collections import defaultdict
from itertools import product
from typing import Any, Dict, List, Tuple


class KnowledgeGraph:
    """Directed multigraph: subject -> [(predicate, object, weight)]."""

    def __init__(self) -> None:
        self._adj: Dict[str, List[Tuple[str, str, float]]] = defaultdict(list)
        self._reverse: Dict[str, List[Tuple[str, str, float]]] = defaultdict(list)

    def add_triple(self, subj: str, pred: str, obj: str, weight: float = 1.0) -> None:
        self._adj[subj].append((pred, obj, weight))
        self._reverse[obj].append((pred, subj, weight))

    def neighbors(self, node: str) -> List[Tuple[str, str, float]]:
        return self._adj.get(node, [])

    def reverse_neighbors(self, node: str) -> List[Tuple[str, str, float]]:
        return self._reverse.get(node, [])

    def nodes(self) -> List[str]:
        return list(set(self._adj.keys()) | set(self._reverse.keys()))

    def edge_count(self) -> int:
        return sum(len(v) for v in self._adj.values())

    def node_out_degree(self, node: str) -> int:
        return len(self._adj.get(node, []))

    def node_in_degree(self, node: str) -> int:
        return len(self._reverse.get(node, []))

    @staticmethod
    def from_tsv(lines: List[str]) -> "KnowledgeGraph":
        kg = KnowledgeGraph()
        for line in lines:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split("\t")
            if len(parts) < 3:
                continue
            subj, pred, obj = parts[0], parts[1], parts[2]
            w = float(parts[3]) if len(parts) >= 4 else 1.0
            kg.add_triple(subj, pred, obj, w)
        return kg


# ── Path discovery: BFS-based multi-hop traversal (SQUIRE-inspired) ──────────

def _find_paths(
    kg: KnowledgeGraph,
    start: str,
    target: str,
    max_hops: int,
    beam_width: int,
    max_paths: int,
) -> List[List[Tuple[str, str, str]]]:
    """
    Beam-search over the KG to enumerate paths from start to target.
    Each path is a list of (subject, predicate, object) triples.

    SQUIRE-style: the encoder transforms the query (start, target) into an
    embedding; the decoder generates candidate paths step-by-step. Here we
    approximate that with constrained BFS + beam pruning.
    """
    if max_hops < 1 or max_paths < 1:
        return []

    # priority queue approximation: beam of (node, path_so_far, cum_weight)
    beam: List[Tuple[str, List[Tuple[str, str, str]], float]] = [
        (start, [], 1.0)
    ]

    complete: List[Tuple[List[Tuple[str, str, str]], float]] = []

    for hop in range(max_hops):
        next_beam: List[Tuple[str, List[Tuple[str, str, str]], float]] = []

        for cur_node, path, cum_w in beam:
            for pred, nxt, w in kg.neighbors(cur_node):
                if nxt in {p[0] for p in path} | {p[2] for p in path}:
                    continue  # no cycles
                new_path = path + [(cur_node, pred, nxt)]
                new_cum = cum_w * w
                if nxt == target:
                    complete.append((new_path, new_cum))
                elif hop + 1 < max_hops:
                    next_beam.append((nxt, new_path, new_cum))

        # beam pruning: keep top beam_width by cumulative weight
        next_beam.sort(key=lambda x: x[2], reverse=True)
        beam = next_beam[:beam_width]

        if len(complete) >= max_paths:
            break

    # sort complete paths by cumulative weight desc
    complete.sort(key=lambda x: x[1], reverse=True)
    return [p for p, _ in complete[:max_paths]]


# ── Scoring functions ────────────────────────────────────────────────────────

def _score_logical_depth(path: List[Tuple[str, str, str]]) -> float:
    """Deeper paths = more reasoning power. Reward hops."""
    return 1.0 - 1.0 / (1.0 + len(path))  # asymptote towards 1


def _score_connectivity_density(
    kg: KnowledgeGraph, path: List[Tuple[str, str, str]]
) -> float:
    """Average (in-degree + out-degree) of internal nodes on the path."""
    if not path:
        return 0.0
    total = 0.0
    count = 0
    seen: set[str] = set()
    for s, _, o in path:
        if s not in seen:
            total += kg.node_out_degree(s) + kg.node_in_degree(s)
            seen.add(s)
            count += 1
        if o not in seen:
            total += kg.node_out_degree(o) + kg.node_in_degree(o)
            seen.add(o)
            count += 1
    return total / max(count, 1)


def _score_edge_reliability(path: List[Tuple[str, str, str]]) -> float:
    """Cumulative product of edge weights — favors high-confidence edges."""
    w = 1.0
    for _, _, w_e in path:
        w *= w_e
    return w


def _score_recursive_reachability(
    kg: KnowledgeGraph, path: List[Tuple[str, str, str]], depth: int = 2
) -> float:
    """
    Recursive reachability: how many nodes can be reached from path nodes
    within `depth` additional hops, normalized.
    """
    if not path:
        return 0.0
    reached: set[str] = set()
    frontier: set[str] = {path[0][0]}  # start node

    for _ in range(depth + len(path)):
        nxt: set[str] = set()
        for n in frontier:
            if n in reached:
                continue
            reached.add(n)
            for _, neighbor, _ in kg.neighbors(n):
                if neighbor not in reached:
                    nxt.add(neighbor)
        frontier = nxt
        if not frontier:
            break

    total_nodes = len(kg.nodes())
    return len(reached) / max(total_nodes, 1)


# ── Composite ranker ─────────────────────────────────────────────────────────

def rank_paths(
    kg: KnowledgeGraph,
    query_entities: List[str],
    max_hops: int = 3,
    beam_width: int = 10,
    max_paths: int = 20,
    depth_weight: float = 0.3,
    density_weight: float = 0.25,
    reliability_weight: float = 0.2,
    reachability_weight: float = 0.25,
) -> List[Dict[str, Any]]:
    """
    Discover and rank multi-hop paths between all pairs of query entities.

    Returns ranked list of dicts with keys: source, target, path, score,
    logical_depth, connectivity_density, edge_reliability, recursive_reachability.
    """
    pairs = [(a, b) for a, b in product(query_entities, query_entities) if a != b]

    all_scored: List[Dict[str, Any]] = []

    for src, tgt in pairs:
        paths = _find_paths(kg, src, tgt, max_hops, beam_width, max_paths)
        for path in paths:
            ld = _score_logical_depth(path)
            cd = _score_connectivity_density(kg, path)
            er = _score_edge_reliability(path)
            rr = _score_recursive_reachability(kg, path)

            composite = (
                depth_weight * ld
                + density_weight * cd
                + reliability_weight * er
                + reachability_weight * rr
            )

            all_scored.append(
                {
                    "source": src,
                    "target": tgt,
                    "path": path,
                    "hop_count": len(path),
                    "score": round(composite, 4),
                    "logical_depth": round(ld, 4),
                    "connectivity_density": round(cd, 4),
                    "edge_reliability": round(er, 4),
                    "recursive_reachability": round(rr, 4),
                }
            )

    all_scored.sort(key=lambda x: x["score"], reverse=True)
    return all_scored


# ── CLI (argparse, Typer-style subcommands) ─────────────────────────────────

def _format_path(path: List[Tuple[str, str, str]]) -> str:
    if not path:
        return "(empty)"
    items = []
    prev_obj = None
    for s, p, o in path:
        if prev_obj is not None and s != prev_obj:
            items.append(f"[gap:{prev_obj}->{s}]")
        items.append(f"({s})-[{p}]->({o})")
        prev_obj = o
    return "  ".join(items)


def cmd_rank(args: argparse.Namespace) -> None:
    kg = KnowledgeGraph.from_tsv(sys.stdin.read().splitlines() if args.file is None
                                 else open(args.file).read().splitlines())
    if not kg.nodes():
        print("Error: empty or unparseable knowledge graph input.", file=sys.stderr)
        sys.exit(1)

    entities = args.entities
    if not entities:
        entities = kg.nodes()
        if len(entities) > 10:
            print(
                "Warning: no --entities given; ranking all {} nodes"
                " (may be slow).".format(len(entities)),
                file=sys.stderr,
            )
            entities = entities[:10]

    results = rank_paths(
        kg,
        entities,
        max_hops=args.max_hops,
        beam_width=args.beam,
        max_paths=args.max_paths,
        depth_weight=args.w_depth,
        density_weight=args.w_density,
        reliability_weight=args.w_reliability,
        reachability_weight=args.w_reachability,
    )

    if args.output == "json":
        out = []
        for r in results:
            entry = dict(r)
            entry["path"] = _format_path(entry["path"])
            out.append(entry)
        print(json.dumps(out, indent=2))
    elif args.output == "tsv":
        print(
            "rank\tscore\tsource\ttarget\thops\tlogical_depth\t"
            "connectivity_density\tedge_reliability\trecursive_reachability\tpath"
        )
        for i, r in enumerate(results, 1):
            print(
                f"{i}\t{r['score']}\t{r['source']}\t{r['target']}\t"
                f"{r['hop_count']}\t{r['logical_depth']}\t"
                f"{r['connectivity_density']}\t{r['edge_reliability']}\t"
                f"{r['recursive_reachability']}\t{_format_path(r['path'])}"
            )
    else:  # pretty
        print(f"\n{'='*70}")
        print(f"  Multi-Hop Knowledge Path Ranking  —  {len(results)} paths found")
        print(f"{'='*70}\n")
        for i, r in enumerate(results[:args.top_n], 1):
            print(f"  #{i:<3}  score={r['score']:.4f}  hops={r['hop_count']}")
            print(f"       {r['source']}  →  {r['target']}")
            print(f"       path:  {_format_path(r['path'])}")
            print(
                f"       ld={r['logical_depth']:.3f}  "
                f"cd={r['connectivity_density']:.3f}  "
                f"er={r['edge_reliability']:.3f}  "
                f"rr={r['recursive_reachability']:.3f}"
            )
            print()


def cmd_stats(args: argparse.Namespace) -> None:
    kg = KnowledgeGraph.from_tsv(sys.stdin.read().splitlines() if args.file is None
                                 else open(args.file).read().splitlines())
    print(f"Nodes:           {len(kg.nodes())}")
    print(f"Edges:           {kg.edge_count()}")
    if kg.nodes():
        degrees = [(n, kg.node_out_degree(n), kg.node_in_degree(n)) for n in kg.nodes()]
        max_out = max(degrees, key=lambda x: x[1])
        max_in  = max(degrees, key=lambda x: x[2])
        print(f"Max out-degree:  {max_out[0]} ({max_out[1]})")
        print(f"Max in-degree:   {max_in[0]} ({max_in[1]})")
        avg_out = sum(d[1] for d in degrees) / len(degrees)
        avg_in  = sum(d[2] for d in degrees) / len(degrees)
        print(f"Avg out-degree:  {avg_out:.2f}")
        print(f"Avg in-degree:   {avg_in:.2f}")


def cmd_sample(args: argparse.Namespace) -> None:
    """Generate a sample knowledge graph for testing."""
    kg = KnowledgeGraph()
    kg.add_triple("Alan_Turing", "born_in", "London", 1.0)
    kg.add_triple("Alan_Turing", "studied_at", "Cambridge", 0.9)
    kg.add_triple("Alan_Turing", "worked_on", "Enigma", 1.0)
    kg.add_triple("Alan_Turing", "field", "Computer_Science", 1.0)
    kg.add_triple("Cambridge", "located_in", "England", 1.0)
    kg.add_triple("London", "located_in", "England", 1.0)
    kg.add_triple("England", "part_of", "UK", 1.0)
    kg.add_triple("Computer_Science", "subfield_of", "Mathematics", 0.8)
    kg.add_triple("Mathematics", "subfield_of", "Science", 0.8)
    kg.add_triple("Enigma", "used_by", "Nazi_Germany", 1.0)
    kg.add_triple("Enigma", "broken_by", "Bletchley_Park", 1.0)
    kg.add_triple("Bletchley_Park", "located_in", "UK", 1.0)
    kg.add_triple("Alan_Turing", "worked_at", "Bletchley_Park", 1.0)
    kg.add_triple("Nazi_Germany", "located_in", "Europe", 1.0)
    kg.add_triple("UK", "located_in", "Europe", 1.0)

    out_lines = []
    for subj, edges in sorted(kg._adj.items()):
        for pred, obj, w in edges:
            out_lines.append(f"{subj}\t{pred}\t{obj}\t{w}")
    print("\n".join(out_lines))


def main() -> None:
    parser = argparse.ArgumentParser(
        prog="mhkpr",
        description="Multi-Hop Knowledge Path Ranking — rank paths in a KG by logical depth & connectivity",
    )
    sub = parser.add_subparsers(dest="command", help="subcommand")

    # rank
    r = sub.add_parser("rank", help="Discover and rank multi-hop paths")
    r.add_argument("--entities", "-e", nargs="+", help="Query entities (default: all nodes)")
    r.add_argument("--file", "-f", help="TSV KG file (default: stdin)")
    r.add_argument("--max-hops", type=int, default=3, help="Maximum path length in hops")
    r.add_argument("--beam", type=int, default=10, help="Beam width for search")
    r.add_argument("--max-paths", type=int, default=20, help="Max paths per pair")
    r.add_argument("--top-n", type=int, default=20, help="Top N results to display")
    r.add_argument("--output", "-o", choices=["pretty", "json", "tsv"], default="pretty")
    r.add_argument("--w-depth", type=float, default=0.3, help="Weight for logical depth")
    r.add_argument("--w-density",