Upload 3 files

models/__init__.py

@@ -1,4 +1,4 @@
-# models package
-from .ai_effector import AIEffector
-
-__all__ = ["AIEffector"]
+# models package
+from .ai_effector import AIEffector
+
+__all__ = ["AIEffector"]

models/ai_effector.py

@@ -0,0 +1,633 @@
+"""
+AI Effector - DiffVox LLM 기반 이펙트 파라미터 예측
+===================================================
+V4: 근본 원인 해결
+- sigmoid 변환 (delay.feedback, delay.mix, distortion_amount)
+- parametrizations.X.original 키 정규화
+- delay.delay_time은 학습 안됨 → 프리셋 보완
+- 동의어 매핑
+"""
+
+import os
+import json
+import re
+import math
+import torch
+import numpy as np
+from typing import Dict, List, Optional, Any
+from pathlib import Path
+from datetime import datetime
+import warnings
+
+warnings.filterwarnings("ignore")
+
+
+def sigmoid(x: float) -> float:
+    """시그모이드 함수"""
+    try:
+        return 1 / (1 + math.exp(-x))
+    except OverflowError:
+        return 0.0 if x < 0 else 1.0
+
+
+# 기본 파라미터 (모델 로드 실패 시 사용)
+DEFAULT_PARAMETERS = {
+    "eq_peak1.params.freq": 1000.0,
+    "eq_peak1.params.gain": 0.0,
+    "eq_peak1.params.Q": 1.0,
+    "eq_peak2.params.freq": 4000.0,
+    "eq_peak2.params.gain": 0.0,
+    "eq_peak2.params.Q": 1.0,
+    "eq_lowshelf.params.freq": 200.0,
+    "eq_lowshelf.params.gain": 0.0,
+    "eq_highshelf.params.freq": 8000.0,
+    "eq_highshelf.params.gain": 0.0,
+    "distortion_amount": 0.0,
+    "delay.delay_time": 0.02,
+    "delay.feedback": 0.3,
+    "delay.mix": 0.2,
+    "final_wet_mix": 0.5
+}
+
+# 파라미터 범위 제한 (변환 후 실제 값 기준)
+PARAM_RANGES = {
+    "eq_peak1.params.freq": (20.0, 20000.0),
+    "eq_peak1.params.gain": (-12.0, 12.0),
+    "eq_peak1.params.Q": (0.1, 10.0),
+    "eq_peak2.params.freq": (20.0, 20000.0),
+    "eq_peak2.params.gain": (-12.0, 12.0),
+    "eq_peak2.params.Q": (0.1, 10.0),
+    "eq_lowshelf.params.freq": (20.0, 2000.0),
+    "eq_lowshelf.params.gain": (-12.0, 12.0),
+    "eq_highshelf.params.freq": (1000.0, 20000.0),
+    "eq_highshelf.params.gain": (-12.0, 12.0),
+    "distortion_amount": (0.0, 0.1),  # sigmoid * 0.1 후
+    "delay.delay_time": (0.01, 1.0),
+    "delay.feedback": (0.0, 0.9),  # sigmoid 후
+    "delay.mix": (0.0, 1.0),  # sigmoid 후
+    "final_wet_mix": (0.0, 1.0),  # sigmoid 후
+}
+
+# 동의어 매핑 (미학습 단어 → 학습된 단어)
+SYNONYM_MAP = {
+    "calm": "warm soft",
+    "relaxed": "warm soft",
+    "chill": "warm soft",
+    "smooth": "warm",
+    "mellow": "warm soft",
+    "breezy": "bright spacious",
+    "airy": "bright spacious",
+    "light": "bright",
+    "crisp": "bright",
+    "clean": "bright",
+    "dreamy": "warm spacious",
+    "ethereal": "bright spacious",
+    "atmospheric": "spacious",
+    "ambient": "spacious warm",
+    "aggressive": "saturated bright",
+    "powerful": "saturated",
+    "punchy": "saturated bright",
+    "hard": "saturated",
+    "gritty": "saturated dark",
+    "soft": "warm",
+    "harsh": "bright saturated",
+    "muddy": "dark",
+    "thin": "bright",
+    "thick": "warm dark",
+    "full": "warm",
+    "reverb": "spacious",
+    "echo": "spacious",
+    "wet": "spacious",
+}
+
+# 스타일 프리셋 (delay.delay_time 보완용)
+STYLE_PRESETS = {
+    "warm": {
+        "eq_lowshelf.params.gain": 3.0,
+        "eq_highshelf.params.gain": -1.0,
+    },
+    "bright": {
+        "eq_highshelf.params.gain": 4.0,
+        "eq_peak2.params.gain": 2.0,
+        "eq_lowshelf.params.gain": -1.0,
+    },
+    "spacious": {
+        "delay.delay_time": 0.05,  # 학습 안된 파라미터 보완
+    },
+    "dark": {
+        "eq_highshelf.params.gain": -4.0,
+        "eq_lowshelf.params.gain": 2.0,
+    },
+    "saturated": {},
+    "soft": {
+        "eq_highshelf.params.gain": -2.0,
+        "eq_lowshelf.params.gain": 1.0,
+    },
+}
+
+
+class CLAPAudioEncoder:
+    """CLAP 기반 오디오 인코더 (학습 시와 동일)"""
+    
+    def __init__(self, output_dim: int = 64, model_name: str = "laion/larger_clap_music"):
+        self.output_dim = output_dim
+        self.model_name = model_name
+        self.target_sr = 48000
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        
+        self.model = None
+        self.processor = None
+        self._load_model()
+    
+    def _load_model(self):
+        try:
+            from transformers import ClapModel, ClapProcessor
+            
+            print(f"[CLAPEncoder] CLAP 모델 로딩 중: {self.model_name}")
+            
+            self.processor = ClapProcessor.from_pretrained(self.model_name)
+            self.model = ClapModel.from_pretrained(self.model_name)
+            self.model = self.model.to(self.device)
+            self.model.eval()
+            
+            print(f"[CLAPEncoder] ✅ CLAP 모델 로드 완료 (512→{self.output_dim} pooling)")
+            
+        except ImportError:
+            print("[CLAPEncoder] ❌ transformers 미설치")
+        except Exception as e:
+            print(f"[CLAPEncoder] ❌ 모델 로드 실패: {e}")
+    
+    def get_audio_features(self, audio_path: str) -> List[float]:
+        if self.model is None:
+            return [0.0] * self.output_dim
+        
+        try:
+            import librosa
+            
+            audio, sr = librosa.load(audio_path, sr=self.target_sr, mono=True)
+            
+            inputs = self.processor(
+                audios=audio,
+                sampling_rate=self.target_sr,
+                return_tensors="pt",
+                padding=True
+            ).to(self.device)
+            
+            with torch.no_grad():
+                outputs = self.model.get_audio_features(**inputs)
+            
+            features_512 = outputs[0].cpu().numpy()
+            features_64 = self._reduce_dimension(features_512)
+            
+            return features_64.tolist()
+            
+        except Exception as e:
+            print(f"[CLAPEncoder] 특징 추출 실패: {e}")
+            return [0.0] * self.output_dim
+    
+    def _reduce_dimension(self, features: np.ndarray) -> np.ndarray:
+        current_dim = len(features)
+        if current_dim == self.output_dim:
+            return features
+        
+        pool_size = current_dim // self.output_dim
+        remainder = current_dim % self.output_dim
+        
+        pooled = []
+        idx = 0
+        for i in range(self.output_dim):
+            size = pool_size + (1 if i < remainder else 0)
+            pooled.append(np.mean(features[idx:idx+size]))
+            idx += size
+        
+        return np.array(pooled)
+    
+    def is_loaded(self) -> bool:
+        return self.model is not None
+
+
+class AIEffector:
+    """AI 기반 이펙터 파라미터 예측 (V4)"""
+    
+    def __init__(
+        self,
+        model_repo_id: str = "heybaeheef/KU_SW_Academy",
+        model_subfolder: str = "checkpoints",
+        base_model_name: str = "Qwen/Qwen3-8B",
+        audio_feature_dim: int = 64,
+        use_huggingface: bool = True
+    ):
+        self.model_repo_id = model_repo_id
+        self.model_subfolder = model_subfolder
+        self.base_model_name = base_model_name
+        self.audio_feature_dim = audio_feature_dim
+        self.use_huggingface = use_huggingface
+        
+        self.model = None
+        self.tokenizer = None
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        
+        print(f"[AIEffector] CLAP 오디오 인코더 초기화 중...")
+        self.audio_encoder = CLAPAudioEncoder(output_dim=audio_feature_dim)
+        
+        self.request_count = 0
+        self._load_model()
+    
+    def _load_model(self):
+        try:
+            from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
+            from peft import PeftModel
+            
+            print(f"[AIEffector] 베이스 모델 로딩 중: {self.base_model_name}")
+            
+            if torch.cuda.is_available():
+                bnb_config = BitsAndBytesConfig(
+                    load_in_4bit=True,
+                    bnb_4bit_quant_type="nf4",
+                    bnb_4bit_compute_dtype=torch.float16,
+                    bnb_4bit_use_double_quant=True
+                )
+                base_model = AutoModelForCausalLM.from_pretrained(
+                    self.base_model_name,
+                    quantization_config=bnb_config,
+                    device_map="auto",
+                    trust_remote_code=True
+                )
+            else:
+                base_model = AutoModelForCausalLM.from_pretrained(
+                    self.base_model_name,
+                    torch_dtype=torch.float32,
+                    device_map="auto",
+                    trust_remote_code=True
+                )
+            
+            self.tokenizer = AutoTokenizer.from_pretrained(
+                self.base_model_name,
+                trust_remote_code=True
+            )
+            
+            if self.tokenizer.pad_token is None:
+                self.tokenizer.pad_token = self.tokenizer.eos_token
+            
+            print(f"[AIEffector] LoRA 어댑터 로딩 중...")
+            
+            if self.use_huggingface:
+                self.model = PeftModel.from_pretrained(
+                    base_model,
+                    self.model_repo_id,
+                    subfolder=self.model_subfolder,
+                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+                )
+            else:
+                local_path = os.path.join(self.model_repo_id, self.model_subfolder)
+                self.model = PeftModel.from_pretrained(
+                    base_model,
+                    local_path,
+                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+                )
+            
+            self.model.eval()
+            print(f"[AIEffector] ✅ 모델 로드 성공!")
+            
+        except Exception as e:
+            print(f"[AIEffector] ❌ 모델 로드 실패: {e}")
+            import traceback
+            traceback.print_exc()
+            self.model = None
+            self.tokenizer = None
+    
+    def is_loaded(self) -> bool:
+        return self.model is not None
+    
+    def _preprocess_text(self, text: str) -> str:
+        """동의어 매핑"""
+        text_lower = text.lower()
+        for synonym, replacement in SYNONYM_MAP.items():
+            if synonym in text_lower:
+                text_lower = text_lower.replace(synonym, replacement)
+                print(f"    [Synonym] '{synonym}' → '{replacement}'")
+        return text_lower
+    
+    def _apply_preset(self, prompt: str) -> Dict[str, float]:
+        """프리셋 매칭 (delay.delay_time 보완용)"""
+        params = {}
+        prompt_lower = prompt.lower()
+        
+        matched = []
+        for style_name, style_params in STYLE_PRESETS.items():
+            if style_name in prompt_lower:
+                params.update(style_params)
+                matched.append(style_name)
+        
+        if matched:
+            print(f"    [Preset] 매칭: {matched}")
+        
+        return params
+    
+    def _format_prompt(self, text_prompt: str, audio_features: List[float]) -> str:
+        """학습 시와 동일한 프롬프트"""
+        audio_state_str = json.dumps(audio_features)
+        return f"""Task: Convert text to audio parameters.
+Audio: {audio_state_str}
+Text: {text_prompt}
+Parameters:"""
+    
+    def _preprocess_json(self, json_str: str) -> str:
+        """JSON 전처리"""
+        # 숫자 언더스코어 제거 (0.30_299 → 0.30299)
+        json_str = re.sub(r'(\d)_(\d)', r'\1\2', json_str)
+        # Trailing comma 제거
+        json_str = re.sub(r',(\s*[}\]])', r'\1', json_str)
+        # NaN, Infinity
+        json_str = re.sub(r'\bNaN\b', '0', json_str)
+        json_str = re.sub(r'\bInfinity\b', '999999', json_str)
+        json_str = re.sub(r'-Infinity\b', '-999999', json_str)
+        return json_str
+    
+    def _normalize_key(self, key: str) -> str:
+        """
+        파라미터 키 정규화
+        eq_peak1.params.parametrizations.freq.original → eq_peak1.params.freq
+        """
+        # parametrizations.X.original → X
+        key = re.sub(r'\.parametrizations\.(\w+)\.original', r'.\1', key)
+        # Q → Q (대문자 유지)
+        return key
+    
+    def _extract_json_object(self, text: str) -> Optional[str]:
+        """JSON 객체 추출"""
+        start = text.find('{')
+        if start == -1:
+            return None
+        
+        depth = 0
+        for i, char in enumerate(text[start:], start):
+            if char == '{':
+                depth += 1
+            elif char == '}':
+                depth -= 1
+                if depth == 0:
+                    return text[start:i+1]
+        return None
+    
+    def _convert_raw_to_actual(self, params: Dict[str, float]) -> Dict[str, float]:
+        """
+        ★★★ 핵심: 학습 데이터의 raw 값을 실제 값으로 변환 ★★★
+        
+        학습 데이터는 nn.Parameter의 raw 값을 저장함.
+        실제 사용 시 sigmoid 등 변환이 적용됨.
+        """
+        result = params.copy()
+        
+        # 1. delay.feedback: sigmoid 변환
+        if 'delay.feedback' in result:
+            raw = result['delay.feedback']
+            actual = sigmoid(raw)
+            print(f"    [Convert] delay.feedback: {raw:.4f} → sigmoid → {actual:.4f}")
+            result['delay.feedback'] = actual
+        
+        # 2. delay.mix: sigmoid 변환
+        if 'delay.mix' in result:
+            raw = result['delay.mix']
+            actual = sigmoid(raw)
+            print(f"    [Convert] delay.mix: {raw:.4f} → sigmoid → {actual:.4f}")
+            result['delay.mix'] = actual
+        
+        # 3. distortion_amount: sigmoid * 0.1
+        if 'distortion_amount' in result:
+            raw = result['distortion_amount']
+            actual = sigmoid(raw) * 0.1
+            print(f"    [Convert] distortion_amount: {raw:.4f} → sigmoid*0.1 → {actual:.4f}")
+            result['distortion_amount'] = actual
+        
+        # 4. final_wet_mix: sigmoid 변환
+        if 'final_wet_mix' in result:
+            raw = result['final_wet_mix']
+            actual = sigmoid(raw)
+            print(f"    [Convert] final_wet_mix: {raw:.4f} → sigmoid → {actual:.4f}")
+            result['final_wet_mix'] = actual
+        
+        return result
+    
+    def _clamp_values(self, params: Dict[str, float]) -> Dict[str, float]:
+        """값 범위 제한"""
+        result = params.copy()
+        
+        for key, (min_val, max_val) in PARAM_RANGES.items():
+            if key in result:
+                original = result[key]
+                clamped = max(min_val, min(max_val, original))
+                if clamped != original:
+                    print(f"    [Clamp] {key}: {original:.4f} → {clamped:.4f}")
+                result[key] = clamped
+        
+        return result
+    
+    def _parse_output(self, output_text: str) -> Dict[str, float]:
+        """LLM 출력 파싱"""
+        
+        print(f"    [Parse] Raw output 길이: {len(output_text)} 문자")
+        
+        json_str = None
+        
+        try:
+            text = output_text
+            
+            # <think> 태그 제거
+            text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
+            
+            # 코드블록 추출
+            code_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
+            if code_match:
+                text = code_match.group(1)
+            
+            # JSON 추출
+            json_str = self._extract_json_object(text)
+            
+            if json_str:
+                print(f"    [Parse] JSON 발견 (길이: {len(json_str)})")
+                
+                # 전처리
+                json_str = self._preprocess_json(json_str)
+                
+                # 파싱
+                raw_params = json.loads(json_str)
+                
+                # 결과 매핑
+                result = DEFAULT_PARAMETERS.copy()
+                parsed_count = 0
+                
+                for key, value in raw_params.items():
+                    try:
+                        # 키 정규화
+                        norm_key = self._normalize_key(key)
+                        float_val = float(value)
+                        
+                        # 매칭되는 기본 키 찾기
+                        matched_key = None
+                        for default_key in DEFAULT_PARAMETERS.keys():
+                            # 정확한 매칭
+                            if norm_key == default_key:
+                                matched_key = default_key
+                                break
+                            # 부분 매칭 (키 끝부분)
+                            if norm_key.endswith(default_key.split('.')[-1]) and \
+                               norm_key.split('.')[0] == default_key.split('.')[0]:
+                                matched_key = default_key
+                                break
+                        
+                        if matched_key:
+                            result[matched_key] = float_val
+                            parsed_count += 1
+                        else:
+                            print(f"    [Parse] 매칭 안됨: {key} → {norm_key}")
+                            
+                    except (ValueError, TypeError) as e:
+                        print(f"    [Parse] 변환 실패: {key}={value} ({e})")
+                
+                print(f"    [Parse] ✅ {parsed_count}개 파라미터 매핑됨")
+                return result
+                
+        except json.JSONDecodeError as e:
+            print(f"    [Parse] ❌ JSON 에러: {e}")
+            if json_str:
+                pos = getattr(e, 'pos', 0)
+                print(f"    [Parse] 위치: ...{json_str[max(0,pos-20):pos+20]}...")
+        except Exception as e:
+            print(f"    [Parse] ❌ 예외: {e}")
+        
+        print(f"    [Parse] ⚠️ 기본값 폴백")
+        return DEFAULT_PARAMETERS.copy()
+    
+    def predict(self, audio_path: str, text_prompt: str = "") -> Dict[str, float]:
+        """파라미터 예측"""
+        
+        self.request_count += 1
+        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        
+        print(f"\n{'='*60}")
+        print(f"[AIEffector V4] 🎵 요청 #{self.request_count} - {timestamp}")
+        print(f"{'='*60}")
+        print(f"  📂 오디오: {Path(audio_path).name}")
+        print(f"  💬 원본 프롬프트: '{text_prompt}'")
+        
+        # 동의어 변환
+        processed_prompt = self._preprocess_text(text_prompt)
+        if processed_prompt != text_prompt.lower():
+            print(f"  💬 변환 프롬프트: '{processed_prompt}'")
+        
+        print(f"  🤖 모델: {'AI' if self.is_loaded() else '프리셋'}")
+        
+        # 모델 없으면 프리셋
+        if not self.is_loaded():
+            print(f"\n  ⚠️ AI 모델 미로드")
+            params = DEFAULT_PARAMETERS.copy()
+            params.update(self._apply_preset(processed_prompt))
+            self._log_parameters(params)
+            return self._convert_to_effect_chain_format(params)
+        
+        try:
+            # 1. CLAP 특징 추출
+            print(f"\n  📊 [Step 1] CLAP 특징 추출...")
+            audio_features = self.audio_encoder.get_audio_features(audio_path)
+            
+            if not audio_features or all(f == 0 for f in audio_features):
+                print(f"    ⚠️ 실패, 프리셋 폴백")
+                params = DEFAULT_PARAMETERS.copy()
+                params.update(self._apply_preset(processed_prompt))
+                self._log_parameters(params)
+                return self._convert_to_effect_chain_format(params)
+            
+            print(f"    ✅ {len(audio_features)}차원")
+            
+            # 2. 프롬프트 생성
+            print(f"\n  🔤 [Step 2] 프롬프트 생성...")
+            prompt = self._format_prompt(processed_prompt, audio_features)
+            
+            # 3. 토큰화
+            print(f"\n  🔢 [Step 3] 토큰화...")
+            inputs = self.tokenizer(
+                prompt,
+                return_tensors="pt",
+                truncation=True,
+                max_length=1500
+            ).to(self.device)
+            print(f"    토큰 수: {inputs['input_ids'].shape[1]}")
+            
+            # 4. LLM 생성
+            print(f"\n  🧠 [Step 4] LLM 추론...")
+            import time
+            start = time.time()
+            
+            with torch.no_grad():
+                outputs = self.model.generate(
+                    **inputs,
+                    max_new_tokens=500,
+                    do_sample=False,
+                    temperature=0.1,
+                    pad_token_id=self.tokenizer.pad_token_id,
+                    eos_token_id=self.tokenizer.eos_token_id,
+                )
+            
+            print(f"    추론 시간: {time.time()-start:.2f}초")
+            
+            # 5. 디코딩
+            print(f"\n  📝 [Step 5] 디코딩...")
+            gen_tokens = outputs[0][inputs['input_ids'].shape[1]:]
+            output_text = self.tokenizer.decode(gen_tokens, skip_special_tokens=True).strip()
+            print(f"    출력 (처음 400자):\n{output_text[:400]}")
+            
+            # 6. 파싱
+            print(f"\n  🔧 [Step 6] 파싱...")
+            raw_params = self._parse_output(output_text)
+            
+            # 7. ★★★ Raw → Actual 변환 ★★★
+            print(f"\n  🔄 [Step 7] Raw → Actual 변환...")
+            actual_params = self._convert_raw_to_actual(raw_params)
+            
+            # 8. 값 클램핑
+            print(f"\n  📐 [Step 8] 값 클램핑...")
+            clamped_params = self._clamp_values(actual_params)
+            
+            # 9. 프리셋 보완 (delay.delay_time은 학습 안됨)
+            print(f"\n  🎛️ [Step 9] 프리셋 보완...")
+            preset = self._apply_preset(processed_prompt)
+            if 'delay.delay_time' in preset:
+                clamped_params['delay.delay_time'] = preset['delay.delay_time']
+                print(f"    delay.delay_time: {preset['delay.delay_time']} (프리셋)")
+            
+            # 10. 로깅
+            self._log_parameters(clamped_params)
+            
+            print(f"\n  ✅ 완료!")
+            print(f"{'='*60}\n")
+            
+            return self._convert_to_effect_chain_format(clamped_params)
+            
+        except Exception as e:
+            print(f"\n  ❌ 실패: {e}")
+            import traceback
+            traceback.print_exc()
+            params = DEFAULT_PARAMETERS.copy()
+            params.update(self._apply_preset(processed_prompt))
+            self._log_parameters(params)
+            return self._convert_to_effect_chain_format(params)
+    
+    def _convert_to_effect_chain_format(self, params: Dict[str, float]) -> Dict[str, float]:
+        """effect_chain.py 형식으로 변환 (Q → q)"""
+        result = {}
+        for key, value in params.items():
+            new_key = key.replace('.Q', '.q')
+            result[new_key] = value
+        return result
+    
+    def _log_parameters(self, params: Dict[str, float]):
+        """파라미터 로깅"""
+        print(f"\n  📋 최종 파라미터:")
+        print(f"    [EQ Peak 1] freq={params.get('eq_peak1.params.freq',0):.0f}Hz, gain={params.get('eq_peak1.params.gain',0):.2f}dB")
+        print(f"    [EQ Peak 2] freq={params.get('eq_peak2.params.freq',0):.0f}Hz, gain={params.get('eq_peak2.params.gain',0):.2f}dB")
+        print(f"    [Low Shelf] gain={params.get('eq_lowshelf.params.gain',0):.2f}dB")
+        print(f"    [High Shelf] gain={params.get('eq_highshelf.params.gain',0):.2f}dB")
+        print(f"    [Distortion] {params.get('distortion_amount',0):.4f}")
+        print(f"    [Delay] time={params.get('delay.delay_time',0):.3f}s, fb={params.get('delay.feedback',0):.2f}, mix={params.get('delay.mix',0):.2f}")
+        print(f"    [Wet Mix] {params.get('final_wet_mix',0):.2f}")

models/audio_encoder.py

@@ -1,189 +1,189 @@
-"""
-Audio Encoder for MagicPath Server
-===================================
-CLAP 모델을 사용하여 오디오 파일에서 특징 벡터 추출
-DiffVox LLM과 동일한 인코더 사용
-"""
-
-import torch
-import numpy as np
-from typing import List, Optional
-import warnings
-
-warnings.filterwarnings("ignore")
-
-
-class AudioEncoder:
-    """CLAP 기반 오디오 인코더"""
-    
-    def __init__(
-        self, 
-        output_dim: int = 64, 
-        reduction_method: str = "pool",
-        model_name: str = "laion/larger_clap_general"
-    ):
-        """
-        오디오 인코더 초기화
-        
-        Args:
-            output_dim: 출력 특징 차원 (기본 64)
-            reduction_method: 차원 축소 방법 ("pool", "pca", "linear")
-            model_name: CLAP 모델 이름
-        """
-        self.output_dim = output_dim
-        self.reduction_method = reduction_method
-        self.model_name = model_name
-        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        
-        self.model = None
-        self.processor = None
-        self.projection = None
-        
-        self._load_model()
-    
-    def _load_model(self):
-        """CLAP 모델 로드"""
-        try:
-            from transformers import ClapModel, ClapProcessor
-            
-            print(f"[AudioEncoder] CLAP 모델 로딩 중: {self.model_name}")
-            
-            self.processor = ClapProcessor.from_pretrained(self.model_name)
-            self.model = ClapModel.from_pretrained(self.model_name)
-            self.model = self.model.to(self.device)
-            self.model.eval()
-            
-            # CLAP 출력 차원 확인 (보통 512)
-            clap_dim = self.model.config.projection_dim
-            print(f"[AudioEncoder] CLAP 출력 차원: {clap_dim}")
-            
-            # 차원 축소를 위한 projection layer
-            if self.reduction_method == "linear" and clap_dim != self.output_dim:
-                self.projection = torch.nn.Linear(clap_dim, self.output_dim)
-                self.projection = self.projection.to(self.device)
-                print(f"[AudioEncoder] Linear projection: {clap_dim} → {self.output_dim}")
-            
-            print("[AudioEncoder] ✅ 모델 로드 완료")
-            
-        except ImportError:
-            print("[AudioEncoder] ❌ transformers 미설치")
-            print("   pip install transformers")
-        except Exception as e:
-            print(f"[AudioEncoder] ❌ 모델 로드 실패: {e}")
-    
-    def get_audio_features(self, audio_path: str) -> List[float]:
-        """
-        오디오 파일에서 특징 벡터 추출
-        
-        Args:
-            audio_path: 오디오 파일 경로
-            
-        Returns:
-            특징 벡터 (output_dim 차원)
-        """
-        if self.model is None:
-            print("[AudioEncoder] 모델이 로드되지 않음")
-            return []
-        
-        try:
-            import librosa
-            
-            # 오디오 로드
-            audio, sr = librosa.load(audio_path, sr=48000, mono=True)
-            
-            # CLAP 입력 준비
-            inputs = self.processor(
-                audios=audio,
-                sampling_rate=48000,
-                return_tensors="pt"
-            )
-            inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            # 특징 추출
-            with torch.no_grad():
-                audio_features = self.model.get_audio_features(**inputs)
-            
-            # CPU로 이동
-            features = audio_features.squeeze().cpu().numpy()
-            
-            # 차원 축소
-            features = self._reduce_dimension(features)
-            
-            return features.tolist()
-            
-        except Exception as e:
-            print(f"[AudioEncoder] 특징 추출 실패: {e}")
-            import traceback
-            traceback.print_exc()
-            return []
-    
-    def _reduce_dimension(self, features: np.ndarray) -> np.ndarray:
-        """특징 벡터 차원 축소"""
-        current_dim = len(features)
-        
-        if current_dim == self.output_dim:
-            return features
-        
-        if self.reduction_method == "pool":
-            # 평균 풀링으로 차원 축소
-            if current_dim > self.output_dim:
-                pool_size = current_dim // self.output_dim
-                remainder = current_dim % self.output_dim
-                
-                pooled = []
-                idx = 0
-                for i in range(self.output_dim):
-                    size = pool_size + (1 if i < remainder else 0)
-                    pooled.append(np.mean(features[idx:idx+size]))
-                    idx += size
-                
-                return np.array(pooled)
-            else:
-                # 차원이 작으면 zero-padding
-                padded = np.zeros(self.output_dim)
-                padded[:current_dim] = features
-                return padded
-        
-        elif self.reduction_method == "linear" and self.projection is not None:
-            # Linear projection
-            with torch.no_grad():
-                features_tensor = torch.tensor(features, dtype=torch.float32).to(self.device)
-                projected = self.projection(features_tensor)
-                return projected.cpu().numpy()
-        
-        else:
-            # 기본: 앞에서부터 자르기
-            return features[:self.output_dim]
-    
-    def get_text_features(self, text: str) -> List[float]:
-        """
-        텍스트에서 특징 벡터 추출 (CLAP text encoder)
-        
-        Args:
-            text: 입력 텍스트
-            
-        Returns:
-            특징 벡터
-        """
-        if self.model is None:
-            return []
-        
-        try:
-            inputs = self.processor(
-                text=text,
-                return_tensors="pt",
-                padding=True
-            )
-            inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            with torch.no_grad():
-                text_features = self.model.get_text_features(**inputs)
-            
-            features = text_features.squeeze().cpu().numpy()
-            features = self._reduce_dimension(features)
-            
-            return features.tolist()
-            
-        except Exception as e:
-            print(f"[AudioEncoder] 텍스트 특징 추출 실패: {e}")
-            return []
+"""
+Audio Encoder for MagicPath Server
+===================================
+CLAP 모델을 사용하여 오디오 파일에서 특징 벡터 추출
+DiffVox LLM과 동일한 인코더 사용
+"""
+
+import torch
+import numpy as np
+from typing import List, Optional
+import warnings
+
+warnings.filterwarnings("ignore")
+
+
+class AudioEncoder:
+    """CLAP 기반 오디오 인코더"""
+    
+    def __init__(
+        self, 
+        output_dim: int = 64, 
+        reduction_method: str = "pool",
+        model_name: str = "laion/larger_clap_general"
+    ):
+        """
+        오디오 인코더 초기화
+        
+        Args:
+            output_dim: 출력 특징 차원 (기본 64)
+            reduction_method: 차원 축소 방법 ("pool", "pca", "linear")
+            model_name: CLAP 모델 이름
+        """
+        self.output_dim = output_dim
+        self.reduction_method = reduction_method
+        self.model_name = model_name
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        
+        self.model = None
+        self.processor = None
+        self.projection = None
+        
+        self._load_model()
+    
+    def _load_model(self):
+        """CLAP 모델 로드"""
+        try:
+            from transformers import ClapModel, ClapProcessor
+            
+            print(f"[AudioEncoder] CLAP 모델 로딩 중: {self.model_name}")
+            
+            self.processor = ClapProcessor.from_pretrained(self.model_name)
+            self.model = ClapModel.from_pretrained(self.model_name)
+            self.model = self.model.to(self.device)
+            self.model.eval()
+            
+            # CLAP 출력 차원 확인 (보통 512)
+            clap_dim = self.model.config.projection_dim
+            print(f"[AudioEncoder] CLAP 출력 차원: {clap_dim}")
+            
+            # 차원 축소를 위한 projection layer
+            if self.reduction_method == "linear" and clap_dim != self.output_dim:
+                self.projection = torch.nn.Linear(clap_dim, self.output_dim)
+                self.projection = self.projection.to(self.device)
+                print(f"[AudioEncoder] Linear projection: {clap_dim} → {self.output_dim}")
+            
+            print("[AudioEncoder] ✅ 모델 로드 완료")
+            
+        except ImportError:
+            print("[AudioEncoder] ❌ transformers 미설치")
+            print("   pip install transformers")
+        except Exception as e:
+            print(f"[AudioEncoder] ❌ 모델 로드 실패: {e}")
+    
+    def get_audio_features(self, audio_path: str) -> List[float]:
+        """
+        오디오 파일에서 특징 벡터 추출
+        
+        Args:
+            audio_path: 오디오 파일 경로
+            
+        Returns:
+            특징 벡터 (output_dim 차원)
+        """
+        if self.model is None:
+            print("[AudioEncoder] 모델이 로드되지 않음")
+            return []
+        
+        try:
+            import librosa
+            
+            # 오디오 로드
+            audio, sr = librosa.load(audio_path, sr=48000, mono=True)
+            
+            # CLAP 입력 준비
+            inputs = self.processor(
+                audios=audio,
+                sampling_rate=48000,
+                return_tensors="pt"
+            )
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+            
+            # 특징 추출
+            with torch.no_grad():
+                audio_features = self.model.get_audio_features(**inputs)
+            
+            # CPU로 이동
+            features = audio_features.squeeze().cpu().numpy()
+            
+            # 차원 축소
+            features = self._reduce_dimension(features)
+            
+            return features.tolist()
+            
+        except Exception as e:
+            print(f"[AudioEncoder] 특징 추출 실패: {e}")
+            import traceback
+            traceback.print_exc()
+            return []
+    
+    def _reduce_dimension(self, features: np.ndarray) -> np.ndarray:
+        """특징 벡터 차원 축소"""
+        current_dim = len(features)
+        
+        if current_dim == self.output_dim:
+            return features
+        
+        if self.reduction_method == "pool":
+            # 평균 풀링으로 차원 축소
+            if current_dim > self.output_dim:
+                pool_size = current_dim // self.output_dim
+                remainder = current_dim % self.output_dim
+                
+                pooled = []
+                idx = 0
+                for i in range(self.output_dim):
+                    size = pool_size + (1 if i < remainder else 0)
+                    pooled.append(np.mean(features[idx:idx+size]))
+                    idx += size
+                
+                return np.array(pooled)
+            else:
+                # 차원이 작으면 zero-padding
+                padded = np.zeros(self.output_dim)
+                padded[:current_dim] = features
+                return padded
+        
+        elif self.reduction_method == "linear" and self.projection is not None:
+            # Linear projection
+            with torch.no_grad():
+                features_tensor = torch.tensor(features, dtype=torch.float32).to(self.device)
+                projected = self.projection(features_tensor)
+                return projected.cpu().numpy()
+        
+        else:
+            # 기본: 앞에서부터 자르기
+            return features[:self.output_dim]
+    
+    def get_text_features(self, text: str) -> List[float]:
+        """
+        텍스트에서 특징 벡터 추출 (CLAP text encoder)
+        
+        Args:
+            text: 입력 텍스트
+            
+        Returns:
+            특징 벡터
+        """
+        if self.model is None:
+            return []
+        
+        try:
+            inputs = self.processor(
+                text=text,
+                return_tensors="pt",
+                padding=True
+            )
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+            
+            with torch.no_grad():
+                text_features = self.model.get_text_features(**inputs)
+            
+            features = text_features.squeeze().cpu().numpy()
+            features = self._reduce_dimension(features)
+            
+            return features.tolist()
+            
+        except Exception as e:
+            print(f"[AudioEncoder] 텍스트 특징 추출 실패: {e}")
+            return []