File size: 5,090 Bytes
979853c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """
Knowledge Graph Visualization with OpenSearch + LightRAG WebUI
This script demonstrates two ways to visualize the knowledge graph
stored in OpenSearch:
1. **WebUI (recommended)**: Opens the LightRAG WebUI in your browser
for interactive graph exploration with search, filtering, and
force-directed layout.
2. **Standalone HTML**: Fetches graph data from the LightRAG Server API
and generates an interactive HTML file using Pyvis, similar to
graph_visual_with_html.py but reading from OpenSearch instead of
a local .graphml file.
Prerequisites:
1. LightRAG Server running with OpenSearch storage:
lightrag-server --host 0.0.0.0 --port 9621
2. Documents already indexed (e.g., via the WebUI or API)
Usage:
# Open WebUI for interactive exploration
python examples/graph_visual_with_opensearch.py
# Generate standalone HTML file
python examples/graph_visual_with_opensearch.py --html
# Custom server URL and output file
python examples/graph_visual_with_opensearch.py --html --server http://localhost:9621 --output my_graph.html
"""
import argparse
import os
import sys
import webbrowser
import pipmaster as pm
if not pm.is_installed("requests"):
pm.install("requests")
if not pm.is_installed("pyvis"):
pm.install("pyvis")
import requests
from pyvis.network import Network
def fetch_graph(server_url: str, label: str = "*", max_nodes: int = 300) -> dict:
"""Fetch knowledge graph data from LightRAG Server API."""
url = f"{server_url}/graphs"
params = {"label": label, "max_nodes": max_nodes}
resp = requests.get(url, params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def generate_html(graph_data: dict, output_file: str) -> str:
"""Generate an interactive HTML visualization from graph data."""
nodes = graph_data.get("nodes", [])
edges = graph_data.get("edges", [])
if not nodes:
print("No nodes found in the graph. Index some documents first.")
sys.exit(1)
print(f"Building visualization: {len(nodes)} nodes, {len(edges)} edges")
net = Network(height="100vh", notebook=False, cdn_resources="in_line")
# Add nodes with colors based on entity type
import hashlib
for node in nodes:
node_id = node.get("id", "")
props = node.get("properties", {})
entity_type = props.get("entity_type", "unknown")
description = props.get("description", "")
# Deterministic color from entity type
color_hash = int(hashlib.md5(entity_type.encode()).hexdigest()[:6], 16)
color = f"#{color_hash:06x}"
net.add_node(
node_id,
label=node_id,
title=f"[{entity_type}] {description[:200]}"
if description
else entity_type,
color=color,
)
# Add edges
for edge in edges:
source = edge.get("source", "")
target = edge.get("target", "")
props = edge.get("properties", {})
rel_type = edge.get("type", "")
description = props.get("description", "")
net.add_edge(
source,
target,
title=f"[{rel_type}] {description[:200]}" if description else rel_type,
label=rel_type,
)
net.save_graph(output_file)
print(f"Graph saved to {output_file}")
return output_file
def main():
parser = argparse.ArgumentParser(
description="Visualize LightRAG knowledge graph from OpenSearch"
)
parser.add_argument(
"--html",
action="store_true",
help="Generate standalone HTML file instead of opening WebUI",
)
parser.add_argument(
"--server",
default="http://localhost:9621",
help="LightRAG Server URL (default: http://localhost:9621)",
)
parser.add_argument(
"--output",
default="knowledge_graph_opensearch.html",
help="Output HTML file (default: knowledge_graph_opensearch.html)",
)
parser.add_argument(
"--label",
default="*",
help="Starting node label, or '*' for all nodes (default: *)",
)
parser.add_argument(
"--max-nodes",
type=int,
default=300,
help="Maximum nodes to fetch (default: 300)",
)
args = parser.parse_args()
# Verify server is running
try:
requests.get(f"{args.server}/health", timeout=5)
except requests.ConnectionError:
print(f"Error: Cannot connect to LightRAG Server at {args.server}")
print("Start the server first: lightrag-server --host 0.0.0.0 --port 9621")
sys.exit(1)
if args.html:
# Generate standalone HTML
graph_data = fetch_graph(args.server, args.label, args.max_nodes)
output = generate_html(graph_data, args.output)
webbrowser.open(f"file://{os.path.abspath(output)}")
else:
# Open WebUI graph explorer
url = f"{args.server}/#/graph"
print(f"Opening LightRAG WebUI graph explorer: {url}")
webbrowser.open(url)
if __name__ == "__main__":
main()
|