monitoring.py

"""
Langfuse를 활용한 모니터링 구현 (선택적)
"""
from typing import Dict, Any, Optional
import time
import os
from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

# 설정 값 가져오기
LANGFUSE_SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY", "")
LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY", "")
LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")

class LangfuseMonitoring:
    def __init__(self):
        """
        Langfuse 모니터링 초기화 (선택적 기능)
        """
        self.enabled = False
        print("모니터링 기능을 초기화합니다...")

        # Langfuse가 설치되어 있는지 확인
        try:
            from langfuse import Langfuse

            if LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY:
                try:
                    self.langfuse = Langfuse(
                        public_key=LANGFUSE_PUBLIC_KEY,
                        secret_key=LANGFUSE_SECRET_KEY,
                        host=LANGFUSE_HOST,
                    )
                    self.enabled = True
                    print("Langfuse 모니터링이 활성화되었습니다.")
                except Exception as e:
                    print(f"Langfuse 초기화 실패: {e}")
            else:
                print("Langfuse API 키가 설정되지 않았습니다. 모니터링은 비활성화됩니다.")
        except ImportError:
            print("langfuse 패키지가 설치되지 않았습니다. 모니터링 기능이 비활성화됩니다.")
            print("pip install langfuse 명령으로 설치할 수 있습니다.")

    def start_trace(self, name: str, user_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> Any:
        """
        새 트레이스 시작

        Args:
            name: 트레이스 이름
            user_id: 사용자 ID (선택적)
            metadata: 추가 메타데이터 (선택적)

        Returns:
            트레이스 객체 또는 None
        """
        if not self.enabled:
            return None

        try:
            return self.langfuse.trace(
                name=name,
                user_id=user_id,
                metadata=metadata or {},
            )
        except Exception as e:
            print(f"트레이스 생성 실패: {e}")
            return None

    def log_generation(self, trace: Any, name: str, prompt: str, response: str, metadata: Optional[Dict[str, Any]] = None) -> None:
        """
        LLM 생성 로깅

        Args:
            trace: 트레이스 객체
            name: 생성 이름
            prompt: 입력 프롬프트
            response: 모델 응답
            metadata: 추가 메타데이터 (선택적)
        """
        if not self.enabled or trace is None:
            return

        try:
            trace.generation(
                name=name,
                model="user-defined-model",
                prompt=prompt,
                completion=response,
                metadata=metadata or {},
            )
        except Exception as e:
            print(f"생성 로깅 실패: {e}")

    def log_span(self, trace: Any, name: str, input_data: Any, output_data: Any, start_time: float, end_time: float) -> None:
        """
        처리 구간 로깅

        Args:
            trace: 트레이스 객체
            name: 구간 이름
            input_data: 입력 데이터
            output_data: 출력 데이터
            start_time: 시작 시간
            end_time: 종료 시간
        """
        if not self.enabled or trace is None:
            return

        try:
            trace.span(
                name=name,
                start_time=start_time,
                end_time=end_time,
                input=input_data,
                output=output_data,
                metadata={"duration_ms": (end_time - start_time) * 1000},
            )
        except Exception as e:
            print(f"구간 로깅 실패: {e}")

    def end_trace(self, trace: Any) -> None:
        """
        트레이스 종료

        Args:
            trace: 종료할 트레이스 객체
        """
        if not self.enabled or trace is None:
            return

        try:
            trace.update(status="success")
        except Exception as e:
            print(f"트레이스 종료 실패: {e}")