Auto commit at 22-2025-08 4:00:04

README.md

@@ -10,6 +10,9 @@ app_file: app.py
 pinned: false
 ---
 
+# 250822
+- polyglot 1.3b lora 파라메터 조절, 응답 품질 향상
+
 # 250821
 - polyplot 5.8b 응답 속도 개선, 모델별 tokenizer config settings json 변수 명시적으로 기입
 

lily_llm_api/models/polyglot_ko_1_3b_chat_250822_0312.py

@@ -1,280 +0,0 @@
-#!/usr/bin/env python3
-"""
-Polyglot-ko-1.3b-chat 모델 프로필
-heegyu/polyglot-ko-1.3b-chat 모델용
-"""
-
-from typing import Dict, Any, Tuple
-import torch
-from transformers import AutoTokenizer, AutoModelForCausalLM
-import logging
-import os
-from pathlib import Path
-import re
-
-
-HF_TOKEN = os.getenv("HF_TOKEN")
-
-logger = logging.getLogger(__name__)
-
-class PolyglotKo13bChatProfile:
-    """Polyglot-ko-1.3b-chat 모델 프로필"""
-    
-    def __init__(self):
-        self.model_name = "heegyu/polyglot-ko-1.3b-chat"
-        self.local_path = "./lily_llm_core/models/polyglot_ko_1_3b_chat"
-        self.display_name = "Polyglot-ko-1.3b-chat"
-        self.description = "한국어 채팅 전용 경량 모델 (1.3B)"
-        self.language = "ko"
-        self.model_size = "1.3B"
-        
-    def load_model(self) -> Tuple[AutoModelForCausalLM, AutoTokenizer]:
-        """모델 로드 (토크나이저 설정 수정)"""
-        logger.info(f"📥 {self.display_name} 모델 로드 중...")
-        try:
-            use_local = Path(self.local_path).exists() and any(Path(self.local_path).iterdir())
-            model_path = self.local_path if use_local else self.model_name
-            logger.info(f"🔍 모델 경로: {model_path} (local={'yes' if use_local else 'no'})")
-
-            # 강제로 Hugging Face에서 다운로드 (로컬 모델 문제 해결)            
-            # use_local = False
-            # model_path = self.model_name
-            # logger.info(f"🔍 모델 경로: {model_path} (local=no - 강제 HF 다운로드)")                                            
-
-            tokenizer = AutoTokenizer.from_pretrained(
-                model_path,
-                token=HF_TOKEN, 
-                use_fast=True,
-                trust_remote_code=True,
-                local_files_only=use_local,
-            )
-            
-            # 토크나이저 설정 수정 - EOS 토큰 문제 해결
-            if tokenizer.eos_token is None:
-                logger.warning("⚠️ EOS 토큰이 없습니다. 모델 공식 문서에 따라 <|endoftext|> 설정")
-                tokenizer.eos_token = "<|endoftext|>"
-            
-            if tokenizer.pad_token is None:
-                logger.warning("⚠️ PAD 토큰이 없습니다. EOS 토큰으로 설정")
-                tokenizer.pad_token = tokenizer.eos_token
-            
-            # 특수 토큰 확인
-            logger.info(f"🔍 토크나이저 설정:")
-            logger.info(f"  - EOS 토큰: {tokenizer.eos_token} (ID: {tokenizer.eos_token_id})")
-            logger.info(f"  - PAD 토큰: {tokenizer.pad_token} (ID: {tokenizer.pad_token_id})")
-            # logger.info(f"  - BOS 토큰: {tokenizer.bos_token} (ID: {tokenizer.bos_token_id})")
-
-            # CPU에서는 float32가 더 안정적, CUDA에서는 float16 사용
-            device = 'cuda' if torch.cuda.is_available() else 'cpu'
-            selected_dtype = torch.float16 if device == 'cuda' else torch.float32
-
-            model = AutoModelForCausalLM.from_pretrained(
-                model_path,
-                token=HF_TOKEN,
-                trust_remote_code=True,
-                torch_dtype=selected_dtype,
-                local_files_only=use_local,
-            ).to(device)
-
-            logger.info(f"✅ {self.display_name} 모델 로드 성공! (device={device}, dtype={selected_dtype})")
-            return model, tokenizer
-        except Exception as e:
-            logger.error(f"❌ {self.display_name} 모델 로드 실패: {e}")
-            raise
-    
-    def format_prompt(self, user_input: str, context: str = None) -> str:
-        """프롬프트 포맷팅 - 시스템 프롬프트 단순화"""
-        
-        # 기본 시스템 프롬프트 (단순화)
-        system_prompt = """당신은 친절하고 도움이 되는 AI 챗봇입니다. 사용자의 질문에 정확하고 유용한 답변을 제공하세요."""
-        
-        # 시스템 프롬프트를 항상 먼저 포함
-        if context:
-            # 컨텍스트가 있을 때
-            if user_input in context:
-                # 중복 방지: 컨텍스트만 사용
-                prompt = f"""{system_prompt}
-
-{context}
-
-### 챗봇:"""
-            else:
-                # 새로운 사용자 입력 추가
-                prompt = f"""{system_prompt}
-
-{context}
-
-### 사용자:
-{user_input}
-
-### 챗봇:"""
-        else:
-            # 컨텍스트가 없어도 시스템 프롬프트는 포함
-            prompt = f"""{system_prompt}
-
-### 사용자:
-{user_input}
-
-### 챗봇:"""
-        return prompt
-    
-    def extract_response(self, full_text: str, formatted_prompt: str = None) -> str:
-        """응답 추출 - 품질 검증 및 개선"""
-        logger.info(f"--- Polyglot 응답 추출 시작 ---")
-        logger.info(f"전체 생성 텍스트 (Raw): \n---\n{full_text}\n---")
-        logger.info(f"사용된 프롬프트: {formatted_prompt}")
-        
-        # 1순위: "### 챗봇:" 태그로 ���출 시도
-        if "### 챗봇:" in full_text:
-            response = full_text.split("### 챗봇:")[-1].strip()
-            logger.info(f"✅ 성공: '### 챗봇:' 태그로 응답 추출")
-            logger.info(f"추출된 응답: {response}")
-            
-            # 응답 품질 검증
-            if self._validate_response_quality(response):
-                return response
-            else:
-                logger.warning("⚠️ 응답 품질이 낮습니다. 품질 개선 제안을 추가합니다.")
-                return self._improve_response_quality(response)
-        
-        # 2순위: 프롬프트 제거로 추출 시도
-        if formatted_prompt and formatted_prompt in full_text:
-            response = full_text.replace(formatted_prompt, "").strip()
-            logger.info(f"✅ 성공: 프롬프트 제거로 응답 추출")
-            logger.info(f"추출된 응답: {response}")
-            
-            if self._validate_response_quality(response):
-                return response
-            else:
-                return self._improve_response_quality(response)
-        
-        # 3순위: <|im_start|>assistant 태그 이후 내용 추출
-        if "<|im_start|>assistant" in full_text:
-            parts = full_text.split("<|im_start|>assistant")
-            if len(parts) > 1:
-                # 마지막 assistant 태그 이후 내용
-                last_assistant_part = parts[-1]
-                # <|im_end|> 태그 제거
-                if "<|im_end|>" in last_assistant_part:
-                    response = last_assistant_part.split("<|im_end|>")[0].strip()
-                else:
-                    response = last_assistant_part.strip()
-                
-                logger.info(f"✅ 성공: '<|im_start|>assistant' 태그로 응답 추출")
-                logger.info(f"추출된 응답: {response}")
-                
-                if self._validate_response_quality(response):
-                    return response
-                else:
-                    return self._improve_response_quality(response)
-        
-        # 4순위: 일반적인 프롬프트 패턴 제거 시도
-        clean_text = full_text.strip()
-        patterns_to_remove = [
-            "(응답이 너무 짧습니다. 더 자세한 답변을 원하시면 다시 질문해주세요.)",
-            "### 사용자:",
-            "### 챗봇:",
-            "사용자:",
-            "챗봇:",
-            "assistant:",
-            "user:",
-            "<|im_start|>user",
-            "<|im_end|>",
-            "<|im_start|>assistant",
-            "<|im_start|>system"
-        ]
-        
-        for pattern in patterns_to_remove:
-            clean_text = clean_text.replace(pattern, "")
-        
-        clean_text = clean_text.strip()
-        
-        if clean_text and clean_text != full_text:
-            logger.info("✅ 성공: 패턴 제거로 응답 정리")
-            logger.info(f"정리된 응답: {clean_text}")
-            
-            if self._validate_response_quality(clean_text):
-                return clean_text
-            else:
-                return self._improve_response_quality(clean_text)
-        
-        # 5순위: 전체 텍스트에서 불필요한 부분만 제거
-        final_response = full_text.strip()
-        logger.warning("⚠️ 경고: 특별한 응답 추출 패턴을 찾지 못했습니다. 전체 텍스트를 정리하여 반환합니다.")
-        logger.info(f"최종 반환 텍스트: {final_response}")
-        
-        if self._validate_response_quality(final_response):
-            return final_response
-        else:
-            return self._improve_response_quality(final_response)
-    
-    def _validate_response_quality(self, response: str) -> bool:
-        """응답 품질 검증"""
-        if not response or len(response.strip()) < 5:
-            return False
-        
-        # 영어가 포함되어 있으면 품질 낮음
-        # if any(char.isascii() and char.isalpha() for char in response):
-        #     return False
-        
-        # 문장이 중간에 끊어진 경우 품질 낮음
-        # if response.endswith(('하', '는', '을', '를', '이', '가', '의', '에', '로')):
-        #     return False
-        
-        # 중복된 단어가 많으면 품질 낮음
-        # words = response.split()
-        # if len(words) > 3 and len(set(words)) / len(words) < 0.7:
-        #     return False
-        
-        return True
-    
-    def _improve_response_quality(self, response: str) -> str:
-        """응답 품질 개선"""
-        # 기본 정리
-        improved = response.strip()
-        
-        # 영어 제거
-        
-        # improved = re.sub(r'[a-zA-Z]+', '', improved)
-        
-        # 중복 공백 제거
-        improved = re.sub(r'\s+', ' ', improved)
-        
-        # 문장이 중간에 끊어진 경우 처리
-        # if improved.endswith(('하', '는', '을', '를', '이', '가', '의', '에', '로')):
-        #     improved += '니다.'
-        
-        # 너무 짧은 경우 기본 응답 추가
-        if len(improved) < 5:
-            improved = f"{improved} (응답이 너무 짧습니다. 더 자세한 답변을 원하시면 다시 질문해주세요.)"
-        
-        logger.info(f"🔧 응답 품질 개선 완료: {improved}")
-        return improved
-    
-    def get_generation_config(self) -> Dict[str, Any]:
-        """생성 설정 - 공식 EOS 토큰 사용, 생성 파라미터 최적화"""
-        return {
-            "max_new_tokens": 128,           # 256 → 128로 줄임 (컨텍스트 길이 고려)
-            "temperature": 0.7,              # 0.9 → 0.7로 조정 (안정성 향상)
-            "do_sample": True,               # 샘플링 활성화
-            "top_k": 50,                     # 100 → 50으로 조정 (품질과 안정성 균형)
-            "top_p": 0.9,                    # 0.95 → 0.9로 조정
-            "repetition_penalty": 1.1,       # 1.05 → 1.1로 조정
-            "no_repeat_ngram_size": 3,       # 2 → 3으로 조정
-            "pad_token_id": 2,               # 공식 설정 사용
-            "eos_token_id": 2,               # 공식 설정 사용
-            "use_cache": True,               # 캐시 활성화 (속도 향상)
-            "early_stopping": False,         # EOS 토큰까지 생성하도록 설정
-        }
-    
-    def get_model_info(self) -> Dict[str, Any]:
-        """모델 정보"""
-        return {
-            "model_name": self.model_name,
-            "display_name": self.display_name,
-            "description": self.description,
-            "language": self.language,
-            "model_size": self.model_size,
-            "local_path": self.local_path,
-            "multimodal": False,
-        } 

lily_llm_core/context_manager_250822_0312.py

@@ -1,702 +0,0 @@
-#!/usr/bin/env python3
-"""
-컨텍스트 관리자 (Context Manager)
-대화 히스토리와 단기 기억을 관리하는 시스템
-"""
-
-import logging
-import time
-from typing import List, Dict, Any, Optional, Tuple
-from dataclasses import dataclass
-from collections import deque
-import json
-
-logger = logging.getLogger(__name__)
-
-@dataclass
-class ConversationTurn:
-    """대화 턴을 나타내는 데이터 클래스"""
-    role: str  # 'user' 또는 'assistant'
-    content: str
-    timestamp: float
-    message_id: str
-    metadata: Optional[Dict[str, Any]] = None
-
-class ContextManager:
-    """대화 컨텍스트를 관리하는 클래스"""
-    
-    def __init__(self, 
-                 max_tokens: int = 2000,      # 4000 → 2000으로 줄임
-                 max_turns: int = 20,         # 20 → 10으로 줄임
-                 strategy: str = "sliding_window"):
-        """
-        Args:
-            max_tokens: 최대 토큰 수
-            max_turns: 최대 대화 턴 수
-            strategy: 컨텍스트 관리 전략 ('sliding_window', 'priority_keep', 'circular')
-        """
-        self.max_tokens = max_tokens
-        self.max_turns = max_turns
-        self.strategy = strategy
-        
-        # 세션별 대화 히스토리 (세션 ID로 분리)
-        self.session_conversations: Dict[str, deque] = {}
-        self.default_session = "default"
-        
-        # 기본 세션 초기화
-        self.session_conversations[self.default_session] = deque(maxlen=max_turns * 2)
-        
-        # 시스템 프롬프트
-        self.system_prompt = ""
-        
-        # 컨텍스트 통계
-        self.total_tokens = 0
-        self.current_context_length = 0
-        
-        # 메모리 최적화 설정
-        self.enable_memory_optimization = True
-        self.compression_threshold = 0.8  # 80% 도달 시 압축 시작
-        
-        # 🔄 자동 정리 주기 설정
-        self.auto_cleanup_enabled = True
-        self.cleanup_interval_turns = 5      # 8 → 5턴마다 정리
-        self.cleanup_interval_time = 180     # 5분 → 3분마다 정리
-        self.cleanup_strategy = "aggressive" # smart → aggressive로 변경
-        self.last_cleanup_time = {}          # 세션별 마지막 정리 시간
-        self.turn_counters = {}              # 세션별 턴 카운터
-        
-        logger.info(f"🔧 컨텍스트 관리자 초기화: max_tokens={max_tokens}, strategy={strategy}, auto_cleanup={self.auto_cleanup_enabled}")
-    
-    def set_system_prompt(self, prompt: str):
-        """시스템 프롬프트 설정"""
-        self.system_prompt = prompt
-        logger.info(f"📝 시스템 프롬프트 설정: {len(prompt)} 문자")
-    
-    def set_auto_cleanup_config(self, 
-                               enabled: bool = True,
-                               interval_turns: int = 8,
-                               interval_time: int = 300,
-                               strategy: str = "smart"):
-        """자동 정리 설정 구성"""
-        self.auto_cleanup_enabled = enabled
-        self.cleanup_interval_turns = max(1, interval_turns)
-        self.cleanup_interval_time = max(60, interval_time)
-        self.cleanup_strategy = strategy
-        
-        logger.info(f"🔄 자동 정리 설정: enabled={enabled}, turns={interval_turns}, time={interval_time}s, strategy={strategy}")
-    
-    def get_auto_cleanup_config(self) -> Dict[str, Any]:
-        """자동 정리 설정 반환"""
-        return {
-            "enabled": self.auto_cleanup_enabled,
-            "interval_turns": self.cleanup_interval_turns,
-            "interval_time": self.cleanup_interval_time,
-            "strategy": self.cleanup_strategy
-        }
-    
-    def add_user_message(self, content: str, message_id: str = None, metadata: Dict[str, Any] = None) -> str:
-        """사용자 메시지 추가"""
-        if not message_id:
-            message_id = f"user_{int(time.time() * 1000)}"
-        
-        # 세션 ID 추출 (metadata에서)
-        session_id = "default"
-        if metadata and "session_id" in metadata:
-            session_id = metadata["session_id"]
-        
-        # 세션이 없으면 생성
-        if session_id not in self.session_conversations:
-            self.session_conversations[session_id] = deque(maxlen=self.max_turns * 2)
-        
-        turn = ConversationTurn(
-            role="user",
-            content=content,
-            timestamp=time.time(),
-            message_id=message_id,
-            metadata=metadata or {}
-        )
-        
-        self.session_conversations[session_id].append(turn)
-        self._update_context_stats(session_id)
-        self._optimize_context(session_id)
-        
-        # 🔄 자동 정리 체크
-        self._check_auto_cleanup(session_id)
-        
-        logger.info(f"👤 사용자 메시지 추가: {len(content)} 문자 (세션: {session_id}, 총 {len(self.session_conversations[session_id])} 턴)")
-        return message_id
-    
-    def add_assistant_message(self, content: str, message_id: str = None, metadata: Dict[str, Any] = None) -> str:
-        """어시스턴트 메시지 추가"""
-        if not message_id:
-            message_id = f"assistant_{int(time.time() * 1000)}"
-        
-        # 세션 ID 추출 (metadata에서)
-        session_id = "default"
-        if metadata and "session_id" in metadata:
-            session_id = metadata["session_id"]
-        
-        # 세션이 없으면 생성
-        if session_id not in self.session_conversations:
-            self.session_conversations[session_id] = deque(maxlen=self.max_turns * 2)
-        
-        turn = ConversationTurn(
-            role="assistant",
-            content=content,
-            timestamp=time.time(),
-            message_id=message_id,
-            metadata=metadata or {}
-        )
-        
-        self.session_conversations[session_id].append(turn)
-        self._update_context_stats(session_id)
-        self._optimize_context(session_id)
-        
-        # 🔄 자동 정리 체크
-        self._check_auto_cleanup(session_id)
-        
-        logger.info(f"🤖 어시스턴트 메시지 추가: {len(content)} 문자 (세션: {session_id}, 총 {len(self.session_conversations[session_id])} 턴)")
-        return message_id
-    
-    def get_context(self, include_system: bool = True, max_length: Optional[int] = None, session_id: str = "default") -> str:
-        """현재 컨텍스트를 문자열로 반환 (세션별)"""
-        context_parts = []
-        
-        # 세션이 없으면 기본 세션 사용
-        if session_id not in self.session_conversations:
-            session_id = "default"
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        # 시스템 프롬프트 포함
-        if include_system and self.system_prompt:
-            context_parts.append(f"<|im_start|>system\n{self.system_prompt}<|im_end|>")
-        
-        # 대화 히스토리 포함
-        for turn in conversation_history:
-            if turn.role == "user":
-                context_parts.append(f"<|im_start|>user\n{turn.content}<|im_end|>")
-            elif turn.role == "assistant":
-                context_parts.append(f"<|im_start|>assistant\n{turn.content}<|im_end|>")
-        
-        # 어시스턴트 응답 시작 토큰 추가
-        context_parts.append("<|im_start|>assistant\n")
-        
-        context = "\n".join(context_parts)
-        
-        # 길이 제한 적용
-        if max_length and len(context) > max_length:
-            context = self._truncate_context(context, max_length)
-        
-        return context
-    
-    def get_context_for_model(self, model_name: str = "default", session_id: str = "default") -> str:
-        """모델별 최적화된 컨텍스트 반환 (세션별)"""
-        # 모델별 특별한 처리 (필요시 확장)
-        if "kanana" in model_name.lower():
-            return self.get_context(include_system=True, session_id=session_id)
-        elif "llama" in model_name.lower():
-            # Llama 형식
-            return self._format_for_llama(session_id)
-        elif "polyglot" in model_name.lower():
-            # Polyglot 형식 - <|im_start|> 태그 사용하지 않음
-            return self._format_for_polyglot(session_id)
-        else:
-            return self.get_context(include_system=True, session_id=session_id)
-    
-    def _format_for_llama(self, session_id: str = "default") -> str:
-        """Llama 모델용 형식으로 변환 (세션별)"""
-        context_parts = []
-        
-        # 세션이 없으면 기본 세션 사용
-        if session_id not in self.session_conversations:
-            session_id = "default"
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        if self.system_prompt:
-            context_parts.append(f"[INST] {self.system_prompt} [/INST]")
-        
-        for turn in conversation_history:
-            if turn.role == "user":
-                context_parts.append(f"[INST] {turn.content} [/INST]")
-            elif turn.role == "assistant":
-                context_parts.append(turn.content)
-        
-        return "\n".join(context_parts)
-    
-    def _format_for_polyglot(self, session_id: str = "default") -> str:
-        """Polyglot 모델용 형식으로 변환 (세션별) - 공식 형식 사용"""
-        context_parts = []
-        
-        # 세션이 없으면 기본 세션 사용
-        if session_id not in self.session_conversations:
-            session_id = "default"
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        # 대화 히스토리만 포함 (공식 형식 사용)
-        for turn in conversation_history:
-            if turn.role == "user":
-                context_parts.append(f"### 사용자:\n{turn.content}")
-            elif turn.role == "assistant":
-                context_parts.append(f"### 챗봇:\n{turn.content}")
-        
-        if context_parts:
-            return "\n\n".join(context_parts)
-        else:
-            return ""
-    
-    def get_recent_context(self, turns: int = 5, session_id: str = "default") -> str:
-        """최근 N개 턴의 컨텍스트만 반환 (세션별)"""
-        # 세션이 없으면 기본 세션 사용
-        if session_id not in self.session_conversations:
-            session_id = "default"
-        
-        conversation_history = self.session_conversations[session_id]
-        recent_turns = list(conversation_history)[-turns:]
-        context_parts = []
-        
-        for turn in recent_turns:
-            if turn.role == "user":
-                context_parts.append(f"<|im_start|>user\n{turn.content}<|im_end|>")
-            elif turn.role == "assistant":
-                context_parts.append(f"<|im_start|>assistant\n{turn.content}<|im_end|>")
-        
-        context_parts.append("<|im_start|>assistant\n")
-        return "\n".join(context_parts)
-    
-    def get_context_summary(self, session_id: str = "default") -> Dict[str, Any]:
-        """컨텍스트 요약 정보 반환 (세션별)"""
-        # 세션이 없으면 기본 세션 사용
-        if session_id not in self.session_conversations:
-            session_id = "default"
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        return {
-            "session_id": session_id,
-            "total_turns": len(conversation_history),
-            "user_messages": len([t for t in conversation_history if t.role == "user"]),
-            "assistant_messages": len([t for t in conversation_history if t.role == "assistant"]),
-            "estimated_tokens": self.total_tokens,
-            "context_length": self.current_context_length,
-            "memory_usage": len(conversation_history) / self.max_turns,
-            "oldest_message": conversation_history[0].timestamp if conversation_history else None,
-            "newest_message": conversation_history[-1].timestamp if conversation_history else None
-        }
-    
-    def clear_context(self, session_id: str = "default"):
-        """컨텍스트 초기화 (세션별)"""
-        if session_id not in self.session_conversations:
-            logger.warning(f"⚠️ 세션 {session_id}가 존재하지 않습니다.")
-            return
-        
-        self.session_conversations[session_id].clear()
-        self.total_tokens = 0
-        self.current_context_length = 0
-        logger.info(f"🗑️ 세션 {session_id} 컨텍스트 초기화 완료")
-    
-    def clear_all_sessions(self):
-        """모든 세션 컨텍스트 초기화"""
-        for session_id in list(self.session_conversations.keys()):
-            self.session_conversations[session_id].clear()
-        self.total_tokens = 0
-        self.current_context_length = 0
-        logger.info("🗑️ 모든 세션 컨텍스트 초기화 완료")
-    
-    def remove_message(self, message_id: str, session_id: str = "default") -> bool:
-        """특정 메시지 제거 (세션별)"""
-        if session_id not in self.session_conversations:
-            return False
-        
-        conversation_history = self.session_conversations[session_id]
-        for i, turn in enumerate(conversation_history):
-            if turn.message_id == message_id:
-                removed_turn = conversation_history.pop(i)
-                self._update_context_stats(session_id)
-                logger.info(f"🗑️ 메시지 제거: {message_id} (세션: {session_id})")
-                return True
-        return False
-    
-    def edit_message(self, message_id: str, new_content: str, session_id: str = "default") -> bool:
-        """메시지 내용 수정 (세션별)"""
-        if session_id not in self.session_conversations:
-            return False
-        
-        conversation_history = self.session_conversations[session_id]
-        for turn in conversation_history:
-            if turn.message_id == message_id:
-                turn.content = new_content
-                turn.timestamp = time.time()
-                self._update_context_stats(session_id)
-                logger.info(f"✏️ 메시지 수정: {message_id} (세션: {session_id})")
-                return True
-        return False
-    
-    def search_context(self, query: str, max_results: int = 5, session_id: str = "default") -> List[Dict[str, Any]]:
-        """컨텍스트 내에서 검색 (세션별)"""
-        if session_id not in self.session_conversations:
-            return []
-        
-        conversation_history = self.session_conversations[session_id]
-        results = []
-        query_lower = query.lower()
-        
-        for turn in conversation_history:
-            if query_lower in turn.content.lower():
-                results.append({
-                    "message_id": turn.message_id,
-                    "role": turn.role,
-                    "content": turn.content,
-                    "timestamp": turn.timestamp,
-                    "relevance_score": self._calculate_relevance(query, turn.content)
-                })
-        
-        # 관련성 점수로 정렬
-        results.sort(key=lambda x: x["relevance_score"], reverse=True)
-        return results[:max_results]
-    
-    def _calculate_relevance(self, query: str, content: str) -> float:
-        """간단한 관련성 점수 계산"""
-        query_words = set(query.lower().split())
-        content_words = set(content.lower().split())
-        
-        if not query_words:
-            return 0.0
-        
-        intersection = query_words.intersection(content_words)
-        return len(intersection) / len(query_words)
-    
-    def _update_context_stats(self, session_id: str = "default"):
-        """컨텍스트 통계 업데이트 (세션별)"""
-        if session_id not in self.session_conversations:
-            return
-        
-        self.current_context_length = len(self.get_context(session_id=session_id))
-        # 간단한 토큰 추정 (실제 토크나이저 사용 권장)
-        self.total_tokens = self.current_context_length // 4
-    
-    def _optimize_context(self, session_id: str = "default"):
-        """컨텍스트 최적화 (세션별)"""
-        if not self.enable_memory_optimization:
-            return
-        
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        # 메모리 사용량이 임계값을 초과하면 압축 시작
-        if len(conversation_history) / self.max_turns > self.compression_threshold:
-            self._compress_context(session_id)
-    
-    def _compress_context(self, session_id: str = "default"):
-        """컨텍스트 압축 (중요한 메시지 유지, 세션별)"""
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        if len(conversation_history) <= self.max_turns:
-            return
-        
-        logger.info(f"🗜️ 세션 {session_id} 컨텍스트 압축 시작: {len(conversation_history)} → {self.max_turns}")
-        
-        # 전략에 따른 압축
-        if self.strategy == "sliding_window":
-            # 슬라이딩 윈도우: 최근 메시지 우선
-            while len(conversation_history) > self.max_turns:
-                conversation_history.popleft()
-        
-        elif self.strategy == "priority_keep":
-            # 우선순위 기반: 시스템 프롬프트와 최근 메시지 우선
-            # 첫 번째와 마지막 메시지는 유지
-            if len(conversation_history) > self.max_turns:
-                # 중간 메시지들 중 일부 제거
-                middle_start = self.max_turns // 2
-                middle_end = len(conversation_history) - self.max_turns // 2
-                
-                # 중간 부분을 요약으로 대체
-                removed_turns = list(conversation_history)[middle_start:middle_end]
-                summary_content = f"[이전 {len(removed_turns)}개 메시지 요약: {len(removed_turns)}개 대화 턴]"
-                
-                # 중간 부분 제거
-                for _ in range(middle_end - middle_start):
-                    conversation_history.pop(middle_start)
-                
-                # 요약 메시지 추가
-                summary_turn = ConversationTurn(
-                    role="system",
-                    content=summary_content,
-                    timestamp=time.time(),
-                    message_id=f"summary_{int(time.time() * 1000)}"
-                )
-                conversation_history.insert(middle_start, summary_turn)
-        
-        elif self.strategy == "circular":
-            # 순환 버퍼: 가장 오래된 메시지 제거
-            while len(conversation_history) > self.max_turns:
-                conversation_history.popleft()
-        
-        self._update_context_stats(session_id)
-        logger.info(f"✅ 세션 {session_id} 컨텍스트 압축 완료: {len(conversation_history)} 턴")
-    
-    def _truncate_context(self, context: str, max_length: int) -> str:
-        """컨텍스트 길이 제한"""
-        if len(context) <= max_length:
-            return context
-        
-        # 가장 최근 메시지부터 유지
-        truncated_context = context[-max_length:]
-        
-        # 메시지 경계 확인
-        if not truncated_context.startswith("<|im_start|>"):
-            # 메시지 경계를 찾아서 자르기
-            start_idx = truncated_context.find("<|im_start|>")
-            if start_idx != -1:
-                truncated_context = truncated_context[start_idx:]
-        
-        return truncated_context
-    
-    def export_context(self, file_path: str = None, session_id: str = "default") -> str:
-        """컨텍스트를 파일로 내보내기 (세션별)"""
-        if not file_path:
-            file_path = f"context_export_{session_id}_{int(time.time())}.json"
-        
-        if session_id not in self.session_conversations:
-            logger.warning(f"⚠️ 세션 {session_id}가 존재하지 않습니다.")
-            return None
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        export_data = {
-            "export_timestamp": time.time(),
-            "session_id": session_id,
-            "system_prompt": self.system_prompt,
-            "conversation_history": [
-                {
-                    "role": turn.role,
-                    "content": turn.content,
-                    "timestamp": turn.timestamp,
-                    "message_id": turn.message_id,
-                    "metadata": turn.metadata
-                }
-                for turn in conversation_history
-            ],
-            "context_stats": self.get_context_summary(session_id)
-        }
-        
-        with open(file_path, 'w', encoding='utf-8') as f:
-            json.dump(export_data, f, ensure_ascii=False, indent=2)
-        
-        logger.info(f"💾 세션 {session_id} 컨텍스트 내보내기 완료: {file_path}")
-        return file_path
-    
-    def import_context(self, file_path: str) -> bool:
-        """파일에서 컨텍스트 가져오기"""
-        try:
-            with open(file_path, 'r', encoding='utf-8') as f:
-                import_data = json.load(f)
-            
-            # 기존 컨텍스트 초기화
-            self.clear_context()
-            
-            # 시스템 프롬프트 복원
-            if "system_prompt" in import_data:
-                self.system_prompt = import_data["system_prompt"]
-            
-            # 대화 히스토리 복원
-            if "conversation_history" in import_data:
-                for turn_data in import_data["conversation_history"]:
-                    turn = ConversationTurn(
-                        role=turn_data["role"],
-                        content=turn_data["content"],
-                        timestamp=turn_data["timestamp"],
-                        message_id=turn_data["message_id"],
-                        metadata=turn_data.get("metadata", {})
-                    )
-                    self.conversation_history.append(turn)
-            
-            self._update_context_stats()
-            logger.info(f"📥 컨텍스트 가져오기 완료: {file_path}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ 컨텍스트 가져오기 실패: {e}")
-            return False
-    
-    def get_memory_efficiency(self, session_id: str = "default") -> Dict[str, float]:
-        """메모리 효율성 지표 반환 (세션별)"""
-        if session_id not in self.session_conversations:
-            return {}
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        return {
-            "session_id": session_id,
-            "context_utilization": len(conversation_history) / self.max_turns,
-            "token_efficiency": self.total_tokens / self.max_tokens if self.max_tokens > 0 else 0,
-            "compression_ratio": 1.0 - (len(conversation_history) / (self.max_turns * 2)),
-            "memory_fragmentation": self._calculate_fragmentation(session_id)
-        }
-    
-    def _calculate_fragmentation(self, session_id: str = "default") -> float:
-        """메모리 단편화 정도 계산 (세션별)"""
-        if session_id not in self.session_conversations:
-            return 0.0
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        if len(conversation_history) <= 1:
-            return 0.0
-        
-        # 연속된 메시지 간의 시간 간격으로 단편화 계산
-        timestamps = [turn.timestamp for turn in conversation_history]
-        intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]
-        
-        if not intervals:
-            return 0.0
-        
-        avg_interval = sum(intervals) / len(intervals)
-        variance = sum((x - avg_interval) ** 2 for x in intervals) / len(intervals)
-        
-        # 정규화된 단편화 점수 (0-1)
-        return min(1.0, variance / (avg_interval ** 2) if avg_interval > 0 else 0.0)
-    
-    def _check_auto_cleanup(self, session_id: str = "default"):
-        """자동 정리 조건 체크 및 실행"""
-        if not self.auto_cleanup_enabled:
-            return
-        
-        current_time = time.time()
-        
-        # 세션별 카운터 초기화
-        if session_id not in self.turn_counters:
-            self.turn_counters[session_id] = 0
-        if session_id not in self.last_cleanup_time:
-            self.last_cleanup_time[session_id] = current_time
-        
-        # 턴 카운터 증가
-        self.turn_counters[session_id] += 1
-        
-        # 정리 조건 체크
-        should_cleanup = False
-        cleanup_reason = ""
-        
-        # 턴 기반 정리
-        if self.turn_counters[session_id] >= self.cleanup_interval_turns:
-            should_cleanup = True
-            cleanup_reason = f"턴 기반 ({self.turn_counters[session_id]} 턴)"
-        
-        # 시간 기반 정리
-        elif current_time - self.last_cleanup_time[session_id] >= self.cleanup_interval_time:
-            should_cleanup = True
-            cleanup_reason = f"시간 기반 ({int(current_time - self.last_cleanup_time[session_id])}초)"
-        
-        # 컨텍스트 길이 기반 정리 (강화)
-        elif len(self.session_conversations.get(session_id, [])) > self.max_turns:
-            should_cleanup = True
-            cleanup_reason = f"길이 기반 ({len(self.session_conversations.get(session_id, []))} > {self.max_turns})"
-        
-        # 자동 정리 실행
-        if should_cleanup:
-            logger.info(f"🔄 세션 {session_id} 자동 정리 시작: {cleanup_reason}")
-            self._execute_auto_cleanup(session_id)
-            
-            # 카운터 및 시간 리셋
-            self.turn_counters[session_id] = 0
-            self.last_cleanup_time[session_id] = current_time
-    
-    def _execute_auto_cleanup(self, session_id: str = "default"):
-        """자동 정리 실행"""
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        original_length = len(conversation_history)
-        
-        if original_length <= self.max_turns:
-            return
-        
-        # 전략별 정리 실행
-        if self.cleanup_strategy == "smart":
-            self._smart_cleanup(session_id)
-        elif self.cleanup_strategy == "aggressive":
-            self._aggressive_cleanup(session_id)
-        elif self.cleanup_strategy == "conservative":
-            self._conservative_cleanup(session_id)
-        
-        final_length = len(conversation_history)
-        removed_count = original_length - final_length
-        
-        if removed_count > 0:
-            logger.info(f"✅ 세션 {session_id} 자동 정리 완료: {original_length} → {final_length} 턴 (제거: {removed_count})")
-    
-    def _smart_cleanup(self, session_id: str = "default"):
-        """스마트 정리: 중요 메시지 유지, 중간 메시지 요약"""
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        if len(conversation_history) <= self.max_turns:
-            return
-        
-        # 중요 메시지 수 계산 (시스템 + 최근)
-        important_count = min(3, self.max_turns // 3)
-        recent_count = min(5, self.max_turns // 2)
-        
-        # 중간 메시지들 제거
-        middle_start = important_count
-        middle_end = len(conversation_history) - recent_count
-        
-        if middle_end > middle_start:
-            removed_turns = list(conversation_history)[middle_start:middle_end]
-            
-            # 요약 메시지 생성
-            summary_content = f"[이전 {len(removed_turns)}개 메시지 요약: {len(removed_turns)}개 대화 턴]"
-            
-            # 중간 부분 제거
-            for _ in range(middle_end - middle_start):
-                conversation_history.pop(middle_start)
-            
-            # 요약 메시지 추가
-            summary_turn = ConversationTurn(
-                role="system",
-                content=summary_content,
-                timestamp=time.time(),
-                message_id=f"summary_{int(time.time() * 1000)}"
-            )
-            conversation_history.insert(middle_start, summary_turn)
-    
-    def _aggressive_cleanup(self, session_id: str = "default"):
-        """적극적 정리: 최근 메시지만 유지"""
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        # 최근 max_turns 개만 유지
-        while len(conversation_history) > self.max_turns:
-            conversation_history.popleft()
-    
-    def _conservative_cleanup(self, session_id: str = "default"):
-        """보수적 정리: 점진적으로 정리"""
-        if session_id not in self.session_conversations:
-            return
-        
-        conversation_history = self.session_conversations[session_id]
-        
-        # 20%씩 점진적으로 제거
-        target_length = int(len(conversation_history) * 0.8)
-        if target_length > self.max_turns:
-            while len(conversation_history) > target_length:
-                conversation_history.popleft()
-
-# 전역 컨텍스트 관리자 인스턴스
-context_manager = ContextManager()
-
-def get_context_manager() -> ContextManager:
-    """전역 컨텍스트 관리자 반환"""
-    return context_manager 

lily_llm_core/lora_manager_250822_0312.py

@@ -1,524 +0,0 @@
-#!/usr/bin/env python3
-"""
-LoRA/QLoRA 관리자 (LoRA Manager)
-LoRA 어댑터를 로드하고 관리하는 시스템
-"""
-
-import logging
-import os
-import json
-import torch
-from typing import Dict, Any, Optional, List, Union
-from pathlib import Path
-import warnings
-import time
-
-# logger를 먼저 정의
-logger = logging.getLogger(__name__)
-
-# PEFT 관련 import (설치되지 않은 경우 경고)
-try:
-    logger.info("🔍 PEFT 라이브러리 import 시도 중...")
-    from peft import (
-        LoraConfig, 
-        get_peft_model, 
-        PeftModel, 
-        TaskType,
-        prepare_model_for_kbit_training
-    )
-    from peft.utils import get_peft_model_state_dict
-    PEFT_AVAILABLE = True
-    logger.info("✅ PEFT 라이브러리 import 성공")
-except ImportError as e:
-    PEFT_AVAILABLE = False
-    logger.error(f"❌ PEFT 라이브러리 import 실패: {e}")
-    logger.error(f"❌ Python 경로: {os.environ.get('PYTHONPATH', 'Not set')}")
-    logger.error(f"❌ 현재 작업 디렉토리: {os.getcwd()}")
-    warnings.warn(f"PEFT 라이브러리가 설치되지 않았습니다. LoRA 기능을 사용할 수 없습니다. 오류: {e}")
-
-# Transformers 관련 import
-try:
-    logger.info("🔍 Transformers 라이브러리 import 시도 중...")
-    from transformers import (
-        AutoModelForCausalLM,
-        AutoTokenizer,
-        BitsAndBytesConfig,
-        TrainingArguments,
-        Trainer,
-        DataCollatorForLanguageModeling
-    )
-    TRANSFORMERS_AVAILABLE = True
-    logger.info("✅ Transformers 라이브러리 import 성공")
-except ImportError as e:
-    TRANSFORMERS_AVAILABLE = False
-    logger.error(f"❌ Transformers 라이브러리 import 실패: {e}")
-    warnings.warn(f"Transformers 라이브러리가 설치되지 않았습니다. 오류: {e}")
-
-class LoRAManager:
-    """LoRA/QLoRA 모델 관리 클래스"""
-    
-    def __init__(self, base_model_path: str = None, device: str = "auto"):
-        """
-        Args:
-            base_model_path: 기본 모델 경로
-            device: 사용할 디바이스 ('auto', 'cpu', 'cuda', 'mps')
-        """
-        logger.info(f"🔧 LoRA 관리자 초기화 시작: PEFT_AVAILABLE={PEFT_AVAILABLE}, TRANSFORMERS_AVAILABLE={TRANSFORMERS_AVAILABLE}")
-        
-        if not PEFT_AVAILABLE:
-            logger.error("❌ PEFT 라이브러리를 사용할 수 없습니다.")
-            logger.error("❌ pip install peft를 실행했는지 확인하세요.")
-            logger.error("❌ 가상환경이 활성화되어 있는지 확인하세요.")
-            raise ImportError("PEFT 라이브러리가 필요합니다. pip install peft를 실행하세요.")
-        
-        if not TRANSFORMERS_AVAILABLE:
-            logger.error("❌ Transformers 라이브러리를 사용할 수 없습니다.")
-            logger.error("❌ pip install transformers를 실행했는지 확인하세요.")
-            raise ImportError("Transformers 라이브러리가 필요합니다. pip install transformers를 실행하세요.")
-        
-        self.base_model_path = base_model_path
-        self.device = self._get_device(device)
-        
-        # 모델 및 토크나이저
-        self.base_model = None
-        self.tokenizer = None
-        self.lora_model = None
-        
-        # LoRA 설정
-        self.lora_config = None
-        self.current_adapter_name = None
-        
-        # 어댑터 저장 경로
-        self.adapters_dir = Path("lora_adapters")
-        self.adapters_dir.mkdir(exist_ok=True)
-        
-        # 로드된 어댑터 목록
-        self.loaded_adapters = {}
-        
-        logger.info(f"🔧 LoRA 관리자 초기화: device={self.device}")
-    
-    def _get_device(self, device: str) -> str:
-        """사용 가능한 디바이스 확인"""
-        if device == "auto":
-            if torch.cuda.is_available():
-                return "cuda"
-            elif torch.backends.mps.is_available():
-                return "mps"
-            else:
-                return "cpu"
-        return device
-    
-    def load_base_model(self, model_path: str = None, model_type: str = "causal_lm") -> bool:
-        """기본 모델 로드"""
-        try:
-            model_path = model_path or self.base_model_path
-            if not model_path:
-                raise ValueError("모델 경로가 지정되지 않았습니다.")
-            
-            logger.info(f"📥 기본 모델 로딩 시작: {model_path}")
-            
-            # 토크나이저 로드
-            self.tokenizer = AutoTokenizer.from_pretrained(
-                model_path,
-                trust_remote_code=True,
-                local_files_only=os.path.exists(model_path)
-            )
-            
-            # 패딩 토큰 설정
-            if self.tokenizer.pad_token is None:
-                self.tokenizer.pad_token = self.tokenizer.eos_token
-            
-            # 모델 로드
-            if model_type == "causal_lm":
-                self.base_model = AutoModelForCausalLM.from_pretrained(
-                    model_path,
-                    trust_remote_code=True,
-                    local_files_only=os.path.exists(model_path),
-                    torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
-                    device_map="auto" if self.device == "cuda" else None
-                )
-            else:
-                raise ValueError(f"지원하지 않는 모델 타입: {model_type}")
-            
-            # 디바이스로 이동
-            if self.device != "cuda":  # cuda는 device_map="auto" 사용
-                self.base_model = self.base_model.to(self.device)
-            
-            self.base_model_path = model_path
-            logger.info(f"✅ 기본 모델 로딩 완료: {model_path}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ 기본 모델 로딩 실패: {e}")
-            return False
-    
-    def create_lora_config(self, 
-                          r: int = 16,
-                          lora_alpha: int = 32,
-                          target_modules: List[str] = None,
-                          lora_dropout: float = 0.1,
-                          bias: str = "none",
-                          task_type: str = "CAUSAL_LM") -> LoraConfig:
-        """LoRA 설정 생성"""
-        if target_modules is None:
-            # 일반적인 모델 아키텍처에 대한 기본값
-            target_modules = ["q_proj", "v_proj", "k_proj", "o_proj", "gate_proj", "up_proj", "down_proj"]
-        
-        # TaskType 변환
-        logger.info(f"🔍 [DEBUG] 입력된 task_type: {task_type}")
-        task_type_map = {
-            "CAUSAL_LM": TaskType.CAUSAL_LM,
-            "SEQ_2_SEQ_LM": TaskType.SEQ_2_SEQ_LM,
-            "SEQUENCE_CLASSIFICATION": TaskType.SEQUENCE_CLASSIFICATION,
-            "TOKEN_CLASSIFICATION": TaskType.TOKEN_CLASSIFICATION,
-            "QUESTION_ANSWERING": TaskType.QUESTION_ANSWERING
-        }
-        
-        logger.info(f"🔍 [DEBUG] 사용 가능한 TaskType: {list(task_type_map.keys())}")
-        task_type_enum = task_type_map.get(task_type, TaskType.CAUSAL_LM)
-        logger.info(f"🔍 [DEBUG] 선택된 TaskType: {task_type_enum}")
-        
-        self.lora_config = LoraConfig(
-            r=r,
-            lora_alpha=lora_alpha,
-            target_modules=target_modules,
-            lora_dropout=lora_dropout,
-            bias=bias,
-            task_type=task_type_enum
-        )
-        
-        logger.info(f"🔧 LoRA 설정 생성: r={r}, alpha={lora_alpha}, target_modules={target_modules}")
-        return self.lora_config
-    
-    def apply_lora_to_model(self, adapter_name: str = "default") -> bool:
-        """LoRA를 기본 모델에 적용"""
-        try:
-            if self.base_model is None:
-                raise ValueError("기본 모델이 로드되지 않았습니다.")
-            
-            if self.lora_config is None:
-                raise ValueError("LoRA 설정이 생성되지 않았습니다.")
-            
-            logger.info(f"🔗 LoRA 어댑터 적용 시작: {adapter_name}")
-            
-            # LoRA 모델 생성
-            self.lora_model = get_peft_model(self.base_model, self.lora_config)
-            
-            # 어댑터 이름 설정
-            self.current_adapter_name = adapter_name
-            
-            # 훈련 모드로 설정
-            self.lora_model.train()
-            
-            # 모델 정보 출력
-            self.lora_model.print_trainable_parameters()
-            
-            logger.info(f"✅ LoRA 어댑터 적용 완료: {adapter_name}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ LoRA 어댑터 적용 실패: {e}")
-            return False
-    
-    def load_lora_adapter(self, adapter_path: str, adapter_name: str = None) -> bool:
-        """저장된 LoRA 어댑터 로드"""
-        try:
-            if not os.path.exists(adapter_path):
-                raise FileNotFoundError(f"어댑터 경로를 찾을 수 없습니다: {adapter_path}")
-            
-            if adapter_name is None:
-                adapter_name = Path(adapter_path).stem
-            
-            logger.info(f"📥 LoRA 어댑터 로딩 시작: {adapter_path}")
-            
-            # 기본 모델이 로드되지 않은 경우 로드
-            if self.base_model is None:
-                # 어댑터 설정 파일에서 기본 모델 경로 확인
-                config_path = os.path.join(adapter_path, "adapter_config.json")
-                if os.path.exists(config_path):
-                    with open(config_path, 'r') as f:
-                        config = json.load(f)
-                    base_model_path = config.get("base_model_name_or_path")
-                    if base_model_path:
-                        self.load_base_model(base_model_path)
-            
-            # LoRA 어댑터 로드
-            self.lora_model = PeftModel.from_pretrained(
-                self.base_model,
-                adapter_path,
-                torch_dtype=torch.float16 if self.device == "cuda" else torch.float32
-            )
-            
-            # 디바이스로 이동
-            if self.device != "cuda":
-                self.lora_model = self.lora_model.to(self.device)
-            
-            self.current_adapter_name = adapter_name
-            self.loaded_adapters[adapter_name] = adapter_path
-            
-            logger.info(f"✅ LoRA 어댑터 로딩 완료: {adapter_name}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ LoRA 어댑터 로딩 실패: {e}")
-            return False
-    
-    def save_lora_adapter(self, adapter_name: str = None, output_dir: str = None) -> bool:
-        """LoRA 어댑터 저장"""
-        try:
-            if self.lora_model is None:
-                raise ValueError("LoRA 모델이 로드되지 않았습니다.")
-            
-            adapter_name = adapter_name or self.current_adapter_name or "default"
-            output_dir = output_dir or str(self.adapters_dir / adapter_name)
-            
-            logger.info(f"💾 LoRA 어댑터 저장 시작: {adapter_name} -> {output_dir}")
-            
-            # 어댑터 저장
-            self.lora_model.save_pretrained(output_dir)
-            
-            # 토크나이저도 저장
-            if self.tokenizer:
-                self.tokenizer.save_pretrained(output_dir)
-            
-            # 어댑터 정보 저장
-            adapter_info = {
-                "adapter_name": adapter_name,
-                "base_model": self.base_model_path,
-                "lora_config": self.lora_config.to_dict() if self.lora_config else None,
-                "created_at": str(torch.tensor(time.time())),
-                "device": self.device
-            }
-            
-            with open(os.path.join(output_dir, "adapter_info.json"), 'w') as f:
-                json.dump(adapter_info, f, indent=2)
-            
-            logger.info(f"✅ LoRA 어댑터 저장 완료: {output_dir}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ LoRA 어댑터 저장 실패: {e}")
-            return False
-    
-    def merge_lora_with_base(self, output_path: str = None) -> bool:
-        """LoRA 어댑터를 기본 모델과 병합"""
-        try:
-            if self.lora_model is None:
-                raise ValueError("LoRA 모델이 로드되지 않았습니다.")
-            
-            output_path = output_path or f"{self.base_model_path}_merged"
-            
-            logger.info(f"🔗 LoRA 어댑터 병합 시작: {output_path}")
-            
-            # 병합된 모델 생성
-            merged_model = self.lora_model.merge_and_unload()
-            
-            # 병합된 모델 저장
-            merged_model.save_pretrained(output_path)
-            
-            # 토크나이저도 저장
-            if self.tokenizer:
-                self.tokenizer.save_pretrained(output_path)
-            
-            logger.info(f"✅ LoRA 어댑터 병합 완료: {output_path}")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ LoRA 어댑터 병합 실패: {e}")
-            return False
-    
-    def list_available_adapters(self) -> List[Dict[str, Any]]:
-        """사용 가능한 어댑터 목록 반환"""
-        adapters = []
-        
-        for adapter_dir in self.adapters_dir.iterdir():
-            if adapter_dir.is_dir():
-                config_path = adapter_dir / "adapter_config.json"
-                info_path = adapter_dir / "adapter_info.json"
-                
-                adapter_info = {
-                    "name": adapter_dir.name,
-                    "path": str(adapter_dir),
-                    "config_exists": config_path.exists(),
-                    "info_exists": info_path.exists()
-                }
-                
-                # 어댑터 정보 로드
-                if info_path.exists():
-                    try:
-                        with open(info_path, 'r') as f:
-                            info = json.load(f)
-                        adapter_info.update(info)
-                    except Exception as e:
-                        logger.warning(f"어댑터 정보 로드 실패: {e}")
-                
-                adapters.append(adapter_info)
-        
-        return adapters
-    
-    def get_adapter_stats(self) -> Dict[str, Any]:
-        """어댑터 통계 정보 반환"""
-        if self.lora_model is None:
-            return {"error": "LoRA 모델이 로드되지 않았습니다."}
-        
-        try:
-            # 훈련 가능한 파라미터 수
-            trainable_params = 0
-            all_param = 0
-            
-            for param in self.lora_model.parameters():
-                all_param += param.numel()
-                if param.requires_grad:
-                    trainable_params += param.numel()
-            
-            return {
-                "adapter_name": self.current_adapter_name,
-                "trainable_params": trainable_params,
-                "all_params": all_param,
-                "trainable_ratio": trainable_params / all_param if all_param > 0 else 0,
-                "device": self.device,
-                "model_type": type(self.lora_model).__name__
-            }
-            
-        except Exception as e:
-            logger.error(f"어댑터 통계 수집 실패: {e}")
-            return {"error": str(e)}
-    
-    def switch_adapter(self, adapter_name: str) -> bool:
-        """다른 어댑터로 전환"""
-        try:
-            if adapter_name not in self.loaded_adapters:
-                # 어댑터 로드
-                adapter_path = self.adapters_dir / adapter_name
-                if not adapter_path.exists():
-                    raise FileNotFoundError(f"어댑터를 찾을 수 없습니다: {adapter_name}")
-                
-                return self.load_lora_adapter(str(adapter_path), adapter_name)
-            else:
-                # 이미 로드된 어댑터 사용
-                self.current_adapter_name = adapter_name
-                logger.info(f"🔄 어댑터 전환: {adapter_name}")
-                return True
-                
-        except Exception as e:
-            logger.error(f"❌ 어댑터 전환 실패: {e}")
-            return False
-    
-    def unload_adapter(self) -> bool:
-        """LoRA 어댑터 언로드"""
-        try:
-            if self.lora_model is None:
-                return True
-            
-            logger.info("🗑️ LoRA 어댑터 언로드 시작")
-            
-            # 어댑터 제거
-            self.lora_model = None
-            self.current_adapter_name = None
-            self.lora_config = None
-            
-            logger.info("✅ LoRA 어댑터 언로드 완료")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ LoRA 어댑터 언로드 실패: {e}")
-            return False
-    
-    def generate_text(self, prompt: str, max_length: int = 100, temperature: float = 0.7) -> str:
-        """LoRA 모델을 사용한 텍스트 생성"""
-        try:
-            if self.lora_model is None:
-                raise ValueError("LoRA 모델이 로드되지 않았습니다.")
-            
-            if self.tokenizer is None:
-                raise ValueError("토크나이저가 로드되지 않았습니다.")
-            
-            # 입력 토크나이징
-            inputs = self.tokenizer(prompt, return_tensors="pt")
-            inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            # 추론 모드로 설정
-            self.lora_model.eval()
-            
-            with torch.no_grad():
-                outputs = self.lora_model.generate(
-                    **inputs,
-                    max_new_tokens=max_length,
-                    temperature=temperature,
-                    do_sample=True,
-                    pad_token_id=self.tokenizer.eos_token_id
-                )
-            
-            # 응답 디코딩
-            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
-            
-            # 프롬프트 제거
-            if response.startswith(prompt):
-                response = response[len(prompt):].strip()
-            
-            return response
-            
-        except Exception as e:
-            logger.error(f"❌ 텍스트 생성 실패: {e}")
-            return f"텍스트 생성 중 오류가 발생했습니다: {str(e)}"
-    
-    def prepare_for_training(self, training_args: TrainingArguments = None) -> bool:
-        """훈련을 위한 모델 준비"""
-        try:
-            if self.lora_model is None:
-                raise ValueError("LoRA 모델이 로드되지 않았습니다.")
-            
-            logger.info("🔧 훈련을 위한 모델 준비 시작")
-            
-            # 기본 훈련 인수
-            if training_args is None:
-                training_args = TrainingArguments(
-                    output_dir="./lora_training_output",
-                    num_train_epochs=3,
-                    per_device_train_batch_size=4,
-                    gradient_accumulation_steps=4,
-                    learning_rate=2e-4,
-                    warmup_steps=100,
-                    logging_steps=10,
-                    save_steps=500,
-                    eval_steps=500,
-                    evaluation_strategy="steps",
-                    save_strategy="steps",
-                    load_best_model_at_end=True,
-                    metric_for_best_model="eval_loss",
-                    greater_is_better=False,
-                    fp16=torch.cuda.is_available(),
-                    dataloader_pin_memory=False,
-                )
-            
-            # 훈련 모드로 설정
-            self.lora_model.train()
-            
-            # 그래디언트 체크포인팅 활성화 (메모리 ���약)
-            self.lora_model.gradient_checkpointing_enable()
-            
-            # 그래디언트 클리핑 설정
-            self.lora_model.enable_input_require_grads()
-            
-            logger.info("✅ 훈련을 위한 모델 준비 완료")
-            return True
-            
-        except Exception as e:
-            logger.error(f"❌ 훈련 준비 실패: {e}")
-            return False
-
-# 전역 LoRA 관리자 인스턴스 (안전한 생성)
-try:
-    if PEFT_AVAILABLE and TRANSFORMERS_AVAILABLE:
-        lora_manager = LoRAManager()
-        logger.info("✅ 전역 LoRA 관리자 인스턴스 생성 완료")
-    else:
-        lora_manager = None
-        logger.warning("⚠️ LoRA 라이브러리가 사용 불가능하여 LoRA 관리자를 생성하지 않았습니다.")
-except Exception as e:
-    lora_manager = None
-    logger.error(f"❌ LoRA 관리자 인스턴스 생성 실패: {e}")
-
-def get_lora_manager() -> Optional[LoRAManager]:
-    """전역 LoRA 관리자 반환 (None일 수 있음)"""
-    return lora_manager