File size: 5,454 Bytes
83c5f9d 8791d59 83c5f9d 8791d59 83c5f9d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
"""Service pour initialiser le serveur MCP avec FastMCP"""
from mcp.server.fastmcp import FastMCP
from typing import Dict, Any
import logging
from fastapi import FastAPI # Ajouté pour résoudre l'erreur Pylance "FastAPI is not defined"
from services.stance_model_manager import stance_model_manager
from services.label_model_manager import kpa_model_manager # Corrigé : import depuis kpa_model_manager (cohérent avec main.py)
from services.stt_service import speech_to_text
from services.tts_service import text_to_speech
from services.chat_service import generate_chat_response
# Note : Adapte les imports models si tes schémas sont dans un fichier unique (ex. models/schemas.py)
# from models.stance import StanceRequest, StanceResponse # Si séparé
# from models.kpa import PredictionRequest, PredictionResponse # Si séparé
# Ou si un seul fichier models.py :
# from models import StanceRequest, StanceResponse, PredictionRequest, PredictionResponse
logger = logging.getLogger(__name__)
# Créer l'instance FastMCP (nom du serveur, JSON pour responses structurées)
mcp_server = FastMCP("NLP-Debater-MCP", json_response=True, stateless_http=False) # Stateful pour sessions voice chat
# Tool pour Stance Detection
@mcp_server.tool()
def detect_stance(topic: str, argument: str) -> Dict[str, Any]:
"""
Détecte la stance (PRO/CON) d'un argument par rapport à un topic.
Args:
topic: Le sujet de débat (ex. "Assisted suicide should be a criminal offence")
argument: L'argument à classifier (ex. "People have the right to choose...")
Returns:
Dict avec predicted_stance, confidence, probabilities.
"""
if not stance_model_manager.model_loaded:
raise ValueError("Modèle stance non chargé")
result = stance_model_manager.predict(topic, argument)
return {
"predicted_stance": result["predicted_stance"],
"confidence": result["confidence"],
"probability_con": result["probability_con"],
"probability_pro": result["probability_pro"]
}
# Tool pour Key-Point Argument Matching (KPA)
@mcp_server.tool()
def match_keypoint_argument(argument: str, key_point: str) -> Dict[str, Any]:
"""
Prédit si un argument matche un key-point (apparie/non_apparie).
Args:
argument: Texte de l'argument
key_point: Le key-point de référence
Returns:
Dict avec prediction (0/1), label, confidence, probabilities.
"""
if not kpa_model_manager.model_loaded:
raise ValueError("Modèle KPA non chargé")
result = kpa_model_manager.predict(argument, key_point)
return {
"prediction": result["prediction"],
"label": result["label"],
"confidence": result["confidence"],
"probabilities": result["probabilities"]
}
# Tool pour STT (Speech-to-Text) - Note : Pour audio, utilise un upload via resource ou adapte
@mcp_server.tool()
def transcribe_audio(audio_path: str) -> str:
"""
Transcrit un fichier audio en texte (via Groq Whisper).
Args:
audio_path: Chemin vers le fichier audio (ex. temp file)
Returns:
Texte transcrit.
"""
return speech_to_text(audio_path)
# Tool pour TTS (Text-to-Speech)
@mcp_server.tool()
def generate_speech(text: str, voice: str = "Aaliyah-PlayAI", format: str = "wav") -> str:
"""
Génère un fichier audio à partir de texte (via Groq TTS).
Args:
text: Texte à synthétiser
voice: Voix (défaut: Aaliyah-PlayAI)
format: wav ou mp3
Returns:
Chemin vers le fichier audio généré.
"""
return text_to_speech(text, voice, format)
# Tool pour Argument Generation (Chatbot)
@mcp_server.tool()
def generate_argument(user_input: str, conversation_id: str = None) -> str:
"""
Génère une réponse argumentative via chatbot (via Groq Llama).
Args:
user_input: Input utilisateur
conversation_id: ID de session (optionnel)
Returns:
Réponse générée.
"""
return generate_chat_response(user_input, conversation_id)
# Resource exemple : Prompt template pour débats
@mcp_server.resource("debate://prompt")
def get_debate_prompt(topic: str) -> str:
"""Récupère un template de prompt pour générer des arguments sur un topic."""
return f"Tu es un expert en débat. Génère 3 arguments PRO pour le topic: {topic}. Sois concis et persuasif."
def init_mcp_server(app: FastAPI) -> None: # Retiré les quotes : FastAPI est maintenant importé
"""
Initialise et monte le serveur MCP sur l'app FastAPI.
Ajoute les routes MCP à /api/v1/mcp (ex. : /tools, /call, /resources).
"""
# Monter l'app MCP sur FastAPI (via Starlette Mount, compatible FastAPI)
from starlette.routing import Mount
# Créer l'app MCP streamable
mcp_app = mcp_server.streamable_http_app(streamable_http_path="/mcp") # Chemin racine /mcp
# Monter directement sur l'app FastAPI (app.mount pour sub-app)
app.mount("/api/v1/mcp", mcp_app) # Monte à /api/v1/mcp - gère /tools, /call, etc.
logger.info("✓ Serveur MCP initialisé et monté sur /api/v1/mcp avec tools NLP/STT/TTS")
# Pour batch ou health, ajoute si besoin (ce tool est déjà décoré)
@mcp_server.tool()
def health_check() -> Dict[str, Any]:
return {"status": "healthy", "tools": list(mcp_server.tools.keys())} |