atareao / knowledge-graph-t09-demo

#!/usr/bin/env python3 """ knowledge_graph.py — Visualización de un grafo de conocimiento con NetworkX Construye un grafo de conocimiento con las herramientas del stack del podcast, muestra consultas sobre el grafo, y genera una imagen PNG del mismo. Propósito para el episodio 828: Demostrar visualmente qué es un "grafo de conocimiento" — la base de GraphRAG. Se ejecuta después de la comparativa RAG vs GraphRAG para mostrar un grafo real. Uso: python knowledge_graph.py # Modo interactivo (consulta + PNG) python knowledge_graph.py --no-png # Solo consultas, sin guardar imagen python knowledge_graph.py --png-only # Solo generar PNG Requisitos: pip install networkx matplotlib """ import argparse import sys from typing import Dict, List, Tuple try: import networkx as nx except ImportError: print("❌ Falta networkx. Instala con: pip install networkx matplotlib") sys.exit(1) try: from networkx.drawing.nx_pydot import graphviz_layout except ImportError: print("❌ Falta pydot. Instala con: pip install pydot") sys.exit(1) try: import matplotlib.pyplot as plt except ImportError: print("❌ Falta matplotlib. Instala con: pip install matplotlib") sys.exit(1) # ─── Datos del grafo ───────────────────────────────────────────────────────── # Nodos: (id, nombre, tipo, descripcion) ENTIDADES: List[Tuple[int, str, str, str]] = [ (1, "Traefik", "software", "Reverse proxy para servicios Docker"), (2, "Docker", "software", "Plataforma de contenedores"), (3, "Podman", "software", "Alternativa a Docker sin daemon"), (4, "Ollama", "software", "Ejecución local de modelos LLM"), (5, "Open WebUI", "software", "Interfaz web para LLMs locales"), (6, "LightRAG", "software", "GraphRAG ligero con actualización incremental"), (7, "sqlite-vec", "libreria", "Extensión vectorial para SQLite"), (8, "Whisper", "software", "Speech-to-text local"), (9, "Piper TTS", "software", "Text-to-speech local"), (10, "LLaVA", "modelo", "Modelo de visión local"), (11, "MCP", "protocolo", "Model Context Protocol"), (12, "OpenCode", "software", "CLI/TUI para desarrollo asistido con IA"), (13, "Hermes Agent", "software", "Agente autónomo con skills y MCP"), (14, "Anacleto", "software", "Motor de orquestación de agentes en Rust"), (15, "CrewAI", "libreria", "Framework Python para equipos de agentes"), (16, "systemd", "software", "Init system y gestor de servicios"), (17, "Quadlets", "software", "Contenedores gestionados por systemd"), (18, "NetworkX", "libreria", "Biblioteca Python para análisis de grafos"), ] # Aristas: (origen_id, destino_id, relacion) RELACIONES: List[Tuple[int, int, str]] = [ # Traefik (1, 2, "gestiona tráfico de"), (1, 3, "también funciona con"), (1, 17, "se despliega con"), # Docker (2, 3, "es alternativa de"), (2, 17, "gestionado por"), # Ollama (4, 5, "tiene interfaz en"), (4, 11, "expone API compatible"), # Open WebUI (5, 11, "soporta"), (5, 6, "integra RAG con"), # LightRAG (6, 7, "puede usar"), (6, 18, "construye grafos con"), # Agentes (12, 11, "implementa"), (13, 11, "implementa"), (14, 11, "implementa"), (12, 4, "usa modelos de"), (13, 4, "usa modelos de"), (14, 4, "usa modelos de"), (15, 12, "comparable con"), # Multimedia (8, 4, "se integra con"), (9, 4, "se integra con"), (10, 4, "se ejecuta vía"), # Infraestructura (16, 17, "gestiona"), (17, 2, "gestiona"), (17, 3, "gestiona"), ] # Colores por tipo de entidad (para el gráfico) COLORES_TIPO = { "software": "#4A90D9", # azul "libreria": "#50C878", # verde "modelo": "#E8A838", # naranja "protocolo": "#9B59B6", # púrpura } # ─── Construcción del grafo ────────────────────────────────────────────────── def construir_grafo() -> nx.DiGraph: """Construye el grafo dirigido con entidades y relaciones.""" G = nx.DiGraph() # Añadir nodos con atributos for eid, nombre, tipo, desc in ENTIDADES: G.add_node(nombre, tipo=tipo, descripcion=desc, id=eid) # Añadir aristas con atributos for origen_id, destino_id, relacion in RELACIONES: # Buscar nombres por ID origen_nombre = next(n for eid, n, _, _ in ENTIDADES if eid == origen_id) destino_nombre = next(n for eid, n, _, _ in ENTIDADES if eid == destino_id) G.add_edge(origen_nombre, destino_nombre, relacion=relacion) return G # ─── Consultas sobre el grafo ──────────────────────────────────────────────── def consultar_relaciones(G: nx.DiGraph, entidad: str) -> None: """ Consulta todas las relaciones de una entidad: qué sale de ella y qué llega. Esta es la consulta equivalente a la del SQL en demo-graphrag.sh. """ print(f"\n{'=' * 60}") print(f" CONSULTA: ¿Qué está relacionado con «{entidad}»?") print(f"{'=' * 60}\n") if entidad not in G: print(f" ⚠ La entidad «{entidad}» no existe en el grafo.\n") return # Relaciones salientes (entidad → destino) salientes = list(G.out_edges(entidad, data=True)) if salientes: print(f" ▶ {entidad} se relaciona CON:") for _, destino, data in salientes: print(f" • {data['relacion']} → {destino}") else: print(f" ▶ {entidad} no tiene relaciones salientes.") # Relaciones entrantes (origen → entidad) entrantes = list(G.in_edges(entidad, data=True)) if entrantes: print(f"\n ▶ {entidad} es relacionado POR:") for origen, _, data in entrantes: print(f" • {origen} → {data['relacion']}") else: print(f"\n ▶ {entidad} no tiene relaciones entrantes.") print() def consultar_camino(G: nx.DiGraph, origen: str, destino: str) -> None: """ Encuentra el camino más corto entre dos entidades en el grafo. Demuestra que el grafo permite navegación semántica. """ print(f"\n{'=' * 60}") print(f" CONSULTA: Camino de «{origen}» a «{destino}»") print(f"{'=' * 60}\n") if origen not in G: print(f" ⚠ La entidad «{origen}» no existe.\n") return if destino not in G: print(f" ⚠ La entidad «{destino}» no existe.\n") return try: camino = nx.shortest_path(G, origen, destino) print(f" Ruta encontrada ({len(camino) - 1} saltos):") for i in range(len(camino) - 1): a, b = camino[i], camino[i + 1] relacion = G[a][b]["relacion"] print(f" {a} ──[{relacion}]──→ {b}") print() except nx.NetworkXNoPath: print(f" ⚠ No hay camino entre «{origen}» y «{destino}».\n") def consultar_vecinos_comunes(G: nx.DiGraph, entidad_a: str, entidad_b: str) -> None: """ Encuentra vecinos comunes entre dos entidades. Útil para descubrir conexiones indirectas. """ print(f"\n{'=' * 60}") print(f" CONSULTA: Vecinos comunes de «{entidad_a}» y «{entidad_b}»") print(f"{'=' * 60}\n") if entidad_a not in G or entidad_b not in G: print(" ⚠ Una de las entidades no existe.\n") return # Vecinos: predecesores + sucesores vecinos_a = set(G.predecessors(entidad_a)) | set(G.successors(entidad_a)) vecinos_b = set(G.predecessors(entidad_b)) | set(G.successors(entidad_b)) comunes = vecinos_a & vecinos_b if comunes: print(f" Entidades conectadas tanto con «{entidad_a}» como con «{entidad_b}»:") for v in sorted(comunes): print(f" • {v}") else: print(f" No hay vecinos comunes entre «{entidad_a}» y «{entidad_b}».") print() def mostrar_estadisticas(G: nx.DiGraph) -> None: """Muestra estadísticas básicas del grafo.""" print(f"\n{'=' * 60}") print(f" ESTADÍSTICAS DEL GRAFO DE CONOCIMIENTO") print(f"{'=' * 60}\n") print(f" • Nodos (entidades): {G.number_of_nodes()}") print(f" • Aristas (relaciones): {G.number_of_edges()}") print(f" • Densidad: {nx.density(G):.4f}") print(f" • ¿Es conexo?: {'Sí' if nx.is_weakly_connected(G) else 'No'}") print( f" • Diámetro: {nx.diameter(G.to_undirected()) if nx.is_weakly_connected(G) else 'N/A'}" ) print() # ─── Visualización ─────────────────────────────────────────────────────────── def generar_png(G: nx.DiGraph, archivo: str = "grafo_conocimiento.png") -> str: """ Genera una imagen PNG del grafo usando matplotlib. Los nodos se colorean por tipo y las aristas se etiquetan con la relación. """ print(f" Generando visualización: {archivo} ...", end=" ") plt.figure(figsize=(16, 12)) # Layout: usar Graphviz si está disponible, sino spring_layout try: pos = graphviz_layout(G, prog="dot") except Exception: pos = nx.spring_layout(G, k=1.5, seed=42, iterations=50) # Dibujar nodos por tipo (cada tipo con su color) for tipo in COLORES_TIPO: nodos_tipo = [n for n, attr in G.nodes(data=True) if attr["tipo"] == tipo] nx.draw_networkx_nodes( G, pos, nodelist=nodos_tipo, node_color=COLORES_TIPO[tipo], node_size=2500, node_shape="o", edgecolors="white", linewidths=1.5, alpha=0.95, ) # Dibujar aristas con flechas nx.draw_networkx_edges( G, pos, edge_color="#888888", arrows=True, arrowsize=20, arrowstyle="->", width=1.5, alpha=0.7, connectionstyle="arc3,rad=0.1", ) # Etiquetas de los nodos nx.draw_networkx_labels( G, pos, font_size=11, font_weight="bold", font_family="sans-serif", ) # Etiquetas de las aristas (relaciones) edge_labels = {(a, b): data["relacion"] for a, b, data in G.edges(data=True)} nx.draw_networkx_edge_labels( G, pos, edge_labels=edge_labels, font_size=8, font_family="sans-serif", alpha=0.8, label_pos=0.5, ) # Leyenda legend_elements = [] for tipo, color in COLORES_TIPO.items(): legend_elements.append( plt.scatter( [], [], c=color, s=150, label=tipo, edgecolors="white", linewidths=1 ) ) plt.legend( handles=legend_elements, title="Tipo de entidad", loc="upper right", fontsize=10, title_fontsize=12, ) plt.title( "Grafo de conocimiento — Stack de herramientas del podcast", fontsize=16, fontweight="bold", pad=20, ) plt.axis("off") plt.tight_layout() plt.savefig(archivo, dpi=150, bbox_inches="tight", facecolor="#FAFAFA") plt.close() print("✅") return archivo # ─── Main ──────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Demo de grafo de conocimiento con NetworkX", ) parser.add_argument( "--no-png", action="store_true", help="No generar PNG, solo mostrar consultas", ) parser.add_argument( "--png-only", action="store_true", help="Solo generar PNG sin consultas", ) args = parser.parse_args() print() print("╔══════════════════════════════════════════════════════════╗") print("║ GRAFO DE CONOCIMIENTO — Demo para el episodio 828 ║") print("║ Construido con NetworkX + matplotlib ║") print("╚══════════════════════════════════════════════════════════╝") # Construir el grafo print("\n Construyendo grafo...") G = construir_grafo() print( f" ✓ Grafo creado: {G.number_of_nodes()} nodos, {G.number_of_edges()} aristas" ) if not args.png_only: # Mostrar estadísticas mostrar_estadisticas(G) # Consulta principal: ¿Qué está relacionado con Traefik? consultar_relaciones(G, "Traefik") # Consulta: camino de OpenCode a Quadlets consultar_camino(G, "OpenCode", "Quadlets") # Consulta: vecinos comunes de Ollama y Open WebUI consultar_vecinos_comunes(G, "Ollama", "Open WebUI") if not args.no_png: archivo = generar_png(G) print(f"\n 📁 Imagen guardada: {archivo}") print(f" Ábrela para ver el grafo visualmente.") print("\n ✅ Demo completada.") print() if __name__ == "__main__": main()