# ========================= # Consenso Democrático - App para HuggingFace Spaces # ========================= import os import io import gc import time import warnings import pandas as pd import gradio as gr from typing import Dict from contextlib import redirect_stdout, redirect_stderr # CrewAI from crewai import Agent, Task, Crew # --- Configuración de entorno --- # IMPORTANTE: En HuggingFace Spaces, las variables de entorno se configuran en la interfaz web # NO pongas tu API key directamente en el código OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if not OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY environment variable is required") os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY os.environ["USER_AGENT"] = os.getenv("USER_AGENT", "ConsensusApp") os.environ['CREWAI_DO_NOT_TELEMETRY'] = 'true' os.environ['CREWAI_SHARE_CREW'] = 'false' # Suprimir warnings ruidosos warnings.filterwarnings("ignore", category=UserWarning, module="crewai") warnings.filterwarnings("ignore", category=DeprecationWarning) # ========================= # Datos base y utilidades # ========================= # Mapeo de partidos políticos a orientaciones PARTY_TO_ORIENTATION = { "FDP": "RIGHT", "jf": "RIGHT", "CVP": "CENTER", "Grüne": "LEFT", "SVP": "RIGHT", "PdA": "LEFT", "SP": "LEFT", "glp": "CENTER", "JUSO": "LEFT", "BDP": "CENTER", "JG": "RIGHT", "JSVP": "RIGHT", "EDU": "RIGHT", "Piraten": "CENTER", "jglp": "CENTER", "EVP": "CENTER", "JCVP": "CENTER", "jevp": "CENTER", "JBDP": "CENTER", "Die": "CENTER", "JFS": "RIGHT", "SD": "RIGHT", "JM": "CENTER" } # Modelos por nivel LLM_BASE = "gpt-4o-mini" LLM_SUPERVISOR = "gpt-4o-mini" LLM_DIRECTOR = "gpt-4o-mini" def get_majority_decision_groups(df: pd.DataFrame) -> Dict: """ Calcula decisiones por mayoría y porcentajes (FAVOR/AGAINST) por: - language - party_reply - orientación (derivada de party_reply via PARTY_TO_ORIENTATION) - FINAL_MAJORITY global """ result = {} def calc_decision_and_percentages(series): total = len(series) favor_count = (series == 'FAVOR').sum() against_count = (series == 'AGAINST').sum() if favor_count > against_count: decision = 'FAVOR' elif against_count > favor_count: decision = 'AGAINST' else: decision = 'AMBIGUOUS' favor_pct = round((favor_count / total) * 100, 2) if total else 0.0 against_pct = round((against_count / total) * 100, 2) if total else 0.0 return {"decision": decision, "percent_favor": favor_pct, "percent_against": against_pct} # Por idioma for lang, group in df.groupby('language')['label']: result[lang] = calc_decision_and_percentages(group) # Por partido for party, group in df.groupby('party_reply')['label']: result[party] = calc_decision_and_percentages(group) # Por orientación df_or = df.copy() df_or['orientation'] = df_or['party_reply'].map(PARTY_TO_ORIENTATION) df_or = df_or.dropna(subset=['orientation']) if not df_or.empty: for orient, group in df_or.groupby('orientation')['label']: result[f"{orient}_MAJORITY"] = calc_decision_and_percentages(group) # Global result['FINAL_MAJORITY'] = calc_decision_and_percentages(df['label']) return result # Silenciar salidas ruidosas de CrewAI def _silent_kickoff(crew: Crew): buf_out, buf_err = io.StringIO(), io.StringIO() with redirect_stdout(buf_out), redirect_stderr(buf_err): return crew.kickoff() # ========================= # Sistema Multiagente (CrewAI) # ========================= class DemocraticConsensusCrewAI: def __init__(self, question: str = ""): self.question = question self.results = {"base_agents": {}, "supervisors": {}, "director": ""} # Un agente por nivel self.base_agent = Agent( role="Democratic Base Representative", goal="Generate democratic consensus for any given group of opinions, weighting all perspectives equally.", backstory=("You are a neutral representative. Given any group's opinions and a majority decision, " "you craft a fair, balanced consensus that aligns with that majority, acknowledging minorities as nuances."), verbose=False, allow_delegation=False, llm=LLM_BASE ) self.supervisor_agent = Agent( role="Democratic Supervisor", goal="Refine and consolidate consensuses for a language or an orientation umbrella.", backstory=("You refine base consensuses for a given umbrella (language or orientation), " "keeping the specified majority as the core, integrating minority views as exceptions."), verbose=False, allow_delegation=False, llm=LLM_SUPERVISOR ) self.director_agent = Agent( role="Digital President - Final Decision Maker", goal="Synthesize all supervisor consensuses into a final consensus that aligns with the absolute majority.", backstory=("You aggregate all umbrellas and issue a final democratic position respecting the global majority."), verbose=False, allow_delegation=False, llm=LLM_DIRECTOR ) # Creación de tareas def create_base_tasks(self, large_strings_by_language: Dict, large_strings_by_party: Dict, majorities_decision: Dict): tasks = [] # Por idioma for language, opinions_text in large_strings_by_language.items(): maj = majorities_decision.get(language, {}).get('decision', 'AMBIGUOUS') t = Task( description=f""" MANDATORY INSTRUCTION: The consensus you generate MUST be based around the MAJORITY DECISION: {maj} Analyze the following {language} language opinions to answer the question: '{self.question}' {opinions_text} CRITICAL REQUIREMENTS: 1. Align with the majority decision: {maj} 2. Treat it as non-negotiable core. 3. Include minority views as implementation nuances. 4. Keep the final stance supporting {maj}. Return only the final consensus text that supports {maj}. """.strip(), agent=self.base_agent, expected_output=f"[BASE][LANG={language}][MAJ={maj}]" ) tasks.append(t) # Por partido for party, opinions_text in large_strings_by_party.items(): maj = majorities_decision.get(party, {}).get('decision', 'AMBIGUOUS') t = Task( description=f""" MANDATORY INSTRUCTION: The consensus you generate MUST be based around the MAJORITY DECISION: {maj} Analyze the following {party} party opinions to answer the question: '{self.question}' {opinions_text} CRITICAL REQUIREMENTS: 1. Align with the majority decision: {maj} 2. Treat it as non-negotiable core. 3. Include minority views as implementation nuances. 4. Keep the final stance supporting {maj}. Return only the final consensus text that supports {maj}. """.strip(), agent=self.base_agent, expected_output=f"[BASE][PARTY={party}][MAJ={maj}]" ) tasks.append(t) return tasks def create_supervisor_tasks(self, base_results: Dict, majorities_decision: Dict): tasks = [] # Idiomas for lang in ["German", "French", "Italian"]: key = ('LANGUAGE', lang) if key in base_results: maj = majorities_decision.get(lang, {}).get('decision', 'AMBIGUOUS') t = Task( description=f""" MANDATORY INSTRUCTION: Your refined consensus MUST be based around the MAJORITY DECISION: {maj} Refine and consolidate the democratic consensus for {lang} language regarding: '{self.question}' {base_results[key]} CRITICAL REQUIREMENTS: 1. Strengthen alignment with {maj}. 2. Minority views only as exceptions or details. 3. Return text in English (<= 150 tokens). Return only the refined consensus in English supporting {maj}. """.strip(), agent=self.supervisor_agent, expected_output=f"[SUPERVISOR][LANG={lang}][MAJ={maj}]" ) tasks.append(t) # Orientaciones (unifica partidos) orientation_consensuses = {"LEFT": [], "CENTER": [], "RIGHT": []} for (kind, name), consensus in base_results.items(): if kind == 'PARTY' and name in PARTY_TO_ORIENTATION: orientation_consensuses[PARTY_TO_ORIENTATION[name]].append(consensus) for orientation in ["LEFT", "CENTER", "RIGHT"]: if orientation_consensuses[orientation]: combined = "\n\n".join([f"Consensus {i+1}: {c}" for i, c in enumerate(orientation_consensuses[orientation], 1)]) maj = majorities_decision.get(f"{orientation}_MAJORITY", {}).get('decision', 'AMBIGUOUS') t = Task( description=f""" MANDATORY INSTRUCTION: Your unified consensus MUST be based around the MAJORITY DECISION: {maj} Create a unified democratic consensus for {orientation} orientation regarding: '{self.question}' {combined} CRITICAL REQUIREMENTS: 1. Align around {maj}. 2. Merge party consensuses with {maj} as core. 3. Return text in English (<= 150 tokens). Return only the meta-consensus in English supporting {maj}. """.strip(), agent=self.supervisor_agent, expected_output=f"[SUPERVISOR][ORIENT={orientation}][MAJ={maj}]" ) tasks.append(t) return tasks def create_director_task(self, supervisor_results: Dict, majorities_decision: Dict): all_consensuses = "\n\n".join([f"{k}: {v}" for k, v in supervisor_results.items()]) final_majority = majorities_decision.get('FINAL_MAJORITY', {}).get('decision', 'AMBIGUOUS') t = Task( description=f""" SUPREME MANDATORY INSTRUCTION: Your FINAL consensus MUST be based around the ABSOLUTE MAJORITY DECISION: {final_majority} As Digital President, synthesize the FINAL DEMOCRATIC CONSENSUS for the question: '{self.question}' Representative consensuses: {all_consensuses} CRITICAL REQUIREMENTS: 1. Align strictly with {final_majority}. 2. Integrate minorities only as nuances. 3. <= 150 tokens. Return only the final consensus supporting {final_majority}. """.strip(), agent=self.director_agent, expected_output=f"[DIRECTOR][MAJ={final_majority}]" ) return t def run_democratic_consensus_system(self, large_strings_by_language: Dict, large_strings_by_party: Dict, majorities_decision: Dict = {}) -> Dict: # Nivel 1: Base base_tasks = self.create_base_tasks(large_strings_by_language, large_strings_by_party, majorities_decision) base_results = {} if base_tasks: base_crew = Crew(agents=[self.base_agent], tasks=base_tasks, verbose=0, share_crew=False) base_out = _silent_kickoff(base_crew) for i, task in enumerate(base_tasks): raw = base_out.tasks_output[i].raw if i < len(base_out.tasks_output) else "" tag = task.expected_output if "[LANG=" in tag: lang = tag.split("[LANG=")[1].split("]")[0] base_results[('LANGUAGE', lang)] = raw elif "[PARTY=" in tag: party = tag.split("[PARTY=")[1].split("]")[0] base_results[('PARTY', party)] = raw self.results["base_agents"] = base_results # Nivel 2: Supervisor supervisor_tasks = self.create_supervisor_tasks(base_results, majorities_decision) supervisor_results = {} if supervisor_tasks: sup_crew = Crew(agents=[self.supervisor_agent], tasks=supervisor_tasks, verbose=0, share_crew=False) sup_out = _silent_kickoff(sup_crew) for i, task in enumerate(supervisor_tasks): raw = sup_out.tasks_output[i].raw if i < len(sup_out.tasks_output) else "" tag = task.expected_output if "[LANG=" in tag: lang = tag.split("[LANG=")[1].split("]")[0] supervisor_results[('LANGUAGE', lang)] = raw elif "[ORIENT=" in tag: orient = tag.split("[ORIENT=")[1].split("]")[0] supervisor_results[('ORIENTATION', orient)] = raw self.results["supervisors"] = supervisor_results # Nivel 3: Director director_task = self.create_director_task(supervisor_results, majorities_decision) dir_crew = Crew(agents=[self.director_agent], tasks=[director_task], verbose=0, share_crew=False) dir_out = _silent_kickoff(dir_crew) final_consensus = dir_out.tasks_output[0].raw if dir_out.tasks_output else "No final consensus generated" self.results["director"] = final_consensus # Limpieza gc.collect() return self.results # Wrapper def run_democratic_consensus_system(large_strings_by_language: Dict, large_strings_by_party: Dict, question: str = "", majorities_decision: Dict = {}) -> Dict: system = DemocraticConsensusCrewAI(question=question) return system.run_democratic_consensus_system(large_strings_by_language, large_strings_by_party, majorities_decision) # ========================= # Función de procesamiento (Excel -> resultados UI) # ========================= def process_excel_and_generate_consensus(excel_file_path): """ Lee el Excel subido, calcula mayorías, corre el sistema de consenso y devuelve los textos + porcentajes para mostrarlos en la UI. Se espera que el Excel tenga: ['language', 'party_reply', 'label', 'comment']. """ if not excel_file_path: return ("Error: No file uploaded", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "") try: df = pd.read_excel(excel_file_path) # Validación mínima required = {'language', 'party_reply', 'label', 'comment'} missing = list(required - set(df.columns)) if missing: return (f"Error: Missing required columns: {missing}", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "") # Extraer pregunta si existe if 'question_y' in df.columns and df['question_y'].dropna().size: question = str(df['question_y'].dropna().iloc[0]) elif 'question' in df.columns and df['question'].dropna().size: question = str(df['question'].dropna().iloc[0]) else: question = "Question not found in Excel" # Strings por idioma y partido large_strings_by_language = ( df.groupby("language")["comment"] .apply(lambda g: "'; ".join([f"Opinion {i+1}: '{c}" for i, c in enumerate(g.dropna().astype(str))])) .to_dict() ) large_strings_by_party = ( df.groupby("party_reply")["comment"] .apply(lambda g: "'; ".join([f"Opinion {i+1}: '{c}" for i, c in enumerate(g.dropna().astype(str))])) .to_dict() ) # Mayorías majorities_decision = get_majority_decision_groups(df) # Ejecutar sistema multiagente results = run_democratic_consensus_system( large_strings_by_language, large_strings_by_party, question, majorities_decision ) # Consensos supervisores german = results['supervisors'].get(('LANGUAGE', 'German'), "No consensus data") french = results['supervisors'].get(('LANGUAGE', 'French'), "No consensus data") italian = results['supervisors'].get(('LANGUAGE', 'Italian'), "No consensus data") left = results['supervisors'].get(('ORIENTATION', 'LEFT'), "No consensus data") center = results['supervisors'].get(('ORIENTATION', 'CENTER'), "No consensus data") right = results['supervisors'].get(('ORIENTATION', 'RIGHT'), "No consensus data") final = results.get('director', "No consensus data") # Porcentajes (FAVOR) german_percentage = f"{majorities_decision.get('German', {}).get('percent_favor', 0):.1f}%" french_percentage = f"{majorities_decision.get('French', {}).get('percent_favor', 0):.1f}%" italian_percentage = f"{majorities_decision.get('Italian', {}).get('percent_favor', 0):.1f}%" left_percentage = f"{majorities_decision.get('LEFT_MAJORITY', {}).get('percent_favor', 0):.1f}%" center_percentage = f"{majorities_decision.get('CENTER_MAJORITY', {}).get('percent_favor', 0):.1f}%" right_percentage = f"{majorities_decision.get('RIGHT_MAJORITY', {}).get('percent_favor', 0):.1f}%" final_percentage = f"{majorities_decision.get('FINAL_MAJORITY', {}).get('percent_favor', 0):.1f}%" status = f"Success: Consensus statements generated from {os.path.basename(excel_file_path)}" return (status, question, german, french, italian, left, center, right, final, german_percentage, french_percentage, italian_percentage, left_percentage, center_percentage, right_percentage, final_percentage) except Exception as e: return (f"Error processing file: {e}", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "") # ========================= # Interfaz de Gradio # ========================= def create_consensus_app(): with gr.Blocks( title="🏛️ Consensus Statements", theme=gr.themes.Base(), css=""" /* Estilos de banderas */ .flag-container { display: flex; justify-content: center; align-items: center; gap: 15px; margin-bottom: 10px; } .flag-image { width: 50px; height: auto; border-radius: 5px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); } /* Estilos de las cajas de consenso */ .consensus-box { padding: 15px; border-radius: 15px; margin: 10px; min-height: 300px; max-height: 500px; overflow-y: auto; word-wrap: break-word; display: flex; align-items: flex-start; justify-content: center; text-align: center; font-family: 'Georgia', 'Times New Roman', serif; font-weight: normal; } .consensus-box-content { font-size: 1.0em; line-height: 1.4; padding: 5px; width: 100%; } .consensus-box-german { background: linear-gradient(135deg, #FFB6C1, #FFC0CB); color: #333333; } .consensus-box-french { background: linear-gradient(135deg, #E6E6FA, #DDA0DD); color: #333333; } .consensus-box-italian { background: linear-gradient(135deg, #F0E68C, #F5DEB3); color: #333333; } .consensus-box-left { background: linear-gradient(135deg, #98FB98, #90EE90); color: #333333; } .consensus-box-center { background: linear-gradient(135deg, #87CEEB, #B0E0E6); color: #333333; } .consensus-box-right { background: linear-gradient(135deg, #FFDAB9, #FFE4B5); color: #333333; } /* Estilo para la caja del consenso final */ .final-consensus-box { background: linear-gradient(135deg, #A5D6A7, #66BB6A); color: #333333; padding: 30px; border-radius: 15px; margin: 20px 0; box-shadow: 0 6px 10px rgba(0,0,0,0.15); min-height: 150px; max-height: 400px; overflow-y: auto; display: flex; align-items: flex-start; justify-content: center; text-align: center; font-family: 'Georgia', 'Times New Roman', serif; font-weight: normal; font-size: 1.1em; } .question-header { background: linear-gradient(135deg, #1e3a5f, #2d5a87); color: white; padding: 15px; border-radius: 10px; text-align: center; margin-bottom: 20px; font-family: 'Georgia', 'Times New Roman', serif; font-weight: normal; } .upload-section { text-align: center; margin: 20px 0; } .gradio-container { font-family: 'Georgia', 'Times New Roman', serif; } /* Botón personalizado */ .custom-button-color { background-color: #2e8b57 !important; color: #ffffff !important; border-color: #2e8b57 !important; } """ ) as demo: # Título y banderas gr.HTML("""