Auto commit at 22-2025-08 3:17:20

lily_llm_api/app_v2.py

@@ -2,7 +2,7 @@
 """
 Lily LLM API 서버 v2 (인터랙티브 선택 복원 및 성능 최적화 최종본)
 """
-from fastapi import FastAPI, HTTPException, UploadFile, File, Form, Depends, WebSocket, WebSocketDisconnect
+from fastapi import FastAPI, HTTPException, Request, UploadFile, File, Form, Depends, WebSocket, WebSocketDisconnect
 from fastapi.security import HTTPAuthorizationCredentials
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
@@ -300,7 +300,7 @@ def load_model_sync(model_id: str):
 
 def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_length: Optional[int] = None, 
                  temperature: Optional[float] = None, top_p: Optional[float] = None, 
-                 do_sample: Optional[bool] = None) -> dict:
+                 do_sample: Optional[bool] = None, use_context: bool = True, session_id: str = None) -> dict:
     """[최적화] 모델 생성을 처리하는 통합 동기 함수"""
     try:
         print(f"🔍 [DEBUG] generate_sync 시작 - prompt 길이: {len(prompt)}")
@@ -377,6 +377,24 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
         # --- 2. 프롬프트 구성 ---
         print(f"🔍 [DEBUG] 프롬프트 구성 시작")
         
+        # 컨텍스트 통합 (대화 기록 포함) - 모델별 최적화
+        context_prompt = ""
+        if use_context and session_id:
+            try:
+                # 수정: 모델별 최적화된 컨텍스트 사용
+                context = context_manager.get_context_for_model(
+                    current_profile.model_name, 
+                    session_id
+                )
+                if context and len(context.strip()) > 0:
+                    context_prompt = context + "\n\n"
+                    print(f"🔍 [DEBUG] 컨텍스트 포함됨 - 길이: {len(context_prompt)} (세션: {session_id})")
+                else:
+                    print(f"🔍 [DEBUG] 컨텍스트 없음 또는 비어있음 (세션: {session_id})")
+            except Exception as e:
+                print(f"⚠️ [DEBUG] 컨텍스트 로드 실패: {e} (세션: {session_id})")
+                context_prompt = ""
+        
         # formatted_prompt 초기화
         formatted_prompt = None
         
@@ -402,14 +420,33 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
             image_tokens = ""
             print(f"🔍 [DEBUG] 이미지 없음 - 텍스트-only 모드")
             
-            # 텍스트-only 모델용 프롬프트 구성
+            # 텍스트-only 모델용 프롬프트 구성 (컨텍스트 포함)
             if hasattr(current_profile, 'format_prompt'):
-                formatted_prompt = current_profile.format_prompt(prompt)
-                print(f"🔍 [DEBUG] 프로필 format_prompt 사용: {formatted_prompt}")
+                # Polyglot 모델일 때는 format_prompt 메서드 사용 (컨텍스트 지원)
+                if "polyglot" in current_profile.model_name.lower():
+                    # 컨텍스트와 프롬프트를 함께 전달
+                    formatted_prompt = current_profile.format_prompt(prompt, context_prompt)
+                else:
+                    # 다른 모델은 기존 방식 사용
+                    base_prompt = current_profile.format_prompt(prompt)
+                    if context_prompt:
+                        formatted_prompt = context_prompt + base_prompt
+                    else:
+                        formatted_prompt = base_prompt
+                print(f"🔍 [DEBUG] 프로필 format_prompt 사용 (컨텍스트 포함): {formatted_prompt}")
             else:
-                # 기본 프롬프트 (fallback)
-                formatted_prompt = f"<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n"
-                print(f"🔍 [DEBUG] 기본 프롬프트 사용: {formatted_prompt}")
+                # 기본 프롬프트 (fallback) - 컨텍스트 포함
+                # Polyglot 모델은 <|im_start|> 태그를 제대로 처리하지 못함
+                if "polyglot" in current_profile.model_name.lower():
+                    base_prompt = f"### 사용자:\n{prompt}\n\n### 챗봇:\n"
+                else:
+                    base_prompt = f"<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n"
+                
+                if context_prompt:
+                    formatted_prompt = context_prompt + base_prompt
+                else:
+                    formatted_prompt = base_prompt
+                print(f"🔍 [DEBUG] 기본 프롬프트 사용 (컨텍스트 포함): {formatted_prompt}")
         
         print(f"🔍 [DEBUG] 프롬프트 구성 완료 - 길이: {len(formatted_prompt) if formatted_prompt else 0}")
         print(f"🔍 [DEBUG] 최종 프롬프트: {formatted_prompt}")
@@ -427,7 +464,7 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
                 return_tensors="pt",
                 padding=True,
                 truncation=True,
-                max_length=256,
+                max_length=2048,
             )
             if 'token_type_ids' in inputs:
                 del inputs['token_type_ids']
@@ -455,7 +492,7 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
                     return_tensors="pt",
                     padding=True,
                     truncation=True,
-                    max_length=256,
+                    max_lengt=2048,
                 )
                 if 'token_type_ids' in inputs:
                     del inputs['token_type_ids']
@@ -537,12 +574,35 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
                     print(f"🔍 [DEBUG] 최종 이미지 텐서 디바이스: {pixel_values.device}")
                     print(f"🔍 [DEBUG] 모델 생성 시작 - 멀티모달")
                     
-                    generated_ids = model.generate(
-                        input_ids=input_ids,
-                        attention_mask=attention_mask,
-                        pixel_values=pixel_values,
-                        **gen_config
-                    )
+                    # LoRA 어댑터가 적용된 모델인지 확인
+                    if LORA_AVAILABLE and lora_manager and hasattr(lora_manager, 'current_adapter_name') and lora_manager.current_adapter_name:
+                        print(f"🔍 [DEBUG] LoRA 어댑터 적용됨 (멀티모달): {lora_manager.current_adapter_name}")
+                        # LoRA가 적용된 모델 사용
+                        lora_model = lora_manager.get_model()
+                        if lora_model:
+                            print(f"🔍 [DEBUG] LoRA 모델로 멀티모달 생성 실행")
+                            generated_ids = lora_model.generate(
+                                input_ids=input_ids,
+                                attention_mask=attention_mask,
+                                pixel_values=pixel_values,
+                                **gen_config
+                            )
+                        else:
+                            print(f"⚠️ [DEBUG] LoRA 모델을 가져올 수 없음, 기본 모델 사용")
+                            generated_ids = model.generate(
+                                input_ids=input_ids,
+                                attention_mask=attention_mask,
+                                pixel_values=pixel_values,
+                                **gen_config
+                            )
+                    else:
+                        print(f"🔍 [DEBUG] LoRA 어댑터 없음 (멀티모달), 기본 모델 사용")
+                        generated_ids = model.generate(
+                            input_ids=input_ids,
+                            attention_mask=attention_mask,
+                            pixel_values=pixel_values,
+                            **gen_config
+                        )
 
                                         # 토큰 설정을 명시적으로 전달하여 EOS 토큰 문제 해결
                     # generate_kwargs = {
@@ -600,23 +660,26 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
                             gen_config['pad_token_id'] = None
                             print(f"🔍 [DEBUG] PAD 토큰 설정: None (토크나이저에 PAD 토큰 없음)")
                     
-                    # EOS 토큰 설정 - 강제로 설정하여 EOS 토큰 문제 해결
-                    if tokenizer.eos_token_id is not None:
-                        gen_config['eos_token_id'] = tokenizer.eos_token_id
-                        print(f"🔍 [DEBUG] EOS 토큰 강제 설정: {tokenizer.eos_token_id}")
-                    else:
-                        gen_config['eos_token_id'] = 2  # <|endoftext|> 기본값
-                        print(f"🔍 [DEBUG] EOS 토큰 기본값 설정: 2")
+                    # 토큰 설정 - 프로필에서 설정된 값 우선 사용
+                    if 'eos_token_id' not in gen_config or gen_config['eos_token_id'] is None:
+                        if tokenizer.eos_token_id is not None:
+                            gen_config['eos_token_id'] = tokenizer.eos_token_id
+                            print(f"🔍 [DEBUG] EOS 토큰 설정: {tokenizer.eos_token_id}")
+                        else:
+                            gen_config['eos_token_id'] = None
+                            print(f"🔍 [DEBUG] EOS 토큰 설정: None (자동 처리)")
                     
-                    # PAD 토큰도 강제 설정
-                    if tokenizer.pad_token_id is not None:
-                        gen_config['pad_token_id'] = tokenizer.pad_token_id
-                    else:
-                        gen_config['pad_token_id'] = tokenizer.eos_token_id or 2
+                    if 'pad_token_id' not in gen_config or gen_config['pad_token_id'] is None:
+                        if tokenizer.pad_token_id is not None:
+                            gen_config['pad_token_id'] = tokenizer.pad_token_id
+                        else:
+                            gen_config['pad_token_id'] = None
                     
-                    # BOS 토큰 설정
-                    if hasattr(tokenizer, 'bos_token_id') and tokenizer.bos_token_id is not None:
-                        gen_config['bos_token_id'] = tokenizer.bos_token_id
+                    if 'bos_token_id' not in gen_config or gen_config['bos_token_id'] is None:
+                        if hasattr(tokenizer, 'bos_token_id') and tokenizer.bos_token_id is not None:
+                            gen_config['bos_token_id'] = tokenizer.bos_token_id
+                        else:
+                            gen_config['bos_token_id'] = None
                     
                     print(f"🔍 [DEBUG] 최종 토큰 설정: EOS={gen_config['eos_token_id']}, PAD={gen_config['pad_token_id']}, BOS={gen_config.get('bos_token_id')}")
                     
@@ -630,11 +693,45 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
                     # 모델 생성 진행 상황 모니터링을 위한 콜백 추가
                     print(f"🔍 [DEBUG] 모델 생성 시작 시간: {time.time()}")
                     
-                    generated_ids = model.generate(
-                        input_ids=input_ids,
-                        attention_mask=attention_mask,
-                        **gen_config
-                    )
+                    # LoRA 어댑터가 적용된 모델인지 확인
+                    if LORA_AVAILABLE and lora_manager and hasattr(lora_manager, 'current_adapter_name') and lora_manager.current_adapter_name:
+                        print(f"🔍 [DEBUG] LoRA 어댑터 적용됨: {lora_manager.current_adapter_name}")
+                        # LoRA가 적용된 모델 사용
+                        lora_model = lora_manager.get_model()
+                        if lora_model:
+                            print(f"🔍 [DEBUG] LoRA 모델로 생성 실행")
+                            generated_ids = lora_model.generate(
+                                input_ids=input_ids,
+                                attention_mask=attention_mask,
+                                **gen_config
+                            )
+                        else:
+                            print(f"⚠️ [DEBUG] LoRA 모델을 가져올 수 없음, 기본 모델 사용")
+                            generated_ids = model.generate(
+                                input_ids=input_ids,
+                                attention_mask=attention_mask,
+                                **gen_config
+                            )
+                    else:
+                        print(f"🔍 [DEBUG] LoRA 어댑터 없음, 기본 모델 사용")
+                        # LoRA 상태 디버깅
+                        if LORA_AVAILABLE:
+                            if lora_manager:
+                                print(f"🔍 [DEBUG] LoRA 매니저 존재: {type(lora_manager)}")
+                                if hasattr(lora_manager, 'current_adapter_name'):
+                                    print(f"🔍 [DEBUG] 현재 어댑터: {lora_manager.current_adapter_name}")
+                                if hasattr(lora_manager, 'base_model'):
+                                    print(f"🔍 [DEBUG] 기본 모델 로드됨: {lora_manager.base_model is not None}")
+                            else:
+                                print(f"🔍 [DEBUG] LoRA 매니저가 None")
+                        else:
+                            print(f"🔍 [DEBUG] LoRA 지원 안됨")
+                        
+                        generated_ids = model.generate(
+                            input_ids=input_ids,
+                            attention_mask=attention_mask,
+                            **gen_config
+                        )
                     
                     # 토큰 설정을 명시적으로 전달하여 EOS 토큰 문제 해결
                     # generate_kwargs = {
@@ -739,8 +836,162 @@ def generate_sync(prompt: str, image_data_list: Optional[List[bytes]], max_lengt
         traceback.print_exc()
         return {"error": str(e)}
 
+@app.get("/lora/status")
+async def get_lora_status():
+    """현재 LoRA 상태 확인"""
+    try:
+        if not LORA_AVAILABLE or lora_manager is None:
+            return {"status": "error", "message": "LoRA 기능이 사용 불가능합니다"}
+        
+        return {
+            "status": "success",
+            "lora_available": True,
+            "current_adapter": lora_manager.current_adapter_name if hasattr(lora_manager, 'current_adapter_name') else None,
+            "base_model_loaded": hasattr(lora_manager, 'base_model') and lora_manager.base_model is not None,
+            "device": getattr(lora_manager, 'device', 'unknown')
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.get("/context/status")
+async def get_context_status():
+    """컨텍스트 관리자 상태 확인"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        # 세션별 정보 수집
+        session_info = {}
+        for session_id, conversation in context_manager.session_conversations.items():
+            session_info[session_id] = {
+                "turns": len(conversation),
+                "user_messages": len([t for t in conversation if t.role == "user"]),
+                "assistant_messages": len([t for t in conversation if t.role == "assistant"])
+            }
+        
+        return {
+            "status": "success",
+            "context_manager_available": True,
+            "total_sessions": len(context_manager.session_conversations),
+            "sessions": session_info,
+            "max_tokens": context_manager.max_tokens,
+            "max_turns": context_manager.max_turns,
+            "strategy": context_manager.strategy
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.get("/context/history")
+async def get_context_history(session_id: str = None):
+    """컨텍스트 히스토리 조회"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        if session_id:
+            # 특정 세션의 컨텍스트만 조회
+            context = context_manager.get_context(include_system=True, max_length=4000, session_id=session_id)
+            session_summary = context_manager.get_context_summary(session_id)
+            return {
+                "status": "success",
+                "session_id": session_id,
+                "context": context,
+                "history_length": session_summary.get("total_turns", 0),
+                "session_summary": session_summary
+            }
+        else:
+            # 전체 컨텍스트 조회
+            context = context_manager.get_context(include_system=True, max_length=4000)
+            return {
+                "status": "success",
+                "context": context,
+                "history_length": len(context_manager.conversation_history),
+                "all_sessions": True
+            }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.get("/context/auto-cleanup")
+async def get_auto_cleanup_config():
+    """자동 정리 설정 조회"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        config = context_manager.get_auto_cleanup_config()
+        return {
+            "status": "success",
+            "auto_cleanup_config": config
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.post("/context/auto-cleanup")
+async def set_auto_cleanup_config(
+    enabled: bool = Form(True),
+    interval_turns: int = Form(8),
+    interval_time: int = Form(300),
+    strategy: str = Form("smart")
+):
+    """자동 정리 설정 변경"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        context_manager.set_auto_cleanup_config(
+            enabled=enabled,
+            interval_turns=interval_turns,
+            interval_time=interval_time,
+            strategy=strategy
+        )
+        
+        return {
+            "status": "success",
+            "message": "자동 정리 설정이 업데이트되었습니다",
+            "new_config": context_manager.get_auto_cleanup_config()
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.post("/context/cleanup/{session_id}")
+async def manual_cleanup_session(session_id: str):
+    """특정 세션 수동 정리"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        # 수동 정리 실행
+        context_manager._execute_auto_cleanup(session_id)
+        
+        return {
+            "status": "success",
+            "message": f"세션 {session_id} 수동 정리 완료",
+            "session_id": session_id
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.post("/context/cleanup-all")
+async def manual_cleanup_all_sessions():
+    """모든 세션 수동 정리"""
+    try:
+        if not context_manager:
+            return {"status": "error", "message": "Context manager not available"}
+        
+        # 모든 세션에 대해 수동 정리 실행
+        for session_id in context_manager.session_conversations.keys():
+            context_manager._execute_auto_cleanup(session_id)
+        
+        return {
+            "status": "success",
+            "message": "모든 세션 수동 정리 완료"
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
 @app.post("/generate", response_model=GenerateResponse)
-async def generate(prompt: str = Form(...), 
+async def generate(request: Request,
+                  prompt: str = Form(...), 
                   image1: UploadFile = File(None),
                   image2: UploadFile = File(None),
                   image3: UploadFile = File(None),
@@ -753,8 +1004,24 @@ async def generate(prompt: str = Form(...),
 
     start_time = time.time()
 
+    # 세션 ID가 없으면 자동 생성 (클라이언트별 고유 세션)
+    if not session_id:
+        # 클라이언트 IP 기반으로 고유한 세션 생성 (같은 클라이언트는 같은 세션 유지)
+        client_ip = "unknown"
+        try:
+            # Request 객체에서 클라이언트 IP 추출
+            client_ip = request.client.host if request.client else "unknown"
+        except:
+            pass
+        
+        # 클라이언트 IP + 시간 기반으로 세션 생성 (하루 동안 유지)
+        day_timestamp = int(time.time() // 86400) * 86400  # 하루 단위로 반올림
+        session_id = f"client_{client_ip}_{day_timestamp}"
+        print(f"🔍 [DEBUG] 자동 세션 ID 생성: {session_id} (클라이언트: {client_ip})")
+    
     if use_context:
         context_manager.add_user_message(prompt, metadata={"session_id": session_id})
+        print(f"🔍 [DEBUG] 사용자 메시지 추가됨 (세션: {session_id})")
 
     # 이미지 데이터 처리
     image_data_list = []
@@ -767,8 +1034,8 @@ async def generate(prompt: str = Form(...),
                 logger.warning(f"이미지 로드 실패: {e}")
 
     try:
-        # generate_sync 함수 호출
-        result = generate_sync(prompt, image_data_list)
+        # generate_sync 함수 호출 (컨텍스트 포함)
+        result = generate_sync(prompt, image_data_list, use_context=use_context, session_id=session_id)
         
         if "error" in result:
             raise HTTPException(status_code=500, detail=result["error"])
@@ -2760,13 +3027,3 @@ async def get_hybrid_rag_status():
     except Exception as e:
         logger.error(f"멀티모달 RAG 상태 확인 오류: {e}")
         return {"status": "error", "error": str(e)}
-
-# run_server_v2.py 에서 직접 실행 시 주석 처리
-# if __name__ == "__main__":
-#     uvicorn.run(
-#         app,
-#         host="0.0.0.0",
-#         port=8001,
-#         reload=False,
-#         log_level="info"
-#     )

lily_llm_api/models/polyglot_ko_1_3b_chat.py

@@ -12,6 +12,9 @@ import os
 from pathlib import Path
 import re
 
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+
 logger = logging.getLogger(__name__)
 
 class PolyglotKo13bChatProfile:
@@ -31,11 +34,16 @@ class PolyglotKo13bChatProfile:
         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,
@@ -54,7 +62,7 @@ class PolyglotKo13bChatProfile:
             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})")
+            # 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'
@@ -62,6 +70,7 @@ class PolyglotKo13bChatProfile:
 
             model = AutoModelForCausalLM.from_pretrained(
                 model_path,
+                token=HF_TOKEN,
                 trust_remote_code=True,
                 torch_dtype=selected_dtype,
                 local_files_only=use_local,
@@ -73,21 +82,40 @@ class PolyglotKo13bChatProfile:
             logger.error(f"❌ {self.display_name} 모델 로드 실패: {e}")
             raise
     
-    def format_prompt(self, user_input: str) -> str:
-        """프롬프트 포맷팅 - 공식 문서와 일치"""
-        # Hugging Face 모델 페이지의 공식 프롬프트 형식 사용
-        # prompt = f"""당신은 AI 챗봇입니다. 사용자에게 도움이 되고 유익한 내용을 제공해야합니다. 답변은 길고 자세하며 친절한 설명을 덧붙여서 작성하세요.
-        prompt = f"""        
-        1. 반드시 한국어로만 응답하세요
-        2. 자연스럽고 일관성 있는 대화를 유지하세요
-        3. 사용자의 질문에 정확하고 도움이 되는 답변을 제공하세요
-        4. 문장이 중간에 끊기지 않도록 완성된 답변을 작성하세요                
-        
-        ### 사용자:
-        {user_input}
-
-        ### 챗봇:
-        """
+    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:
@@ -120,19 +148,40 @@ class PolyglotKo13bChatProfile:
             else:
                 return self._improve_response_quality(response)
         
-        # 3순위: 일반적인 프롬프트 패턴 제거 시도
+        # 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 = [            
-            "1. 반드시 한국어로만 응답하세요",
-            "2. 자연스럽고 일관성 있는 대화를 유지하세요",
-            "3. 사용자의 질문에 정확하고 도움이 되는 답변을 제공하세요",
-            "4. 문장이 중간에 끊기지 않도록 완성된 답변을 작성하세요",            
+        patterns_to_remove = [
+            "(응답이 너무 짧습니다. 더 자세한 답변을 원하시면 다시 질문해주세요.)",
             "### 사용자:",
             "### 챗봇:",
             "사용자:",
             "챗봇:",
             "assistant:",
-            "user:"
+            "user:",
+            "<|im_start|>user",
+            "<|im_end|>",
+            "<|im_start|>assistant",
+            "<|im_start|>system"
         ]
         
         for pattern in patterns_to_remove:
@@ -149,7 +198,7 @@ class PolyglotKo13bChatProfile:
             else:
                 return self._improve_response_quality(clean_text)
         
-        # 4순위: 전체 텍스트에서 불필요한 부분만 제거
+        # 5순위: 전체 텍스트에서 불필요한 부분만 제거
         final_response = full_text.strip()
         logger.warning("⚠️ 경고: 특별한 응답 추출 패턴을 찾지 못했습니다. 전체 텍스트를 정리하여 반환합니다.")
         logger.info(f"최종 반환 텍스트: {final_response}")
@@ -191,7 +240,7 @@ class PolyglotKo13bChatProfile:
         # 중복 공백 제거
         improved = re.sub(r'\s+', ' ', improved)
         
-        # # 문장이 중간에 끊어진 경우 처리
+        # 문장이 중간에 끊어진 경우 처리
         # if improved.endswith(('하', '는', '을', '를', '이', '가', '의', '에', '로')):
         #     improved += '니다.'
         
@@ -203,21 +252,19 @@ class PolyglotKo13bChatProfile:
         return improved
     
     def get_generation_config(self) -> Dict[str, Any]:
-        """생성 설정 - 공식 EOS 토큰 사용"""
+        """생성 설정 - 공식 EOS 토큰 사용, 생성 파라미터 최적화"""
         return {
-            "max_new_tokens": 128,           # 64에서 128로 증가하여 완성된 답변 생성
-            "temperature": 0.3,              # 일관성 향상
+            "max_new_tokens": 128,           # 256 → 128로 줄임 (컨텍스트 길이 고려)
+            "temperature": 0.7,              # 0.9 → 0.7로 조정 (안정성 향상)
             "do_sample": True,               # 샘플링 활성화
-            "top_k": 20,                     # 품질 향상
-            "top_p": 0.8,                    # 일관성 향상
-            "repetition_penalty": 1.2,       # 반복 방지
-            "no_repeat_ngram_size": 4,       # 반복 방지
-            "pad_token_id": 2,            # 모델 기본값 사용
-            "eos_token_id": 2,               # <|endoftext|> 토큰 ID 명시적 설정
-            "use_cache": True,               # 캐시 사용으로 속도 향상
-            # "max_time": 60.0,                # 60초 타임아웃
-            # "early_stopping": False,         # False로 설정하여 <|endoftext|>까지 생성
-            "stopping_criteria": None,       # 기본 정지 기준 사용
+            "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]:

lily_llm_api/models/polyglot_ko_1_3b_chat_250822_0312.py

@@ -0,0 +1,280 @@
+#!/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_api/models/polyglot_ko_5_8b_chat.py

@@ -187,9 +187,9 @@ class PolyglotKo58bChatProfile:
             # 핵심 생성 설정
             "max_new_tokens": 128,            # 1024→256으로 줄여서 EOS 토큰을 빨리 만나도록
             # "min_new_tokens": 16,
-            "temperature": 0.7,               # 0.8→0.7로 낮춰서 일관성 향상
+            "temperature": 0.3,               # 0.8→0.7로 낮춰서 일관성 향상
             "do_sample": True,                # 샘플링 활성화
-            "top_k": 50,                      # 40→50으로 다양성 증가
+            "top_k": 20,                      # 40→50으로 다양성 증가
             "top_p": 0.9,                     # 0.95→0.9로 일관성 향상
             
             # 반복 방지 설정
@@ -199,7 +199,7 @@ class PolyglotKo58bChatProfile:
             # 토큰 설정 (중요!)
             "pad_token_id": 2,                # <|endoftext|> 토큰 ID
             "eos_token_id": 2,                # <|endoftext|> 토큰 ID (핵심!)
-            "bos_token_id": 0,                # <|startoftext|> 토큰 ID
+            # "bos_token_id": 0,                # <|startoftext|> 토큰 ID
             
             # 생성 제어 설정
             "use_cache": True,                # 캐시 사용으로 속도 향상

lily_llm_core/context_manager.py

@@ -26,8 +26,8 @@ class ContextManager:
     """대화 컨텍스트를 관리하는 클래스"""
     
     def __init__(self, 
-                 max_tokens: int = 4000,
-                 max_turns: int = 20,
+                 max_tokens: int = 2000,      # 4000 → 2000으로 줄임
+                 max_turns: int = 20,         # 20 → 10으로 줄임
                  strategy: str = "sliding_window"):
         """
         Args:
@@ -39,8 +39,12 @@ class ContextManager:
         self.max_turns = max_turns
         self.strategy = strategy
         
-        # 대화 히스토리 (deque 사용으로 효율적인 양방향 접근)
-        self.conversation_history: deque = deque(maxlen=max_turns * 2)
+        # 세션별 대화 히스토리 (세션 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 = ""
@@ -53,18 +57,57 @@ class ContextManager:
         self.enable_memory_optimization = True
         self.compression_threshold = 0.8  # 80% 도달 시 압축 시작
         
-        logger.info(f"🔧 컨텍스트 관리자 초기화: max_tokens={max_tokens}, strategy={strategy}")
+        # 🔄 자동 정리 주기 설정
+        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,
@@ -73,11 +116,14 @@ class ContextManager:
             metadata=metadata or {}
         )
         
-        self.conversation_history.append(turn)
-        self._update_context_stats()
-        self._optimize_context()
+        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)} 문자 (총 {len(self.conversation_history)} 턴)")
+        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:
@@ -85,6 +131,15 @@ class ContextManager:
         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,
@@ -93,23 +148,32 @@ class ContextManager:
             metadata=metadata or {}
         )
         
-        self.conversation_history.append(turn)
-        self._update_context_stats()
-        self._optimize_context()
+        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)} 문자 (총 {len(self.conversation_history)} 턴)")
+        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) -> str:
-        """현재 컨텍스트를 문자열로 반환"""
+    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 self.conversation_history:
+        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":
@@ -126,25 +190,34 @@ class ContextManager:
         
         return context
     
-    def get_context_for_model(self, model_name: str = "default") -> str:
-        """모델별 최적화된 컨텍스트 반환"""
+    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)
+            return self.get_context(include_system=True, session_id=session_id)
         elif "llama" in model_name.lower():
             # Llama 형식
-            return self._format_for_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)
+            return self.get_context(include_system=True, session_id=session_id)
     
-    def _format_for_llama(self) -> str:
-        """Llama 모델용 형식으로 변환"""
+    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 self.conversation_history:
+        for turn in conversation_history:
             if turn.role == "user":
                 context_parts.append(f"[INST] {turn.content} [/INST]")
             elif turn.role == "assistant":
@@ -152,9 +225,36 @@ class ContextManager:
         
         return "\n".join(context_parts)
     
-    def get_recent_context(self, turns: int = 5) -> str:
-        """최근 N개 턴의 컨텍스트만 반환"""
-        recent_turns = list(self.conversation_history)[-turns:]
+    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:
@@ -166,53 +266,84 @@ class ContextManager:
         context_parts.append("<|im_start|>assistant\n")
         return "\n".join(context_parts)
     
-    def get_context_summary(self) -> Dict[str, Any]:
-        """컨텍스트 요약 정보 반환"""
+    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 {
-            "total_turns": len(self.conversation_history),
-            "user_messages": len([t for t in self.conversation_history if t.role == "user"]),
-            "assistant_messages": len([t for t in self.conversation_history if t.role == "assistant"]),
+            "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(self.conversation_history) / self.max_turns,
-            "oldest_message": self.conversation_history[0].timestamp if self.conversation_history else None,
-            "newest_message": self.conversation_history[-1].timestamp if self.conversation_history else None
+            "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):
-        """컨텍스트 초기화"""
-        self.conversation_history.clear()
+    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("🗑️ 컨텍스트 초기화 완료")
+        logger.info(f"🗑️ 세션 {session_id} 컨텍스트 초기화 완료")
     
-    def remove_message(self, message_id: str) -> bool:
-        """특정 메시지 제거"""
-        for i, turn in enumerate(self.conversation_history):
+    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 = self.conversation_history.pop(i)
-                self._update_context_stats()
-                logger.info(f"🗑️ 메시지 제거: {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) -> bool:
-        """메시지 내용 수정"""
-        for turn in self.conversation_history:
+    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()
-                logger.info(f"✏️ 메시지 수정: {message_id}")
+                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) -> List[Dict[str, Any]]:
-        """컨텍스트 내에서 검색"""
+    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 self.conversation_history:
+        for turn in conversation_history:
             if query_lower in turn.content.lower():
                 results.append({
                     "message_id": turn.message_id,
@@ -237,49 +368,62 @@ class ContextManager:
         intersection = query_words.intersection(content_words)
         return len(intersection) / len(query_words)
     
-    def _update_context_stats(self):
-        """컨텍스트 통계 업데이트"""
-        self.current_context_length = len(self.get_context())
+    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):
-        """컨텍스트 최적화"""
+    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(self.conversation_history) / self.max_turns > self.compression_threshold:
-            self._compress_context()
+        if len(conversation_history) / self.max_turns > self.compression_threshold:
+            self._compress_context(session_id)
     
-    def _compress_context(self):
-        """컨텍스트 압축 (중요한 메시지 유지)"""
-        if len(self.conversation_history) <= self.max_turns:
+    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"🗜️ 컨텍스트 압축 시작: {len(self.conversation_history)} → {self.max_turns}")
+        logger.info(f"🗜️ 세션 {session_id} 컨텍스트 압축 시작: {len(conversation_history)} → {self.max_turns}")
         
         # 전략에 따른 압축
         if self.strategy == "sliding_window":
             # 슬라이딩 윈도우: 최근 메시지 우선
-            while len(self.conversation_history) > self.max_turns:
-                self.conversation_history.popleft()
+            while len(conversation_history) > self.max_turns:
+                conversation_history.popleft()
         
         elif self.strategy == "priority_keep":
             # 우선순위 기반: 시스템 프롬프트와 최근 메시지 우선
             # 첫 번째와 마지막 메시지는 유지
-            if len(self.conversation_history) > self.max_turns:
+            if len(conversation_history) > self.max_turns:
                 # 중간 메시지들 중 일부 제거
                 middle_start = self.max_turns // 2
-                middle_end = len(self.conversation_history) - self.max_turns // 2
+                middle_end = len(conversation_history) - self.max_turns // 2
                 
                 # 중간 부분을 요약으로 대체
-                removed_turns = list(self.conversation_history)[middle_start:middle_end]
+                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):
-                    self.conversation_history.pop(middle_start)
+                    conversation_history.pop(middle_start)
                 
                 # 요약 메시지 추가
                 summary_turn = ConversationTurn(
@@ -288,15 +432,15 @@ class ContextManager:
                     timestamp=time.time(),
                     message_id=f"summary_{int(time.time() * 1000)}"
                 )
-                self.conversation_history.insert(middle_start, summary_turn)
+                conversation_history.insert(middle_start, summary_turn)
         
         elif self.strategy == "circular":
             # 순환 버퍼: 가장 오래된 메시지 제거
-            while len(self.conversation_history) > self.max_turns:
-                self.conversation_history.popleft()
+            while len(conversation_history) > self.max_turns:
+                conversation_history.popleft()
         
-        self._update_context_stats()
-        logger.info(f"✅ 컨텍스트 압축 완료: {len(self.conversation_history)} 턴")
+        self._update_context_stats(session_id)
+        logger.info(f"✅ 세션 {session_id} 컨텍스트 압축 완료: {len(conversation_history)} 턴")
     
     def _truncate_context(self, context: str, max_length: int) -> str:
         """컨텍스트 길이 제한"""
@@ -315,13 +459,20 @@ class ContextManager:
         
         return truncated_context
     
-    def export_context(self, file_path: str = None) -> str:
-        """컨텍스트를 파일로 내보내기"""
+    def export_context(self, file_path: str = None, session_id: str = "default") -> str:
+        """컨텍스트를 파일로 내보내기 (세션별)"""
         if not file_path:
-            file_path = f"context_export_{int(time.time())}.json"
+            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": [
                 {
@@ -331,15 +482,15 @@ class ContextManager:
                     "message_id": turn.message_id,
                     "metadata": turn.metadata
                 }
-                for turn in self.conversation_history
+                for turn in conversation_history
             ],
-            "context_stats": self.get_context_summary()
+            "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"💾 컨텍스트 내보내기 완료: {file_path}")
+        logger.info(f"💾 세션 {session_id} 컨텍스트 내보내기 완료: {file_path}")
         return file_path
     
     def import_context(self, file_path: str) -> bool:
@@ -375,22 +526,33 @@ class ContextManager:
             logger.error(f"❌ 컨텍스트 가져오기 실패: {e}")
             return False
     
-    def get_memory_efficiency(self) -> Dict[str, float]:
-        """메모리 효율성 지표 반환"""
+    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 {
-            "context_utilization": len(self.conversation_history) / self.max_turns,
+            "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(self.conversation_history) / (self.max_turns * 2)),
-            "memory_fragmentation": self._calculate_fragmentation()
+            "compression_ratio": 1.0 - (len(conversation_history) / (self.max_turns * 2)),
+            "memory_fragmentation": self._calculate_fragmentation(session_id)
         }
     
-    def _calculate_fragmentation(self) -> float:
-        """메모리 단편화 정도 계산"""
-        if len(self.conversation_history) <= 1:
+    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 self.conversation_history]
+        timestamps = [turn.timestamp for turn in conversation_history]
         intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]
         
         if not intervals:
@@ -401,6 +563,136 @@ class ContextManager:
         
         # 정규화된 단편화 점수 (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()

lily_llm_core/context_manager_250822_0312.py

@@ -0,0 +1,702 @@
+#!/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.py

@@ -165,6 +165,7 @@ class LoRAManager:
             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,
@@ -173,7 +174,9 @@ class LoRAManager:
             "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,

lily_llm_core/lora_manager_250822_0312.py

@@ -0,0 +1,524 @@
+#!/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 

requirements_full_lily_250821_2206_lora.txt

@@ -0,0 +1,133 @@
+accelerate==1.10.0
+aiohappyeyeballs==2.6.1
+aiohttp==3.12.15
+aiosignal==1.4.0
+amqp==5.3.1
+annotated-types==0.7.0
+anyio==4.10.0
+attrs==25.3.0
+bcrypt==4.3.0
+billiard==4.2.1
+bitsandbytes==0.47.0
+celery==5.5.3
+certifi==2025.8.3
+cffi==1.17.1
+charset-normalizer==3.4.3
+click==8.2.1
+click-didyoumean==0.3.1
+click-plugins==1.1.1.2
+click-repl==0.3.0
+colorama==0.4.6
+cryptography==45.0.6
+dataclasses-json==0.6.7
+easyocr==1.7.2
+ecdsa==0.19.1
+einops==0.8.1
+faiss-cpu==1.12.0
+fastapi==0.116.1
+filelock==3.19.1
+frozenlist==1.7.0
+fsspec==2025.7.0
+greenlet==3.2.4
+h11==0.16.0
+httpcore==1.0.9
+httptools==0.6.4
+httpx==0.28.1
+httpx-sse==0.4.1
+huggingface-hub==0.34.4
+idna==3.10
+imageio==2.37.0
+intel-openmp==2021.4.0
+Jinja2==3.1.6
+joblib==1.5.1
+jsonpatch==1.33
+jsonpointer==3.0.0
+kombu==5.5.4
+langchain==0.3.27
+langchain-community==0.3.27
+langchain-core==0.3.74
+langchain-text-splitters==0.3.9
+langsmith==0.4.14
+lazy_loader==0.4
+lxml==6.0.0
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+marshmallow==3.26.1
+mdurl==0.1.2
+mkl==2021.4.0
+mpmath==1.3.0
+multidict==6.6.4
+mypy_extensions==1.1.0
+networkx==3.5
+ninja==1.13.0
+nltk==3.9.1
+numpy==1.26.4
+opencv-python-headless==4.11.0.86
+orjson==3.11.2
+packaging==25.0
+pandas==2.3.1
+passlib==1.7.4
+peft==0.15.0
+pillow==11.3.0
+prompt_toolkit==3.0.51
+propcache==0.3.2
+psutil==7.0.0
+pyasn1==0.6.1
+pyclipper==1.3.0.post6
+pycparser==2.22
+pydantic==2.11.7
+pydantic-settings==2.10.1
+pydantic_core==2.33.2
+PyJWT==2.10.1
+PyMuPDF==1.26.3
+pytesseract==0.3.13
+python-bidi==0.6.6
+python-dateutil==2.9.0.post0
+python-docx==1.2.0
+python-dotenv==1.1.1
+python-jose==3.5.0
+python-json-logger==3.3.0
+python-multipart==0.0.20
+python-pptx==1.0.2
+pytz==2025.2
+PyYAML==6.0.2
+redis==6.4.0
+regex==2025.7.34
+requests==2.32.5
+requests-toolbelt==1.0.0
+rsa==4.9.1
+safetensors==0.6.2
+scikit-image==0.25.2
+scikit-learn==1.7.1
+scipy==1.16.1
+sentence-transformers==2.2.2
+sentencepiece==0.2.1
+shapely==2.1.1
+six==1.17.0
+sniffio==1.3.1
+SQLAlchemy==2.0.43
+starlette==0.47.2
+sympy==1.14.0
+tbb==2021.13.1
+tenacity==9.1.2
+threadpoolctl==3.6.0
+tifffile==2025.6.11
+timm==1.0.19
+tokenizers==0.21.4
+torch==2.3.1
+torchvision==0.18.1
+tqdm==4.67.1
+transformers==4.55.2
+typing-inspect==0.9.0
+typing-inspection==0.4.1
+typing_extensions==4.14.1
+tzdata==2025.2
+urllib3==2.5.0
+uvicorn==0.35.0
+vine==5.1.0
+watchfiles==1.1.0
+wcwidth==0.2.13
+websockets==15.0.1
+xlsxwriter==3.2.5
+yarl==1.20.1
+zstandard==0.24.0

test_auto_cleanup.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""
+자동 정리 기능 테스트 스크립트
+"""
+
+import requests
+import time
+import json
+
+# API 기본 URL
+BASE_URL = "http://localhost:8001"
+
+def test_auto_cleanup_config():
+    """자동 정리 설정 테스트"""
+    print("🔧 자동 정리 설정 테스트")
+    print("=" * 50)
+    
+    # 1. 현재 설정 조회
+    print("1. 현재 자동 정리 설정 조회:")
+    response = requests.get(f"{BASE_URL}/context/auto-cleanup")
+    if response.status_code == 200:
+        config = response.json()
+        print(f"   ✅ 설정 조회 성공: {json.dumps(config, indent=2, ensure_ascii=False)}")
+    else:
+        print(f"   ❌ 설정 조회 실패: {response.status_code}")
+        return
+    
+    # 2. 설정 변경 테스트
+    print("\n2. 자동 정리 설정 변경 (4턴마다, 2분마다):")
+    new_config = {
+        "enabled": True,
+        "interval_turns": 4,
+        "interval_time": 120,
+        "strategy": "smart"
+    }
+    
+    response = requests.post(f"{BASE_URL}/context/auto-cleanup", data=new_config)
+    if response.status_code == 200:
+        result = response.json()
+        print(f"   ✅ 설정 변경 성공: {json.dumps(result, indent=2, ensure_ascii=False)}")
+    else:
+        print(f"   ❌ 설정 변경 실패: {response.status_code}")
+        return
+    
+    # 3. 변경된 설정 확인
+    print("\n3. 변경된 설정 확인:")
+    response = requests.get(f"{BASE_URL}/context/auto-cleanup")
+    if response.status_code == 200:
+        config = response.json()
+        print(f"   ✅ 설정 확인: {json.dumps(config, indent=2, ensure_ascii=False)}")
+    else:
+        print(f"   ❌ 설정 확인 실패: {response.status_code}")
+
+def test_context_generation():
+    """컨텍스트 생성 테스트 (자동 정리 트리거)"""
+    print("\n🔄 컨텍스트 생성 테스트 (자동 정리 트리거)")
+    print("=" * 50)
+    
+    session_id = f"test_session_{int(time.time())}"
+    
+    # 6턴의 대화 생성 (4턴마다 정리되도록 설정했으므로)
+    for i in range(6):
+        print(f"\n--- 턴 {i+1} ---")
+        
+        # 사용자 메시지 전송
+        user_message = f"테스트 메시지 {i+1}: 이것은 자동 정리 테스트를 위한 메시지입니다."
+        print(f"사용자: {user_message}")
+        
+        response = requests.post(f"{BASE_URL}/generate", data={
+            "prompt": user_message,
+            "use_context": True,
+            "session_id": session_id
+        })
+        
+        if response.status_code == 200:
+            result = response.json()
+            print(f"AI 응답: {result['generated_text'][:100]}...")
+        else:
+            print(f"❌ 생성 실패: {response.status_code}")
+            return
+        
+        # 컨텍스트 상태 확인
+        response = requests.get(f"{BASE_URL}/context/status")
+        if response.status_code == 200:
+            status = response.json()
+            if session_id in status.get("sessions", {}):
+                turns = status["sessions"][session_id]["turns"]
+                print(f"   📊 현재 턴 수: {turns}")
+        
+        # 잠시 대기
+        time.sleep(1)
+
+def test_manual_cleanup():
+    """수동 정리 테스트"""
+    print("\n🧹 수동 정리 테스트")
+    print("=" * 50)
+    
+    # 1. 특정 세션 수동 정리
+    session_id = f"test_session_{int(time.time())}"
+    print(f"1. 세션 {session_id} 수동 정리:")
+    
+    response = requests.post(f"{BASE_URL}/context/cleanup/{session_id}")
+    if response.status_code == 200:
+        result = response.json()
+        print(f"   ✅ 수동 정리 성공: {json.dumps(result, indent=2, ensure_ascii=False)}")
+    else:
+        print(f"   ❌ 수동 정리 실패: {response.status_code}")
+    
+    # 2. 모든 세션 수동 정리
+    print("\n2. 모든 세션 수동 정리:")
+    response = requests.post(f"{BASE_URL}/context/cleanup-all")
+    if response.status_code == 200:
+        result = response.json()
+        print(f"   ✅ 전체 정리 성공: {json.dumps(result, indent=2, ensure_ascii=False)}")
+    else:
+        print(f"   ❌ 전체 정리 실패: {response.status_code}")
+
+def test_context_status():
+    """컨텍스트 상태 확인"""
+    print("\n📊 컨텍스트 상태 확인")
+    print("=" * 50)
+    
+    response = requests.get(f"{BASE_URL}/context/status")
+    if response.status_code == 200:
+        status = response.json()
+        print(f"✅ 상태 조회 성공:")
+        print(f"   - 총 세션 수: {status.get('total_sessions', 0)}")
+        print(f"   - 최대 턴 수: {status.get('max_turns', 0)}")
+        print(f"   - 전략: {status.get('strategy', 'unknown')}")
+        
+        if "sessions" in status:
+            print("   - 세션별 정보:")
+            for session_id, session_info in status["sessions"].items():
+                print(f"     * {session_id}: {session_info['turns']} 턴")
+    else:
+        print(f"❌ 상태 조회 실패: {response.status_code}")
+
+def main():
+    """메인 테스트 함수"""
+    print("🚀 자동 정리 기능 테스트 시작")
+    print("=" * 60)
+    
+    try:
+        # 1. 자동 정리 설정 테스트
+        test_auto_cleanup_config()
+        
+        # 2. 컨텍스트 생성 테스트 (자동 정리 트리거)
+        test_context_generation()
+        
+        # 3. 수동 정리 테스트
+        test_manual_cleanup()
+        
+        # 4. 최종 상태 확인
+        test_context_status()
+        
+        print("\n🎉 모든 테스트 완료!")
+        
+    except Exception as e:
+        print(f"\n❌ 테스트 중 오류 발생: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    main()