Cómo Tree-KG permite gráficos de conocimiento jerárquicos para la navegación contextual y el razonamiento de múltiples saltos explicable más allá del RAG tradicional

!pip instalar networkx matplotlib transformadores de oraciones antrópicos scikit-learn numpy importar networkx como nx importar matplotlib.pyplot como plt desde escribir importar Lista, Dict, Tuple, Opcional, configurar importar numpy como np desde sklearn.metrics.pairwise importar cosine_similarity desde Sentence_transformers importar SentenceTransformer desde colecciones importar defaultdict, deque importar json

class TreeKnowledgeGraph: """ Gráfico de conocimiento jerárquico que imita los patrones de aprendizaje humano. Admite razonamiento de saltos múltiples y navegación contextual. """ 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: Opcional[Dict] = Ninguno): """Agregar un nodo con incrustación semántica y metadatos.""" embedding = self.embedder.encode(content, convert_to_tensor=False) self.graph.add_node(node_id, content=content, node_type=node_type, metadata=metadata o {}) self.node_embeddings[node_id] = incrustación self.node_metadata[node_id] = { 'content': contenido, 'type': node_type, 'metadata': metadata o {} } def add_edge(self, parent: str, child: str, relación: str="contains", peso: float = 1.0): """Agregar borde jerárquico o asociativo entre nodos.""" self.graph.add_edge(padre, hijo, relación=relación, peso=peso) def get_ancestors(self, node_id: str, max_profundidad: int = 5) -> Lista[str]: """Obtener todos los nodos ancestros (contexto jerárquico).""" ancestros =[]actual = node_id profundidad = 0 mientras que profundidad < max_profundidad: predecesores = lista (self.graph.predecessors (actual)) si no predecesores: interrupción actual = predecesores[0]ancestros.append(actual) profundidad += 1 return ancestros def get_dedescendientes(self, node_id: str, max_profundidad: int = 2) -> Lista[str]: """Obtener todos los nodos descendientes.""" descendientes =[]cola = deque([(node_id, 0)]) visitado = {node_id} mientras que cola: actual, profundidad = queue.popleft() si profundidad >= max_profundidad: continuar para el niño en self.graph.successors(actual): si el niño no está visitado: visitado.add(child) dedescendientes.append(child) queue.append((child, profundidad + 1)) return descendientes def semantic_search(self, query: str, top_k: int = 5) -> Lista[Tuple[str, float]]: """Encontrar los nodos más semánticamente similares para consultar.""" query_embedding = self.embedder.encode(query, convert_to_tensor=False) similitudes =[]para node_id, incrustar en self.node_embeddings.items(): sim = cosine_similarity( query_embedding.reshape(1, -1), embedding.reshape(1, -1) )[0][0]similitudes.append((node_id, float(sim))) similitudes.sort(key=lambda x: x[1], inversa=Verdadero) return similitudes[:top_k] def get_subgraph_context(self, node_id: str, profundidad: int = 2) -> Dict: """Obtener información contextual rica alrededor de un nodo.""" context = { 'node': self.node_metadata.get(node_id, {}), 'ancestors':[], 'descendientes':[], 'hermanos':[], 'relacionado':[]} ancestros = self.get_ancestors(node_id) contexto['ancestros'] = [ self.node_metadata.get(a, {}) para a en ancestros ] descendientes = self.get_dedescendientes(nodo_id, profundidad) contexto['descendientes'] = [ self.node_metadata.get(d, {}) para d en descendientes ] padres = list(self.graph.predecessors(node_id)) si padres: hermanos = list(self.graph.successors(padres[0])) hermanos = [s para s en hermanos si s! = node_id] contexto['hermanos'] = [ self.node_metadata.get(s, {}) para s en hermanos ] devolver contexto

class MultiHopReasoningAgent: """ Agente que realiza razonamiento inteligente de múltiples saltos a través del gráfico de conocimiento. """ def __init__(self, kg: TreeKnowledgeGraph): self.kg = kg self.reasoning_history =[]def Reason(self, query: str, max_hops: int = 3, explore_width: int = 3) -> Dict: """ Realizar razonamiento de múltiples saltos para responder una consulta. Estrategia: 1. Encontrar nodos relevantes iniciales (búsqueda semántica) 2. Explorar el contexto del gráfico alrededor de estos nodos 3. Realizar exploración en amplitud con puntuación de relevancia 4. Agregar información de múltiples saltos """ Reasoning_trace = { 'consulta': consulta, 'saltos':[], 'contexto_final': {}, 'ruta_razonamiento':[]} nodos_iniciales = self.kg.semantic_search(query, top_k=ancho_exploración) razonamiento_trace['saltos'].append({ 'número_salto': 0, 'acción': 'búsqueda_semántica', 'nodos_encontrados': nodos_iniciales }) visitado = set() frontera_actual = [id_nodo para id_nodo, _ en nodos_iniciales] todos_nodos_relevantes = set(current_frontier) para salto en rango(1, max_hops + 1): next_frontier =[]hop_info = { 'número_salto': salto, 'nodos_explorados':[], 'nuevos_descubrimientos':[]} para node_id en current_frontier: si node_id está visitado: continuar visitado.add(node_id) contexto = self.kg.get_subgraph_context(node_id, profundidad=1) conectado_nodos =[]para ancestro en contexto['ancestros']: si 'contenido' en ancestro: conectado_nodes.append(ancestro) para descendiente en contexto['descendientes']: si 'contenido' en descendiente: conectado_nodes.append(descendiente) para hermano en contexto['hermanos']: si 'contenido' en hermano: conectado_nodes.append(hermano) conexiones_relevantes = self._score_relevance( consulta, nodos_conectados, 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) }) para conn_content, puntuación en conexiones_relevantes: para nid, meta en self.kg.node_metadata.items(): si meta['content'] == conn_content y nid no están visitados: next_frontier.append(nid) all_relevant_nodes.add(nid) hop_info['new_discoveries'].append({ 'node_id': nid, 'relevance_score': puntuación }) break Reasoning_trace['hops'].append(hop_info) current_frontier = next_frontier si no current_frontier: romper 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 razonamiento_trace def _score_relevance(self, consulta: str, candidatos: Lista[Dict], top_k: int = 3) -> Lista[Tuple[str, float]]: """Calificar los nodos candidatos por relevancia para la consulta.""" si no son candidatos: devolver[]query_embedding = self.kg.embedder.encode(consulta) puntuaciones =[]para candidato en candidatos: contenido = candidato.get('contenido', '') si no contenido: continuar candidato_embedding = self.kg.embedder.encode(contenido) similitud = coseno_similaridad( query_embedding.reshape(1, -1), candidato_embedding.reshape(1, -1) )[0][0]puntuaciones.append((contenido, float(similitud))) puntuaciones.sort(clave=lambda x: x[1], reverse=True) return puntuaciones[:top_k] def _aggregate_context(self, query: str, node_ids: Set[str]) -> Dict: """Agregar y clasificar información de todos los nodos descubiertos.""" agregado = { 'total_nodes': len(node_ids), 'hierarchical_paths':[], 'conceptos_clave':[], 'respuesta_sintetizada':[]} para node_id en node_ids: ancestros = self.kg.get_ancestors(node_id) si ancestros: ruta = ancestros[::-1] + [node_id] path_contents = [ self.kg.node_metadata[n]['content'] for n en la ruta si n en self.kg.node_metadata ] agregado['hierarchical_paths'].append(path_contents) para node_id en node_ids: meta = self.kg.node_metadata.get(node_id, {}) agregado['key_concepts'].append({ 'id': node_id, 'content': meta.get('content', ''), 'type': meta.get('type', 'unknown') }) para node_id en node_ids: content = self.kg.node_metadata.get(node_id, {}).get('content', '') if content: agregado['synthesized_answer'].append(content) return agregado def explicó_reasoning(self, trace: Dict) -> str: """Generar una explicación legible por humanos del proceso de razonamiento.""" explicación = [f"Query: {trace['query']}n"] explicación.append(f"Total de saltos realizados: {len(trace['hops']) – 1}n") explicación.append(f"Total de nodos relevantes descubiertos: {len(trace['reasoning_path'])}nn") para hop_info en trace['hops']: hop_num = hop_info['hop_number'] explicación.append(f"— Hop {hop_num} —") if hop_num == 0: explicación.append(f"Acción: Búsqueda semántica inicial") explicación.append(f"Encontrado {len(hop_info['nodes_found'])} nodos candidatos") para node_id, puntuación en hop_info['nodes_found'][:3]: explicación.append(f" – {node_id} (relevancia: {score:.3f})") else: explicación.append(f"Explorado {len(hop_info['explored_nodes'])} nodos") explicación.append(f"Descubrió {len(hop_info['new_discoveries'])} nuevos nodos relevantes") explicación.append("") explicación.append("n— Contexto agregado final —") contexto = trace['final_context'] explicación.append(f"Conceptos totales integrados: {context['total_nodes']}") explicación.append(f"Rutas jerárquicas encontradas: {len(context['hierarchical_paths'])}") return "n".join(explicación)

def build_software_development_kb() -> TreeKnowledgeGraph: """Construya un gráfico integral de conocimientos sobre desarrollo de software.""" kg = TreeKnowledgeGraph() kg.add_node('root', 'Desarrollo de software e informática', 'dominio') kg.add_node('programación', 'La programación abarca escribir, probar y mantener código para crear aplicaciones de software', 'dominio') kg.add_node('arquitectura', 'La arquitectura de software implica el diseño de la estructura de alto nivel y los componentes de los sistemas de software', 'dominio') kg.add_node('dominio') kg.add_edge('raíz', 'programación', 'contiene') kg.add_edge('raíz', 'arquitectura', 'contiene') kg.add_edge('raíz', 'devops', 'contiene') kg.add_node('python', 'idioma') kg.add_node('javascript', 'JavaScript es un lenguaje dinámico utilizado principalmente para el desarrollo web, que permite aplicaciones interactivas del lado del cliente y del servidor', 'idioma') kg.add_node('rust', 'language') kg.add_edge('programación', 'python', 'incluye') kg.add_edge('programación', 'javascript', 'incluye') kg.add_edge('programación', 'rust', 'incluye') kg.add_node('python_basics', 'Los conceptos básicos de Python incluyen variables, tipos de datos, flujo de control, funciones y fundamentos de programación orientada a objetos', 'concepto') kg.add_node('python_rendimiento', 'Python La optimización del rendimiento implica técnicas como creación de perfiles, almacenamiento en caché, uso de extensiones C y aprovechamiento de la programación asíncrona', 'concept') kg.add_node('python_data', 'Python for Data Science utiliza bibliotecas como NumPy, Pandas y Scikit-learn para manipulación de datos, análisis y aprendizaje automático', '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', 'La E/S asíncrona en Python permite operaciones sin bloqueo usando la sintaxis async/await con la biblioteca asyncio para tareas concurrentes', 'technique') kg.add_node('multiprocessing', 'Python Multiprocessing utiliza procesos separados para evitar GIL, lo que permite una verdadera ejecución paralela para tareas vinculadas a la CPU', 'técnica') kg.add_node('cython', 'Cython compila Python en C para obtener ganancias de rendimiento significativas, especialmente en cálculos numéricos y bucles cerrados', 'herramienta') kg.add_node('profiling', 'Python Profiling identifica cuellos de botella en el rendimiento usando herramientas como cProfile, line_profiler y Memory_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', 'profile', 'contains') kg.add_node('event_loop', 'Event Loop es el núcleo de asyncio que administra y programa tareas asincrónicas, maneja devoluciones de llamadas y corrutinas', 'concept') kg.add_node('coroutines', 'Las corrutinas son funciones especiales definidas con async def que pueden pausar la ejecución con await, lo que permite la cooperación multitarea', 'concept') kg.add_node('asyncio_patterns', 'Los patrones AsyncIO incluyen recopilación para ejecución simultánea, create_task para tareas en segundo plano y colas para productor-consumidor', 'pattern') kg.add_edge('async_io', 'event_loop', 'contains') kg.add_edge('async_io', 'coroutines', 'contiene') kg.add_edge('async_io', 'asyncio_patterns', 'contiene') kg.add_node('microservicios', 'La arquitectura de microservicios descompone las aplicaciones en servicios pequeños e independientes que se comunican a través de API', 'patrón') kg.add_edge('arquitectura', 'microservicios', 'contiene') kg.add_edge('async_io', 'microservices', ' related_to') kg.add_node('containers', 'Los contenedores empaquetan aplicaciones con dependencias en unidades aisladas, asegurando la coherencia entre entornos', 'tecnología') kg.add_edge('devops', 'containers', 'contains') kg.add_edge('microservices', 'containers', 'deployed_with') kg.add_node('numpy_optimization', 'La optimización de NumPy utiliza vectorización y transmisión para evitar bucles de Python, aprovechando las bibliotecas optimizadas de C y Fortran', 'technique') kg.add_edge('python_data', 'numpy_optimization', 'contains') kg.add_edge('python_performance', 'numpy_optimization', 'relacionado_con') devolver kg

def visualize_knowledge_graph(kg: TreeKnowledgeGraph, resaltado_nodes: Opcional[List[str]] = Ninguno): """Visualizar la estructura del gráfico de conocimiento.""" plt.figure(figsize=(16, 12)) pos = nx.spring_layout(kg.graph, k=2, iterations=50, seed=42) node_colors =[]para el nodo en kg.graph.nodes(): si resalta_nodos y nodo en resaltado_nodes: node_colors.append('amarillo') else: node_type = kg.graph.nodes[node].get('node_type', 'concept') color_map = { 'dominio': 'azul claro', 'idioma': 'verde claro', 'concepto': 'coral claro', 'técnica': 'amarillo claro', 'herramienta': 'rosa claro', 'patrón': 'lavanda', 'tecnología': '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", flechas=True, tamaño de flecha=20, alfa=0.6, ancho=2) nx.draw_networkx_labels(kg.graph, pos, font_size=8, font_weight="bold") plt.title("Tree-KG: gráfico de conocimiento jerárquico", fontsize=16, fontweight="bold") plt.axis('off') plt.tight_layout() plt.show() def run_demo(): """Ejecutar una demostración completa del sistema Tree-KG.""" print("=" * 80) print("Tree-KG: Demostración de gráfico de conocimiento jerárquico") print("=" * 80) print() print("Construyendo un gráfico de conocimiento…") kg = build_software_development_kb() print(f" ✓ Gráfico creado con {kg.graph.number_of_nodes()} nodos y {kg.graph.number_of_edges()} bordesn") print("Visualizando gráfico de conocimiento…") visualize_knowledge_graph(kg) agent = MultiHopReasoningAgent(kg) queries = [ "¿Cómo puedo mejorar el rendimiento de Python para tareas vinculadas a IO?", "¿Cuáles son las mejores prácticas para la programación asíncrona?", "¿Cómo se relaciona la arquitectura de microservicios con Python?" ] para i, consulta en enumerate(queries, 1): print(f"n{'=' * 80}") print(f"QUERY {i}: {query}") print('=' * 80) trace = agent.reason(query, max_hops=3, explore_width=3) explicación = agent.explain_reasoning(trace) print(explanation) print("n— Ejemplo jerárquico Rutas —") para j, ruta en enumerate(trace['final_context']['hierarchical_paths'][:3], 1): print(f"nPath {j}:") para k, concepto en enumerate(path): sangría = " " * k print(f"{sangría}→ {concepto[:80]}…") print("n— Contexto sintetizado —") respuestas_partes = trace['final_context']['synthesized_answer'][:5] para parte en respuestas_partes: print(f"• {part[:150]}…") print() print("nVisualizando la ruta de razonamiento para la última consulta…") last_trace = agent.reasoning_history[-1] visualize_knowledge_graph(kg, resaltado_nodes=last_trace['reasoning_path']) print("n" + "=" * 80) print("¡Demo completada!") print("=" * 80)

class AdvancedTreeKG(TreeKnowledgeGraph): """Extended Tree-KG con funciones avanzadas.""" def __init__(self, embedding_model: str="all-MiniLM-L6-v2"): super().__init__(embedding_model) self.node_importance = {} def Compute_node_importance(self): """Calcular puntuaciones de importancia utilizando un algoritmo similar a PageRank.""" if self.graph.number_of_nodes() == 0: devolver pagerank = nx.pagerank(self.graph) intermediación = nx.betweenness_centrality(self.graph) para el nodo en self.graph.nodes(): self.node_importance[nodo] = { 'pagerank': pagerank.get(nodo, 0), 'entremedia': intermedia.get(nodo, 0), 'combinado': pagerank.get(nodo, 0) * 0.7 + Betweenness.get(nodo, 0) * 0.3 } def find_shortest_path_with_context(self, source: str, target: str) -> Dict: """Encuentra la ruta más corta y extrae todo el contexto a lo largo del camino.""" try: path = nx.shortest_path(self.graph, source, target) context = { 'path': path, 'path_length': len(ruta) – 1, 'nodos_detalle':[]} para el nodo en la ruta: detalle = { 'id': nodo, 'contenido': self.node_metadata.get(nodo, {}).get('content', ''), 'importancia': self.node_importance.get(nodo, {}).get('combined', 0) } contexto['nodes_detail'].append(detalle) contexto de retorno excepto nx.NetworkXNoPath: retorno {'ruta':[], 'error': 'No existe ninguna ruta'}

if __name__ == "__main__": run_demo() print("nn" + "=" * 80) print("DEMOSTRACIÓN DE CARACTERÍSTICAS AVANZADAS") print("=" * 80) print("nConstrucción de Tree-KG avanzado…") 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("Puntuaciones de importancia de los nodos informáticos…") adv_kg_new.compute_node_importance() print("nTop 5 nodos más importantes:") sorted_nodes = sorted( adv_kg_new.node_importance.items(), clave=lambda x: x[1]['combinado'], reverso=True )[:5] para nodo, puntuaciones en sorted_nodes: content = adv_kg_new.node_metadata[nodo]['content'][:60] print(f" {nodo}: {content}…") print(f" Puntuación combinada: {scores['combined']:.4f}") print("n ✓ ¡Tutorial de Tree-KG completo!") print("nKey Conclusiones:") print("1. Tree-KG permite la navegación contextual frente a la recuperación de fragmentos simples") print("2. El razonamiento de múltiples saltos descubre información relevante en la estructura del gráfico") print("3. La organización jerárquica refleja los patrones de aprendizaje humano") print("4. Búsqueda semántica + recorrido de gráficos = potente alternativa RAG")