An Implementation of IWE’s Context Bridge as an AI-Powered Knowledge Graph with Agentic RAG, OpenAI Function Calling, and Graph Traversal
In this tutorial, we implement IWE: an open-source, Rust-powered personal knowledge management system that treats markdown notes as a navigable knowledge graph. Since IWE is a CLI/LSP tool designed for local editors. We build a realistic developer knowledge base from scratch, wire up wiki-links and markdown links into a directed graph, and then walk through every major IWE operation: fuzzy search with find, context-aware retrieval with retrieve, hierarchy display with tree, document consolidation with squash, statistics with stats, and DOT graph export for visualization. We then go beyond the CLI by integrating OpenAI to power IWE-style AI transforms: summarization, link suggestion, and todo extraction, directly against our knowledge graph. Finally, we construct a full agentic RAG pipeline where an AI agent navigates the graph using function-calling tools, performs multi-hop reasoning across interconnected documents, identifies knowledge gaps, and even generates new notes that slot into the existing structure. Copy CodeCopiedUse a different Browser import subprocess, sys def _install(pkg): subprocess.check_call([sys.executable, “-m”, “pip”, “install”, “-q”, pkg]) _install(“openai”) _install(“graphviz”) import re, json, textwrap, os, getpass from collections import defaultdict from dataclasses import dataclass, field from typing import Optional from datetime import datetime try: from google.colab import userdata OPENAI_API_KEY = userdata.get(“OPENAI_API_KEY”) if not OPENAI_API_KEY: raise ValueError print(” Loaded OPENAI_API_KEY from Colab secrets.”) except Exception: OPENAI_API_KEY = getpass.getpass(” Enter your OpenAI API key: “) print(” API key received.”) os.environ[“OPENAI_API_KEY”] = OPENAI_API_KEY from openai import OpenAI client = OpenAI(api_key=OPENAI_API_KEY) print(“n” + “=” * 72) print(” IWE Advanced Tutorial — Knowledge Graph + AI Agents”) print(“=” * 72) @dataclass class Section: level: int title: str content: str children: list = field(default_factory=list) @dataclass class Document: key: str title: str raw_content: str sections: list = field(default_factory=list) outgoing_links: list = field(default_factory=list) tags: list = field(default_factory=list) created: str = “” modified: str = “” class KnowledgeGraph: def __init__(self): self.documents: dict[str, Document] = {} self.backlinks: dict[str, set] = defaultdict(set) _WIKI_LINK = re.compile(r”[[([^]|]+)(?:|([^]]+))?]]”) _MD_LINK = re.compile(r”[([^]]+)](([^)]+))”) _HEADER = re.compile(r”^(#{1,6})s+(.+)”, re.MULTILINE) _TAG = re.compile(r”#([a-zA-Z][w/-]*)”) def _extract_links(self, text: str) -> list[str]: links = [] for match in self._WIKI_LINK.finditer(text): links.append(match.group(1).strip()) for match in self._MD_LINK.finditer(text): target = match.group(2).strip() if not target.startswith(“http”): target = target.replace(“.md”, “”) links.append(target) return links def _parse_sections(self, text: str) -> list[Section]: sections = [] parts = self._HEADER.split(text) i = 1 while i < len(parts) – 1: level = len(parts[i]) title = parts[i + 1].strip() body = parts[i + 2] if i + 2 < len(parts) else “” sections.append(Section(level=level, title=title, content=body.strip())) i += 3 return sections def _extract_tags(self, text: str) -> list[str]: tags = set() for line in text.split(“n”): if line.strip().startswith(“#”) and ” ” in line.strip(): stripped = re.sub(r”^#{1,6}s+.*”, “”, line) for m in self._TAG.finditer(stripped): tags.add(m.group(1)) else: for m in self._TAG.finditer(line): tags.add(m.group(1)) return sorted(tags) def add_document(self, key: str, content: str) -> Document: sections = self._parse_sections(content) title = sections[0].title if sections else key links = self._extract_links(content) tags = self._extract_tags(content) now = datetime.now().strftime(“%Y-%m-%d %H:%M”) doc = Document( key=key, title=title, raw_content=content, sections=sections, outgoing_links=links, tags=tags, created=now, modified=now, ) self.documents[key] = doc for target in links: self.backlinks[target].add(key) return doc def get(self, key: str) -> Optional[Document]: return self.documents.get(key) def find(self, query: str, roots_only: bool = False, limit: int = 10) -> list[str]: q = query.lower() scored = [] for key, doc in self.documents.items(): score = 0 if q in doc.title.lower(): score += 10 if q in doc.raw_content.lower(): score += doc.raw_content.lower().count(q) if q in key.lower(): score += 5 for tag in doc.tags: if q in tag.lower(): score += 3 if score > 0: scored.append((key, score)) scored.sort(key=lambda x: -x[1]) results = [k for k, _ in scored[:limit]] if roots_only: results = [k for k in results if not self.backlinks.get(k)] return results def retrieve(self, key: str, depth: int = 1, context: int = 1, exclude: set = None) -> str: exclude = exclude or set() parts = [] if context > 0: parents_of = list(self.backlinks.get(key, set()) – exclude) for p in parents_of[:context]: pdoc = self.get(p) if pdoc: parts.append(f”[CONTEXT: {pdoc.title}]n{pdoc.raw_content[:300]}…n”) exclude.add(p) doc = self.get(key) if not doc: return f” Document ‘{key}’ not found.” parts.append(doc.raw_content) exclude.add(key) if depth > 0: for link in doc.outgoing_links: if link not in exclude: child = self.get(link) if child: parts.append(f”n—n[LINKED: {child.title}]n”) parts.append( self.retrieve(link, depth=depth – 1, context=0, exclude=exclude) ) return “n”.join(parts) def tree(self, key: str, indent: int = 0, _visited: set = None) -> str: _visited = _visited if _visited is not None else set() doc = self.get(key) if not doc: return “” prefix = ” ” * indent + (“└─ ” if indent else “”) if key in _visited: return f”{prefix}{doc.title} ({key}) (circular ref)” _visited.add(key) lines = [f”{prefix}{doc.title} ({key})”] for link in doc.outgoing_links: if self.get(link): lines.append(self.tree(link, indent + 1, _visited)) return “n”.join(lines) def squash(self, key: str, visited: set = None) -> str: visited = visited or set() doc = self.get(key) if not doc or key in visited: return “” visited.add(key) parts = [doc.raw_content] for link in doc.outgoing_links: child_content = self.squash(link, visited) if child_content: parts.append(f”n{‘─’ * 40}n”) parts.append(child_content) return “n”.join(parts) def stats(self) -> dict: total_words = sum(len(d.raw_content.split()) for d in self.documents.values()) total_links = sum(len(d.outgoing_links) for d in self.documents.values()) orphans = [k for k in self.documents if not self.backlinks.get(k) and not self.documents[k].outgoing_links] all_tags = set() for d in self.documents.values(): all_tags.update(d.tags) return { “total_documents”: len(self.documents), “total_words”: total_words, “total_links”: total_links, “unique_tags”: len(all_tags), “tags”: sorted(all_tags), “orphan_notes”: orphans, “avg_words_per_doc”: total_words // max(len(self.documents), 1), } def export_dot(self, highlight_key: str = None) -> str: lines = [‘digraph KnowledgeGraph {‘, ‘ rankdir=LR;’, ‘ node [shape=box, style=”rounded,filled”, fillcolor=”#f0f4ff”, ‘ ‘fontname=”Helvetica”, fontsize=10];’, ‘ edge [color=”#666666″, arrowsize=0.7];’] for key, doc in self.documents.items(): label = doc.title[:30] color = ‘#ffe4b5’ if highlight_key == key else ‘#f0f4ff’ lines.append(f’ “{key}” [label=”{label}”, fillcolor=”{color}”];’) for key, doc in self.documents.items(): for link in doc.outgoing_links: if link in self.documents: lines.append(f’ “{key}” -> “{link}”;’) lines.append(“}”) return “n”.join(lines) print(“n Section 1 complete — KnowledgeGraph class defined.n”) We install the required dependencies, securely accept the OpenAI API key through Colab secrets or a password prompt, and initialize the OpenAI client. We then define the three foundational data classes, Section, Document, and KnowledgeGraph, that mirror IWE’s arena-based graph architecture where every markdown file is a node and every link is a directed edge. We implement the full suite of IWE CLI operations on the




