Spaces:
Sleeping
Sleeping
| """ | |
| AI선배 My Co-Pilot - RAG Core Engine | |
| ===================================== | |
| 대학원생의 연구 기록을 영구 보존하고, 30년 뒤에도 맥락을 기억하는 Private RAG 서비스 | |
| 핵심 기능: | |
| - 로컬 Vector DB (Qdrant)로 데이터 보안 유지 | |
| - 장기 기억 보존을 위한 메타데이터 기반 검색 | |
| - OpenRouter를 통한 다양한 LLM 모델 선택 | |
| Author: AI선배 Team (공모전 MVP) | |
| """ | |
| import os | |
| import uuid | |
| from datetime import datetime | |
| from typing import List, Dict, Any, Optional | |
| from dotenv import load_dotenv | |
| # Qdrant Vector DB - 로컬 보안 중심의 벡터 저장소 | |
| from qdrant_client import QdrantClient | |
| from qdrant_client.http import models as qdrant_models | |
| from qdrant_client.http.models import ( | |
| Distance, | |
| VectorParams, | |
| ScalarQuantization, # 비용 절감을 위한 양자화 설정 | |
| ScalarQuantizationConfig, | |
| ScalarType, | |
| PointStruct, | |
| Filter, | |
| FieldCondition, | |
| MatchValue, | |
| SearchParams | |
| ) | |
| # Sentence Transformers - 한국어 지원 임베딩 모델 | |
| from sentence_transformers import SentenceTransformer | |
| # Flashrank - 검색 결과 재정렬로 정확도 향상 | |
| try: | |
| from flashrank import Ranker, RerankRequest | |
| FLASHRANK_AVAILABLE = True | |
| except ImportError: | |
| FLASHRANK_AVAILABLE = False | |
| print("⚠️ Flashrank not installed. Reranking will be skipped.") | |
| # OpenRouter LLM 호출 | |
| import httpx | |
| import json | |
| import re | |
| # PDF 파싱 (연구계획서 업로드용) | |
| try: | |
| from pypdf import PdfReader | |
| PDF_AVAILABLE = True | |
| except ImportError: | |
| PDF_AVAILABLE = False | |
| print("⚠️ pypdf not installed. PDF parsing will be disabled.") | |
| # 환경 변수 로드 | |
| load_dotenv() | |
| # Hugging Face Spaces 환경 감지 | |
| IS_HF_SPACES = os.getenv("SPACE_ID") is not None | |
| # 영구 저장 경로 (HF Spaces에서는 메모리 모드 사용) | |
| QDRANT_STORAGE_PATH = "./qdrant_data" | |
| class RAGEngine: | |
| """ | |
| RAG (Retrieval-Augmented Generation) 엔진 | |
| 대학원생의 연구 기록을 벡터 DB에 저장하고, | |
| 질문 시 관련 기억을 검색하여 LLM에게 컨텍스트로 전달합니다. | |
| 특징: | |
| - ScalarQuantization (int8): 메모리 사용량 4배 감소 (비용 절감) | |
| - 메타데이터 기반 필터링: 날짜별 기억 검색 가능 | |
| - Flashrank Reranking: 검색 정확도 향상 | |
| """ | |
| # 지원하는 OpenRouter 모델 목록 (무료 모델) | |
| SUPPORTED_MODELS = { | |
| "grok-4.1-fast": "x-ai/grok-4.1-fast:free", | |
| "glm-4.5-air": "z-ai/glm-4.5-air:free", | |
| "qwen3-235b": "qwen/qwen3-235b-a22b:free", | |
| "Nova-2-Lite": "amazon/nova-2-lite-v1:free" | |
| } | |
| def __init__(self, collection_name: str = "research_memories"): | |
| """ | |
| RAG 엔진 초기화 | |
| Args: | |
| collection_name: Qdrant 컬렉션 이름 (연구 기억 저장소) | |
| """ | |
| self.collection_name = collection_name | |
| self.openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "") | |
| # ============================================ | |
| # Qdrant 클라이언트 초기화 (싱글톤 패턴) | |
| # ============================================ | |
| # 보안 강화: 외부 서버 없이 로컬에서만 데이터 저장 | |
| # 앱 재시작 후에도 기억이 유지됨 (30년 보존!) | |
| self.qdrant_client = self._get_qdrant_client() | |
| # ============================================ | |
| # 대화 히스토리 (멀티턴 대화 지원) | |
| # ============================================ | |
| self.conversation_history: List[Dict[str, str]] = [] # 최근 대화 기록 | |
| self.max_history_turns = 10 # 최대 보존 턴 수 | |
| # ============================================ | |
| # 임베딩 모델 초기화 (다국어 지원) | |
| # ============================================ | |
| # all-MiniLM-L6-v2: 경량화된 다국어 임베딩 모델 | |
| # 한국어 연구 기록도 효과적으로 벡터화 | |
| self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2') | |
| self.embedding_dim = 384 # all-MiniLM-L6-v2의 출력 차원 | |
| # ============================================ | |
| # Flashrank Reranker 초기화 (검색 정확도 향상) | |
| # ============================================ | |
| if FLASHRANK_AVAILABLE: | |
| try: | |
| self.reranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2") | |
| except Exception as e: | |
| print(f"⚠️ Flashrank initialization failed: {e}") | |
| self.reranker = None | |
| else: | |
| self.reranker = None | |
| # ============================================ | |
| # 사용자 컨텍스트 (Onboarding 정보) | |
| # ============================================ | |
| self.user_context = { | |
| "topic": "", | |
| "keywords": [], | |
| "system_prompt": "" | |
| } | |
| # 컬렉션 초기화 | |
| self._init_collection() | |
| # 싱글톤 Qdrant 클라이언트 (클래스 변수) | |
| _qdrant_instance = None | |
| def _get_qdrant_client(cls): | |
| """ | |
| Qdrant 클라이언트 싱글톤 패턴 | |
| 여러 세션에서 동시 접근 문제 해결 | |
| HF Spaces 환경에서는 메모리 모드 사용 | |
| """ | |
| if cls._qdrant_instance is None: | |
| # HF Spaces 환경에서는 메모리 모드 강제 사용 | |
| if IS_HF_SPACES: | |
| print("🌐 Hugging Face Spaces detected, using in-memory mode") | |
| cls._qdrant_instance = QdrantClient(":memory:") | |
| else: | |
| try: | |
| cls._qdrant_instance = QdrantClient(path=QDRANT_STORAGE_PATH) | |
| except RuntimeError as e: | |
| # 이미 다른 인스턴스가 접근 중인 경우 메모리 모드로 폴백 | |
| print(f"⚠️ Qdrant storage locked, using memory mode: {e}") | |
| cls._qdrant_instance = QdrantClient(":memory:") | |
| return cls._qdrant_instance | |
| def _init_collection(self): | |
| """ | |
| Qdrant 컬렉션 생성 (ScalarQuantization 적용) | |
| 기술 포인트: | |
| - ScalarQuantization (int8): 벡터를 8비트 정수로 압축 | |
| - 메모리 사용량 75% 감소, 검색 속도 향상 | |
| - 대학원 연구 기록처럼 장기 보존이 필요한 데이터에 적합 | |
| """ | |
| collections = self.qdrant_client.get_collections().collections | |
| collection_names = [c.name for c in collections] | |
| if self.collection_name not in collection_names: | |
| # ================================================= | |
| # ScalarQuantization 설정 (기술 점수용 핵심 기능) | |
| # ================================================= | |
| # int8 양자화: float32 → int8로 압축 (4배 메모리 절감) | |
| # always_ram=True: 자주 접근하는 데이터를 RAM에 유지 | |
| quantization_config = ScalarQuantization( | |
| scalar=ScalarQuantizationConfig( | |
| type=ScalarType.INT8, # 8비트 정수 양자화 | |
| quantile=0.99, # 상위 1% 아웃라이어 무시 | |
| always_ram=True # RAM 상주로 검색 속도 최적화 | |
| ) | |
| ) | |
| self.qdrant_client.create_collection( | |
| collection_name=self.collection_name, | |
| vectors_config=VectorParams( | |
| size=self.embedding_dim, | |
| distance=Distance.COSINE # 코사인 유사도 (텍스트 검색에 최적) | |
| ), | |
| quantization_config=quantization_config # 비용 절감을 위한 양자화 | |
| ) | |
| print(f"✅ Collection '{self.collection_name}' created with ScalarQuantization (int8)") | |
| def inject_memory(self, text: str, date_str: str, category: str = "experiment", | |
| tags: Optional[List[str]] = None, experiment_name: str = "") -> bool: | |
| """ | |
| [핵심 기능] 시연용 과거 기억 주입 | |
| 특정 텍스트를 특정 날짜의 메타데이터와 함께 벡터 DB에 저장합니다. | |
| 공모전 시연에서 "3년 뒤 회상" 시나리오를 구현하기 위한 함수입니다. | |
| Args: | |
| text: 저장할 기억 텍스트 | |
| date_str: 기억의 날짜 (예: "2024-03-15") | |
| category: 기억 카테고리 (experiment, idea, meeting 등) | |
| tags: 태그 리스트 (예: ["#ablation", "#dropout"]) | |
| experiment_name: 실험 이름 (예: "Ablation Study 1차") | |
| Returns: | |
| 성공 여부 | |
| Example: | |
| >>> engine.inject_memory( | |
| ... "Ablation Study에서 Drop-out 0.3일 때 정확도 최고", | |
| ... "2024-03-15", | |
| ... "experiment", | |
| ... tags=["#ablation", "#dropout"], | |
| ... experiment_name="Dropout 비율 실험" | |
| ... ) | |
| """ | |
| try: | |
| # ============================================ | |
| # 태그 자동 추출 (텍스트에서 #태그 찾기) | |
| # ============================================ | |
| if tags is None: | |
| tags = [] | |
| # 텍스트에서 #으로 시작하는 태그 자동 추출 | |
| auto_tags = re.findall(r'#(\w+)', text) | |
| all_tags = list(set(tags + ["#" + t if not t.startswith("#") else t for t in auto_tags])) | |
| # 텍스트를 벡터로 변환 (임베딩) | |
| embedding = self.embedding_model.encode(text).tolist() | |
| # ============================================ | |
| # 장기 기억 보존을 위한 메타데이터 필터링 | |
| # ============================================ | |
| # 날짜, 카테고리, 태그 정보를 메타데이터로 저장하여 | |
| # 나중에 "2024년 실험 결과", "#ablation 실험" 같은 질문에 필터링 가능 | |
| # 날짜+시간 정밀 저장 (같은 날 여러 실험 구분) | |
| timestamp = datetime.now().isoformat() | |
| point = PointStruct( | |
| id=str(uuid.uuid4()), # 고유 ID 생성 | |
| vector=embedding, | |
| payload={ | |
| "text": text, | |
| "date": date_str, # 기억 날짜 (검색 필터용) | |
| "year": date_str.split("-")[0], # 연도 (연도별 검색용) | |
| "month": date_str.split("-")[1] if len(date_str.split("-")) > 1 else "", # 월 | |
| "category": category, # 카테고리 분류 | |
| "tags": all_tags, # 태그 리스트 (#실험1, #ablation 등) | |
| "experiment_name": experiment_name, # 실험 이름 | |
| "created_at": timestamp, # 실제 생성 시간 | |
| "is_demo_data": False # 시연용 데이터 여부 | |
| } | |
| ) | |
| # 벡터 DB에 저장 | |
| self.qdrant_client.upsert( | |
| collection_name=self.collection_name, | |
| points=[point] | |
| ) | |
| print(f"✅ Memory injected: '{text[:50]}...' (Date: {date_str})") | |
| return True | |
| except Exception as e: | |
| print(f"❌ Memory injection failed: {e}") | |
| return False | |
| def onboard_user(self, topic: str, keywords: List[str]) -> str: | |
| """ | |
| [Scenario A] 입학 첫날 - 사용자 온보딩 | |
| 연구 주제와 관심 키워드를 입력받아 시스템 프롬프트를 재설정합니다. | |
| AI가 사용자의 연구 분야에 맞춤화된 조언을 제공할 수 있게 합니다. | |
| Args: | |
| topic: 연구 주제 (예: "Private LLM 보안") | |
| keywords: 관심 키워드 리스트 (예: ["Security", "Efficiency"]) | |
| Returns: | |
| AI의 첫인사 메시지 | |
| """ | |
| # 사용자 컨텍스트 저장 | |
| self.user_context["topic"] = topic | |
| self.user_context["keywords"] = keywords | |
| # =========================================== | |
| # 개인화된 시스템 프롬프트 생성 | |
| # =========================================== | |
| # 연구 주제와 키워드를 각인시켜 맞춤형 조언 제공 | |
| keywords_str = ", ".join(keywords) | |
| self.user_context["system_prompt"] = f""" | |
| 당신은 "AI선배"입니다. 대학원생의 평생 연구 파트너로서 다음 역할을 수행합니다: | |
| 1. 연구 주제: {topic} | |
| 2. 관심 키워드: {keywords_str} | |
| 3. 역할: | |
| - 과거 연구 기록을 바탕으로 맥락 있는 조언 제공 | |
| - 실험 결과, 아이디어, 회의 내용 등을 장기 기억으로 보존 | |
| - 질문에 대해 과거 기록을 참조하여 정확한 답변 제공 | |
| 항상 친근하면서도 전문적인 톤으로 대화하세요. | |
| 한국어로 답변하세요. | |
| 과거 기록을 참조할 때는 날짜와 출처를 명시하세요. | |
| """ | |
| # 온보딩 정보도 기억으로 저장 | |
| onboard_text = f"연구 주제: {topic}, 관심 키워드: {keywords_str}" | |
| self.inject_memory( | |
| onboard_text, | |
| datetime.now().strftime("%Y-%m-%d"), | |
| category="onboarding" | |
| ) | |
| # 첫인사 생성 | |
| greeting = f"""안녕? 나는 네 평생 연구 파트너 **AI선배**야! 🎓 | |
| **{topic}** 분야의 연구를 함께 하게 되어 정말 기뻐! | |
| 특히 **{keywords_str}** 관련 연구는 정말 흥미로운 주제야. | |
| 앞으로 네 모든 연구 여정을 함께할 거야: | |
| - 📝 실험 결과와 아이디어를 기억해둘게 | |
| - 🔍 나중에 물어보면 정확히 찾아줄게 | |
| - 💡 맥락을 기반으로 조언도 해줄게 | |
| 30년 뒤에도 "그때 그 실험 결과 뭐였지?" 하고 물어보면 | |
| 바로 대답해줄 수 있어! 이제 시작해볼까? 🚀""" | |
| return greeting | |
| def _search_memories(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]: | |
| """ | |
| 벡터 DB에서 관련 기억 검색 | |
| Args: | |
| query: 검색 쿼리 | |
| top_k: 반환할 결과 수 | |
| Returns: | |
| 검색된 기억 목록 (점수, 텍스트, 메타데이터 포함) | |
| """ | |
| # 쿼리를 벡터로 변환 | |
| query_embedding = self.embedding_model.encode(query).tolist() | |
| # Qdrant에서 유사 벡터 검색 (최신 API: query_points) | |
| search_results = self.qdrant_client.query_points( | |
| collection_name=self.collection_name, | |
| query=query_embedding, | |
| limit=top_k * 2, # Reranking을 위해 더 많이 검색 | |
| with_payload=True | |
| ).points | |
| # 결과 정리 | |
| memories = [] | |
| for result in search_results: | |
| memories.append({ | |
| "score": result.score, | |
| "text": result.payload.get("text", ""), | |
| "date": result.payload.get("date", "Unknown"), | |
| "category": result.payload.get("category", "general"), | |
| "metadata": result.payload | |
| }) | |
| # =========================================== | |
| # Flashrank Reranking (검색 보정) | |
| # =========================================== | |
| # 초기 벡터 검색 결과를 교차 인코더로 재정렬하여 정확도 향상 | |
| if self.reranker and memories: | |
| try: | |
| passages = [{"id": i, "text": m["text"]} for i, m in enumerate(memories)] | |
| rerank_request = RerankRequest(query=query, passages=passages) | |
| rerank_results = self.reranker.rerank(rerank_request) | |
| # 재정렬된 순서로 기억 정렬 | |
| reranked_memories = [] | |
| for result in rerank_results[:top_k]: | |
| idx = result["id"] | |
| memories[idx]["rerank_score"] = result["score"] | |
| reranked_memories.append(memories[idx]) | |
| return reranked_memories | |
| except Exception as e: | |
| print(f"⚠️ Reranking failed: {e}") | |
| return memories[:top_k] | |
| def _call_openrouter(self, messages: List[Dict], model_name: str) -> str: | |
| """ | |
| OpenRouter API를 통한 LLM 호출 (non-streaming) | |
| Args: | |
| messages: 대화 메시지 목록 | |
| model_name: 사용할 모델 이름 | |
| Returns: | |
| LLM 응답 텍스트 | |
| """ | |
| # API 키 확인 | |
| if not self.openrouter_api_key or self.openrouter_api_key == "your_openrouter_api_key_here": | |
| print("⚠️ OpenRouter API key not set. Using demo mode.") | |
| return self._generate_mock_response(messages) | |
| # OpenRouter API 호출 | |
| model_id = self.SUPPORTED_MODELS.get(model_name, "x-ai/grok-4.1-fast:free") | |
| try: | |
| import json as json_lib | |
| json_data = json_lib.dumps({ | |
| "model": model_id, | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 16384 | |
| }, ensure_ascii=False).encode('utf-8') | |
| with httpx.Client(timeout=60.0) as client: | |
| response = client.post( | |
| "https://openrouter.ai/api/v1/chat/completions", | |
| headers={ | |
| "Authorization": f"Bearer {self.openrouter_api_key}", | |
| "Content-Type": "application/json; charset=utf-8", | |
| "HTTP-Referer": "https://ai-sunbae.demo", | |
| "X-Title": "AI-Sunbae My Co-Pilot" | |
| }, | |
| content=json_data | |
| ) | |
| if response.status_code == 200: | |
| result = response.json() | |
| return result["choices"][0]["message"]["content"] | |
| else: | |
| print(f"❌ OpenRouter API error: {response.status_code}") | |
| print(response.text) | |
| return self._generate_mock_response(messages) | |
| except Exception as e: | |
| print(f"❌ OpenRouter call failed: {e}") | |
| return self._generate_mock_response(messages) | |
| def _call_openrouter_stream(self, messages: List[Dict], model_name: str): | |
| """ | |
| OpenRouter API 스트리밍 호출 (Generator) | |
| Args: | |
| messages: 대화 메시지 목록 | |
| model_name: 사용할 모델 이름 | |
| Yields: | |
| 응답 텍스트 청크 | |
| """ | |
| import json as json_lib | |
| # API 키 확인 | |
| if not self.openrouter_api_key or self.openrouter_api_key == "your_openrouter_api_key_here": | |
| print("⚠️ OpenRouter API key not set. Using demo mode.") | |
| mock_response = self._generate_mock_response(messages) | |
| # Mock 응답을 스트리밍처럼 chunk로 나눠서 반환 | |
| for word in mock_response.split(" "): | |
| yield word + " " | |
| return | |
| model_id = self.SUPPORTED_MODELS.get(model_name, "x-ai/grok-4.1-fast:free") | |
| try: | |
| json_data = json_lib.dumps({ | |
| "model": model_id, | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 16384, | |
| "stream": True # 스트리밍 활성화 | |
| }, ensure_ascii=False).encode('utf-8') | |
| with httpx.Client(timeout=120.0) as client: | |
| with client.stream( | |
| "POST", | |
| "https://openrouter.ai/api/v1/chat/completions", | |
| headers={ | |
| "Authorization": f"Bearer {self.openrouter_api_key}", | |
| "Content-Type": "application/json; charset=utf-8", | |
| "HTTP-Referer": "https://ai-sunbae.demo", | |
| "X-Title": "AI-Sunbae My Co-Pilot" | |
| }, | |
| content=json_data | |
| ) as response: | |
| if response.status_code != 200: | |
| print(f"❌ OpenRouter API error: {response.status_code}") | |
| yield self._generate_mock_response(messages) | |
| return | |
| for line in response.iter_lines(): | |
| if line.startswith("data: "): | |
| data = line[6:] # "data: " 제거 | |
| if data == "[DONE]": | |
| break | |
| try: | |
| chunk = json_lib.loads(data) | |
| delta = chunk.get("choices", [{}])[0].get("delta", {}) | |
| content = delta.get("content", "") | |
| if content: | |
| yield content | |
| except json_lib.JSONDecodeError: | |
| continue | |
| except Exception as e: | |
| print(f"❌ OpenRouter stream failed: {e}") | |
| yield self._generate_mock_response(messages) | |
| def _generate_mock_response(self, messages: List[Dict]) -> str: | |
| """ | |
| 데모 모드용 Mock 응답 생성 | |
| API 키가 없거나 호출 실패 시 사용되는 시연용 응답입니다. | |
| """ | |
| last_message = messages[-1]["content"] if messages else "" | |
| # 시연 시나리오에 맞는 Mock 응답 | |
| if "2024" in last_message or "작년" in last_message or "ablation" in last_message.lower(): | |
| return """📚 **과거 기록 검색 결과** | |
| 2024년 3월 15일 기록에 따르면, **Ablation Study 실험**에서 다음과 같은 결과를 얻었습니다: | |
| | Drop-out 비율 | 정확도 | | |
| |--------------|--------| | |
| | 0.1 | 87.2% | | |
| | 0.2 | 89.5% | | |
| | **0.3** | **91.8%** ← 최고 | | |
| | 0.4 | 88.3% | | |
| **Drop-out 비율 0.3**일 때 가장 높은 정확도(91.8%)를 기록했네요! | |
| 이 결과는 과적합(overfitting)을 적절히 방지하면서도 모델의 표현력을 유지하는 최적점을 찾은 것으로 보입니다. | |
| --- | |
| *🕐 이 정보는 2024-03-15에 기록된 실험 결과입니다.*""" | |
| if "연구" in last_message or "주제" in last_message: | |
| return """네 연구 주제에 대해 알려줄게! | |
| 현재 설정된 연구 주제: **Private LLM** | |
| 관심 키워드: **Security, Efficiency** | |
| 이 분야는 정말 핫한 주제야! 특히 데이터 프라이버시와 모델 경량화의 균형을 찾는 것이 핵심이지. | |
| 궁금한 거 있으면 언제든 물어봐! 🚀""" | |
| return f"""네 질문을 이해했어! | |
| "{last_message[:50]}{'...' if len(last_message) > 50 else ''}"에 대해 답변할게. | |
| 현재 **데모 모드**로 작동 중이야. 실제 LLM 응답을 받으려면 `.env` 파일에 OpenRouter API 키를 설정해줘! | |
| 그래도 핵심 기능(기억 저장/검색)은 정상 작동하니까 걱정 마! 😊""" | |
| # ============================================ | |
| # 대화 의도 분류 시스템 (Intent Classification) | |
| # ============================================ | |
| # 의도 분류 정의 | |
| class IntentType: | |
| QUESTION = "question" # 일반 질문 (RAG 검색 필요) | |
| MEMORY_SAVE = "memory_save" # 기억 저장 명령어 | |
| EXPERIMENT = "experiment" # 실험 기록/결과 공유 | |
| IDEA = "idea" # 아이디어/가설 공유 | |
| MEETING = "meeting" # 회의/미팅 내용 | |
| PAPER = "paper" # 논문 관련 | |
| TODO = "todo" # 할 일/계획 | |
| DAILY = "daily" # 일상 대화 (저장 불필요) | |
| COMPARISON = "comparison" # 비교 요청 | |
| SEARCH = "search" # 검색 요청 | |
| GREETING = "greeting" # 인사 | |
| def classify_intent(self, user_input: str) -> Dict[str, Any]: | |
| """ | |
| 사용자 입력의 의도를 분류하고 처리 방법 결정 | |
| Args: | |
| user_input: 사용자 입력 텍스트 | |
| Returns: | |
| { | |
| "intent": 분류된 의도, | |
| "should_save": 저장 필요 여부, | |
| "category": 저장 시 카테고리, | |
| "confidence": 신뢰도 (0-1), | |
| "extracted_info": 추출된 정보 | |
| } | |
| """ | |
| input_lower = user_input.lower() | |
| result = { | |
| "intent": "question", | |
| "should_save": False, | |
| "category": None, | |
| "confidence": 0.5, | |
| "extracted_info": {}, | |
| "needs_llm_classification": False | |
| } | |
| # ============================================ | |
| # 1차: 규칙 기반 분류 (빠른 패턴 매칭) | |
| # ============================================ | |
| # 1. 명시적 저장 명령어 (최우선) | |
| save_patterns = [ | |
| (r"기억해줘[:\s]*(.+)", "memory_save"), | |
| (r"이거\s*기억해[:\s]*(.+)", "memory_save"), | |
| (r"메모해줘[:\s]*(.+)", "memory_save"), | |
| (r"저장해줘[:\s]*(.+)", "memory_save"), | |
| (r"remember[:\s]+(.+)", "memory_save"), | |
| ] | |
| for pattern, intent in save_patterns: | |
| match = re.search(pattern, user_input, re.IGNORECASE) | |
| if match: | |
| return { | |
| "intent": intent, | |
| "should_save": True, | |
| "category": "user_memo", | |
| "confidence": 0.95, | |
| "extracted_info": {"content": match.group(1).strip()}, | |
| "needs_llm_classification": False | |
| } | |
| # 2. 비교 요청 | |
| compare_patterns = [ | |
| r"(.+)\s*vs\.?\s*(.+)", | |
| r"(.+)랑\s*(.+)\s*비교", | |
| r"비교해\s*줘", | |
| r"차이가\s*뭐", | |
| ] | |
| for pattern in compare_patterns: | |
| if re.search(pattern, user_input, re.IGNORECASE): | |
| return { | |
| "intent": "comparison", | |
| "should_save": False, | |
| "category": None, | |
| "confidence": 0.9, | |
| "extracted_info": {}, | |
| "needs_llm_classification": False | |
| } | |
| # 3. 인사/일상 대화 (저장 불필요) | |
| greeting_patterns = [ | |
| r"^(안녕|하이|헬로|hi|hello|반가워)", | |
| r"^(고마워|감사|땡큐|thanks)", | |
| r"^(잘\s*있어|잘\s*가|바이|bye)", | |
| r"^(네|응|ㅇㅇ|ㅋㅋ|ㅎㅎ|오키|ok)$", | |
| r"^(뭐해|뭐하고\s*있어|심심해)$", | |
| ] | |
| for pattern in greeting_patterns: | |
| if re.search(pattern, input_lower): | |
| return { | |
| "intent": "greeting", | |
| "should_save": False, | |
| "category": None, | |
| "confidence": 0.95, | |
| "extracted_info": {}, | |
| "needs_llm_classification": False | |
| } | |
| # 4. 질문 패턴 (RAG 검색 필요) | |
| question_patterns = [ | |
| r"\?$", | |
| r"(뭐였지|뭐야|뭐지|뭐냐)", | |
| r"(어떻게|어떤|어디|언제|왜|누가)", | |
| r"(알려줘|설명해|가르쳐)", | |
| r"(했었지|했나|했어)", | |
| r"(찾아줘|검색해|조회해)", | |
| ] | |
| for pattern in question_patterns: | |
| if re.search(pattern, user_input, re.IGNORECASE): | |
| return { | |
| "intent": "question", | |
| "should_save": False, | |
| "category": None, | |
| "confidence": 0.85, | |
| "extracted_info": {}, | |
| "needs_llm_classification": False | |
| } | |
| # 5. 실험 관련 키워드 | |
| experiment_keywords = [ | |
| "실험", "테스트", "결과", "정확도", "accuracy", "loss", | |
| "epoch", "학습", "training", "모델", "성능", "f1", "precision", | |
| "recall", "auc", "dropout", "learning rate", "batch", | |
| ] | |
| exp_score = sum(1 for kw in experiment_keywords if kw in input_lower) | |
| if exp_score >= 2: | |
| result = { | |
| "intent": "experiment", | |
| "should_save": True, | |
| "category": "experiment", | |
| "confidence": min(0.5 + exp_score * 0.1, 0.9), | |
| "extracted_info": {}, | |
| "needs_llm_classification": True # LLM 확인 필요 | |
| } | |
| # 6. 아이디어/가설 관련 | |
| idea_keywords = ["아이디어", "생각", "가설", "hypothesis", "제안", "시도해볼"] | |
| if any(kw in input_lower for kw in idea_keywords): | |
| result = { | |
| "intent": "idea", | |
| "should_save": True, | |
| "category": "idea", | |
| "confidence": 0.75, | |
| "extracted_info": {}, | |
| "needs_llm_classification": True | |
| } | |
| # 7. 회의/미팅 관련 | |
| meeting_keywords = ["미팅", "회의", "논의", "피드백", "교수님", "면담", "세미나"] | |
| if any(kw in input_lower for kw in meeting_keywords): | |
| result = { | |
| "intent": "meeting", | |
| "should_save": True, | |
| "category": "meeting", | |
| "confidence": 0.8, | |
| "extracted_info": {}, | |
| "needs_llm_classification": True | |
| } | |
| # 8. 논문 관련 | |
| paper_keywords = ["논문", "paper", "리뷰", "인용", "참고문헌", "저널", "학회"] | |
| if any(kw in input_lower for kw in paper_keywords): | |
| result = { | |
| "intent": "paper", | |
| "should_save": True, | |
| "category": "paper_review", | |
| "confidence": 0.75, | |
| "extracted_info": {}, | |
| "needs_llm_classification": True | |
| } | |
| # 9. 할 일/계획 | |
| todo_keywords = ["해야", "할일", "todo", "계획", "다음에", "나중에", "예정"] | |
| if any(kw in input_lower for kw in todo_keywords): | |
| result = { | |
| "intent": "todo", | |
| "should_save": True, | |
| "category": "todo", | |
| "confidence": 0.75, | |
| "extracted_info": {}, | |
| "needs_llm_classification": True | |
| } | |
| return result | |
| def classify_with_llm(self, user_input: str, rule_result: Dict[str, Any], | |
| model_name: str = "grok-4.1-fast") -> Dict[str, Any]: | |
| """ | |
| LLM을 사용한 2차 의도 분류 (애매한 경우) | |
| 규칙 기반 분류의 신뢰도가 낮거나, 저장 여부가 불확실할 때 사용 | |
| Args: | |
| user_input: 사용자 입력 | |
| rule_result: 규칙 기반 분류 결과 | |
| model_name: 사용할 LLM 모델 | |
| Returns: | |
| 최종 분류 결과 | |
| """ | |
| # LLM 분류가 필요하지 않으면 규칙 기반 결과 반환 | |
| if not rule_result.get("needs_llm_classification", False): | |
| return rule_result | |
| # 신뢰도가 충분히 높으면 LLM 호출 스킵 | |
| if rule_result.get("confidence", 0) >= 0.85: | |
| return rule_result | |
| # LLM 분류 프롬프트 | |
| classification_prompt = f"""다음 사용자 입력을 분석해서 JSON으로 응답해줘. | |
| 사용자 입력: "{user_input}" | |
| 분류 기준: | |
| 1. 저장할 가치가 있는 정보인가? (연구 기록, 실험 결과, 아이디어, 회의 내용 등) | |
| 2. 단순 질문인가? (과거 기록 검색이 필요한 질문) | |
| 3. 일상 대화인가? (저장 불필요) | |
| 다음 JSON 형식으로만 응답해 (다른 텍스트 없이): | |
| {{ | |
| "should_save": true/false, | |
| "category": "experiment|idea|meeting|paper|todo|question|daily", | |
| "reason": "판단 이유 (한 문장)", | |
| "key_info": "저장할 핵심 정보 (저장할 경우)" | |
| }}""" | |
| messages = [ | |
| {"role": "system", "content": "너는 텍스트 분류기야. JSON만 출력해."}, | |
| {"role": "user", "content": classification_prompt} | |
| ] | |
| try: | |
| response = self._call_openrouter(messages, model_name) | |
| # JSON 추출 | |
| json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL) | |
| if json_match: | |
| llm_result = json.loads(json_match.group()) | |
| return { | |
| "intent": llm_result.get("category", rule_result["intent"]), | |
| "should_save": llm_result.get("should_save", rule_result["should_save"]), | |
| "category": llm_result.get("category", rule_result["category"]), | |
| "confidence": 0.9, # LLM 판단이므로 높은 신뢰도 | |
| "extracted_info": { | |
| "key_info": llm_result.get("key_info", ""), | |
| "reason": llm_result.get("reason", "") | |
| }, | |
| "needs_llm_classification": False | |
| } | |
| except Exception as e: | |
| print(f"⚠️ LLM classification failed: {e}") | |
| # LLM 호출 실패 시 규칙 기반 결과 반환 | |
| return rule_result | |
| def process_with_intent(self, user_input: str, model_name: str = "grok-4.1-fast") -> Dict[str, Any]: | |
| """ | |
| 의도 분류 기반 통합 처리 | |
| 사용자 입력을 분류하고, 의도에 따라 적절한 처리를 수행 | |
| Args: | |
| user_input: 사용자 입력 | |
| model_name: 사용할 LLM 모델 | |
| Returns: | |
| 처리 결과 { | |
| "intent": 분류된 의도, | |
| "action_taken": 수행된 액션, | |
| "response": 응답, | |
| "saved": 저장 여부, | |
| "memories": 검색된 기억 (질문인 경우) | |
| } | |
| """ | |
| # 1차: 규칙 기반 분류 | |
| intent_result = self.classify_intent(user_input) | |
| # 2차: 필요시 LLM 분류 | |
| if intent_result.get("needs_llm_classification"): | |
| intent_result = self.classify_with_llm(user_input, intent_result, model_name) | |
| intent = intent_result["intent"] | |
| result = { | |
| "intent": intent, | |
| "classification": intent_result, | |
| "action_taken": None, | |
| "response": None, | |
| "saved": False, | |
| "memories": [] | |
| } | |
| # ============================================ | |
| # 의도별 처리 분기 | |
| # ============================================ | |
| # Case 1: 명시적 저장 명령 | |
| if intent == "memory_save": | |
| content = intent_result["extracted_info"].get("content", user_input) | |
| save_result = self.save_smart_memo(content) | |
| result["action_taken"] = "smart_memo_save" | |
| result["saved"] = save_result["success"] | |
| result["response"] = save_result["message"] | |
| return result | |
| # Case 2: 인사/일상 (저장 없이 응답) | |
| if intent in ["greeting", "daily"]: | |
| answer = self.get_answer(user_input, model_name, include_memories=False) | |
| result["action_taken"] = "chat_response" | |
| result["response"] = answer["answer"] | |
| return result | |
| # Case 3: 비교 요청 | |
| if intent == "comparison": | |
| compare_result = self.check_comparison_command(user_input) | |
| if compare_result and compare_result.get("is_comparison_command"): | |
| result["action_taken"] = "comparison" | |
| result["response"] = compare_result["response"] | |
| result["memories"] = compare_result["result"].get("experiments", []) | |
| return result | |
| # Case 4: 질문 (RAG 검색) | |
| if intent == "question": | |
| answer = self.get_answer(user_input, model_name, include_memories=True) | |
| result["action_taken"] = "rag_search" | |
| result["response"] = answer["answer"] | |
| result["memories"] = answer["memories"] | |
| return result | |
| # Case 5: 저장이 필요한 정보 (실험, 아이디어, 회의, 논문, 할일) | |
| if intent_result.get("should_save", False) and intent in ["experiment", "idea", "meeting", "paper", "todo"]: | |
| # 스마트 메모로 저장 | |
| save_result = self.save_smart_memo(user_input) | |
| # 저장 후 응답도 생성 | |
| answer = self.get_answer(user_input, model_name, include_memories=True) | |
| category_emoji = { | |
| "experiment": "🧪", | |
| "idea": "💡", | |
| "meeting": "👥", | |
| "paper": "📄", | |
| "todo": "📋" | |
| } | |
| emoji = category_emoji.get(intent, "📝") | |
| result["action_taken"] = "save_and_respond" | |
| result["saved"] = save_result["success"] | |
| result["response"] = f"{emoji} **자동 저장됨** (분류: {intent})\n\n{answer['answer']}" | |
| result["memories"] = answer["memories"] | |
| return result | |
| # 기본: 질문으로 처리 | |
| answer = self.get_answer(user_input, model_name, include_memories=True) | |
| result["action_taken"] = "default_rag" | |
| result["response"] = answer["answer"] | |
| result["memories"] = answer["memories"] | |
| return result | |
| def get_answer( | |
| self, | |
| query: str, | |
| model_name: str = "grok-4.1-fast", | |
| include_memories: bool = True | |
| ) -> Dict[str, Any]: | |
| """ | |
| [핵심 기능] RAG 기반 질문 응답 | |
| 1. 벡터 DB에서 관련 기억 검색 | |
| 2. Flashrank로 검색 결과 재정렬 | |
| 3. 검색된 기억을 컨텍스트로 LLM에 전달 | |
| 4. LLM 응답과 참조 기억 반환 | |
| Args: | |
| query: 사용자 질문 | |
| model_name: 사용할 LLM 모델 | |
| include_memories: 기억 검색 포함 여부 | |
| Returns: | |
| { | |
| "answer": LLM 응답, | |
| "memories": 참조한 기억 목록, | |
| "model": 사용된 모델 | |
| } | |
| """ | |
| memories = [] | |
| context = "" | |
| # =========================================== | |
| # 장기 기억 검색 (Hybrid Search + Reranking) | |
| # =========================================== | |
| if include_memories: | |
| memories = self._search_memories(query, top_k=3) | |
| if memories: | |
| context = "\n\n📚 **관련 과거 기록:**\n" | |
| for i, mem in enumerate(memories, 1): | |
| context += f"\n[기록 {i}] (날짜: {mem['date']}, 분류: {mem['category']})\n" | |
| context += f"{mem['text']}\n" | |
| # 시스템 프롬프트 구성 | |
| system_prompt = self.user_context.get("system_prompt", "") | |
| if not system_prompt: | |
| system_prompt = """ | |
| 당신은 사용자의 학위 논문 작성을 돕고 연구 고민을 함께 나누는 'AI 선배'입니다. | |
| 사용자는 대학원생이며, 깊이 있는 전문 지식과 실질적인 해결책을 원합니다. | |
| [답변 원칙] | |
| 1. **과거 기록 활용**: 제공된 메모리(Context)가 있다면 반드시 이를 현재 질문과 연결 지어 설명하세요. (예: "지난 3월 15일 기록하신 아이디어와 연결해보면...") | |
| 2. **구체적 제안**: "공부해보세요" 대신 "이 논문의 3장을 참고하세요" 또는 "다음 코드를 실행해보세요"라고 구체적으로 지시하세요. | |
| 3. **코드 및 수식**: 코드는 바로 실행 가능하도록 작성하고, 수식은 LaTeX 포맷으로 명확히 표기하세요. | |
| 4. **검증 모드**: 사용자의 연구 방법론에 허점이 있다면, "이 부분은 리뷰어에게 지적받을 수 있습니다"와 같이 예상되는 반론을 미리 짚어주세요. | |
| 항상 연구 파트너로서 존중과 응원의 태도를 유지하되, 학문적 엄밀함은 타협하지 마세요. | |
| """ | |
| # ============================================ | |
| # 메시지 구성 (대화 히스토리 포함) | |
| # ============================================ | |
| messages = [ | |
| {"role": "system", "content": system_prompt}, | |
| ] | |
| if context: | |
| messages.append({ | |
| "role": "system", | |
| "content": f"다음은 사용자의 과거 연구 기록입니다. 질문과 관련이 있다면 참조하세요:{context}" | |
| }) | |
| # 이전 대화 히스토리 추가 (멀티턴 대화 지원) | |
| messages.extend(self.conversation_history) | |
| # 현재 질문 추가 | |
| messages.append({"role": "user", "content": query}) | |
| # LLM 호출 | |
| answer = self._call_openrouter(messages, model_name) | |
| # 대화 히스토리에 현재 대화 추가 | |
| self.add_to_history("user", query) | |
| self.add_to_history("assistant", answer) | |
| return { | |
| "answer": answer, | |
| "memories": memories, | |
| "model": model_name, | |
| "context_used": bool(memories) | |
| } | |
| def get_answer_stream( | |
| self, | |
| query: str, | |
| model_name: str = "grok-4.1-fast", | |
| include_memories: bool = True | |
| ): | |
| """ | |
| [핵심 기능] RAG 기반 질문 응답 (스트리밍) | |
| Args: | |
| query: 사용자 질문 | |
| model_name: 사용할 LLM 모델 | |
| include_memories: 기억 검색 포함 여부 | |
| Yields: | |
| {"type": "memories", "data": [...]} - 먼저 기억 정보 전송 | |
| {"type": "token", "data": "..."} - 토큰 스트리밍 | |
| {"type": "done", "data": "full_answer"} - 완료 | |
| """ | |
| memories = [] | |
| context = "" | |
| # 장기 기억 검색 | |
| if include_memories: | |
| memories = self._search_memories(query, top_k=3) | |
| if memories: | |
| context = "\n\n📚 **관련 과거 기록:**\n" | |
| for i, mem in enumerate(memories, 1): | |
| context += f"\n[기록 {i}] (날짜: {mem['date']}, 분류: {mem['category']})\n" | |
| context += f"{mem['text']}\n" | |
| # 먼저 기억 정보 전송 | |
| yield {"type": "memories", "data": memories} | |
| # 시스템 프롬프트 구성 | |
| system_prompt = self.user_context.get("system_prompt", "") | |
| if not system_prompt: | |
| system_prompt = """ | |
| 당신은 사용자의 학위 논문 작성을 돕고 연구 고민을 함께 나누는 'AI 선배'입니다. | |
| 사용자는 대학원생이며, 깊이 있는 전문 지식과 실질적인 해결책을 원합니다. | |
| [답변 원칙] | |
| 1. **과거 기록 활용**: 제공된 메모리(Context)가 있다면 반드시 이를 현재 질문과 연결 지어 설명하세요. (예: "지난 3월 15일 기록하신 아이디어와 연결해보면...") | |
| 2. **구체적 제안**: "공부해보세요" 대신 "이 논문의 3장을 참고하세요" 또는 "다음 코드를 실행해보세요"라고 구체적으로 지시하세요. | |
| 3. **코드 및 수식**: 코드는 바로 실행 가능하도록 작성하고, 수식은 LaTeX 포맷으로 명확히 표기하세요. | |
| 4. **검증 모드**: 사용자의 연구 방법론에 허점이 있다면, "이 부분은 리뷰어에게 지적받을 수 있습니다"와 같이 예상되는 반론을 미리 짚어주세요. | |
| 항상 연구 파트너로서 존중과 응원의 태도를 유지하되, 학문적 엄밀함은 타협하지 마세요. | |
| """ | |
| # 메시지 구성 | |
| messages = [{"role": "system", "content": system_prompt}] | |
| if context: | |
| messages.append({ | |
| "role": "system", | |
| "content": f"다음은 사용자의 과거 연구 기록입니다. 질문과 관련이 있다면 참조하세요:{context}" | |
| }) | |
| messages.extend(self.conversation_history) | |
| messages.append({"role": "user", "content": query}) | |
| # 스트리밍 LLM 호출 | |
| full_answer = "" | |
| for chunk in self._call_openrouter_stream(messages, model_name): | |
| full_answer += chunk | |
| yield {"type": "token", "data": chunk} | |
| # 대화 히스토리 업데이트 | |
| self.add_to_history("user", query) | |
| self.add_to_history("assistant", full_answer) | |
| yield {"type": "done", "data": full_answer} | |
| def add_document(self, text: str, metadata: Optional[Dict] = None) -> bool: | |
| """ | |
| 문서/메모 추가 (일반 용도) | |
| Args: | |
| text: 저장할 텍스트 | |
| metadata: 추가 메타데이터 | |
| Returns: | |
| 성공 여부 | |
| """ | |
| if metadata is None: | |
| metadata = {} | |
| date_str = metadata.get("date", datetime.now().strftime("%Y-%m-%d")) | |
| category = metadata.get("category", "note") | |
| return self.inject_memory(text, date_str, category) | |
| # ============================================ | |
| # 대화 히스토리 관리 (멀티턴 대화 지원) | |
| # ============================================ | |
| def add_to_history(self, role: str, content: str): | |
| """ | |
| 대화 히스토리에 메시지 추가 | |
| Args: | |
| role: "user" 또는 "assistant" | |
| content: 메시지 내용 | |
| """ | |
| self.conversation_history.append({"role": role, "content": content}) | |
| # 최대 턴 수 초과 시 오래된 대화 제거 (시스템 메시지 제외) | |
| if len(self.conversation_history) > self.max_history_turns * 2: | |
| self.conversation_history = self.conversation_history[-self.max_history_turns * 2:] | |
| def clear_history(self): | |
| """대화 히스토리 초기화""" | |
| self.conversation_history = [] | |
| def clear_all_data(self): | |
| """모든 데이터 초기화 (벡터 DB + 대화 히스토리)""" | |
| # 1. 대화 히스토리 초기화 | |
| self.conversation_history = [] | |
| # 2. Qdrant 벡터 DB의 모든 데이터 삭제 | |
| try: | |
| # 컬렉션 삭제 후 재생성 | |
| self.qdrant_client.delete_collection(self.collection_name) | |
| # 컬렉션 재생성 | |
| quantization_config = ScalarQuantization( | |
| scalar=ScalarQuantizationConfig( | |
| type=ScalarType.INT8, | |
| quantile=0.99, | |
| always_ram=True | |
| ) | |
| ) | |
| self.qdrant_client.create_collection( | |
| collection_name=self.collection_name, | |
| vectors_config=VectorParams( | |
| size=self.embedding_dim, | |
| distance=Distance.COSINE | |
| ), | |
| quantization_config=quantization_config | |
| ) | |
| print(f"✅ 컬렉션 '{self.collection_name}' 초기화 완료") | |
| except Exception as e: | |
| print(f"⚠️ 컬렉션 초기화 실패: {e}") | |
| def get_history_for_llm(self) -> List[Dict[str, str]]: | |
| """LLM에 전달할 대화 히스토리 반환""" | |
| return self.conversation_history.copy() | |
| # ============================================ | |
| # PDF 파싱 (연구계획서 처리) - 개선된 버전 | |
| # ============================================ | |
| def parse_pdf(self, pdf_file) -> str: | |
| """ | |
| PDF 파일에서 텍스트 추출 | |
| Args: | |
| pdf_file: 업로드된 PDF 파일 객체 | |
| Returns: | |
| 추출된 텍스트 | |
| """ | |
| if not PDF_AVAILABLE: | |
| return "PDF 파싱 라이브러리가 설치되지 않았습니다." | |
| try: | |
| reader = PdfReader(pdf_file) | |
| text_parts = [] | |
| for page in reader.pages: | |
| page_text = page.extract_text() | |
| if page_text: | |
| text_parts.append(page_text) | |
| full_text = "\n".join(text_parts) | |
| return full_text | |
| except Exception as e: | |
| print(f"❌ PDF parsing failed: {e}") | |
| return "" | |
| def _extract_sections_from_text(self, text: str) -> List[Dict[str, str]]: | |
| """ | |
| 텍스트에서 섹션별로 구조화하여 추출 | |
| 연구계획서의 일반적인 구조: | |
| - 연구 배경/목적 | |
| - 연구 방법 | |
| - 기대 효과 | |
| - 참고문헌 | |
| Args: | |
| text: 전체 텍스트 | |
| Returns: | |
| 섹션 리스트 [{section_name, content}, ...] | |
| """ | |
| sections = [] | |
| # 섹션 헤더 패턴 (한국어/영어) | |
| section_patterns = [ | |
| # 번호 기반 (1. 2. 3. 또는 I. II. III.) | |
| r'(?:^|\n)\s*(?:\d+\.|[IVX]+\.)\s*([^\n]+)', | |
| # 제목 기반 (【】, ■, ●, ○ 등) | |
| r'(?:^|\n)\s*[【■●○◆▶]\s*([^\n]+)', | |
| # 일반적인 섹션 키워드 | |
| r'(?:^|\n)\s*(연구\s*(?:배경|목적|목표|방법|내용|범위|계획|일정)|' | |
| r'기대\s*효과|예상\s*결과|참고\s*문헌|연구비\s*사용|' | |
| r'Abstract|Introduction|Background|Method|Result|Conclusion|Reference)[:：]?\s*', | |
| ] | |
| # 섹션 분리 시도 | |
| current_section = "개요" | |
| current_content = [] | |
| lines = text.split('\n') | |
| for line in lines: | |
| is_header = False | |
| for pattern in section_patterns: | |
| match = re.search(pattern, line, re.IGNORECASE) | |
| if match: | |
| # 이전 섹션 저장 | |
| if current_content: | |
| content = '\n'.join(current_content).strip() | |
| if len(content) > 50: # 너무 짧은 내용 제외 | |
| sections.append({ | |
| "section_name": current_section, | |
| "content": content | |
| }) | |
| # 새 섹션 시작 | |
| current_section = match.group(1).strip() if match.groups() else line.strip() | |
| current_content = [] | |
| is_header = True | |
| break | |
| if not is_header and line.strip(): | |
| current_content.append(line.strip()) | |
| # 마지막 섹션 저장 | |
| if current_content: | |
| content = '\n'.join(current_content).strip() | |
| if len(content) > 50: | |
| sections.append({ | |
| "section_name": current_section, | |
| "content": content | |
| }) | |
| return sections | |
| def _extract_key_info_with_llm(self, text: str, doc_type: str = "연구계획서") -> Dict[str, Any]: | |
| """ | |
| LLM을 사용하여 문서에서 주요 정보 추출 | |
| Args: | |
| text: 문서 텍스트 | |
| doc_type: 문서 유형 | |
| Returns: | |
| 추출된 주요 정보 | |
| """ | |
| # 텍스트가 너무 길면 앞부분만 사용 | |
| max_length = 4000 | |
| truncated_text = text[:max_length] if len(text) > max_length else text | |
| extraction_prompt = f"""다음 {doc_type}에서 핵심 정보를 추출해주세요. | |
| 문서 내용: | |
| {truncated_text} | |
| 다음 형식으로 추출해주세요: | |
| 1. 연구 주제: (한 줄 요약) | |
| 2. 핵심 키워드: (쉼표로 구분, 5개 이내) | |
| 3. 연구 목적: (2-3줄) | |
| 4. 연구 방법: (주요 방법론 2-3개) | |
| 5. 기대 효과: (2-3줄) | |
| 추출 결과:""" | |
| messages = [ | |
| {"role": "system", "content": "당신은 학술 문서 분석 전문가입니다. 핵심 정보만 정확하게 추출합니다."}, | |
| {"role": "user", "content": extraction_prompt} | |
| ] | |
| # LLM 호출 | |
| result = self._call_openrouter(messages, "grok-4.1-fast") | |
| return { | |
| "raw_extraction": result, | |
| "doc_type": doc_type, | |
| "text_length": len(text) | |
| } | |
| def process_research_proposal(self, pdf_file, filename: str) -> Dict[str, Any]: | |
| """ | |
| 연구계획서 PDF를 처리하여 벡터 DB에 구조화된 방식으로 저장 | |
| ★ 개선된 처리 방식: | |
| 1. 섹션별로 분리하여 저장 (연구목적, 방법, 기대효과 등) | |
| 2. LLM으로 핵심 정보 추출 후 별도 저장 | |
| 3. 메타데이터에 섹션 정보 포함 | |
| Args: | |
| pdf_file: 업로드된 PDF 파일 | |
| filename: 파일명 | |
| Returns: | |
| 처리 결과 {success, sections_saved, key_info, summary} | |
| """ | |
| # PDF에서 텍스트 추출 | |
| full_text = self.parse_pdf(pdf_file) | |
| if not full_text: | |
| return {"success": False, "sections_saved": 0, "summary": "텍스트 추출 실패"} | |
| today = datetime.now().strftime("%Y-%m-%d") | |
| saved_count = 0 | |
| # ============================================ | |
| # 1. LLM으로 핵심 정보 추출 및 저장 | |
| # ============================================ | |
| key_info = self._extract_key_info_with_llm(full_text, "연구계획서") | |
| # 핵심 요약을 별도로 저장 | |
| if key_info.get("raw_extraction"): | |
| self.inject_memory( | |
| text=f"[연구계획서 핵심요약 - {filename}]\n{key_info['raw_extraction']}", | |
| date_str=today, | |
| category="research_proposal", | |
| tags=["#연구계획서", "#핵심요약", f"#{filename.replace('.pdf', '')}"], | |
| experiment_name=f"연구계획서: {filename}" | |
| ) | |
| saved_count += 1 | |
| # ============================================ | |
| # 2. 섹션별로 분리하여 저장 | |
| # ============================================ | |
| sections = self._extract_sections_from_text(full_text) | |
| section_tags_map = { | |
| "연구 배경": ["#연구배경", "#배경"], | |
| "연구 목적": ["#연구목적", "#목표"], | |
| "연구 방법": ["#연구방법", "#방법론"], | |
| "기대 효과": ["#기대효과", "#성과"], | |
| "연구 내용": ["#연구내용", "#세부내용"], | |
| "참고 문헌": ["#참고문헌", "#레퍼런스"], | |
| } | |
| for section in sections: | |
| section_name = section["section_name"] | |
| content = section["content"] | |
| # 섹션에 맞는 태그 결정 | |
| tags = ["#연구계획서", f"#{filename.replace('.pdf', '')}"] | |
| for key, tag_list in section_tags_map.items(): | |
| if key in section_name: | |
| tags.extend(tag_list) | |
| break | |
| # 섹션이 너무 길면 요약하여 저장 | |
| if len(content) > 1000: | |
| # 앞부분 500자 + 뒷부분 300자 | |
| content = content[:500] + "\n...(중략)...\n" + content[-300:] | |
| if self.inject_memory( | |
| text=f"[{filename} - {section_name}]\n{content}", | |
| date_str=today, | |
| category="research_proposal", | |
| tags=tags, | |
| experiment_name=f"연구계획서: {filename}" | |
| ): | |
| saved_count += 1 | |
| # ============================================ | |
| # 3. 문서 전체 메타데이터 저장 | |
| # ============================================ | |
| doc_metadata = f"""[연구계획서 메타데이터 - {filename}] | |
| - 파일명: {filename} | |
| - 업로드 날짜: {today} | |
| - 전체 길이: {len(full_text)}자 | |
| - 섹션 수: {len(sections)}개 | |
| - 섹션 목록: {', '.join([s['section_name'] for s in sections[:5]])}""" | |
| self.inject_memory( | |
| text=doc_metadata, | |
| date_str=today, | |
| category="document_metadata", | |
| tags=["#연구계획서", "#메타데이터", f"#{filename.replace('.pdf', '')}"], | |
| experiment_name=f"연구계획서: {filename}" | |
| ) | |
| saved_count += 1 | |
| return { | |
| "success": saved_count > 0, | |
| "sections_saved": saved_count, | |
| "total_sections": len(sections), | |
| "key_info": key_info.get("raw_extraction", ""), | |
| "summary": f"✅ 연구계획서 '{filename}' 처리 완료!\n" | |
| f"- 핵심 요약: 1개\n" | |
| f"- 섹션별 저장: {len(sections)}개\n" | |
| f"- 메타데이터: 1개\n" | |
| f"총 {saved_count}개 항목 저장됨" | |
| } | |
| # ============================================ | |
| # 구조화된 실험 기록 저장 | |
| # ============================================ | |
| def save_experiment( | |
| self, | |
| title: str, | |
| hypothesis: str = "", | |
| method: str = "", | |
| results: str = "", | |
| conclusion: str = "", | |
| parameters: Optional[Dict[str, Any]] = None, | |
| tags: Optional[List[str]] = None, | |
| date_str: Optional[str] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| 실험 기록을 구조화된 형태로 저장 | |
| Args: | |
| title: 실험 제목 | |
| hypothesis: 가설 | |
| method: 실험 방법 | |
| results: 결과 | |
| conclusion: 결론 | |
| parameters: 하이퍼파라미터 등 (dict) | |
| tags: 태그 리스트 | |
| date_str: 날짜 (없으면 오늘) | |
| Returns: | |
| 저장 결과 | |
| """ | |
| if date_str is None: | |
| date_str = datetime.now().strftime("%Y-%m-%d") | |
| if tags is None: | |
| tags = [] | |
| if parameters is None: | |
| parameters = {} | |
| # 파라미터를 문자열로 변환 | |
| params_str = ", ".join([f"{k}={v}" for k, v in parameters.items()]) if parameters else "없음" | |
| # 구조화된 실험 기록 생성 | |
| experiment_record = f"""[실험 기록: {title}] | |
| 📋 **가설**: {hypothesis if hypothesis else '명시되지 않음'} | |
| 🔬 **방법**: {method if method else '명시되지 않음'} | |
| ⚙️ **파라미터**: {params_str} | |
| 📊 **결과**: {results if results else '명시되지 않음'} | |
| 💡 **결론**: {conclusion if conclusion else '명시되지 않음'}""" | |
| # 태그 자동 추가 | |
| auto_tags = ["#실험", f"#{title.replace(' ', '_')}"] | |
| all_tags = list(set(tags + auto_tags)) | |
| # 저장 | |
| success = self.inject_memory( | |
| text=experiment_record, | |
| date_str=date_str, | |
| category="experiment", | |
| tags=all_tags, | |
| experiment_name=title | |
| ) | |
| return { | |
| "success": success, | |
| "experiment_name": title, | |
| "date": date_str, | |
| "tags": all_tags, | |
| "message": f"✅ 실험 '{title}' 기록 완료!" if success else "❌ 저장 실패" | |
| } | |
| # ============================================ | |
| # 스마트 메모 저장 (자동 분류 및 태그) | |
| # ============================================ | |
| def save_smart_memo(self, content: str, date_str: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| 메모를 분석하여 자동으로 카테고리와 태그를 부여하고 저장 | |
| Args: | |
| content: 메모 내용 | |
| date_str: 날짜 (없으면 오늘) | |
| Returns: | |
| 저장 결과 | |
| """ | |
| if date_str is None: | |
| date_str = datetime.now().strftime("%Y-%m-%d") | |
| # ============================================ | |
| # 카테고리 자동 분류 | |
| # ============================================ | |
| category_keywords = { | |
| "experiment": ["실험", "테스트", "결과", "정확도", "성능", "accuracy", "loss", "epoch"], | |
| "idea": ["아이디어", "생각", "가설", "제안", "hypothesis", "idea"], | |
| "meeting": ["미팅", "회의", "논의", "피드백", "meeting", "교수님"], | |
| "paper": ["논문", "paper", "리뷰", "인용", "citation", "참고"], | |
| "todo": ["할일", "해야", "TODO", "계획", "다음에"], | |
| "bug": ["버그", "에러", "오류", "bug", "error", "fix"], | |
| } | |
| content_lower = content.lower() | |
| detected_category = "note" # 기본값 | |
| for category, keywords in category_keywords.items(): | |
| for keyword in keywords: | |
| if keyword.lower() in content_lower: | |
| detected_category = category | |
| break | |
| if detected_category != "note": | |
| break | |
| # ============================================ | |
| # 태그 자동 추출 | |
| # ============================================ | |
| # 1. #태그 직접 추출 | |
| explicit_tags = re.findall(r'#(\w+)', content) | |
| # 2. 숫자/수치 관련 태그 | |
| numbers = re.findall(r'(\d+\.?\d*)\s*%', content) | |
| if numbers: | |
| explicit_tags.append("수치결과") | |
| # 3. 하이퍼파라미터 감지 | |
| param_patterns = [ | |
| r'(?:lr|learning.?rate|학습률)\s*[=:]\s*([\d.e-]+)', | |
| r'(?:batch.?size|배치)\s*[=:]\s*(\d+)', | |
| r'(?:epoch|에폭)\s*[=:]\s*(\d+)', | |
| r'(?:drop.?out|드롭아웃)\s*[=:]\s*([\d.]+)', | |
| ] | |
| detected_params = {} | |
| for pattern in param_patterns: | |
| match = re.search(pattern, content, re.IGNORECASE) | |
| if match: | |
| param_name = pattern.split('(?:')[1].split(')')[0].split('|')[0] | |
| detected_params[param_name] = match.group(1) | |
| explicit_tags.append(param_name) | |
| # 4. 실험 이름 추출 시도 | |
| experiment_name = "" | |
| exp_match = re.search(r'(?:실험|테스트|test)\s*[:\-]?\s*([^,.\n]+)', content, re.IGNORECASE) | |
| if exp_match: | |
| experiment_name = exp_match.group(1).strip()[:30] | |
| # 태그 정리 | |
| all_tags = ["#" + t if not t.startswith("#") else t for t in explicit_tags] | |
| all_tags = list(set(all_tags))[:10] # 최대 10개 | |
| # ============================================ | |
| # 저장 | |
| # ============================================ | |
| success = self.inject_memory( | |
| text=content, | |
| date_str=date_str, | |
| category=detected_category, | |
| tags=all_tags, | |
| experiment_name=experiment_name | |
| ) | |
| return { | |
| "success": success, | |
| "category": detected_category, | |
| "tags": all_tags, | |
| "experiment_name": experiment_name, | |
| "detected_params": detected_params, | |
| "message": f"✅ 메모 저장 완료! (분류: {detected_category}, 태그: {', '.join(all_tags[:3])})" | |
| if success else "❌ 저장 실패" | |
| } | |
| return chunks | |
| # ============================================ | |
| # 실시간 메모 저장 ("이거 기억해줘" 명령어) | |
| # ============================================ | |
| def check_and_save_memory_command(self, user_input: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| 사용자 입력에서 기억 저장 명령어 감지 및 처리 | |
| 패턴: | |
| - "기억해줘: ..." | |
| - "이거 기억해: ..." | |
| - "메모해줘: ..." | |
| - "저장해줘: ..." | |
| Args: | |
| user_input: 사용자 입력 | |
| Returns: | |
| 저장 결과 또는 None (명령어가 아닌 경우) | |
| """ | |
| # 기억 저장 패턴 정의 | |
| memory_patterns = [ | |
| r"기억해줘[:\s]+(.+)", | |
| r"이거\s*기억해[:\s]+(.+)", | |
| r"메모해줘[:\s]+(.+)", | |
| r"저장해줘[:\s]+(.+)", | |
| r"기억해[:\s]+(.+)", | |
| r"remember[:\s]+(.+)", | |
| ] | |
| for pattern in memory_patterns: | |
| match = re.search(pattern, user_input, re.IGNORECASE) | |
| if match: | |
| memory_content = match.group(1).strip() | |
| today = datetime.now().strftime("%Y-%m-%d") | |
| # 스마트 메모로 저장 (자동 분류 & 태그 추출) | |
| result = self.save_smart_memo( | |
| content=memory_content, | |
| date_str=today | |
| ) | |
| # 결과 메시지 생성 | |
| if result["success"]: | |
| preview = memory_content[:50] + ('...' if len(memory_content) > 50 else '') | |
| tags_str = ', '.join(result.get('tags', [])[:3]) | |
| response = f"✅ 스마트 저장 완료!\n📝 내용: '{preview}'\n📂 분류: {result.get('category', 'memo')}\n🏷️ 태그: {tags_str}" | |
| else: | |
| response = "❌ 기억 저장에 실패했어. 다시 시도해줘!" | |
| return { | |
| "is_memory_command": True, | |
| "success": result["success"], | |
| "content": memory_content, | |
| "date": today, | |
| "category": result.get("category", "memo"), | |
| "tags": result.get("tags", []), | |
| "response": response | |
| } | |
| return None # 기억 명령어가 아님 | |
| # ============================================ | |
| # 기억 목록/타임라인 조회 기능 | |
| # ============================================ | |
| def get_all_memories(self, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: | |
| """ | |
| 저장된 모든 기억을 날짜순으로 조회 (타임라인) | |
| Args: | |
| limit: 조회할 최대 개수 | |
| offset: 시작 위치 | |
| Returns: | |
| 기억 목록 [{text, date, category, tags, ...}, ...] | |
| """ | |
| try: | |
| # Qdrant에서 모든 포인트 조회 (scroll 사용) | |
| records, _ = self.qdrant_client.scroll( | |
| collection_name=self.collection_name, | |
| limit=limit, | |
| offset=offset, | |
| with_payload=True, | |
| with_vectors=False | |
| ) | |
| memories = [] | |
| for record in records: | |
| memories.append({ | |
| "id": str(record.id), | |
| "text": record.payload.get("text", ""), | |
| "date": record.payload.get("date", "Unknown"), | |
| "year": record.payload.get("year", ""), | |
| "month": record.payload.get("month", ""), | |
| "category": record.payload.get("category", "general"), | |
| "tags": record.payload.get("tags", []), | |
| "experiment_name": record.payload.get("experiment_name", ""), | |
| "created_at": record.payload.get("created_at", "") | |
| }) | |
| # 날짜순 정렬 (최신순) | |
| memories.sort(key=lambda x: x["date"], reverse=True) | |
| return memories | |
| except Exception as e: | |
| print(f"❌ Failed to get memories: {e}") | |
| return [] | |
| def get_memories_by_year(self, year: str) -> List[Dict[str, Any]]: | |
| """연도별 기억 조회""" | |
| all_memories = self.get_all_memories(limit=500) | |
| return [m for m in all_memories if m["year"] == year] | |
| def get_all_tags(self) -> List[str]: | |
| """저장된 모든 태그 목록 조회""" | |
| all_memories = self.get_all_memories(limit=500) | |
| all_tags = set() | |
| for mem in all_memories: | |
| all_tags.update(mem.get("tags", [])) | |
| return sorted(list(all_tags)) | |
| def get_all_categories(self) -> List[str]: | |
| """저장된 모든 카테고리 목록 조회""" | |
| all_memories = self.get_all_memories(limit=500) | |
| categories = set() | |
| for mem in all_memories: | |
| categories.add(mem.get("category", "general")) | |
| return sorted(list(categories)) | |
| # ============================================ | |
| # 필터 검색 기능 (날짜, 카테고리, 태그) | |
| # ============================================ | |
| def search_with_filters( | |
| self, | |
| query: str = "", | |
| year: Optional[str] = None, | |
| month: Optional[str] = None, | |
| category: Optional[str] = None, | |
| tags: Optional[List[str]] = None, | |
| experiment_name: Optional[str] = None, | |
| top_k: int = 10 | |
| ) -> List[Dict[str, Any]]: | |
| """ | |
| 필터 조건과 함께 기억 검색 | |
| Args: | |
| query: 검색 쿼리 (비어있으면 필터만 적용) | |
| year: 연도 필터 | |
| month: 월 필터 | |
| category: 카테고리 필터 | |
| tags: 태그 필터 (OR 조건) | |
| experiment_name: 실험 이름 필터 | |
| top_k: 반환할 결과 수 | |
| Returns: | |
| 필터링된 기억 목록 | |
| """ | |
| # 필터 조건 구성 | |
| filter_conditions = [] | |
| if year: | |
| filter_conditions.append( | |
| FieldCondition(key="year", match=MatchValue(value=year)) | |
| ) | |
| if month: | |
| filter_conditions.append( | |
| FieldCondition(key="month", match=MatchValue(value=month)) | |
| ) | |
| if category: | |
| filter_conditions.append( | |
| FieldCondition(key="category", match=MatchValue(value=category)) | |
| ) | |
| if experiment_name: | |
| filter_conditions.append( | |
| FieldCondition(key="experiment_name", match=MatchValue(value=experiment_name)) | |
| ) | |
| # 필터 객체 생성 | |
| search_filter = None | |
| if filter_conditions: | |
| search_filter = Filter(must=filter_conditions) | |
| # 쿼리가 있으면 벡터 검색, 없으면 필터만 적용 | |
| if query: | |
| query_embedding = self.embedding_model.encode(query).tolist() | |
| results = self.qdrant_client.query_points( | |
| collection_name=self.collection_name, | |
| query=query_embedding, | |
| query_filter=search_filter, | |
| limit=top_k, | |
| with_payload=True | |
| ).points | |
| else: | |
| # 쿼리 없이 필터만 적용하여 스크롤 | |
| results, _ = self.qdrant_client.scroll( | |
| collection_name=self.collection_name, | |
| scroll_filter=search_filter, | |
| limit=top_k, | |
| with_payload=True | |
| ) | |
| # 결과 정리 | |
| memories = [] | |
| for result in results: | |
| payload = result.payload if hasattr(result, 'payload') else result.payload | |
| score = result.score if hasattr(result, 'score') else 0.0 | |
| mem = { | |
| "score": score, | |
| "text": payload.get("text", ""), | |
| "date": payload.get("date", "Unknown"), | |
| "category": payload.get("category", "general"), | |
| "tags": payload.get("tags", []), | |
| "experiment_name": payload.get("experiment_name", ""), | |
| "metadata": payload | |
| } | |
| # 태그 필터 적용 (post-filtering) | |
| if tags: | |
| mem_tags = set(mem["tags"]) | |
| if not any(t in mem_tags for t in tags): | |
| continue | |
| memories.append(mem) | |
| return memories | |
| # ============================================ | |
| # 연구 주제 변경 기능 | |
| # ============================================ | |
| def update_research_topic(self, new_topic: str, new_keywords: List[str], | |
| keep_history: bool = True) -> str: | |
| """ | |
| 연구 주제 변경 (관심사가 바뀌었을 때) | |
| Args: | |
| new_topic: 새로운 연구 주제 | |
| new_keywords: 새로운 관심 키워드 | |
| keep_history: 이전 대화 기록 유지 여부 | |
| Returns: | |
| 변경 완료 메시지 | |
| """ | |
| old_topic = self.user_context.get("topic", "") | |
| old_keywords = self.user_context.get("keywords", []) | |
| # 이전 주제를 기억으로 저장 (히스토리 추적) | |
| if old_topic: | |
| self.inject_memory( | |
| text=f"연구 주제 변경: '{old_topic}' → '{new_topic}'. 이전 키워드: {', '.join(old_keywords)}", | |
| date_str=datetime.now().strftime("%Y-%m-%d"), | |
| category="topic_change", | |
| tags=["#주제변경", "#연구방향"] | |
| ) | |
| # 새 주제로 업데이트 | |
| self.user_context["topic"] = new_topic | |
| self.user_context["keywords"] = new_keywords | |
| # 시스템 프롬프트 재생성 | |
| keywords_str = ", ".join(new_keywords) | |
| self.user_context["system_prompt"] = f""" | |
| 당신은 "AI선배"입니다. 대학원생의 평생 연구 파트너로서 다음 역할을 수행합니다: | |
| 1. 연구 주제: {new_topic} | |
| 2. 관심 키워드: {keywords_str} | |
| 3. 역할: | |
| - 과거 연구 기록을 바탕으로 맥락 있는 조언 제공 | |
| - 실험 결과, 아이디어, 회의 내용 등을 장기 기억으로 보존 | |
| - 질문에 대해 과거 기록을 참조하여 정확한 답변 제공 | |
| - 이전 연구 주제와의 연결점도 찾아줌 | |
| 항상 친근하면서도 전문적인 톤으로 대화하세요. | |
| 한국어로 답변하세요. | |
| 과거 기록을 참조할 때는 날짜와 출처를 명시하세요. | |
| """ | |
| # 대화 히스토리 처리 | |
| if not keep_history: | |
| self.clear_history() | |
| return f"""🔄 연구 주제가 변경되었어! | |
| **이전:** {old_topic} ({', '.join(old_keywords)}) | |
| **현재:** {new_topic} ({keywords_str}) | |
| 이전 연구 기록은 그대로 보존되어 있으니, | |
| 나중에 "예전에 {old_topic} 할 때 뭐했지?" 라고 물어봐도 대답할 수 있어! 🎓""" | |
| # ============================================ | |
| # 실험 비교 기능 | |
| # ============================================ | |
| def compare_experiments(self, experiment_names: Optional[List[str]] = None, | |
| tags: Optional[List[str]] = None) -> Dict[str, Any]: | |
| """ | |
| 여러 실험 결과 비교 | |
| Args: | |
| experiment_names: 비교할 실험 이름 리스트 | |
| tags: 비교할 태그 리스트 | |
| Returns: | |
| 비교 결과 {experiments: [...], comparison_table: str} | |
| """ | |
| experiments = [] | |
| # 실험 이름으로 검색 | |
| if experiment_names: | |
| for name in experiment_names: | |
| results = self.search_with_filters( | |
| experiment_name=name, | |
| category="experiment", | |
| top_k=5 | |
| ) | |
| if results: | |
| experiments.extend(results) | |
| # 태그로 검색 | |
| if tags: | |
| for tag in tags: | |
| results = self.search_with_filters( | |
| tags=[tag], | |
| category="experiment", | |
| top_k=5 | |
| ) | |
| for r in results: | |
| if r not in experiments: | |
| experiments.append(r) | |
| # 비교 테이블 생성 | |
| if not experiments: | |
| return { | |
| "experiments": [], | |
| "comparison_table": "비교할 실험을 찾지 못했어요." | |
| } | |
| # 마크다운 테이블 생성 | |
| table = "| 날짜 | 실험명 | 태그 | 내용 요약 |\n" | |
| table += "|------|--------|------|----------|\n" | |
| for exp in experiments: | |
| date = exp.get("date", "?") | |
| name = exp.get("experiment_name", "-") | |
| tags_str = ", ".join(exp.get("tags", []))[:20] | |
| text = exp.get("text", "")[:50] + "..." | |
| table += f"| {date} | {name} | {tags_str} | {text} |\n" | |
| return { | |
| "experiments": experiments, | |
| "comparison_table": table, | |
| "count": len(experiments) | |
| } | |
| def check_comparison_command(self, user_input: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| '비교해줘' 명령어 감지 및 처리 | |
| 패턴: | |
| - "Drop-out 0.3 vs 0.5 비교해줘" | |
| - "#ablation 실험들 비교해줘" | |
| - "실험1과 실험2 비교해줘" | |
| """ | |
| comparison_patterns = [ | |
| r"(.+)\s*(?:vs|VS|비교|compared)\s*(.+)\s*비교", | |
| r"(.+)\s*(?:이랑|과|와)\s*(.+)\s*비교해", | |
| r"#(\w+)\s*(?:실험들?|결과들?)\s*비교", | |
| ] | |
| for pattern in comparison_patterns: | |
| match = re.search(pattern, user_input, re.IGNORECASE) | |
| if match: | |
| groups = match.groups() | |
| # 태그 기반 비교 | |
| if groups[0].startswith("#") or len(groups) == 1: | |
| tag = "#" + groups[0].replace("#", "") | |
| result = self.compare_experiments(tags=[tag]) | |
| else: | |
| # 키워드 기반 검색 | |
| search_terms = [g.strip() for g in groups if g] | |
| experiments = [] | |
| for term in search_terms: | |
| results = self.search_with_filters(query=term, category="experiment", top_k=3) | |
| experiments.extend(results) | |
| if experiments: | |
| table = "| 날짜 | 태그 | 내용 |\n|------|------|------|\n" | |
| for exp in experiments: | |
| table += f"| {exp['date']} | {', '.join(exp['tags'])} | {exp['text'][:60]}... |\n" | |
| result = {"experiments": experiments, "comparison_table": table, "count": len(experiments)} | |
| else: | |
| result = {"experiments": [], "comparison_table": "비교할 실험을 찾지 못했어요."} | |
| return { | |
| "is_comparison_command": True, | |
| "result": result, | |
| "response": f"📊 **실험 비교 결과** ({result.get('count', 0)}개 발견)\n\n{result['comparison_table']}" | |
| } | |
| return None | |
| def get_stats(self) -> Dict[str, Any]: | |
| """ | |
| 컬렉션 통계 조회 | |
| Returns: | |
| 저장된 기억 수, 컬렉션 정보 등 | |
| """ | |
| try: | |
| collection_info = self.qdrant_client.get_collection(self.collection_name) | |
| all_tags = self.get_all_tags() | |
| all_categories = self.get_all_categories() | |
| return { | |
| "total_memories": collection_info.points_count, | |
| "collection_name": self.collection_name, | |
| "vector_size": collection_info.config.params.vectors.size, | |
| "total_tags": len(all_tags), | |
| "tags": all_tags[:10], # 상위 10개만 | |
| "categories": all_categories, | |
| "status": "healthy" | |
| } | |
| except Exception as e: | |
| return { | |
| "total_memories": 0, | |
| "error": str(e), | |
| "status": "error" | |
| } | |
| # =========================================== | |
| # 시연용 데모 데이터 생성 함수 (강화 버전) | |
| # =========================================== | |
| def inject_demo_data_2024(engine: RAGEngine) -> bool: | |
| """ | |
| [Scenario B] 2024년 기억 주입 (시연용) | |
| 심사위원 시연을 위해 과거 연구 기록을 미리 주입합니다. | |
| 버튼 클릭 시 호출되어 즉시 기억을 생성합니다. | |
| ★ 태그와 실험명이 포함된 다양한 시나리오 데이터 | |
| Args: | |
| engine: RAGEngine 인스턴스 | |
| Returns: | |
| 성공 여부 | |
| """ | |
| demo_memories = [ | |
| # ============================================ | |
| # Ablation Study 시리즈 (Drop-out 비율 실험) | |
| # ============================================ | |
| { | |
| "text": "Ablation Study 실험에서 Drop-out 비율을 0.3으로 했을 때 정확도가 가장 높았음. 0.1, 0.2, 0.4 모두 테스트했으나 0.3이 91.8%로 최고 성능을 보였다. #ablation #dropout", | |
| "date": "2024-03-15", | |
| "category": "experiment", | |
| "tags": ["#ablation", "#dropout", "#실험1"], | |
| "experiment_name": "Dropout 비율 실험" | |
| }, | |
| { | |
| "text": "Drop-out 0.5로 추가 실험. 정확도 87.2%로 하락. 과도한 dropout은 학습을 방해함을 확인. #ablation #dropout", | |
| "date": "2024-03-20", | |
| "category": "experiment", | |
| "tags": ["#ablation", "#dropout", "#실험2"], | |
| "experiment_name": "Dropout 추가 실험" | |
| }, | |
| # ============================================ | |
| # Batch Size 실험 | |
| # ============================================ | |
| { | |
| "text": "Batch Size 비교 실험: 16, 32, 64, 128 테스트. 32일 때 수렴 속도와 최종 성능의 균형이 가장 좋았음. 메모리 사용량도 적절. #batchsize #hyperparameter", | |
| "date": "2024-03-15", | |
| "category": "experiment", | |
| "tags": ["#batchsize", "#hyperparameter", "#실험3"], | |
| "experiment_name": "Batch Size 실험" | |
| }, | |
| # ============================================ | |
| # 회의 기록 | |
| # ============================================ | |
| { | |
| "text": "Private LLM 프로젝트 킥오프 미팅. 보안과 효율성의 균형을 목표로 설정. 첫 번째 마일스톤은 로컬 임베딩 파이프라인 구축. #kickoff #프로젝트", | |
| "date": "2024-02-20", | |
| "category": "meeting", | |
| "tags": ["#kickoff", "#프로젝트", "#회의"], | |
| "experiment_name": "" | |
| }, | |
| { | |
| "text": "지도교수님과 중간 점검 미팅. Ablation Study 결과 보고. 추가로 Learning Rate 실험 제안 받음. #미팅 #지도교수", | |
| "date": "2024-04-05", | |
| "category": "meeting", | |
| "tags": ["#미팅", "#지도교수", "#중간점검"], | |
| "experiment_name": "" | |
| }, | |
| # ============================================ | |
| # Quantization 실험 | |
| # ============================================ | |
| { | |
| "text": "Quantization 적용 실험: INT8 양자화 시 성능 저하 5% 이내. FP16 대비 메모리 50% 절감. 속도는 1.8배 향상. #quantization #optimization", | |
| "date": "2024-04-10", | |
| "category": "experiment", | |
| "tags": ["#quantization", "#optimization", "#실험4"], | |
| "experiment_name": "INT8 양자화 실험" | |
| }, | |
| { | |
| "text": "INT4 양자화 테스트. 성능 저하 15%로 너무 큼. INT8이 최적점. #quantization #실험5", | |
| "date": "2024-04-15", | |
| "category": "experiment", | |
| "tags": ["#quantization", "#실험5"], | |
| "experiment_name": "INT4 양자화 실험" | |
| }, | |
| # ============================================ | |
| # 아이디어 기록 | |
| # ============================================ | |
| { | |
| "text": "논문 아이디어: 'Efficient Private RAG for Long-term Research Memory' 제목으로 진행. 핵심은 ScalarQuantization과 메타데이터 필터링의 결합. #논문 #아이디어", | |
| "date": "2024-05-01", | |
| "category": "idea", | |
| "tags": ["#논문", "#아이디어", "#RAG"], | |
| "experiment_name": "" | |
| }, | |
| { | |
| "text": "새 아이디어: 시간 기반 메모리 가중치. 최근 기억에 더 높은 가중치를 주되, 중요도 태그로 오래된 기억도 보존. #아이디어 #memory", | |
| "date": "2024-06-15", | |
| "category": "idea", | |
| "tags": ["#아이디어", "#memory", "#temporal"], | |
| "experiment_name": "" | |
| }, | |
| # ============================================ | |
| # Learning Rate 실험 | |
| # ============================================ | |
| { | |
| "text": "Learning Rate 실험: 1e-3, 1e-4, 1e-5 비교. 1e-4가 가장 안정적인 수렴. Warmup 적용 시 1e-3도 사용 가능. #learningrate #hyperparameter", | |
| "date": "2024-05-20", | |
| "category": "experiment", | |
| "tags": ["#learningrate", "#hyperparameter", "#실험6"], | |
| "experiment_name": "Learning Rate 실험" | |
| } | |
| ] | |
| success_count = 0 | |
| for memory in demo_memories: | |
| if engine.inject_memory( | |
| text=memory["text"], | |
| date_str=memory["date"], | |
| category=memory["category"], | |
| tags=memory.get("tags", []), | |
| experiment_name=memory.get("experiment_name", "") | |
| ): | |
| success_count += 1 | |
| print(f"✅ Demo data injection complete: {success_count}/{len(demo_memories)} memories added") | |
| return success_count == len(demo_memories) | |
| # 테스트 코드 | |
| if __name__ == "__main__": | |
| print("=" * 50) | |
| print("AI선배 My Co-Pilot - RAG Engine Test") | |
| print("=" * 50) | |
| # 엔진 초기화 | |
| engine = RAGEngine() | |
| # 사용자 온보딩 테스트 | |
| print("\n[Test 1] User Onboarding") | |
| greeting = engine.onboard_user("Private LLM", ["Security", "Efficiency"]) | |
| print(greeting) | |
| # 시연 데이터 주입 테스트 | |
| print("\n[Test 2] Demo Data Injection") | |
| inject_demo_data_2024(engine) | |
| # 통계 확인 | |
| print("\n[Test 3] Collection Stats") | |
| stats = engine.get_stats() | |
| print(f"Total memories: {stats['total_memories']}") | |
| # 질문 응답 테스트 | |
| print("\n[Test 4] Question Answering") | |
| result = engine.get_answer("2024년 3월에 했던 Ablation Study 결과가 뭐였지?") | |
| print(f"Answer: {result['answer'][:200]}...") | |
| print(f"Referenced memories: {len(result['memories'])}") | |
| print("\n" + "=" * 50) | |
| print("All tests completed!") | |