feat(reasoning): implement Ontological Multi-hop Traversal Engine and MCP tool (v0.9.0)

docs/API.md

@@ -151,6 +151,7 @@ asyncio.run(main())
 | `literature` | `biocontext literature <query> [-l INT]` | Search supporting peer-reviewed publications from Europe PMC / PubMed. |
 | `paper` | `biocontext paper <identifier>` | Fetch detailed paper metadata and abstract by PMID, PMCID, or DOI. |
 | `graph` | `biocontext graph <query> [--taxon INT]` | Construct connected biological Knowledge Subgraph (PRD-21). |
+| `traverse` | `biocontext traverse <source> <target> [--max-hops INT]` | Find and explain deterministic multi-hop reasoning paths (PRD-22). |
 | `mouse` | `biocontext mouse <MGI_ID>` | Lookup mouse gene model via MGI. |
 | `cache` | `biocontext cache [stats|clear]` | Inspect or clear SQLite cache. |
 | `serve` | `biocontext serve` | Start stdio MCP server for AI clients. |

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "biocontext-mcp"
-version = "0.8.0"
+version = "0.9.0"
 description = "Authoritative Biological Entity Resolution & Contextual Intelligence Framework"
 readme = "README.md"
 authors = [

src/biocontext/cli.py

@@ -249,6 +249,18 @@ async def run_cli_async(args: argparse.Namespace) -> int:
         print(subgraph.model_dump_json(indent=2))
         return 0
 
+    elif args.command == "traverse":
+        from biocontext.graph.traversal import TraversalEngine
+        engine = TraversalEngine(resolver=resolver)
+        res = await engine.traverse(
+            source_query=args.source,
+            target_query=args.target,
+            max_hops=args.max_hops,
+            min_score=args.min_score,
+        )
+        print(res.model_dump_json(indent=2))
+        return 0
+
     elif args.command == "cache":
         cache = resolver.cache
         if args.cache_action == "clear":
@@ -284,7 +296,7 @@ def main():
     elif args.command in (
         "resolve", "batch", "protein", "transcripts", "ortholog", "mouse",
         "annotate", "go", "pathway", "pathway-info", "disease", "targets",
-        "literature", "paper", "graph", "cache"
+        "literature", "paper", "graph", "traverse", "cache"
     ):
         sys.exit(asyncio.run(run_cli_async(args)))
     else:

src/biocontext/config.py

@@ -209,6 +209,16 @@ def get_headers(cls, email: Optional[str] = None) -> Dict[str, str]:
             {"flags": ["--no-literature"], "dest": "include_literature", "action": "store_false", "help": "Exclude literature citations"}
         ]
     },
+    {
+        "name": "traverse",
+        "help": "Find and explain deterministic multi-hop reasoning paths (PRD-22)",
+        "arguments": [
+            {"flags": ["source"], "help": "Source entity (e.g. TP53, BRCA1)"},
+            {"flags": ["target"], "help": "Target entity or keyword (e.g. cancer, apoptosis)"},
+            {"flags": ["--max-hops"], "type": int, "default": 3, "help": "Max hops (default: 3)"},
+            {"flags": ["--min-score"], "type": float, "default": 0.0, "help": "Minimum cumulative score (default: 0.0)"}
+        ]
+    },
 
 
 

src/biocontext/graph/__init__.py

@@ -6,10 +6,14 @@
     GraphEdge,
     GraphNode,
     NodeType,
+    PathStep,
     Provenance,
     SubGraphContext,
+    TraversalPath,
+    TraversalResult,
 )
 from biocontext.graph.store import InMemoryGraphStore
+from biocontext.graph.traversal import TraversalEngine
 
 __all__ = [
     "NodeType",
@@ -18,6 +22,11 @@
     "GraphNode",
     "GraphEdge",
     "SubGraphContext",
+    "PathStep",
+    "TraversalPath",
+    "TraversalResult",
     "InMemoryGraphStore",
     "KnowledgeGraphBuilder",
+    "TraversalEngine",
 ]
+

src/biocontext/graph/builder.py

@@ -179,40 +179,40 @@ async def build_gene_subgraph(
         # 5. Gene Ontology (GO Terms)
         if include_go and primary_accession:
             try:
-                go_ctx = await self.resolver.get_functional_annotations(primary_accession)
-                # Sample 2 biological processes and 2 molecular functions
-                sampled_go = (go_ctx.biological_processes[:2] + go_ctx.molecular_functions[:2])
-                for go in sampled_go:
-                    go_node_id = f"go:{go.go_id}"
-                    go_node = GraphNode(
-                        id=go_node_id,
-                        name=go.name or go.go_id,
-                        node_type=NodeType.GO_TERM,
-                        properties={"go_id": go.go_id, "aspect": go.aspect},
-                        provenance=Provenance(
-                            source_database="GeneOntology",
-                            confidence=1.0,
-                            source_id=go.go_id,
-                            evidence_code=go.evidence_code,
-                        ),
-                    )
-                    store.add_node(go_node)
-                    protein_node_id = f"protein:{primary_accession}"
-                    src_id = protein_node_id if store.get_node(protein_node_id) else gene_node_id
-                    edge_type = (
-                        EdgeType.INVOLVED_IN
-                        if go.aspect == "biological_process"
-                        else EdgeType.HAS_FUNCTION
-                    )
-                    store.add_edge(
-                        GraphEdge(
-                            source_id=src_id,
-                            target_id=go_node_id,
-                            edge_type=edge_type,
-                            weight=1.0,
-                            provenance=Provenance(source_database="GeneOntology", confidence=1.0),
+                go_ctx = await self.resolver.annotate_gene(primary_accession, max_terms_per_aspect=2)
+                if go_ctx:
+                    sampled_go = (go_ctx.biological_processes[:2] + go_ctx.molecular_functions[:2])
+                    for go in sampled_go:
+                        go_node_id = f"go:{go.go_id}"
+                        go_node = GraphNode(
+                            id=go_node_id,
+                            name=go.name or go.go_id,
+                            node_type=NodeType.GO_TERM,
+                            properties={"go_id": go.go_id, "aspect": go.aspect},
+                            provenance=Provenance(
+                                source_database="GeneOntology",
+                                confidence=1.0,
+                                source_id=go.go_id,
+                                evidence_code=go.evidence_code,
+                            ),
+                        )
+                        store.add_node(go_node)
+                        protein_node_id = f"protein:{primary_accession}"
+                        src_id = protein_node_id if store.get_node(protein_node_id) else gene_node_id
+                        edge_type = (
+                            EdgeType.INVOLVED_IN
+                            if go.aspect == "biological_process"
+                            else EdgeType.HAS_FUNCTION
+                        )
+                        store.add_edge(
+                            GraphEdge(
+                                source_id=src_id,
+                                target_id=go_node_id,
+                                edge_type=edge_type,
+                                weight=1.0,
+                                provenance=Provenance(source_database="GeneOntology", confidence=1.0),
+                            )
                         )
-                    )
             except Exception as e:
                 logger.warning(f"Failed to fetch GO annotations for graph: {e}")
 

src/biocontext/graph/models.py

@@ -66,3 +66,29 @@ class SubGraphContext(BaseModel):
     edges: List[GraphEdge] = Field(default_factory=list, description="Relationships connecting nodes in the subgraph")
     total_nodes: int = Field(0, description="Total node count")
     total_edges: int = Field(0, description="Total edge count")
+
+
+class PathStep(BaseModel):
+    """Single transition along an ontological traversal path."""
+    source_node: GraphNode = Field(..., description="Origin node of this hop")
+    edge: GraphEdge = Field(..., description="Connecting edge and relation")
+    target_node: GraphNode = Field(..., description="Destination node of this hop")
+
+
+class TraversalPath(BaseModel):
+    """An explainable, multi-hop path connecting two biological entities (PRD-22)."""
+    start_node_id: str = Field(..., description="Starting entity ID")
+    target_node_id: str = Field(..., description="Destination entity ID")
+    steps: List[PathStep] = Field(default_factory=list, description="Sequence of hops in the path")
+    hop_count: int = Field(0, description="Number of edges traversed")
+    cumulative_score: float = Field(1.0, ge=0.0, le=1.0, description="Product of confidence scores along path")
+    human_explanation: str = Field(..., description="Deterministic, zero-hallucination natural language narrative of the biological path")
+
+
+class TraversalResult(BaseModel):
+    """Result of a multi-hop graph traversal query."""
+    source_query: str = Field(..., description="Initial source query or symbol")
+    target_query: str = Field(..., description="Initial target query or disease/pathway")
+    paths_found: List[TraversalPath] = Field(default_factory=list, description="Discovered deterministic biological paths")
+    total_paths: int = Field(0, description="Count of valid paths discovered")
+

src/biocontext/graph/store.py

@@ -125,10 +125,46 @@ def get_subgraph(self, root_id: str, max_depth: int = 1) -> SubGraphContext:
             total_edges=len(collected_edges),
         )
 
+    def find_paths(
+        self,
+        start_id: str,
+        target_id: str,
+        max_hops: int = 3,
+        min_cumulative_score: float = 0.0,
+    ) -> List[List[GraphEdge]]:
+        """Find deterministic directed paths between start_id and target_id using DFS."""
+        if start_id not in self._nodes or target_id not in self._nodes:
+            return []
+
+        results: List[List[GraphEdge]] = []
+
+        def dfs(current_id: str, current_path: List[GraphEdge], visited: Set[str], current_score: float):
+            if current_id == target_id and current_path:
+                results.append(list(current_path))
+                return
+
+            if len(current_path) >= max_hops:
+                return
+
+            for edge in self.get_outgoing_edges(current_id):
+                next_id = edge.target_id
+                if next_id not in visited:
+                    next_score = current_score * edge.weight
+                    if next_score >= min_cumulative_score:
+                        visited.add(next_id)
+                        current_path.append(edge)
+                        dfs(next_id, current_path, visited, next_score)
+                        current_path.pop()
+                        visited.remove(next_id)
+
+        dfs(start_id, [], {start_id}, 1.0)
+        return results
+
     @property
     def node_count(self) -> int:
         return len(self._nodes)
 
     @property
     def edge_count(self) -> int:
         return sum(len(edges) for edges in self._out_edges.values())
+

src/biocontext/graph/traversal.py

@@ -0,0 +1,137 @@
+"""Ontological Multi-hop Traversal and Reasoning Engine (PRD-22)."""
+
+from typing import List, Optional
+from biocontext.graph.builder import KnowledgeGraphBuilder
+from biocontext.graph.models import (
+    GraphEdge,
+    GraphNode,
+    PathStep,
+    TraversalPath,
+    TraversalResult,
+)
+from biocontext.graph.store import InMemoryGraphStore
+from biocontext.logging import get_logger
+from biocontext.resolver import EntityResolver
+
+logger = get_logger("graph.traversal")
+
+
+class TraversalEngine:
+    """Performs deterministic, evidence-based multi-hop graph traversal and explanation."""
+
+    def __init__(
+        self,
+        resolver: Optional[EntityResolver] = None,
+        builder: Optional[KnowledgeGraphBuilder] = None,
+    ):
+        self.resolver = resolver or EntityResolver()
+        self.builder = builder or KnowledgeGraphBuilder(resolver=self.resolver)
+
+    def _generate_explanation(self, steps: List[PathStep], cumulative_score: float) -> str:
+        """Construct a deterministic natural language explanation from traversal steps."""
+        if not steps:
+            return "No connecting path found."
+
+        narrative_parts = []
+        for i, step in enumerate(steps, 1):
+            src = f"{step.source_node.name} ({step.source_node.node_type.value})"
+            rel = step.edge.edge_type.value.replace("_", " ")
+            tgt = f"{step.target_node.name} ({step.target_node.node_type.value})"
+            prov_str = ""
+            if step.edge.provenance:
+                prov_str = f" [via {step.edge.provenance.source_database}"
+                if step.edge.provenance.source_id:
+                    prov_str += f":{step.edge.provenance.source_id}"
+                prov_str += f", weight={step.edge.weight:.2f}]"
+
+            narrative_parts.append(f"Hop {i}: {src} {rel} {tgt}{prov_str}")
+
+        start_name = steps[0].source_node.name
+        end_name = steps[-1].target_node.name
+        overall_summary = (
+            f"Evidence path connects {start_name} to {end_name} in {len(steps)} hops "
+            f"(cumulative confidence score: {cumulative_score:.2f})."
+        )
+        return f"{overall_summary}\n" + "\n".join(f"- {part}" for part in narrative_parts)
+
+    async def traverse(
+        self,
+        source_query: str,
+        target_query: str,
+        max_hops: int = 3,
+        min_score: float = 0.0,
+    ) -> TraversalResult:
+        """Find deterministic multi-hop reasoning paths connecting two biological entities."""
+        # 1. Build subgraph around source entity
+        subgraph = await self.builder.build_gene_subgraph(
+            query=source_query,
+            include_pathways=True,
+            include_go=True,
+            include_diseases=True,
+            include_literature=True,
+        )
+
+        store = InMemoryGraphStore()
+        for node in subgraph.nodes:
+            store.add_node(node)
+        for edge in subgraph.edges:
+            store.add_edge(edge)
+
+        start_id = subgraph.root_node_id
+
+        # 2. Identify target node(s) matching target_query
+        target_q_clean = target_query.strip().lower()
+        candidate_target_ids = []
+        for node in subgraph.nodes:
+            if (
+                target_q_clean in node.id.lower()
+                or target_q_clean in node.name.lower()
+                or target_q_clean in str(node.properties).lower()
+            ):
+                candidate_target_ids.append(node.id)
+
+        traversal_paths: List[TraversalPath] = []
+
+        for target_id in candidate_target_ids:
+            if target_id == start_id:
+                continue
+
+            raw_paths: List[List[GraphEdge]] = store.find_paths(
+                start_id=start_id,
+                target_id=target_id,
+                max_hops=max_hops,
+                min_cumulative_score=min_score,
+            )
+
+            for edge_list in raw_paths:
+                steps: List[PathStep] = []
+                cumulative_score = 1.0
+                for edge in edge_list:
+                    src_node = store.get_node(edge.source_id)
+                    tgt_node = store.get_node(edge.target_id)
+                    if src_node and tgt_node:
+                        steps.append(PathStep(source_node=src_node, edge=edge, target_node=tgt_node))
+                        cumulative_score *= edge.weight
+
+                if steps:
+                    explanation = self._generate_explanation(steps, cumulative_score)
+                    traversal_paths.append(
+                        TraversalPath(
+                            start_node_id=start_id,
+                            target_node_id=target_id,
+                            steps=steps,
+                            hop_count=len(steps),
+                            cumulative_score=cumulative_score,
+                            human_explanation=explanation,
+                        )
+                    )
+
+        # Sort paths by highest cumulative score and lowest hop count
+        traversal_paths.sort(key=lambda p: (-p.cumulative_score, p.hop_count))
+
+        return TraversalResult(
+            source_query=source_query,
+            target_query=target_query,
+            paths_found=traversal_paths,
+            total_paths=len(traversal_paths),
+        )