Graph Traversal

The GraphTraversalEngine provides advanced graph traversal algorithms for the Context Graph, including depth-first search (DFS), breadth-first search (BFS), shortest path finding, cycle detection, and subgraph extraction. It integrates with the ContextGraphStore for production-ready traversals.

Overview

Graph traversal is the foundation for lineage exploration, impact analysis, and dependency mapping in the Context Graph. The engine supports configurable depth limits, relationship filters, and traversal callbacks.

Source: data-plane/ai-service/src/context_graph/services/graph_traversal.py

Traversal Directions

Direction	Description	Use Case
`UPSTREAM`	Follow relationships backward to find sources	"Where does this data come from?"
`DOWNSTREAM`	Follow relationships forward to find dependents	"What depends on this dataset?"
`BOTH`	Bidirectional traversal	"Show me the full neighborhood"

Traversal Strategies

Strategy	Description	Performance
`DFS`	Depth-first search for deep lineage exploration	Memory-efficient, good for deep chains
`BFS`	Breadth-first search for neighbor discovery	Finds closest entities first
`SHORTEST_PATH`	Find shortest path between two entities	Optimal for connection discovery
`ALL_PATHS`	Find all paths between two entities	Expensive, use with depth limits

Configuration

from context_graph.services.graph_traversal import (
    TraversalConfig,
    TraversalDirection,
    TraversalStrategy,
)
 
config = TraversalConfig(
    max_depth=10,          # Maximum traversal depth
    max_nodes=1000,        # Maximum nodes to visit
    strategy=TraversalStrategy.BFS,
    direction=TraversalDirection.UPSTREAM,
    include_properties=True,
    relationship_types=["DERIVED_FROM", "CONSUMED_BY"],
)

Basic Traversal

engine = get_traversal_engine()
 
result = await engine.traverse(
    start_urn="urn:matih:model:acme:churn_predictor",
    tenant_id="acme",
    direction=TraversalDirection.UPSTREAM,
    max_depth=5,
)
 
for node in result.nodes:
    print(f"  {node.urn} (depth: {node.depth})")

Path Finding

Find the shortest path between two entities:

path = await engine.find_path(
    source_urn="urn:matih:dataset:acme:raw_events",
    target_urn="urn:matih:dashboard:acme:sales_overview",
    tenant_id="acme",
    max_depth=10,
)

Cycle Detection

Detect circular dependencies in the graph:

cycles = await engine.detect_cycles(
    start_urn="urn:matih:pipeline:acme:etl_main",
    tenant_id="acme",
    max_depth=20,
)
 
for cycle in cycles:
    print(f"Cycle: {' -> '.join(cycle.path)}")

Subgraph Extraction

Extract an isolated subgraph around an entity for visualization or analysis:

subgraph = await engine.extract_subgraph(
    center_urn="urn:matih:dataset:acme:sales_data",
    tenant_id="acme",
    radius=3,
)

REST API Endpoints

Endpoint	Method	Description
`/api/v1/context-graph/traverse`	POST	Execute a graph traversal
`/api/v1/context-graph/traverse/paths`	POST	Find paths between entities
`/api/v1/context-graph/traverse/cycles`	POST	Detect circular dependencies

Performance Limits

Parameter	Default	Maximum
`max_depth`	10	50
`max_nodes`	1000	10000
Timeout	30 seconds	120 seconds

Discovery Search Agent Thinking Overview