We implement Tree-KG in this tutorial. It is an advanced hierarchical graph knowledge system. Tree-KG goes beyond the traditional retrieval-augmented-generation by combining graph structure and explicit semantic embeddings. This tutorial shows how knowledge can be organized in a hierarchical tree structure that mimics human learning, starting with broad concepts to finer-grained ones, then reasoning across the hierarchy using controlled multihop exploration. The graph is built from scratch and the nodes are enhanced with embedded information. A reasoning agent navigates related concepts and ancestors. Visit the FULL CODES here.
!pip install networkx matplotlib anthropic sentence-transformers scikit-learn numpy
Import networkx as Nx
Import matplotlib.pyplot into plt
Import List, Dict. Tuple. Optional. Set
Numpy can be imported as a np
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer
Deque import collections defaultdict
Download jsonThe Tree-KG System is built using the Tree KG Core Libraries. The tools we set up include graph visualization and construction, similarity and semantic search, efficient data handling, traversal, and scoring. Visit the FULL CODES here.
TreeKnowledgeGraph class:
"""
Hierarchical Knowledge Diagram that mimics the human learning pattern.
Supports contextual navigation, multi-hop reasoning.
"""
Def __init__ (self, embedding_model str="all-MiniLM-L6-v2"):
self.graph = nx.DiGraph()
self.embedder = SentenceTransformer(embedding_model)
self.node_embeddings = {}
self.node_metadata = {}
Def add_node (self)
node_id: str,
content: str,
node_type: str="concept",
Metadata: Optional[Dict] = None):
"""Add a node with semantic embedding and metadata."""
embedding = self.embedder.encode(content, convert_to_tensor=False)
self.graph.add_node(node_id,
content=content,
node_type=node_type,
metadata=metadata or {})
self.node_embeddings[node_id] Embedding
self.node_metadata[node_id] = {
'content': content,
'type': node_type,
'metadata': metadata or {}
}
Add_edge (self)
parent: str,
child: str,
relationship: str="contains",
Weight: float = 1
"""Add hierarchical or associative edge between nodes."""
self.graph.add_edge(parent, child,
relationship=relationship,
weight=weight)
def get_ancestors(self, node_id: str, max_depth: int = 5) -> List[str]:
"""Get all ancestor nodes (hierarchical context)."""
Ancestors []
Node_id = current
depth = 0
While depth List[str]:
"""Get all descendant nodes."""
descendants = []
queue = deque([(node_id, 0)])
visited = {node_id}
While you wait
current, depth = queue.popleft()
if depth >= max_depth:
You can continue reading
for child in self.graph.successors(current):
If child was not visited
visited.add(child)
descendants.append(child)
queue.append((child, depth + 1))
return descendants
def semantic_search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
"""Find most semantically similar nodes to query."""
query_embedding = self.embedder.encode(query, convert_to_tensor=False)
Similarities = []
for node_id, embedding in self.node_embeddings.items():
sim = cosine_similarity
query_embedding.reshape(1, -1),
embedding.reshape(1, -1)
)[0][0]
similarities.append((node_id, float(sim)))
similarities.sort(key=lambda x: x[1], reverse=True)
Return SimilaritiesContext = ""[:top_k]
def get_subgraph_context(self, node_id: str, depth: int = 2) -> Dict:
"""Get rich contextual information around a node."""
The following is a list of other words and phrases that you can use. = {
'node': self.node_metadata.get(node_id, {}),
'ancestors': [],
'descendants': [],
'siblings': [],
'related': []
}
ancestors = self.get_ancestors(node_id)
The following is a list of other words and phrases that you can use.['ancestors'] = [
self.node_metadata.get(a, {}) for a in ancestors
]
descendants = self.get_descendants(node_id, depth)
context['descendants'] = [
self.node_metadata.get(d, {}) for d in descendants
]
parents = list(self.graph.predecessors(node_id))
Parents:
Sisters list(self.graph.successors(parents[0]))
siblings = [s for s in siblings if s != node_id]
You can also find out more about the following:['siblings'] = [
self.node_metadata.get(s, {}) for s in siblings
]
Return to contextTreeKnowledgeGraph defines the class core that organizes knowledge into a hierarchy of directed hierarchies enriched with embeddings. The graph representations and relationships are stored to allow navigation of concepts in a structural way. We also perform similarity-based retrieval. Visit the FULL CODES here.
class MultiHopReasoningAgent:
"""
Intelligent multi-hop reasoning agent that traverses the knowledge graph.
"""
def __init__(self, kg: TreeKnowledgeGraph):
kg
self.reasoning_history = []
Def. self.
query: str,
Maximum number of hops is 3
exploration_width: int = 3) -> Dict:
"""
Use multi-hop logic to solve a question.
Strategy:
1. Search for initial relevant nodes
2. See the context in graphs around these nodes
3. Explore the world in a broad-first manner with a focus on relevance.
4. Averaging information from multiple hops
"""
reasoning_trace = {
'query': query,
'hops': [],
'final_context': {},
'reasoning_path': []
}
initial_nodes = self.kg.semantic_search(query, top_k=exploration_width)
reasoning_trace['hops'].append({
'hop_number': 0,
'action': 'semantic_search',
'nodes_found': initial_nodes
})
Visit = Set()
current_frontier = [node_id for node_id, _ in initial_nodes]
all_relevant_nodes = set(current_frontier)
Hop in range (max_hops + 1)
next_frontier = []
hop_info = {
'hop_number': hop,
'explored_nodes': [],
'new_discoveries': []
}
Node_id for current_frontier
Node_id if visited
Continue reading
visited.add(node_id)
context = self.kg.get_subgraph_context(node_id, depth=1)
connected_nodes = []
In context, what is meant by ancestor?['ancestors']:
If the content is ancestor's:
connected_nodes.append(ancestor)
In context, descendant['descendants']:
if 'content' in descendant:
connected_nodes.append(descendant)
In context, sibling is used to refer to a person.['siblings']:
if 'content' in sibling:
connected_nodes.append(sibling)
relevant_connections = self._score_relevance(
query, connected_nodes, top_k=exploration_width
)
hop_info['explored_nodes'].append({
'node_id': node_id,
'content': self.kg.node_metadata[node_id]['content'][:100],
'connections_found': len(relevant_connections)
})
for conn_content, score in relevant_connections:
for nid, meta in self.kg.node_metadata.items():
Meta['content'] Conn_content not visited and nid is missing
next_frontier.append(nid)
all_relevant_nodes.add(nid)
hop_info['new_discoveries'].append({
'node_id': nid,
'relevance_score: score
})
Break.
reasoning_trace['hops'].append(hop_info)
current_frontier = next_frontier
If the current_frontier is not available:
Break:
final_context = self._aggregate_context(query, all_relevant_nodes)
reasoning_trace['final_context'] = final_context
reasoning_trace['reasoning_path'] = list(all_relevant_nodes)
self.reasoning_history.append(reasoning_trace)
Return Reasoning_Trace
def _score_relevance(self,
query: str,
Candidates' List[Dict],
top_k: int = 3) -> List[Tuple[str, float]]:
"""Score candidate nodes by relevance to query."""
If not candidates
You can return to your original language by clicking here. []
query_embedding = self.kg.embedder.encode(query)
scores = []
Candidate in Candidates
content = candidate.get('content', '')
If not satisfied:
You can continue reading
candidate_embedding = self.kg.embedder.encode(content)
Similarity = cosine_similarity
query_embedding.reshape(1, -1),
candidate_embedding.reshape(1, -1)
)[0][0]
scores.append((content, float(similarity)))
scores.sort(key=lambda x: x[1], reverse=True)
Return ScoresAggregate =[:top_k]
def _aggregate_context(self, query: str, node_ids: Set[str]) -> Dict:
"""Aggregate and rank information from all discovered nodes."""
aggregated = {
'total_nodes': len(node_ids),
'hierarchical_paths': [],
'key_concepts': [],
'synthesized_answer': []
}
Node_id is the node_id.
ancestors = self.kg.get_ancestors(node_id)
If ancestors
Path = ancestors[::-1] + [node_id]
path_contents = [
self.kg.node_metadata[n]['content']
for n in path if n in self.kg.node_metadata
]
ариму дл ариму['hierarchical_paths'].append(path_contents)
For node_id:
meta = self.kg.node_metadata.get(node_id, {})
Aggregate['key_concepts'].append({
'id': node_id,
'content': meta.get('content', ''),
'type': meta.get('type', 'unknown')
})
Node_id is the node_id.
content = self.kg.node_metadata.get(node_id, {}).get('content', '')
If you are content
аадание на аадание['synthesized_answer'].append(content)
Return to aggregate
def explain_reasoning(self, trace: Dict) -> str:
"""Generate human-readable explanation of reasoning process."""
explanation ="Query: {trace['query']}n"]
explanation.append(f"Total hops performed: {len(trace['hops']) - 1}n")
explanation.append(f"Total relevant nodes discovered: {len(trace['reasoning_path'])}nn")
Hop_info is a trace element.['hops']:
hop_num = hop_info['hop_number']
explanation.append(f"--- Hop {hop_num} ---")
if hop_num == 0:
explanation.append(f"Action: Initial semantic search")
explanation.append(f"Found {len(hop_info['nodes_found'])} candidate nodes")
Score in hop_info for node_id['nodes_found'][:3]:
explanation.append(f" - {node_id} (relevance: {score:.3f})")
else:
explanation.append(f"Explored {len(hop_info['explored_nodes'])} nodes")
explanation.append(f"Discovered {len(hop_info['new_discoveries'])} new relevant nodes")
explanation.append("")
explanation.append("n--- Final Aggregated Context ---")
Context = Trace['final_context']
explanation.append(f"Total concepts integrated: {context['total_nodes']}")
explanation.append(f"Hierarchical paths found: {len(context['hierarchical_paths'])}")
Return to the Homepage "n".join(explanation)Our multi-hop reasoning agents actively explore the knowledge graph, instead of passively retrieving all nodes. Starting with semantically-relevant concepts, we expand to ancestors and descendants and then score the connections iteratively in order to guide exploration. We produce a coherent answer and an explanation of the reasoning by combining hierarchical pathways and synthesising content. Click here to see the FULL CODES here.
def build_software_development_kb() -> TreeKnowledgeGraph:
"""Build a comprehensive software development knowledge graph."""
TreeKnowledgeGraph()
The domain name is the same as the "root" of a node.
kg.add_node('programming',
"Programming is the process of writing, maintaining, and testing code in order to develop software applications".
'domain')
kg.add_node('architecture',
Software Architecture is the design of high-level components and structures for software systems.
'domain')
kg.add_node('domain')
kg.add_edge('root', 'programming', 'contains')
kg.add_edge('root', 'architecture', 'contains')
kg.add_edge('root', 'devops', 'contains')
kg.add_node('python',
'language')
kg.add_node('javascript',
JavaScript is an interactive language that allows both server and client side applications to be developed.
'language')
kg.add_node('rust',
'language')
kg.add_edge('programming', 'python', 'includes')
kg.add_edge('programming', 'javascript', 'includes')
kg.add_edge('programming', 'rust', 'includes')
kg.add_node('python_basics',
Python fundamentals include variables, types of data, flow control, functions and object-oriented programing.
'concept')
kg.add_node('python_performance',
Python Performance Optimization involves techniques such as caching, profiling and using C Extensions, along with leveraging Async Programming.
'concept')
kg.add_node('python_data',
Python for Data Science uses NumPy and Pandas for data analysis and machine learning.
'concept')
kg.add_edge('python', 'python_basics', 'contains')
kg.add_edge('python', 'python_performance', 'contains')
kg.add_edge('python', 'python_data', 'contains')
kg.add_node('async_io',
'Asynchronous IO for Python allows non blocking operations with async/await syntactics using the asyncio libraries'
'technique')
kg.add_node('multiprocessing',
The Python Multiprocessing system uses different processes to avoid GIL. This allows true parallel processing for tasks that are CPU bound.
'technique')
kg.add_node('cython',
"Cython compiles Python into C for significant gains in performance, particularly numerical computations, and tight loops"
'tool')
kg.add_node('profiling',
Python Profiling is a way to identify performance bottlenecks by using tools such as cProfile and line_profiler.
'technique')
kg.add_edge('python_performance', 'async_io', 'contains')
kg.add_edge('python_performance', 'multiprocessing', 'contains')
kg.add_edge('python_performance', 'cython', 'contains')
kg.add_edge('python_performance', 'profiling', 'contains')
kg.add_node('event_loop',
Event Loop manages asynchronous tasks and handles callbacks, coroutines, and scheduling.
'concept')
kg.add_node('coroutines',
The term 'Coroutine' is used to describe a special function that allows the execution of a task to be stopped with an await. This is useful for multi-tasking.
'concept')
kg.add_node('asyncio_patterns',
"AsyncIO pattern include gathering for simultaneous execution, create_tasks for background tasks and queues as producer-consumers"
'pattern')
kg.add_edge('async_io', 'event_loop', 'contains')
kg.add_edge('async_io', 'coroutines', 'contains')
kg.add_edge('async_io', 'asyncio_patterns', 'contains')
kg.add_node('microservices',
Microservices architecture is a way to decompose applications into smaller, more independent services, which communicate using APIs.
'pattern')
kg.add_edge('architecture', 'microservices', 'contains')
kg.add_edge('async_io', 'microservices', 'related_to')
kg.add_node('containers',
"Containers packages dependencies of applications and units into isolated containers, ensuring consistency in environments"
'technology')
kg.add_edge('devops', 'containers', 'contains')
kg.add_edge('microservices', 'containers', 'deployed_with')
kg.add_node('numpy_optimization',
NumPy optimizes C and Fortran libraries by using vectorization, broadcasting, and optimized C.
'technique')
kg.add_edge('python_data', 'numpy_optimization', 'contains')
kg.add_edge('python_performance', 'numpy_optimization', 'related_to')
Return kgThe knowledge base we build is hierarchical and rich. It progresses through high-level domains to techniques and tool specifics. We explicitly encode parent–child and cross-domain relationships so that concepts such as Python performance, async I/O, and microservices are structurally connected rather than isolated. We can simulate knowledge being learned, revisited at different levels and use this to enable multi-hop reasoning about real software topics. Visit the FULL CODES here.
def visualize_knowledge_graph(kg: TreeKnowledgeGraph,
Highlight_nodes Optional[List[str]] = None):
"""Visualize the knowledge graph structure."""
plt.figure(figsize=(16, 12))
pos = nx.spring_layout(kg.graph, k=2, iterations=50, seed=42)
node_colors = []
For node in kg.graph.nodes():
If highlight_nodes is a node within highlight_nodes
node_colors.append('yellow')
else:
node_type = kg.graph.nodes[node].get('node_type', 'concept')
color_map = {
'domain': 'lightblue',
'language': 'lightgreen',
'concept': 'lightcoral',
'technique': 'lightyellow',
'tool': 'lightpink',
'pattern': 'lavender',
'technology': 'peachpuff'
}
node_colors.append(color_map.get(node_type, 'lightgray'))
nx.draw_networkx_nodes(kg.graph, pos,
node_color=node_colors,
node_size=2000,
alpha=0.9)
nx.draw_networkx_edges(kg.graph, pos,
edge_color="gray",
arrows=True,
arrowsize=20,
alpha=0.6,
width=2)
nx.draw_networkx_labels(kg.graph, pos,
font_size=8,
font_weight="bold")
plt.title("Tree-KG: Hierarchical Knowledge Graph", fontsize=16, fontweight="bold")
plt.axis('off')
plt.tight_layout()
plt.show()
def run_demo():
"""Run complete demonstration of Tree-KG system."""
Print("=" * 80)
print("Tree-KG: Hierarchical Knowledge Graph Demo")
print("=" * 80)
print()
print("Building knowledge graph...")
kg = build_software_development_kb()
print(f"✓ Created graph with {kg.graph.number_of_nodes()} nodes and {kg.graph.number_of_edges()} edgesn")
print("Visualizing knowledge graph...")
visualize_knowledge_graph(kg)
agent = MultiHopReasoningAgent(kg)
queries = [
"How can I improve Python performance for IO-bound tasks?",
"What are the best practices for async programming?",
"How does microservices architecture relate to Python?"
]
For i, query, in enumerate (queries, 1)
print(f"n{'=' * 80}")
print(f"QUERY {i}: {query}")
print('=' * 80)
trace = query.reason.max_hops=3, exploring_width=3
explanation = agent.explain_reasoning(trace)
print(explanation)
print("n--- Sample Hierarchical Paths ---")
Path j in enumerate()['final_context']['hierarchical_paths'][:3], 1):
print(f"nPath {j}:")
Concept k in the enumerate path:
Indent = " " * k
print(f"{indent}→ {concept[:80]}...")
print("n--- Synthesized Context ---")
Answer_parts = trace['final_context']['synthesized_answer'][:5]
Part of answer_parts
You can also print(f"• {part[:150]}...")
print()
print("nVisualizing reasoning path for last query...")
last_trace = agent.reasoning_history[-1]
visualize_knowledge_graph(kg, highlight_nodes=last_trace['reasoning_path'])
print("n" + "=" * 80)
print("Demo complete!")
print("=" * 80)The hierarchical structure is visualized using colors and layouts to differentiate domains, concepts and techniques. We can also highlight the reasoning pathway. Then we run an end to end demo wherein we create the graph, perform multi-hop reasoning using realistic queries and then print the reasoning trail and synthesized context. We can see the way the agent explores the network, surfaces the hierarchical paths and provides its explanations in a clear and understandable fashion. Visit the FULL CODES here.
class AdvancedTreeKG(TreeKnowledgeGraph):
"""Extended Tree-KG with advanced features."""
Def __init__ (self, embedding_model str="all-MiniLM-L6-v2"):
You can also check out our supersized().__init__(embedding_model)
self.node_importance = {}
def compute_node_importance(self):
"""Compute importance scores using PageRank-like algorithm."""
if self.graph.number_of_nodes() == 0:
You can return to your original language by clicking here.
pagerank = nx.pagerank(self.graph)
betweenness = nx.betweenness_centrality(self.graph)
For node in graph.nodesContext = ""():
self.node_importance[node] = {
'pagerank': pagerank.get(node, 0),
'betweenness': betweenness.get(node, 0),
'combined': pagerank.get(node, 0) * 0.7 + betweenness.get(node, 0) * 0.3
}
def find_shortest_path_with_context(self,
source: str,
target: str) -> Dict:
"""Find shortest path and extract all context along the way."""
try:
path = nx.shortest_path(self.graph, source, target)
context = {
'path': path,
'path_length': len(path) - 1,
'nodes_detail': []
}
Node on path
detail = {
'id': node,
'content': self.node_metadata.get(node, {}).get('content', ''),
'importance': self.node_importance.get(node, {}).get('combined', 0)
}
You can also find out more about the following:['nodes_detail'].append(detail)
Return to context
NetworkXNoPath:
return {Return "path": [], 'error': 'No path exists'}We add graph-level intelligence to the Tree-KG by calculating node importance with centrality measures. PageRank scores and betweenness are combined to determine concepts which play an important structural role in the connection of knowledge within a graph. We can also retrieve the shortest path enriched by contextual and importance data, making it easier to explain and reason between concepts. Visit the FULL CODES here.
If __name__ is equal to "__main__":
run_demo()
print("nn" + "=" * 80)
print("ADVANCED FEATURES DEMO")
print("=" * 80)
print("nBuilding advanced Tree-KG...")
adv_kg= AdvancedTreeKG()
adv_kg = build_software_development_kb()
ADV_KG_NEW = AdvancedTreeKG()
adv_kg_new.graph = adv_kg.graph
adv_kg_new.node_embeddings = adv_kg.node_embeddings
adv_kg_new.node_metadata = adv_kg.node_metadata
print("Computing node importance scores...")
adv_kg_new.compute_node_importance()
print("nTop 5 most important nodes:")
sorted_nodes = sorted(
adv_kg_new.node_importance.items(),
Key=lambda: x[1]['combined'],
reverse=True
)[:5]
Scores for each node in the sorted_nodes
content = adv_kg_new.node_metadata[node]['content'][:60]
print(f" {node}: {content}...")
print(f" Combined score: {scores['combined']:.4f}")
print("n✓ Tree-KG Tutorial Complete!")
print("nKey Takeaways:")
print("1. Tree-KG enables contextual navigation vs simple chunk retrieval")
print("2. Multi-hop reasoning discovers relevant information across graph structure")
print("3. Hierarchical organization mirrors human learning patterns")
print("4. Semantic search + graph traversal = powerful RAG alternative")The demo is completed and we then demonstrate the features that are available to the user. Node importance scores are computed to highlight the most important concepts within the graph. We also examine how the structural centrality of the graph aligns with the semantic relevance.
Tree-KG enables a richer understanding of the world by combining semantic search with hierarchical context and multi-hop logic within a unified framework. Instead of retrieving text fragments we could traverse meaningful knowledge pathways, aggregate insights at different levels and provide explanations which reflect the way conclusions were formed. We showed that by extending Tree-KG with context-aware path extraction and importance scoring, it can be used as a solid foundation to build intelligent agents, research assistances or domain-specific systems which demand transparency, depth and structure beyond RAG.
Click here to find out more FULL CODES here. Also, feel free to follow us on Twitter Don’t forget about our 100k+ ML SubReddit Subscribe now our Newsletter. Wait! Are you using Telegram? now you can join us on telegram as well.
