music_llm_agent.py

import os
import json
import numpy as np
import librosa
import warnings
import pandas as pd
from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from huggingface_hub import InferenceClient

warnings.filterwarnings("ignore")

# Enhanced LLM support with multiple providers
def generate_music_insight(analysis_text, provider="groq", model="llama3-70b-8192"):
    try:
        prompt_template = PromptTemplate(
            input_variables=["analysis_text"],
            template="""
            You're an AI music expert. Based on the musical analysis below, provide insight on:
            - Genre and mood
            - Instrumentation and music type
            - Suggestions for improvement
            - Commercial/viral potential
            - Suitable platforms or audience
            Musical Analysis: {analysis_text}
            Answer in a concise paragraph.
            """
        )
        final_prompt = prompt_template.format(analysis_text=analysis_text)
        
        # Support for multiple LLM providers
        if provider.lower() == "groq":
            llm = ChatGroq(
                model_name=model, 
                groq_api_key=os.getenv("GROQ_API_KEY", "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")
            )
        elif provider.lower() == "openai":
            llm = ChatOpenAI(
                model_name=model,
                openai_api_key=os.getenv("OPENAI_API_KEY", "sk-demo-key")
            )
        elif provider.lower() == "huggingface":
            # For HuggingFace models, we'll use their Inference API
            client = InferenceClient(
                model=model,
                token=os.getenv("HF_API_TOKEN", "hf_demo_token")
            )
            return client.text_generation(
                final_prompt,
                max_new_tokens=512,
                temperature=0.7,
                return_full_text=False
            )
        else:
            raise ValueError(f"Unsupported provider: {provider}")
            
        return llm.invoke(final_prompt)
    except Exception as e:
        return f"LLM Error: {str(e)}"

# AudioFeatureExtractor (unchanged)
class AudioFeatureExtractor:
    def __init__(self, file_path):
        self.file_path = file_path
        self.y = None
        self.sr = None
        self.features = {}

    def load_audio(self):
        try:
            self.y, self.sr = librosa.load(self.file_path, sr=None)
            return True
        except Exception as e:
            print("Audio loading error:", e)
            return False

    def extract_basic_features(self):
        duration = len(self.y) / self.sr
        tempo, beat_frames = librosa.beat.beat_track(y=self.y, sr=self.sr)
        rms = librosa.feature.rms(y=self.y)[0]
        zcr = librosa.feature.zero_crossing_rate(self.y)[0]
        y_harmonic, y_percussive = librosa.effects.hpss(self.y)
        
        self.features['basic'] = {
            'duration': float(duration),
            'tempo': float(tempo),
            'num_beats': len(beat_frames),
            'avg_rms_energy': float(np.mean(rms)),
            'avg_zero_crossing_rate': float(np.mean(zcr)),
            'harmonic_percussive_ratio': float(np.sum(np.abs(y_harmonic)) / (np.sum(np.abs(y_percussive)) + 1e-10))
        }

    def extract_spectral_features(self):
        centroid = librosa.feature.spectral_centroid(y=self.y, sr=self.sr)[0]
        contrast = librosa.feature.spectral_contrast(y=self.y, sr=self.sr)
        rolloff = librosa.feature.spectral_rolloff(y=self.y, sr=self.sr)[0]
        mfccs = librosa.feature.mfcc(y=self.y, sr=self.sr, n_mfcc=13)
        
        self.features['spectral'] = {
            'avg_spectral_centroid': float(np.mean(centroid)),
            'avg_spectral_contrast': [float(np.mean(band)) for band in contrast],
            'avg_spectral_rolloff': float(np.mean(rolloff)),
            'avg_mfccs': [float(np.mean(m)) for m in mfccs]
        }

    def extract_harmonic_features(self):
        chroma = librosa.feature.chroma_stft(y=self.y, sr=self.sr)
        tonnetz = librosa.feature.tonnetz(y=self.y, sr=self.sr)
        chroma_avg = np.mean(librosa.feature.chroma_cqt(y=self.y, sr=self.sr), axis=1)
        key_names = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
        
        self.features['harmonic'] = {
            'chroma_energy': [float(np.mean(c)) for c in chroma],
            'tonnetz_features': [float(np.mean(t)) for t in tonnetz],
            'estimated_key': key_names[np.argmax(chroma_avg)],
            'key_strength': float(np.max(chroma_avg) / (np.sum(chroma_avg) + 1e-10))
        }

    def detect_instruments(self):
        perc_ratio = self.features['basic']['harmonic_percussive_ratio']
        centroid = self.features['spectral']['avg_spectral_centroid']
        mfccs = self.features['spectral']['avg_mfccs']
        
        instruments = []
        if perc_ratio < 0.8:
            instruments.append("Drums/Percussion")
        if centroid < 1000:
            instruments.append("Bass")
        if 1000 <= centroid <= 3000:
            instruments.append("Guitar/Piano")
        if centroid > 3000:
            instruments.append("High-pitched instruments")
        if 1500 < centroid < 4000 and abs(mfccs[2]) > 5:
            instruments.append("Vocals")
            
        self.features['detected_instruments'] = instruments

    def analyze_mood(self):
        tempo = self.features['basic']['tempo']
        energy = self.features['basic']['avg_rms_energy']
        centroid = self.features['spectral']['avg_spectral_centroid']
        
        moods = []
        if tempo < 80:
            moods.append("Slow/Relaxed")
        elif tempo <= 120:
            moods.append("Moderate/Balanced")
        else:
            moods.append("Fast/Energetic")
            
        if energy < 0.1:
            moods.append("Calm")
        elif energy > 0.2:
            moods.append("Intense")
            
        if centroid < 1500:
            moods.append("Dark/Warm")
        elif centroid > 3000:
            moods.append("Bright/Sharp")
            
        self.features['mood_indicators'] = moods

    def extract_all_features(self):
        if not self.load_audio():
            return False
        self.extract_basic_features()
        self.extract_spectral_features()
        self.extract_harmonic_features()
        self.detect_instruments()
        self.analyze_mood()
        return True

    def to_json(self):
        summary = {
            "file_name": os.path.basename(self.file_path),
            "duration": self.features['basic']['duration'],
            "tempo": self.features['basic']['tempo'],
            "key": self.features['harmonic']['estimated_key'],
            "energy": self.features['basic']['avg_rms_energy'],
            "detected_instruments": self.features['detected_instruments'],
            "mood_indicators": self.features['mood_indicators'],
            "spectral_centroid": self.features['spectral']['avg_spectral_centroid'],
            "harmonic_percussive_ratio": self.features['basic']['harmonic_percussive_ratio']
        }
        return json.dumps(summary, indent=2)

class MusicAnalysisAgent:
    def __init__(self, model="llama3-70b-8192", provider="groq"):
        self.model = model
        self.provider = provider
        
        # Configure LLM based on provider
        if provider.lower() == "groq":
            groq_api_key = os.getenv("GROQ_API_KEY", "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")
            self.llm = ChatGroq(model_name=model, groq_api_key=groq_api_key)
        elif provider.lower() == "openai":
            openai_api_key = os.getenv("OPENAI_API_KEY", "sk-demo-key")
            self.llm = ChatOpenAI(model_name=model, openai_api_key=openai_api_key)
        elif provider.lower() == "huggingface":
            # For Hugging Face, we'll initialize during the chain execution
            self.hf_client = InferenceClient(
                model=model,
                token=os.getenv("HF_API_TOKEN", "hf_demo_token")
            )
            self.llm = None
        else:
            raise ValueError(f"Unsupported provider: {provider}")
    
    def _run_chain(self, template, features):
        prompt = PromptTemplate.from_template(template)
        
        if self.provider.lower() == "huggingface":
            # For Hugging Face, use direct inference
            try:
                full_prompt = prompt.format(features=features)
                return self.hf_client.text_generation(
                    full_prompt,
                    max_new_tokens=512,
                    temperature=0.7,
                    return_full_text=False
                )
            except Exception as e:
                print(f"HuggingFace inference error: {str(e)}")
                return f"Error with HuggingFace inference: {str(e)}"
        else:
            # For other providers, use LangChain
            chain = prompt | self.llm | StrOutputParser()
            try:
                return chain.invoke({"features": features})
            except Exception as e:
                print(f"Chain execution error: {str(e)}")
                return f"Error in LLM chain: {str(e)}"

    def analyze_song_features(self, features):
        try:
            return self._run_chain("""
            You are a music producer. Analyze: {features}
            Discuss sound profile, genre, emotional tone, and how features interact.
            """, features)
        except Exception as e:
            print(f"Song feature analysis error: {str(e)}")
            return f"Song analysis error. Please check the {self.provider} API connection."

    def get_song_improvement_suggestions(self, features):
        try:
            return self._run_chain("""
            Suggest improvements in production, instrumentation, mix, structure, and effects: {features}
            """, features)
        except Exception as e:
            print(f"Improvement suggestions error: {str(e)}")
            return f"Improvement suggestions unavailable. Please check the {self.provider} API connection."

    def assess_workout_playlist_fit(self, features):
        try:
            return self._run_chain("""
            Evaluate if suitable for workout playlists. Consider tempo, energy, emotion, instruments, context: {features}
            """, features)
        except Exception as e:
            print(f"Workout playlist assessment error: {str(e)}")
            return f"Workout playlist assessment unavailable. Please check the {self.provider} API connection."

    def suggest_marketing_channels(self, features):
        try:
            return self._run_chain("""
            Recommend marketing channels: platforms, social media, audience targeting, playlist strategy: {features}
            """, features)
        except Exception as e:
            print(f"Marketing channels suggestion error: {str(e)}")
            return f"Marketing channel suggestions unavailable. Please check the {self.provider} API connection."

    def recommend_genre_classification(self, features):
        try:
            return self._run_chain("""
            Suggest a musical genre based on the features below. Be specific and explain: {features}
            """, features)
        except Exception as e:
            print(f"Genre classification error: {str(e)}")
            return f"Genre classification unavailable. Please check the {self.provider} API connection."

    def recommend_music_category(self, features):
        try:
            return self._run_chain("""
            Classify this music into a category like Chill, Romantic, Party, Study, Sad, Workout, etc. {features}
            """, features)
        except Exception as e:
            print(f"Music category recommendation error: {str(e)}")
            return f"Music category classification unavailable. Please check the {self.provider} API connection."

    # New enhanced methods for more specific music analysis
    def analyze_lyric_improvement(self, features, current_lyrics=None):
        """Suggests improvements for song lyrics based on the audio analysis"""
        try:
            return self._run_chain("""
            Based on this audio analysis: {features}
            
            Provide recommendations for lyrical content and style that would complement 
            the musical qualities. Consider the mood, tempo, and emotional tone.
            """+ (f"\n\nCurrent lyrics: {current_lyrics}" if current_lyrics else ""), features)
        except Exception as e:
            print(f"Lyric improvement analysis error: {str(e)}")
            return f"Lyric improvement suggestions unavailable. Please check the {self.provider} API connection."
            
    def analyze_commercial_potential(self, features):
        """Analyzes commercial potential of the track"""
        try:
            return self._run_chain("""
            Analyze the commercial potential of this track based on: {features}
            
            Consider:
            1. Current music market trends
            2. Similar successful tracks
            3. Target audience size and engagement
            4. Streaming potential
            5. Licensing opportunities
            
            Provide a detailed assessment with specific recommendations.
            """, features)
        except Exception as e:
            print(f"Commercial potential analysis error: {str(e)}")
            return f"Commercial potential analysis unavailable. Please check the {self.provider} API connection."