Creación de TranscriptionAgent - Separación de Responsabilidades

[Charly-bite]

Creación de TranscriptionAgent - Separación de Responsabilidades

🎯 Razones para crear TranscriptionAgent separado:

✅ Ventajas de la Separación:

1. Principio de Responsabilidad Única: Cada agente tiene una función específica
2. Mantenimiento más fácil: Cambios en transcripción no afectan procesamiento
3. Escalabilidad: Puedes tener múltiples engines de transcripción
4. Testeo independiente: Cada agente se puede probar por separado
5. Configuración específica: Cada agente tiene sus propios parámetros

📁 Estructura Actual de /agents/:

agents/
├── __pycache__/
├── analysis_agent.py       # NLP y análisis de texto
├── obsidian_agent.py       # Generación de notas Markdown
├── processing_agent.py     # Procesamiento general de audio
├── recording_agent.py      # Grabación de audio
└── transcription_agent.py  # 🆕 NUEVO: Transcripción especializada

🔄 División de Responsabilidades:

📹 RecordingAgent (Existente)

• Captura de audio desde micrófono
• Selección de dispositivos
• Guardado de archivos WAV
• Metadatos de grabación

🎙️ TranscriptionAgent (Nuevo)

• Transcripción con whisper-amd nativo
• Gestión de modelos (tiny, base, small)
• Optimización para AMD A4-9125
• Formatos de salida (txt, srt, vtt)

⚙️ ProcessingAgent (Refactorizado)

• Orquestación del pipeline completo
• Control de calidad de audio
• Coordinación entre agentes
• Validación de resultados

🧠 AnalysisAgent (Existente)

• Análisis NLP con spaCy
• Extracción de conceptos clave
• Identificación de términos de ciberseguridad

📝 ObsidianAgent (Existente)

• Generación de notas estructuradas
• Templates Jinja2
• Integración con Obsidian

📝 Implementación de TranscriptionAgent

#!/usr/bin/env python3
# agents/transcription_agent.py

import subprocess
import sys
import time
from pathlib import Path
from datetime import datetime

# Add the project root to Python path
project_root = Path(__file__).parent.parent
if str(project_root) not in sys.path:
    sys.path.insert(0, str(project_root))

from mcp.agent_framework import AgentFramework

class TranscriptionAgent(AgentFramework):
    """
    Agente especializado en transcripción de audio usando whisper-amd
    optimizado para AMD A4-9125.
    
    Responsabilidades:
    - Transcripción con whisper.cpp nativo
    - Gestión de modelos optimizados
    - Configuración específica de hardware
    - Formatos de salida múltiples
    """
    
    def __init__(self):
        super().__init__("TranscriptionAgent")
        self.config = self._load_config()
        
        # Configuración específica del agente
        self.transcription_config = self.config.get('transcription_settings', {})
        
        # Paths
        self.transcripts_dir = Path('recordings/transcripts')
        self.transcripts_dir.mkdir(parents=True, exist_ok=True)
        
        # Configuración whisper-amd
        self.whisper_binary = "/usr/local/bin/whisper-amd"
        self.models_dir = Path.home() / "whisper_models"
        
        # Modelos disponibles con configuración específica para A4-9125
        self.models = {
            "tiny": {
                "file": self.models_dir / "ggml-tiny.bin",
                "threads": 2,
                "processors": 1,
                "speed": "fastest",
                "use_case": "tiempo_real"
            },
            "base": {
                "file": self.models_dir / "ggml-base.bin", 
                "threads": 2,
                "processors": 1,
                "speed": "balanced",
                "use_case": "produccion"
            },
            "small": {
                "file": self.models_dir / "ggml-small.bin",
                "threads": 1,  # Usar menos threads para modelo más grande
                "processors": 1,
                "speed": "accurate",
                "use_case": "precision"
            }
        }
        
        # Verificar instalación
        self._verify_installation()
        
        print(f"[{self.agent_name}] Initialized with whisper-amd native support")
        print(f"[{self.agent_name}] Available models: {list(self.models.keys())}")

    def _verify_installation(self):
        """Verifica que whisper-amd y modelos estén disponibles."""
        # Verificar whisper-amd
        if not Path(self.whisper_binary).exists():
            print(f"[{self.agent_name}] ERROR: whisper-amd not found at {self.whisper_binary}")
            self.whisper_available = False
            return
        
        try:
            result = subprocess.run([self.whisper_binary, "--help"], 
                                  capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                self.whisper_available = True
                print(f"[{self.agent_name}] whisper-amd verified and functional")
            else:
                self.whisper_available = False
                print(f"[{self.agent_name}] WARNING: whisper-amd not responding correctly")
        except Exception as e:
            self.whisper_available = False
            print(f"[{self.agent_name}] ERROR testing whisper-amd: {e}")
        
        # Verificar modelos disponibles
        self.available_models = []
        for name, config in self.models.items():
            if config["file"].exists():
                size_mb = config["file"].stat().st_size // (1024 * 1024)
                self.available_models.append(name)
                print(f"[{self.agent_name}] ✅ Model {name}: {size_mb}MB ({config['use_case']})")
            else:
                print(f"[{self.agent_name}] ❌ Model {name} not found: {config['file']}")

    def transcribe_audio_file(self, 
                            audio_filepath: Path,
                            model: str = "base",
                            language: str = "es",
                            output_formats: list = None) -> dict:
        """
        Transcribe un archivo de audio usando whisper-amd optimizado.
        
        Args:
            audio_filepath (Path): Ruta del archivo de audio
            model (str): Modelo a usar ('tiny', 'base', 'small')
            language (str): Código de idioma ('es', 'en', 'auto')
            output_formats (list): Formatos de salida ['txt', 'srt', 'vtt']
        
        Returns:
            dict: Información de la transcripción y archivos generados
        """
        if not self.whisper_available:
            print(f"[{self.agent_name}] whisper-amd not available")
            return None
            
        if not audio_filepath.exists():
            print(f"[{self.agent_name}] Audio file not found: {audio_filepath}")
            return None
            
        if model not in self.available_models:
            print(f"[{self.agent_name}] Model {model} not available. Using 'base'")
            model = "base" if "base" in self.available_models else self.available_models[0]
        
        # Configuración por defecto
        if output_formats is None:
            output_formats = ['txt', 'srt', 'vtt']
            
        model_config = self.models[model]
        
        print(f"[{self.agent_name}] Starting transcription...")
        print(f"[{self.agent_name}] Audio: {audio_filepath.name}")
        print(f"[{self.agent_name}] Model: {model} ({model_config['speed']})")
        print(f"[{self.agent_name}] Language: {language}")
        print(f"[{self.agent_name}] Formats: {output_formats}")
        
        # Construir comando whisper-amd
        cmd = [
            self.whisper_binary,
            "-m", str(model_config["file"]),
            "-t", str(model_config["threads"]),
            "-p", str(model_config["processors"]),
            "--output-dir", str(self.transcripts_dir)
        ]
        
        # Agregar formatos de salida
        if 'txt' in output_formats:
            cmd.append("--output-txt")
        if 'srt' in output_formats:
            cmd.append("--output-srt")
        if 'vtt' in output_formats:
            cmd.append("--output-vtt")
            
        # Agregar idioma
        if language and language != "auto":
            cmd.extend(["-l", language])
            
        # Agregar archivo de audio
        cmd.append(str(audio_filepath))
        
        try:
            start_time = time.time()
            
            # Ejecutar transcripción
            print(f"[{self.agent_name}] Executing whisper-amd...")
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=600  # 10 minutos máximo
            )
            
            end_time = time.time()
            transcription_time = end_time - start_time
            
            if result.returncode == 0:
                # Recopilar archivos generados
                base_name = audio_filepath.stem
                generated_files = {}
                transcript_text = ""
                
                # Verificar archivos de salida
                for fmt in output_formats:
                    output_file = self.transcripts_dir / f"{base_name}.{fmt}"
                    if output_file.exists():
                        generated_files[fmt] = output_file
                        
                        # Leer texto principal del archivo .txt
                        if fmt == 'txt':
                            with open(output_file, 'r', encoding='utf-8') as f:
                                transcript_text = f.read().strip()
                
                if transcript_text:
                    word_count = len(transcript_text.split())
                    
                    print(f"[{self.agent_name}] ✅ Transcription successful!")
                    print(f"[{self.agent_name}] Duration: {transcription_time:.2f}s")
                    print(f"[{self.agent_name}] Words: {word_count}")
                    print(f"[{self.agent_name}] Files: {list(generated_files.keys())}")
                    
                    return {
                        "success": True,
                        "text": transcript_text,
                        "files": generated_files,
                        "model": model,
                        "language": language,
                        "transcription_time": transcription_time,
                        "word_count": word_count,
                        "audio_file": audio_filepath,
                        "method": "whisper-amd-native"
                    }
                else:
                    print(f"[{self.agent_name}] Transcription completed but no text found")
            else:
                print(f"[{self.agent_name}] Transcription failed:")
                print(f"[{self.agent_name}] Error: {result.stderr}")
                
        except subprocess.TimeoutExpired:
            print(f"[{self.agent_name}] Transcription timeout (10 minutes)")
        except Exception as e:
            print(f"[{self.agent_name}] Error during transcription: {e}")
            
        return {"success": False, "error": "Transcription failed"}

    def transcribe_multiple_files(self, audio_files: list, **kwargs) -> list:
        """
        Transcribe múltiples archivos en lote.
        
        Args:
            audio_files (list): Lista de rutas de archivos de audio
            **kwargs: Argumentos para transcribe_audio_file
            
        Returns:
            list: Lista de resultados de transcripción
        """
        results = []
        total_files = len(audio_files)
        
        print(f"[{self.agent_name}] Starting batch transcription of {total_files} files")
        
        for i, audio_file in enumerate(audio_files, 1):
            print(f"[{self.agent_name}] Processing file {i}/{total_files}: {audio_file.name}")
            
            result = self.transcribe_audio_file(audio_file, **kwargs)
            results.append({
                "file": audio_file,
                "result": result,
                "index": i
            })
            
            if result and result.get("success"):
                print(f"[{self.agent_name}] ✅ File {i} completed successfully")
            else:
                print(f"[{self.agent_name}] ❌ File {i} failed")
        
        successful = sum(1 for r in results if r["result"] and r["result"].get("success"))
        print(f"[{self.agent_name}] Batch completed: {successful}/{total_files} successful")
        
        return results

    def get_model_recommendation(self, audio_duration: float = None, 
                               priority: str = "balanced") -> str:
        """
        Recomienda el mejor modelo basado en duración y prioridad.
        
        Args:
            audio_duration (float): Duración del audio en segundos
            priority (str): 'speed', 'balanced', 'accuracy'
            
        Returns:
            str: Nombre del modelo recomendado
        """
        if priority == "speed":
            return "tiny" if "tiny" in self.available_models else self.available_models[0]
        elif priority == "accuracy":
            return "small" if "small" in self.available_models else "base"
        else:  # balanced
            return "base" if "base" in self.available_models else self.available_models[0]

    def run(self):
        """Ejecución de prueba del TranscriptionAgent."""
        print(f"[{self.agent_name}] Running TranscriptionAgent test...")
        
        # Buscar archivos de audio para probar
        test_audio_dirs = [
            Path('recordings/raw'),
            Path('recordings/live'),
            Path('/tmp')
        ]
        
        test_audio = None
        for audio_dir in test_audio_dirs:
            if audio_dir.exists():
                audio_files = list(audio_dir.glob('*.wav'))
                if audio_files:
                    test_audio = max(audio_files, key=lambda x: x.stat().st_mtime)
                    break
        
        if test_audio:
            print(f"[{self.agent_name}] Testing with: {test_audio.name}")
            
            # Probar con modelo rápido
            result = self.transcribe_audio_file(
                test_audio,
                model="tiny",
                language="es",
                output_formats=['txt', 'srt']
            )
            
            if result and result.get("success"):
                print(f"[{self.agent_name}] Test successful!")
                print(f"[{self.agent_name}] Preview: {result['text'][:100]}...")
            else:
                print(f"[{self.agent_name}] Test failed")
        else:
            print(f"[{self.agent_name}] No audio files found for testing")


if __name__ == "__main__":
    transcription_agent = TranscriptionAgent()
    transcription_agent.run()

🔧 Configuración adicional

# config/transcription_settings.yaml
transcription_settings:
  default_model: "base"
  default_language: "es"
  default_formats: ["txt", "srt", "vtt"]
  
  # Configuración específica por modelo para AMD A4-9125
  model_configs:
    tiny:
      threads: 2
      processors: 1
      timeout: 300
      use_case: "tiempo_real"
    base:
      threads: 2  
      processors: 1
      timeout: 600
      use_case: "produccion"
    small:
      threads: 1
      processors: 1
      timeout: 900
      use_case: "precision"
      
  # Configuración de idiomas
  languages:
    spanish: "es"
    english: "en"
    auto_detect: "auto"

🔄 Refactorización de ProcessingAgent

# agents/processing_agent.py (refactorizado)

from agents.transcription_agent import TranscriptionAgent

class ProcessingAgent(AgentFramework):
    """
    Orquestador del pipeline de procesamiento de audio.
    Coordina RecordingAgent, TranscriptionAgent y AnalysisAgent.
    """
    
    def __init__(self):
        super().__init__("ProcessingAgent")
        
        # Inicializar agentes especializados
        self.transcription_agent = TranscriptionAgent()
        
    def process_audio_file(self, audio_file: Path) -> dict:
        """
        Procesa un archivo de audio completo usando agentes especializados.
        """
        print(f"[{self.agent_name}] Processing audio file: {audio_file.name}")
        
        # 1. Transcripción con agente especializado
        transcription_result = self.transcription_agent.transcribe_audio_file(
            audio_file,
            model="base",
            language="es"
        )
        
        if not transcription_result or not transcription_result.get("success"):
            print(f"[{self.agent_name}] Transcription failed")
            return None
            
        # 2. Aquí irían otros procesamientos (análisis de calidad, etc.)
        
        return {
            "audio_file": audio_file,
            "transcription": transcription_result,
            "processing_method": "specialized_agents"
        }