feat: init rag service

2026-06-13 05:59:11 +00:00 · 2025-10-29 05:54:08 +09:00
parent 44ae9c546f
commit 5d897cb845
54 changed files with 6425 additions and 0 deletions
@@ -0,0 +1,119 @@
+"""
+설정 관리 유틸리티
+"""
+import os
+import yaml
+from typing import Any, Dict
+from pathlib import Path
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """애플리케이션 설정"""
+
+    # PostgreSQL
+    POSTGRES_HOST: str = "localhost"
+    POSTGRES_PORT: int = 5432
+    POSTGRES_DATABASE: str = "meeting_db"
+    POSTGRES_USER: str = "postgres"
+    POSTGRES_PASSWORD: str = ""
+
+    # Azure OpenAI
+    AZURE_OPENAI_API_KEY: str = ""
+    AZURE_OPENAI_ENDPOINT: str = ""
+
+    # Azure AI Search
+    AZURE_SEARCH_ENDPOINT: str = ""
+    AZURE_SEARCH_API_KEY: str = ""
+
+    # Claude AI
+    CLAUDE_API_KEY: str = ""
+
+    # Redis
+    REDIS_PASSWORD: str = ""
+
+    # Azure Event Hub
+    EVENTHUB_CONNECTION_STRING: str = ""
+    EVENTHUB_NAME: str = ""
+    AZURE_EVENTHUB_CONSUMER_GROUP: str = "$Default"
+    AZURE_STORAGE_CONNECTION_STRING: str = ""
+    AZURE_STORAGE_CONTAINER_NAME: str = ""
+
+    class Config:
+        # rag 디렉토리 기준으로 .env 파일 경로 설정
+        env_file = str(Path(__file__).parent.parent.parent / ".env")
+        case_sensitive = True
+
+
+def load_config(config_path: str = "config.yaml") -> Dict[str, Any]:
+    """
+    설정 파일 로딩
+
+    Args:
+        config_path: 설정 파일 경로
+
+    Returns:
+        설정 딕셔너리
+    """
+    # 환경변수 로딩
+    settings = Settings()
+
+    # YAML 파일 로딩
+    config_file = Path(config_path)
+    if not config_file.exists():
+        raise FileNotFoundError(f"설정 파일을 찾을 수 없습니다: {config_path}")
+
+    with open(config_file, "r", encoding="utf-8") as f:
+        config = yaml.safe_load(f)
+
+    # 환경변수로 대체
+    def replace_env_vars(obj: Any) -> Any:
+        """재귀적으로 환경변수 치환"""
+        if isinstance(obj, dict):
+            return {k: replace_env_vars(v) for k, v in obj.items()}
+        elif isinstance(obj, list):
+            return [replace_env_vars(item) for item in obj]
+        elif isinstance(obj, str) and obj.startswith("${") and obj.endswith("}"):
+            env_var = obj[2:-1]
+            return getattr(settings, env_var, "")
+        return obj
+
+    config = replace_env_vars(config)
+
+    return config
+
+
+def get_database_url(config: Dict[str, Any]) -> str:
+    """
+    PostgreSQL 데이터베이스 URL 생성
+
+    Args:
+        config: 설정 딕셔너리
+
+    Returns:
+        데이터베이스 URL
+    """
+    pg = config["postgres"]
+    return (
+        f"postgresql://{pg['user']}:{pg['password']}"
+        f"@{pg['host']}:{pg['port']}/{pg['database']}"
+    )
+
+
+def get_redis_url(config: Dict[str, Any]) -> str:
+    """
+    Redis URL 생성
+
+    Args:
+        config: 설정 딕셔너리
+
+    Returns:
+        Redis URL
+    """
+    redis = config["redis"]
+    password = redis.get("password", "")
+
+    if password:
+        return f"redis://:{password}@{redis['host']}:{redis['port']}/{redis['db']}"
+    else:
+        return f"redis://{redis['host']}:{redis['port']}/{redis['db']}"
@@ -0,0 +1,180 @@
+"""
+임베딩 생성 유틸리티
+"""
+import openai
+from typing import List, Union
+from tenacity import retry, stop_after_attempt, wait_exponential
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class EmbeddingGenerator:
+    """OpenAI Embedding 생성기"""
+
+    def __init__(
+        self,
+        api_key: str,
+        endpoint: str = None,
+        model: str = "text-embedding-ada-002",
+        dimension: int = 1536,
+        api_version: str = None
+    ):
+        """
+        초기화
+
+        Args:
+            api_key: OpenAI API 키
+            endpoint: 엔드포인트 (선택사항, Azure 전용)
+            model: 임베딩 모델명
+            dimension: 임베딩 차원
+            api_version: API 버전 (선택사항, Azure 전용)
+        """
+        # Azure OpenAI 또는 일반 OpenAI 자동 선택
+        if endpoint and "azure" in endpoint.lower():
+            # Azure OpenAI 사용
+            self.client = openai.AzureOpenAI(
+                api_key=api_key,
+                azure_endpoint=endpoint,
+                api_version=api_version
+            )
+        else:
+            # 일반 OpenAI 사용
+            self.client = openai.OpenAI(
+                api_key=api_key
+            )
+        self.model = model
+        self.dimension = dimension
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10)
+    )
+    def generate_embedding(self, text: str) -> List[float]:
+        """
+        단일 텍스트의 임베딩 생성
+
+        Args:
+            text: 입력 텍스트
+
+        Returns:
+            임베딩 벡터 (1536차원)
+        """
+        try:
+            response = self.client.embeddings.create(
+                model=self.model,
+                input=text
+            )
+            embedding = response.data[0].embedding
+
+            # 차원 검증
+            if len(embedding) != self.dimension:
+                raise ValueError(
+                    f"임베딩 차원 불일치: 예상 {self.dimension}, 실제 {len(embedding)}"
+                )
+
+            return embedding
+
+        except Exception as e:
+            logger.error(f"임베딩 생성 실패: {str(e)}")
+            raise
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10)
+    )
+    def generate_embeddings_batch(
+        self,
+        texts: List[str],
+        batch_size: int = 50
+    ) -> List[List[float]]:
+        """
+        배치 텍스트의 임베딩 생성
+
+        Args:
+            texts: 입력 텍스트 리스트
+            batch_size: 배치 크기 (최대 50)
+
+        Returns:
+            임베딩 벡터 리스트
+        """
+        if not texts:
+            return []
+
+        all_embeddings = []
+
+        # 배치 단위로 처리
+        for i in range(0, len(texts), batch_size):
+            batch = texts[i:i + batch_size]
+
+            try:
+                response = self.client.embeddings.create(
+                    model=self.model,
+                    input=batch
+                )
+
+                batch_embeddings = [item.embedding for item in response.data]
+
+                # 차원 검증
+                for embedding in batch_embeddings:
+                    if len(embedding) != self.dimension:
+                        raise ValueError(
+                            f"임베딩 차원 불일치: 예상 {self.dimension}, 실제 {len(embedding)}"
+                        )
+
+                all_embeddings.extend(batch_embeddings)
+
+                logger.info(f"배치 {i//batch_size + 1}: {len(batch)}개 임베딩 생성 완료")
+
+            except Exception as e:
+                logger.error(f"배치 임베딩 생성 실패: {str(e)}")
+                raise
+
+        return all_embeddings
+
+    def get_token_count(self, text: str) -> int:
+        """
+        텍스트의 토큰 수 계산 (근사치)
+
+        Args:
+            text: 입력 텍스트
+
+        Returns:
+            토큰 수
+        """
+        # 간단한 추정: 한글은 1글자당 약 1.5 토큰, 영어는 0.75 토큰
+        korean_chars = sum(1 for c in text if ord(c) >= 0xAC00 and ord(c) <= 0xD7A3)
+        other_chars = len(text) - korean_chars
+
+        estimated_tokens = int(korean_chars * 1.5 + other_chars * 0.75)
+
+        return estimated_tokens
+
+
+def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
+    """
+    코사인 유사도 계산
+
+    Args:
+        vec1: 벡터 1
+        vec2: 벡터 2
+
+    Returns:
+        코사인 유사도 (0.0 ~ 1.0)
+    """
+    import numpy as np
+
+    vec1_np = np.array(vec1)
+    vec2_np = np.array(vec2)
+
+    dot_product = np.dot(vec1_np, vec2_np)
+    norm1 = np.linalg.norm(vec1_np)
+    norm2 = np.linalg.norm(vec2_np)
+
+    if norm1 == 0 or norm2 == 0:
+        return 0.0
+
+    similarity = dot_product / (norm1 * norm2)
+
+    # -1 ~ 1 범위를 0 ~ 1로 변환
+    return (similarity + 1) / 2
@@ -0,0 +1,74 @@
+"""
+텍스트 처리 유틸리티 모듈
+"""
+from typing import List
+import logging
+from kiwipiepy import Kiwi
+
+logger = logging.getLogger(__name__)
+
+# Kiwi 인스턴스 (싱글톤)
+_kiwi = None
+
+
+def get_kiwi():
+    """Kiwi 형태소 분석기 인스턴스 반환"""
+    global _kiwi
+    if _kiwi is None:
+        _kiwi = Kiwi()
+        logger.info("Kiwi 형태소 분석기 초기화 완료")
+    return _kiwi
+
+
+def extract_nouns(text: str) -> List[str]:
+    """
+    텍스트에서 명사 추출
+
+    Args:
+        text: 입력 텍스트
+
+    Returns:
+        추출된 명사 리스트
+    """
+    if not text or not text.strip():
+        return []
+
+    try:
+        kiwi = get_kiwi()
+
+        # 형태소 분석
+        result = kiwi.analyze(text)
+
+        # 명사 추출 (NNG: 일반명사, NNP: 고유명사, SL: 외국어, SH: 한자, SN: 숫자)
+        nouns = []
+        for token, pos, _, _ in result[0][0]:
+            if pos in ['NNG', 'NNP', 'SL', 'SH', 'SN']:
+                nouns.append(token)
+
+        logger.debug(f"원본 텍스트: {text}")
+        logger.debug(f"추출된 명사: {nouns}")
+
+        return nouns
+
+    except Exception as e:
+        logger.error(f"명사 추출 실패: {str(e)}")
+        # 오류 발생 시 원본 텍스트를 공백으로 분리하여 반환
+        return text.split()
+
+
+def extract_nouns_as_query(text: str) -> str:
+    """
+    텍스트에서 명사를 추출하여 검색 쿼리로 변환
+
+    Args:
+        text: 입력 텍스트
+
+    Returns:
+        공백으로 연결된 명사 문자열
+    """
+    nouns = extract_nouns(text)
+    query = ' '.join(nouns)
+
+    logger.info(f"Query 변환: '{text}' → '{query}'")
+
+    return query if query else text