from smolagents import CodeAgent, HfApiModel, tool import yaml from tools.final_answer import FinalAnswerTool import gradio as gr from textblob import TextBlob from transformers import pipeline import re from nltk.corpus import stopwords from nltk.tokenize import word_tokenize import nltk from typing import List, Dict import os # Variable global para almacenar el historial de la conversación conversation_history: List[Dict[str, str]] = [] # Descargar recursos de NLTK (solo la primera vez) nltk.download('punkt') nltk.download('stopwords') # Cargar el modelo de detección de emociones en español emotion_classifier = pipeline("text-classification", model="finiteautomata/beto-emotion-analysis") # Preprocesar el texto def preprocess_text(text: str) -> str: """Preprocesa el texto para mejorar la detección de emociones.""" text = text.lower() # Convertir a minúsculas text = re.sub(r'[^\w\s]', '', text) # Eliminar puntuación stop_words = set(stopwords.words('spanish')) # Eliminar stopwords en español words = word_tokenize(text) words = [word for word in words if word not in stop_words] return " ".join(words) # Herramienta para análisis de sentimientos @tool def sentiment_analysis(text: str) -> str: """Analiza el sentimiento del texto proporcionado por el usuario. Args: text: El texto del usuario. """ try: analysis = TextBlob(text) sentiment = analysis.sentiment.polarity if sentiment > 0: return "Positivo" elif sentiment < 0: return "Negativo" else: return "Neutral" except Exception as e: return f"Error en el análisis de sentimientos: {str(e)}" # Herramienta para detección de emociones @tool def emotion_detection(text: str) -> str: """Detecta las emociones predominantes en el texto del usuario. Args: text: El texto del usuario. """ try: text = preprocess_text(text) # Preprocesar el texto result = emotion_classifier(text) predominant_emotion = result[0]['label'] return predominant_emotion except Exception as e: return f"Error en la detección de emociones: {str(e)}" # Herramienta para actualizar el perfil del usuario @tool def user_profile_update(emotion: str, sentiment: str) -> str: """Actualiza el perfil del usuario con la información emocional recopilada. Args: emotion: La emoción predominante detectada. sentiment: El sentimiento detectado. """ try: with open("user_profile.txt", "a") as f: f.write(f"Emotion: {emotion}, Sentiment: {sentiment}\n") return "Perfil actualizado correctamente." except Exception as e: return f"Error al actualizar el perfil: {str(e)}" final_answer = FinalAnswerTool() # Cargar el modelo y el agente with open("prompts.yaml", 'r') as stream: prompt_templates = yaml.safe_load(stream) model = HfApiModel( max_tokens=2096, temperature=0.5, model_id='Qwen/Qwen2.5-Coder-32B-Instruct', custom_role_conversions=None, token=os.getenv('hf_token'), ) agent = CodeAgent( model=model, tools=[sentiment_analysis, emotion_detection, user_profile_update, final_answer], max_steps=3, # Aumentado para permitir más pasos verbosity_level=1, grammar=None, planning_interval=None, name="Emotional Support Agent", description="Un agente de IA que ofrece apoyo emocional y crecimiento personal.", prompt_templates=prompt_templates ) # Variable global para almacenar el rol actual del agente current_role = "emotional_support" # Valores posibles: "emotional_support" o "wellness_coach" def switch_role(new_role: str): """Cambia el rol del agente.""" global current_role if new_role in ["emotional_support", "wellness_coach"]: current_role = new_role return f"Rol cambiado a: {new_role}" else: return "Rol no válido. Los roles disponibles son: 'emotional_support' y 'wellness_coach'." # Función para interactuar con el agente def interact_with_agent(messages): """Función que interactúa con el agente y devuelve una respuesta.""" global conversation_history, current_role try: # Extraer la pregunta del usuario user_input = messages[-1]["content"] # Detectar si el usuario quiere cambiar el rol if "cambiar a modo coach" in user_input.lower(): switch_role("wellness_coach") return {"response": "Has cambiado al modo Coach de Bienestar y Objetivos. ¿En qué puedo ayudarte?"} elif "cambiar a modo apoyo emocional" in user_input.lower(): switch_role("emotional_support") return {"response": "Has cambiado al modo Asistente de Apoyo Emocional. ¿En qué puedo ayudarte?"} # Añadir el mensaje del usuario al historial de la conversación conversation_history.append({"role": "user", "content": user_input}) # Pasar el historial de la conversación al agente como contexto context = "\n".join([f"{msg['role']}: {msg['content']}" for msg in conversation_history]) # Ejecutar el agente según el rol actual if current_role == "emotional_support": result = agent.run(f"Contexto de la conversación:\n{context}\n\nPregunta del usuario: {user_input}") elif current_role == "wellness_coach": result = agent.run(f"Modo Coach de Bienestar y Objetivos. Contexto de la conversación:\n{context}\n\nPregunta del usuario: {user_input}") # Verificar si result es una cadena de texto no vacía if result and isinstance(result, str): # Añadir la respuesta del agente al historial de la conversación conversation_history.append({"role": "assistant", "content": result}) return {"response": result} # Devuelve un JSON con la respuesta else: return {"response": "El agente no devolvió una respuesta válida."} except Exception as e: return {"response": f"Error: {str(e)}"} # Crear la interfaz de Gradio iface = gr.Interface( fn=interact_with_agent, inputs=gr.JSON(), # Acepta un JSON como entrada outputs=gr.JSON(), # Devuelve un JSON como salida title="Emotional Support Agent", description="Un agente de IA que ofrece apoyo emocional y crecimiento personal.", ) # Lanzar la aplicación Gradio iface.launch(show_error=True)