import os import grad```as gr import numpy as np import torch import asyn``` import networkx``` nx from transformers import```toModel, AutoTokenizer```om groq import```oq from datetime import datetime from enum import Enum from dataclasses import dat```ass from typing import```ct, List, Optional```ny import plotly.express``` px import matplotlib```plot as plt from sklearn.```ifold import```NE from sentence```ansformers import Sentence```nsformer # Configura```n avanzada NEXUS_VERSION = "1.0.0" COLLECTIVE_INTELL```NCE_THRESHOLD = 0.78 GROQ_MODEL```"llama-3.3-70b-versatile" DISC_MODEL = SentenceTransformer```ll-MiniLM-L6-v2') # Tipos de personalidad DISC class DISCTYPE```um): DOMINANT =```" INFL```TIAL = "I" STEADY =```" CONSCIENTIOUS = "C```@dataclass class CognitiveProfile``` """Perfil```gnitivo basado``` DISC""" primary: DIS```PE secondary:```tional[DISCTYPE] = None vector: np.nd```ay = None traits: Dict[str, float] = None @dataclass class CollectiveAgent: """Agente autón``` especializado``` id: str role: str knowledge_base:```st[str] last_activation:```tetime = field(default_factory=datetime.now```class EmergentSystem: """Motor de inteligencia```lectiva""" def __init__(self): self.```nts = self._```t_agents() self.knowledge_graph```nx.Graph() self.cognitive_network```[] self.g```_client = Groq```i_key=os.getenv("GROQ_API_KEY")) def _init_agents(self) -> Dict[str, CollectiveAgent]: return { "analyzer": Collective```nt("A1", "Analista", ["Pattern Recognition"]), "```thesizer": Collective```nt("S1", "Sintetizador```["Cross-Domain Integration"]), "innovator```CollectiveAgent("I1", "Innovador", ["Divergent Thinking"]), "validator```CollectiveAgent("V1", "Validador",```Critical Analysis"]) } def _```ate_knowledge_graph(self, concept:```r, connections:```st[str]): self```owledge_graph.add_node(concept) for connection in```nnections: self.k```ledge_graph.add_edge(concept,```nnection) async```f collective_reason```(self, input_data```tr) -> Dict[str, Any]: """Proceso```laborativo multi-agente""" # Pro```amiento paralelo``` tasks =``` [ self._agent_processing(agent, input_data) for agent in self.agents.values() ] results```await asyncio.g```er(*tasks) # Síntesis emerg```e collective_```ight = self```mergent_synthesis(results) return { "timestamp":```tetime.now().isoformat``` "insights":```llective_insight, "knowledge_graph": self```isualize_knowledge_graph() } async def _agent_```cessing(self, agent: Collective```nt, input_data: str```> Dict: """Procesamiento especial```do por agente""" prompt = f"```o {agent.role} especializado en {', '.join(agent.knowledge_base)}, analiza: {input_data}" response = self.g```_client.chat.completions.create( messages=[{"role": "user", "content": prompt}], model=GROQ```DEL, temperature```7, max_tokens=1024 ) agent.last_activation```datetime.now() return``` "agent_id":```ent.id, "```e": agent.role``` "analysis":```sponse.choices[0].message.content } def _```rgent_synthesis(self, agent_output```List[Dict]) -> str: """Generación``` conocimiento emergente""" syntheses = [f"{output['role']}: {output['analysis']}" for output in agent_outputs] synthesis```ompt = "Sintet``` las siguientes perspectivas en una```nclusión unificada:\n" + "\n\n".join(syntheses) response```self.groq_client.chat```mpletions.create( messages```[{"role": "user", "content": synthesis_prompt}], model```OQ_MODEL, temperature=```, max_t```ns=2048 ) return response```oices[0].message.content def _visualize_know```ge_graph(self): """Generación```námica de grafo de conocimiento``` plt.figure```gsize=(12, 12)) pos```nx.spring_layout(self.knowledge_graph, k```5) nx.draw(self.knowledge_graph, pos,```th_labels=True, node_size```00, font_size```) plt```tle("Evolución del Conocimiento Co```tivo") return plt``` Evaluación DISC avanz``` class DISC```filer: """Sistema de perfil```ento cognitivo""" def __init__(self): self.questions```self._load```estions() self```ait_vectors = self._init_t```t_vectors() def _load_questions(self) -> List[Dict]: return [ {"id": 1, "text": "En situaciones de presión, tiendo a:", "options": { "D": "Tomar el control y dirigir acciones", "I": "Motivar al equipo con energía", "S": "Mantener la calma y buscar consenso", "C": "Analizar sistemáticamente las opciones" }}, # 11 preguntas adicionales con diseños psicométricos ] def _init_trait```ctors(self) -> Dict[str, np.ndarray]: return { "D```np.array([0.9, 0.1, 0.2, 0.3]), "``` np.array([0.1, 0.9, 0.4, 0.2]), "``` np.array([0.3, 0.3, 0.9, 0.1]), "C": np```ray([0.2, 0.1, 0.1, 0.9]) } def assess_profile(self, responses```ict[int, str]) -> CognitiveProfile``` """Calcula```rfil cognitivo con```bedding sem```ico""" response_text =```".join([self.questions[q]["options"][r] for q, r in responses```ems()]) semantic```ctor = DISC_MODEL```code([response_text])[0] #```peo a espacio```SC disc_s```es = {dtype: np.dot(semantic_vector, vec) for dtype, vec in```lf.trait_vectors```ems()} primary```max(disc_scores```ey=disc_scores```t) secondary```sorted(disc_scores```ey=disc_scores```t, reverse=True```] if disc_scores[primary] > 0.7 else None return Cognitive```file( primary=```CTYPE(primary), secondary=DISCT```(secondary) if secondary```se None, vector```mantic_vector, traits=disc_scores``` ) def visualize_profile(self, profile: CognitiveProfile): """Visualización interactiva```l perfil""" traits = list(profile.traits.keys()) scores = [profile.traits[t] for t in traits] fig```go.Figure(data=[go.Bar(x=traits, y=scores)]) fig.update_layout( title="Per``` DISC Cognit```", yaxis```tle="Intensidad del```sgo", template```lotly_white" ) return fig # Interfaz adaptativa```f create_adaptive_```erface(): """Interf```que se modifica```gún perfil```SC""" with gr```ocks(theme=gr.themes.Soft(primary_hue="pur```")) as interface: #```cción de evaluación DISC with```.Row(): ```sc_responses```{} for question``` DISCProfiler().questions: ```th gr.Column(): gr.Mark```n(f"**{question['text']}**") for opt_key```pt_text in question["options"].items(): btn =```.Button(opt_text, variant="secondary") disc_res```ses[question["id"]] = btn.click( fn=lambda```question["id"], v=opt_key: {k: v}, outputs=None``` ) # Proces```ento colectivo with gr.Row``` input_text = gr```xtbox(label="Consulta Co```tiva", lines``` submit_``` = gr.Button("Engendrar```teligencia", variant```rimary") # Visualización```aptativa with```.Row(): profile```z = gr.Plot() knowledge_graph```gr.Plot() ```sight_output = gr.M```down() # Eventos``` submit_btn.click``` fn=```cess_collective_inquiry``` inputs=[input_text, disc_responses], ```tputs=[insight_output, knowledge_graph, profile_viz] ) return interface```sync def process_collective```quiry(query: str, responses: dict): """Flujo```mpleto de procesamiento""" profiler = DISCProf```r() profile =```ofiler.assess```ofile(responses) profile```z = profiler.```ualize_profile(profile) system = Emergent```tem() collective_```ight = await```stem.collective_reason```(query) knowledge```aph = system._```ualize_knowledge_graph() return collective_```ight["insights"], knowledge_graph, profile```z # Punto de entrada if __name```== "__main```: nexus_interface```create_adaptive_interface``` nexus_```erface.launch( server_name="0.0.0.0", server_port=7860, share```ue )