import os import json import sqlite3 import tempfile import numpy as np import pandas as pd import networkx as nx import plotly.graph_objects as go from typing import List, Dict, Any, Tuple, Optional from sentence_transformers import SentenceTransformer from datetime import datetime import gradio as gr from groq import Groq import logging from huggingface_hub import HfApi # Configuración de logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Configuración de variables de entorno GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") DB_PATH = os.environ.get("DB_PATH", "collective_intelligence.db") SYNC_TO_HF = os.environ.get("SYNC_TO_HF", "False").lower() == "true" HF_REPO_ID = os.environ.get("HF_REPO_ID", "") # Constantes DISC_TYPES = ["D", "I", "S", "C"] COLORS = { "D": "red", "I": "yellow", "S": "green", "C": "blue" } # Preguntas DISC basadas en el modelo de "Surrounded by Idiots" DISC_QUESTIONS = [ { "id": 1, "question": "Cuando enfrento un desafío, prefiero:", "options": { "D": "Tomar el control y decidir rápidamente", "I": "Involucrar a todos y generar entusiasmo", "S": "Escuchar opiniones y buscar consenso", "C": "Analizar datos y evaluar todas las opciones" } }, { "id": 2, "question": "En una conversación, suelo:", "options": { "D": "Ir al punto y ser directo", "I": "Hablar con energía y expresivamente", "S": "Escuchar atentamente y responder con calma", "C": "Preguntar detalles y ser preciso" } }, { "id": 3, "question": "Cuando trabajo en equipo, me siento más cómodo:", "options": { "D": "Liderando y dando instrucciones claras", "I": "Motivando al grupo y generando ideas", "S": "Apoyando a otros y manteniendo la armonía", "C": "Verificando la calidad y corrigiendo errores" } }, { "id": 4, "question": "Frente a un problema, mi primera reacción es:", "options": { "D": "Buscar soluciones inmediatas y actuar", "I": "Compartir el problema y buscar ideas en grupo", "S": "Mantener la calma y evaluar la situación", "C": "Investigar las causas y analizar datos" } }, { "id": 5, "question": "En situaciones de presión:", "options": { "D": "Me vuelvo más decidido y directo", "I": "Trato de aliviar la tensión con optimismo", "S": "Mantengo la estabilidad y sigo procedimientos", "C": "Me enfoco en los detalles y la precisión" } }, { "id": 6, "question": "Cuando alguien presenta una idea nueva:", "options": { "D": "Evalúo rápidamente si es práctica y factible", "I": "Me entusiasmo y pienso en posibilidades", "S": "Considero cómo afectará al equipo y la estabilidad", "C": "Analizo críticamente sus fortalezas y debilidades" } }, { "id": 7, "question": "Mi ritmo de trabajo preferido es:", "options": { "D": "Rápido, orientado a resultados", "I": "Dinámico, con variedad de tareas", "S": "Estable, metódico y consistente", "C": "Estructurado, con tiempo para verificar" } }, { "id": 8, "question": "Al comunicar información importante, prefiero:", "options": { "D": "Ser breve y directo, enfocándome en lo esencial", "I": "Hacer la presentación interactiva y animada", "S": "Asegurarme que todos se sienten incluidos", "C": "Proporcionar datos completos y precisos" } }, { "id": 9, "question": "Mi mayor fortaleza en el trabajo es:", "options": { "D": "Tomar decisiones difíciles y asumir desafíos", "I": "Inspirar a otros y generar entusiasmo", "S": "Ser paciente, confiable y buen oyente", "C": "Ser minucioso, preciso y organizado" } }, { "id": 10, "question": "Cuando recibo críticas:", "options": { "D": "Quiero saber el punto principal sin rodeos", "I": "Me preocupa cómo afecta mi imagen ante otros", "S": "Puedo sentirme herido pero no lo demuestro", "C": "Analizo detalladamente la validez de la crítica" } }, { "id": 11, "question": "En mi tiempo libre, prefiero:", "options": { "D": "Actividades competitivas o desafiantes", "I": "Eventos sociales y conocer gente nueva", "S": "Tiempo tranquilo con amigos cercanos o familia", "C": "Actividades estructuradas o aprender algo nuevo" } }, { "id": 12, "question": "Mi mayor temor podría ser:", "options": { "D": "Perder el control o ser aprovechado", "I": "Ser ignorado o rechazado socialmente", "S": "Enfrentar conflictos o cambios repentinos", "C": "Ser criticado por errores o imprecisiones" } } ] class DISCProfiler: """Clase para el perfilamiento DISC basado en embeddings semánticos""" def __init__(self): """Inicializa el modelo de embeddings y prepara el sistema de perfilamiento""" self.model = SentenceTransformer('all-MiniLM-L6-v2') self.db_manager = DatabaseManager() def calculate_profile(self, responses: List[str]) -> Dict[str, float]: """Calcula el perfil DISC basado en las respuestas seleccionadas""" if len(responses) != len(DISC_QUESTIONS): raise ValueError(f"Se esperaban {len(DISC_QUESTIONS)} respuestas, pero se recibieron {len(responses)}") # Conteo simple de tipos seleccionados profile_counts = {"D": 0, "I": 0, "S": 0, "C": 0} for response in responses: profile_counts[response] += 1 # Normalizar a vector unitario total = sum(profile_counts.values()) profile_vector = {k: v/total for k, v in profile_counts.items()} return profile_vector def get_semantic_profile(self, responses: List[str]) -> Dict[str, float]: """Obtiene un perfil DISC basado en embeddings semánticos""" # Convertir respuestas de tipo (D, I, S, C) a texto completo response_texts = [] for i, resp_type in enumerate(responses): question_idx = i % len(DISC_QUESTIONS) # En caso de que haya más respuestas que preguntas response_texts.append(DISC_QUESTIONS[question_idx]["options"][resp_type]) # Obtener embedding para todas las respuestas combinadas combined_text = " ".join(response_texts) embedding = self.model.encode(combined_text) # Obtener embeddings para las descripciones prototípicas de cada tipo DISC disc_descriptions = { "D": "Dominante, directo, orientado a resultados, decisivo, competitivo, asertivo", "I": "Influyente, expresivo, entusiasta, optimista, sociable, comunicativo", "S": "Estable, paciente, confiable, predecible, cooperativo, calmado", "C": "Concienzudo, analítico, preciso, reservado, sistemático, metódico" } disc_embeddings = {d_type: self.model.encode(desc) for d_type, desc in disc_descriptions.items()} # Calcular similitud coseno con cada tipo similarities = {} for d_type, type_emb in disc_embeddings.items(): sim = np.dot(embedding, type_emb) / (np.linalg.norm(embedding) * np.linalg.norm(type_emb)) similarities[d_type] = float(sim) # Normalizar a suma 1 total = sum(similarities.values()) normalized = {k: v/total for k, v in similarities.items()} return normalized def save_profile(self, user_id: str, profile_vector: Dict[str, float], responses: List[str]): """Guarda el perfil del usuario en la base de datos""" return self.db_manager.save_user_profile(user_id, profile_vector, responses) class CollectiveIntelligenceEngine: """Motor de síntesis colectiva basado en LLM""" def __init__(self): """Inicializa el cliente de Groq y el gestor de base de datos""" self.client = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None self.db_manager = DatabaseManager() self.kg = KnowledgeGraph() def analyze_context(self, text: str) -> List[str]: """Analiza el contexto y extrae conceptos clave""" if not self.client: logger.warning("No se ha configurado GROQ_API_KEY. Utilizando análisis simulado.") # Simulación simple de extracción de conceptos para demostración words = text.lower().split() concepts = [w for w in words if len(w) > 4][:5] return concepts[:5] # Limitar a 5 conceptos try: prompt = f""" Analiza el siguiente texto e identifica los 5 conceptos clave más importantes. Retorna solo los conceptos como una lista de términos individuales, sin numeración ni explicación: Texto: {text} """ response = self.client.chat.completions.create( model="llama3-70b-8192", messages=[{"role": "user", "content": prompt}] ) concepts_text = response.choices[0].message.content # Limpiar y procesar la respuesta concepts = [c.strip() for c in concepts_text.split('\n') if c.strip()] # Eliminar posibles numeraciones, puntos, etc. concepts = [c.split('. ')[-1].split(' - ')[-1].strip() for c in concepts] concepts = [c for c in concepts if c] return concepts[:5] # Asegurar que sean 5 o menos except Exception as e: logger.error(f"Error al analizar contexto con Groq: {e}") return ["error de análisis"] def synthesize_perspectives(self, text: str, user_profile: Dict[str, float]) -> Dict[str, str]: """Sintetiza el texto desde múltiples perspectivas DISC""" if not self.client: logger.warning("No se ha configurado GROQ_API_KEY. Utilizando síntesis simulada.") # Simulación simple de síntesis para demostración return { "D": f"Perspectiva D (dominante): {text[:50]}...", "I": f"Perspectiva I (influyente): {text[10:60]}...", "S": f"Perspectiva S (estable): {text[20:70]}...", "C": f"Perspectiva C (concienzudo): {text[30:80]}..." } perspectives = {} disc_characteristics = { "D": "dominante, orientada a resultados, directa, decidida y enfocada en la eficiencia", "I": "influyente, entusiasta, optimista, social y enfocada en las relaciones", "S": "estable, paciente, confiable, cooperativa y enfocada en la armonía", "C": "concienzuda, analítica, precisa, detallista y enfocada en la calidad" } # Determinar el perfil dominante para personalizar la síntesis dominant_type = max(user_profile.items(), key=lambda x: x[1])[0] for disc_type in DISC_TYPES: try: prompt = f""" Analiza el siguiente texto desde una perspectiva {disc_characteristics[disc_type]}. Dado que el perfil dominante del usuario es de tipo {dominant_type} ({disc_characteristics[dominant_type]}), adapta tu respuesta para que sea particularmente relevante y útil para esta perspectiva. Texto a analizar: {text} Proporciona un análisis conciso (máximo 200 palabras) que destaque los aspectos que una persona con perspectiva {disc_type} consideraría más importantes. """ response = self.client.chat.completions.create( model="llama3-70b-8192", messages=[{"role": "user", "content": prompt}] ) perspectives[disc_type] = response.choices[0].message.content except Exception as e: logger.error(f"Error al sintetizar perspectiva {disc_type} con Groq: {e}") perspectives[disc_type] = f"Error al procesar perspectiva {disc_type}" return perspectives def generate_universal_principles(self, text: str, perspectives: Dict[str, str]) -> str: """Genera principios universales basados en la síntesis multi-perspectiva""" if not self.client: logger.warning("No se ha configurado GROQ_API_KEY. Utilizando generación simulada.") return "Principios universales simulados basados en las perspectivas DISC." try: # Combinar todas las perspectivas all_perspectives = "\n\n".join([f"Perspectiva {k}: {v}" for k, v in perspectives.items()]) prompt = f""" Basándote en el texto original y las diferentes perspectivas DISC proporcionadas, genera 3-5 principios universales que capturen la esencia del contenido de manera que sea valiosa para personas con cualquier perfil DISC. Texto original: {text} Perspectivas DISC: {all_perspectives} Proporciona estos principios universales en formato de lista numerada, explicando brevemente cada uno (1-2 oraciones por principio). """ response = self.client.chat.completions.create( model="llama3-70b-8192", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content except Exception as e: logger.error(f"Error al generar principios universales con Groq: {e}") return "Error al generar principios universales." def process_contribution(self, user_id: str, text: str) -> Dict[str, Any]: """Procesa una contribución del usuario y actualiza el grafo de conocimiento""" # Obtener perfil del usuario user_profile = self.db_manager.get_user_profile(user_id) if not user_profile: logger.warning(f"Usuario {user_id} no tiene perfil. Usando perfil equilibrado.") user_profile = {d_type: 0.25 for d_type in DISC_TYPES} # Analizar contexto concepts = self.analyze_context(text) # Generar síntesis multi-perspectiva perspectives = self.synthesize_perspectives(text, user_profile) # Generar principios universales principles = self.generate_universal_principles(text, perspectives) # Actualizar grafo de conocimiento contribution_id = self.db_manager.save_contribution( user_id, text, concepts, perspectives, principles ) # Actualizar el grafo de conocimiento self.kg.add_contribution(contribution_id, user_id, concepts, text) return { "contribution_id": contribution_id, "concepts": concepts, "perspectives": perspectives, "principles": principles } def get_knowledge_graph(self) -> nx.Graph: """Obtiene el grafo de conocimiento actual""" return self.kg.get_graph() class KnowledgeGraph: """Gestión del grafo de conocimiento con NetworkX""" def __init__(self): """Inicializa el grafo de conocimiento""" self.G = nx.Graph() self.db_manager = DatabaseManager() self.model = SentenceTransformer('all-MiniLM-L6-v2') self._load_from_db() def _load_from_db(self): """Carga el grafo desde la base de datos""" # Obtener todos los nodos y relaciones de la BD nodes = self.db_manager.get_all_nodes() edges = self.db_manager.get_all_edges() # Agregar nodos al grafo for node in nodes: node_type = node["type"] node_id = node["id"] if node_type == "user": profile = self.db_manager.get_user_profile(node_id) dominant_type = max(profile.items(), key=lambda x: x[1])[0] if profile else "balanced" self.G.add_node( node_id, type=node_type, label=node_id, disc_type=dominant_type, color=COLORS.get(dominant_type, "gray") ) elif node_type == "concept": self.G.add_node( node_id, type=node_type, label=node_id, color="purple" ) elif node_type == "contribution": self.G.add_node( node_id, type=node_type, label=f"Contrib-{node_id}", color="orange" ) # Agregar aristas al grafo for edge in edges: self.G.add_edge( edge["source"], edge["target"], weight=edge.get("weight", 1.0), label=edge.get("label", "") ) def add_contribution(self, contribution_id: str, user_id: str, concepts: List[str], text: str): """Agrega una contribución al grafo de conocimiento""" # Agregar nodo de contribución si no existe if contribution_id not in self.G.nodes: self.G.add_node( contribution_id, type="contribution", label=f"Contrib-{contribution_id}", color="orange" ) # Conectar con el usuario if user_id in self.G.nodes: self.G.add_edge( user_id, contribution_id, weight=1.0, label="authored" ) # Conectar con conceptos for concept in concepts: # Agregar concepto si no existe if concept not in self.G.nodes: self.G.add_node( concept, type="concept", label=concept, color="purple" ) self.db_manager.save_node(concept, "concept") # Conectar contribución con concepto self.G.add_edge( contribution_id, concept, weight=1.0, label="contains" ) self.db_manager.save_edge(contribution_id, concept, "contains", 1.0) # Buscar similitudes con otras contribuciones self._find_similar_contributions(contribution_id, text) def _find_similar_contributions(self, contribution_id: str, text: str): """Encuentra contribuciones similares y las conecta en el grafo""" # Obtener todas las contribuciones excepto la actual contributions = self.db_manager.get_all_contributions() contributions = [c for c in contributions if c["id"] != contribution_id] if not contributions: return # Calcular embedding de la contribución actual embedding = self.model.encode(text) # Calcular similitud con otras contribuciones similarities = [] for contrib in contributions: contrib_text = contrib["text"] contrib_embedding = self.model.encode(contrib_text) # Calcular similitud coseno sim = np.dot(embedding, contrib_embedding) / (np.linalg.norm(embedding) * np.linalg.norm(contrib_embedding)) similarities.append((contrib["id"], float(sim))) # Conectar con las contribuciones más similares (similitud > 0.6) threshold = 0.6 for other_id, sim in similarities: if sim > threshold: self.G.add_edge( contribution_id, other_id, weight=sim, label="similar" ) self.db_manager.save_edge(contribution_id, other_id, "similar", sim) def get_graph(self) -> nx.Graph: """Obtiene el grafo actual""" return self.G def get_visualization_data(self) -> Dict[str, Any]: """Prepara los datos para visualización con Plotly""" G = self.G # Usar el layout de spring para posicionar los nodos pos = nx.spring_layout(G, seed=42) # Preparar datos de nodos node_x = [] node_y = [] node_text = [] node_color = [] node_size = [] for node in G.nodes(): x, y = pos[node] node_x.append(x) node_y.append(y) # Texto del nodo node_attrs = G.nodes[node] node_label = node_attrs.get("label", str(node)) node_type = node_attrs.get("type", "unknown") node_text.append(f"{node_label} ({node_type})") # Color del nodo node_color.append(node_attrs.get("color", "gray")) # Tamaño del nodo basado en centralidad if len(G.edges()) > 0: centrality = nx.degree_centrality(G) size = 10 + centrality[node] * 50 else: size = 10 node_size.append(size) # Preparar datos de aristas edge_x = [] edge_y = [] edge_text = [] for edge in G.edges(): x0, y0 = pos[edge[0]] x1, y1 = pos[edge[1]] edge_x.extend([x0, x1, None]) edge_y.extend([y0, y1, None]) # Texto de la arista edge_attrs = G.edges[edge] edge_label = edge_attrs.get("label", "") edge_weight = edge_attrs.get("weight", 1.0) edge_text.append(f"{edge_label} ({edge_weight:.2f})") return { "node_x": node_x, "node_y": node_y, "node_text": node_text, "node_color": node_color, "node_size": node_size, "edge_x": edge_x, "edge_y": edge_y, "edge_text": edge_text } class DatabaseManager: """Gestión de la persistencia con SQLite""" def __init__(self): """Inicializa la conexión a la base de datos""" self.db_path = DB_PATH self._init_db() # Inicializar sincronización con Hugging Face si está configurado self.hf_sync_enabled = SYNC_TO_HF and HF_TOKEN and HF_REPO_ID if self.hf_sync_enabled: try: self.hf_api = HfApi(token=HF_TOKEN) logger.info(f"Sincronización con Hugging Face habilitada para {HF_REPO_ID}") except Exception as e: logger.error(f"Error al configurar sincronización con Hugging Face: {e}") self.hf_sync_enabled = False def _init_db(self): """Inicializa las tablas en la base de datos si no existen""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Tabla de usuarios y perfiles cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, profile TEXT, responses TEXT, created_at TEXT ) """) # Tabla de contribuciones cursor.execute(""" CREATE TABLE IF NOT EXISTS contributions ( id TEXT PRIMARY KEY, user_id TEXT, text TEXT, concepts TEXT, perspectives TEXT, principles TEXT, created_at TEXT, FOREIGN KEY (user_id) REFERENCES users (id) ) """) # Tabla de nodos del grafo cursor.execute(""" CREATE TABLE IF NOT EXISTS graph_nodes ( id TEXT PRIMARY KEY, type TEXT, created_at TEXT ) """) # Tabla de aristas del grafo cursor.execute(""" CREATE TABLE IF NOT EXISTS graph_edges ( source TEXT, target TEXT, label TEXT, weight REAL, created_at TEXT, PRIMARY KEY (source, target, label) ) """) # Tabla para caché de embeddings cursor.execute(""" CREATE TABLE IF NOT EXISTS embedding_cache ( text TEXT PRIMARY KEY, embedding BLOB, created_at TEXT ) """) conn.commit() conn.close() def save_user_profile(self, user_id: str, profile: Dict[str, float], responses: List[str]) -> bool: """Guarda el perfil de usuario en la base de datos""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Convertir datos a formato JSON para almacenamiento profile_json = json.dumps(profile) responses_json = json.dumps(responses) timestamp = datetime.now().isoformat() # Insertar o actualizar usuario cursor.execute(""" INSERT OR REPLACE INTO users (id, profile, responses, created_at) VALUES (?, ?, ?, ?) """, (user_id, profile_json, responses_json, timestamp)) # Guardar nodo de usuario en el grafo cursor.execute(""" INSERT OR IGNORE INTO graph_nodes (id, type, created_at) VALUES (?, ?, ?) """, (user_id, "user", timestamp)) conn.commit() conn.close() # Sincronizar con Hugging Face si está habilitado if self.hf_sync_enabled: self._sync_to_huggingface() return True except Exception as e: logger.error(f"Error al guardar perfil de usuario: {e}") return False def get_user_profile(self, user_id: str) -> Optional[Dict[str, float]]: """Obtiene el perfil de usuario desde la base de datos""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" SELECT profile FROM users WHERE id = ? """, (user_id,)) result = cursor.fetchone() conn.close() if result: return json.loads(result[0]) return None except Exception as e: logger.error(f"Error al obtener perfil de usuario: {e}") return None def save_contribution(self, user_id: str, text: str, concepts: List[str], perspectives: Dict[str, str], principles: str) -> str: """Guarda una contribución en la base de datos""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Generar ID único para la contribución contribution_id = f"contrib_{int(datetime.now().timestamp())}" timestamp = datetime.now().isoformat() # Convertir datos a formato JSON concepts_json = json.dumps(concepts) perspectives_json = json.dumps(perspectives) # Insertar contribución cursor.execute(""" INSERT INTO contributions (id, user_id, text, concepts, perspectives, principles, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, (contribution_id, user_id, text, concepts_json, perspectives_json, principles, timestamp)) # Guardar nodo de contribución en el grafo cursor.execute(""" INSERT INTO graph_nodes (id, type, created_at) VALUES (?, ?, ?) """, (contribution_id, "contribution", timestamp)) # Guardar conceptos como nodos y conectarlos for concept in concepts: # Guardar nodo de concepto si no existe cursor.execute(""" INSERT OR IGNORE INTO graph_nodes (id, type, created_at) VALUES (?, ?, ?) """, (concept, "concept", timestamp)) # Conectar contribución con concepto cursor.execute(""" INSERT OR IGNORE INTO graph_edges (source, target, label, weight, created_at) VALUES (?, ?, ?, ?, ?) """, (contribution_id, concept, "contains", 1.0, timestamp)) # Conectar contribución con usuario cursor.execute(""" INSERT OR IGNORE INTO graph_edges (source, target, label, weight, created_at) VALUES (?, ?, ?, ?, ?) """, (user_id, contribution_id, "authored", 1.0, timestamp)) conn.commit() conn.close() # Sincronizar con Hugging Face si está habilitado if self.hf_sync_enabled: self._sync_to_huggingface() return contribution_id except Exception as e: logger.error(f"Error al guardar contribución: {e}") return f"error_{int(datetime.now().timestamp())}" def save_node(self, node_id: str, node_type: str) -> bool: """Guarda un nodo en la base de datos""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() timestamp = datetime.now().isoformat() cursor.execute(""" INSERT OR IGNORE INTO graph_nodes (id, type, created_at) VALUES (?, ?, ?) """, (node_id, node_type, timestamp)) conn.commit() conn.close() return True except Exception as e: logger.error(f"Error al guardar nodo: {e}") return False def save_edge(self, source: str, target: str, label: str, weight: float) -> bool: """Guarda una arista en la base de datos""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() timestamp = datetime.now().isoformat() cursor.execute(""" INSERT OR REPLACE INTO graph_edges (source, target, label, weight, created_at) VALUES (?, ?, ?, ?, ?) """, (source, target, label, weight, timestamp)) conn.commit() conn.close() return True except Exception as e: logger.error(f"Error al guardar arista: {e}") return False def get_all_nodes(self) -> List[Dict[str, Any]]: """Obtiene todos los nodos del grafo""" try: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT id, type FROM graph_nodes """) results = cursor.fetchall() conn.close() return [dict(row) for row in results] except Exception as e: logger.error(f"Error al obtener nodos: {e}") return [] def get_all_edges(self) -> List[Dict[str, Any]]: """Obtiene todas las aristas del grafo""" try: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT source, target, label, weight FROM graph_edges """) results = cursor.fetchall() conn.close() return [dict(row) for row in results] except Exception as e: logger.error(f"Error al obtener aristas: {e}") return [] def get_all_contributions(self) -> List[Dict[str, Any]]: """Obtiene todas las contribuciones""" try: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT id, user_id, text FROM contributions """) results = cursor.fetchall() conn.close() return [dict(row) for row in results] except Exception as e: logger.error(f"Error al obtener contribuciones: {e}") return [] def save_embedding(self, text: str, embedding: np.ndarray) -> bool: """Guarda un embedding en la caché""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() timestamp = datetime.now().isoformat() embedding_bytes = embedding.tobytes() cursor.execute(""" INSERT OR REPLACE INTO embedding_cache (text, embedding, created_at) VALUES (?, ?, ?) """, (text, embedding_bytes, timestamp)) conn.commit() conn.close() return True except Exception as e: logger.error(f"Error al guardar embedding: {e}") return False def get_embedding(self, text: str) -> Optional[np.ndarray]: """Obtiene un embedding desde la caché""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" SELECT embedding FROM embedding_cache WHERE text = ? """, (text,)) result = cursor.fetchone() conn.close() if result: embedding_bytes = result[0] return np.frombuffer(embedding_bytes, dtype=np.float32) return None except Exception as e: logger.error(f"Error al obtener embedding: {e}") return None def _sync_to_huggingface(self) -> bool: """Sincroniza la base de datos con Hugging Face Hub""" if not self.hf_sync_enabled: return False try: # Crear archivo temporal con copia de la BD with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as temp_file: temp_path = temp_file.name # Copiar BD actual al archivo temporal conn = sqlite3.connect(self.db_path) temp_conn = sqlite3.connect(temp_path) conn.backup(temp_conn) conn.close() temp_conn.close() # Subir archivo a Hugging Face self.hf_api.upload_file( path_or_fileobj=temp_path, path_in_repo=os.path.basename(self.db_path), repo_id=HF_REPO_ID, repo_type="dataset" ) # Eliminar archivo temporal os.unlink(temp_path) logger.info(f"Base de datos sincronizada con Hugging Face: {HF_REPO_ID}") return True except Exception as e: logger.error(f"Error al sincronizar con Hugging Face: {e}") return False class GradioInterface: """Interfaz de usuario con Gradio""" def __init__(self): """Inicializa los componentes de la interfaz""" self.profiler = DISCProfiler() self.ci_engine = CollectiveIntelligenceEngine() self.knowledge_graph = KnowledgeGraph() def create_interface(self): """Crea la interfaz de usuario con Gradio""" with gr.Blocks(title="Plataforma de Inteligencia Colectiva Aumentada") as app: gr.Markdown("# 🧠 Plataforma de Inteligencia Colectiva Aumentada") gr.Markdown("### Sistema de perfilamiento cognitivo y síntesis colectiva basado en DISC") with gr.Tabs() as tabs: # Tab de Perfilamiento DISC with gr.TabItem("Perfilamiento DISC") as profile_tab: with gr.Row(): with gr.Column(scale=3): gr.Markdown("## Sistema de Perfilamiento DISC") gr.Markdown("Basado en 'Surrounded by Idiots' de Thomas Erikson") # Formulario de perfilamiento user_id_input = gr.Textbox(label="Tu nombre o identificador", placeholder="Ingresa tu nombre o ID") # Crear componentes para las preguntas question_components = [] for q in DISC_QUESTIONS: with gr.Group(): gr.Markdown(f"**{q['question']}**") radio = gr.Radio( choices=[ f"D: {q['options']['D']}", f"I: {q['options']['I']}", f"S: {q['options']['S']}", f"C: {q['options']['C']}" ], label="" ) question_components.append(radio) profile_submit = gr.Button("Calcular mi perfil DISC") with gr.Column(scale=2): gr.Markdown("## Tu perfil DISC") profile_output = gr.JSON(label="Resultado del perfil") gr.Markdown("## Visualización del perfil") profile_plot = gr.Plot(label="Visualización DISC") profile_description = gr.Markdown() # Tab de Contribución Colectiva with gr.TabItem("Contribución Colectiva") as contribute_tab: with gr.Row(): with gr.Column(scale=2): gr.Markdown("## Contribuye al conocimiento colectivo") contributor_id = gr.Textbox(label="Tu nombre o identificador", placeholder="Ingresa tu nombre o ID") contribution_text = gr.Textbox( label="Tu contribución", placeholder="Escribe tu contribución, idea o reflexión aquí...", lines=10 ) contribute_button = gr.Button("Contribuir") with gr.Column(scale=3): gr.Markdown("## Síntesis colectiva") concepts_output = gr.JSON(label="Conceptos clave") gr.Markdown("### Perspectivas DISC") perspectives_output = gr.Dataframe( headers=["Tipo", "Perspectiva"], label="Síntesis multi-perspectiva" ) gr.Markdown("### Principios universales") principles_output = gr.Markdown() # Tab de Visualización del Grafo with gr.TabItem("Grafo de Conocimiento") as graph_tab: with gr.Row(): with gr.Column(): gr.Markdown("## Grafo de Conocimiento Colectivo") graph_plot = gr.Plot(label="Visualización del grafo") refresh_graph = gr.Button("Actualizar grafo") # Funciones para manejar eventos def calculate_profile(user_id, *question_responses): if not user_id: return {"error": "Por favor, ingresa un nombre o identificador"}, None, "Por favor, completa el formulario." # Verificar que todas las preguntas tengan respuesta if any(r is None for r in question_responses): return {"error": "Por favor, responde todas las preguntas"}, None, "Por favor, completa todas las preguntas." # Extraer solo la letra de la respuesta (D, I, S, C) responses = [r[0] for r in question_responses] # Calcular perfil profile_vector = self.profiler.get_semantic_profile(responses) # Guardar perfil self.profiler.save_profile(user_id, profile_vector, responses) # Crear visualización fig = self._create_profile_visualization(profile_vector) # Crear descripción description = self._create_profile_description(profile_vector) return profile_vector, fig, description def submit_contribution(user_id, text): if not user_id or not text: return ( {"error": "Por favor, completa todos los campos"}, [], "Por favor, ingresa tu nombre y tu contribución." ) # Procesar contribución result = self.ci_engine.process_contribution(user_id, text) # Preparar salida de perspectivas perspectives_df = [[k, v] for k, v in result["perspectives"].items()] return result["concepts"], perspectives_df, result["principles"] def update_graph(): # Obtener datos del grafo graph_data = self.knowledge_graph.get_visualization_data() # Crear visualización fig = self._create_graph_visualization(graph_data) return fig # Conectar eventos profile_submit.click( fn=calculate_profile, inputs=[user_id_input] + question_components, outputs=[profile_output, profile_plot, profile_description] ) contribute_button.click( fn=submit_contribution, inputs=[contributor_id, contribution_text], outputs=[concepts_output, perspectives_output, principles_output] ) refresh_graph.click( fn=update_graph, inputs=[], outputs=[graph_plot] ) # Inicializar el grafo al cargar app.load( fn=update_graph, inputs=[], outputs=[graph_plot] ) return app def _create_profile_visualization(self, profile: Dict[str, float]) -> go.Figure: """Crea una visualización del perfil DISC""" # Preparar datos categories = list(profile.keys()) values = list(profile.values()) # Crear gráfico de radar fig = go.Figure() fig.add_trace(go.Scatterpolar( r=values, theta=categories, fill='toself', name='Perfil DISC', line_color='rgba(255, 65, 54, 0.8)', fillcolor='rgba(255, 65, 54, 0.3)' )) fig.update_layout( polar=dict( radialaxis=dict( visible=True, range=[0, max(values) * 1.1] ) ), showlegend=False, title="Perfil DISC" ) return fig def _create_profile_description(self, profile: Dict[str, float]) -> str: """Crea una descripción textual del perfil DISC""" # Encontrar el tipo dominante dominant_type = max(profile.items(), key=lambda x: x[1])[0] descriptions = { "D": """ ### Perfil Dominante (D) Tu perfil muestra una tendencia hacia el estilo **Dominante**. Eres una persona: - Orientada a resultados y eficiencia - Decidida y directa en tu comunicación - Con capacidad para tomar decisiones rápidas - Enfocada en desafíos y soluciones prácticas Cuando trabajas en equipo, aportas determinación y capacidad para tomar acción. Para mejorar tu comunicación con otros perfiles, considera ser más paciente con procesos que requieren tiempo y mostrar más empatía hacia las necesidades emocionales de los demás. """, "I": """ ### Perfil Influyente (I) Tu perfil muestra una tendencia hacia el estilo **Influyente**. Eres una persona: - Entusiasta y expresiva en tu comunicación - Sociable y orientada a las relaciones - Optimista y motivadora - Creativa y generadora de ideas Cuando trabajas en equipo, aportas energía positiva y habilidades para inspirar a otros. Para mejorar tu comunicación con otros perfiles, considera dedicar más tiempo a los detalles y la estructura, y ser más concisa en tus explicaciones. """, "S": """ ### Perfil Estable (S) Tu perfil muestra una tendencia hacia el estilo **Estable**. Eres una persona: - Paciente y confiable - Colaborativa y enfocada en el equipo - Buena escuchando y apoyando a otros - Constante y predecible en tu comportamiento Cuando trabajas en equipo, aportas estabilidad y capacidad para mantener la armonía. Para mejorar tu comunicación con otros perfiles, considera expresar más abiertamente tus opiniones y adaptarte con más facilidad a los cambios repentinos. """, "C": """ ### Perfil Concienzudo (C) Tu perfil muestra una tendencia hacia el estilo **Concienzudo**. Eres una persona: - Analítica y precisa en tu enfoque - Orientada a la calidad y los detalles - Sistemática y metódica en tu trabajo - Cautelosa en la toma de decisiones Cuando trabajas en equipo, aportas precisión y pensamiento crítico. Para mejorar tu comunicación con otros perfiles, considera ser más flexible con los procedimientos y expresar más abiertamente tus emociones y pensamientos. """ } return descriptions.get(dominant_type, "### Perfil Equilibrado\n\nTu perfil muestra un balance entre los diferentes estilos DISC.") def _create_graph_visualization(self, graph_data: Dict[str, Any]) -> go.Figure: """Crea una visualización del grafo de conocimiento""" # Crear figura fig = go.Figure() # Agregar aristas fig.add_trace(go.Scatter( x=graph_data["edge_x"], y=graph_data["edge_y"], line=dict(width=0.5, color='#888'), hoverinfo='text', mode='lines', text=graph_data["edge_text"] * 3 if graph_data["edge_text"] else [], name='Conexiones' )) # Agregar nodos fig.add_trace(go.Scatter( x=graph_data["node_x"], y=graph_data["node_y"], mode='markers', hoverinfo='text', text=graph_data["node_text"], marker=dict( showscale=False, color=graph_data["node_color"], size=graph_data["node_size"], line=dict(width=1, color='#000') ), name='Nodos' )) # Configurar diseño fig.update_layout( title='Grafo de Conocimiento Colectivo', showlegend=False, hovermode='closest', margin=dict(b=0, l=0, r=0, t=40), xaxis=dict(showgrid=False, zeroline=False, showticklabels=False), yaxis=dict(showgrid=False, zeroline=False, showticklabels=False), height=600 ) return fig # Crear y lanzar la aplicación if __name__ == "__main__": interface = GradioInterface() app = interface.create_interface() app.launch()