🔧 Gemini API 시작하기: 기본 설정

자, 이제 본격적으로 시작해볼까? 먼저 Gemini API를 사용하기 위한 기본 설정부터 알아보자. 생각보다 간단하니까 걱정하지 마! 😎

가장 먼저 해야 할 일은 구글 AI Studio에서 API 키를 발급받는 거야.

발급 절차:
① Google AI Studio (makersuite.google.com/app/apikey) 접속
② 구글 계정으로 로그인
③ "Get API Key" 버튼 클릭
④ 새 프로젝트 생성 또는 기존 프로젝트 선택
⑤ API 키 복사 및 안전한 곳에 보관

여기서 중요한 건, API 키는 절대 공개 저장소에 올리면 안 돼. 환경 변수나 별도의 설정 파일로 관리해야 해. 💾

Gemini API는 다양한 프로그래밍 언어를 지원해. Python, JavaScript, Go, Java 등등... 오늘은 가장 많이 사용되는 Python을 기준으로 설명할게.

필요한 패키지 설치:

pip install google-generativeai
pip install python-dotenv
pip install pandas  # 데이터 처리용
pip install asyncio  # 비동기 처리용

기본 설정 코드는 이렇게 작성하면 돼:

import google.generativeai as genai
import os
from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

# API 키 설정
GOOGLE_API_KEY = os.getenv('GOOGLE_API_KEY')
genai.configure(api_key=GOOGLE_API_KEY)

# 모델 초기화
model = genai.GenerativeModel('gemini-pro')

이렇게 하면 기본 설정은 끝! 정말 간단하지? 🎉

📝 자동요약 에이전트 구축하기

이제 본격적으로 자동요약 기능을 만들어보자. 자동요약은 긴 텍스트를 핵심 내용만 추출해서 짧게 만드는 거야. 뉴스 기사, 보고서, 회의록 등 다양한 곳에 활용할 수 있어.

가장 기본적인 요약 함수부터 시작해보자:

def summarize_text(text, max_length=200):
    """
    텍스트를 요약하는 함수
    
    Args:
        text: 요약할 원본 텍스트
        max_length: 요약문의 최대 길이 (단어 수)
    
    Returns:
        요약된 텍스트
    """
    prompt = f"""
    다음 텍스트를 {max_length}자 이내로 핵심 내용만 요약해주세요.
    요약할 때는 다음 규칙을 따라주세요:
    1. 가장 중요한 정보를 우선적으로 포함
    2. 객관적이고 중립적인 톤 유지
    3. 불필요한 수식어 제거
    4. 명확하고 간결한 문장 사용
    
    텍스트:
    {text}
    
    요약:
    """
    
    response = model.generate_content(prompt)
    return response.text

이 함수는 기본적인 요약 기능을 제공해. 하지만 실제 프로덕션 환경에서는 더 정교한 처리가 필요해. 예를 들어 에러 핸들링, 토큰 제한 관리, 재시도 로직 같은 거 말이야.

이제 좀 더 실용적인 요약 시스템을 만들어보자. 이 시스템은 다양한 요약 스타일을 지원하고, 긴 문서도 처리할 수 있어:

class AdvancedSummarizer:
    def __init__(self, model_name='gemini-pro'):
        self.model = genai.GenerativeModel(model_name)
        self.max_input_tokens = 30000  # Gemini Pro의 토큰 제한
        
    def chunk_text(self, text, chunk_size=5000):
        """긴 텍스트를 청크로 나누기"""
        words = text.split()
        chunks = []
        current_chunk = []
        current_length = 0
        
        for word in words:
            current_chunk.append(word)
            current_length += len(word) + 1
            
            if current_length >= chunk_size:
                chunks.append(' '.join(current_chunk))
                current_chunk = []
                current_length = 0
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks
    
    def summarize(self, text, style='balanced', length='medium'):
        """
        다양한 스타일로 텍스트 요약
        
        Args:
            text: 요약할 텍스트
            style: 'balanced', 'bullet', 'executive', 'technical'
            length: 'short', 'medium', 'long'
        """
        length_map = {
            'short': 100,
            'medium': 300,
            'long': 500
        }
        
        style_instructions = {
            'balanced': '균형잡힌 요약문을 작성하되, 모든 주요 포인트를 포함하세요.',
            'bullet': '핵심 내용을 불릿 포인트 형식으로 정리하세요.',
            'executive': '경영진을 위한 요약으로, 결론과 액션 아이템을 강조하세요.',
            'technical': '기술적 세부사항을 유지하면서 전문가를 위한 요약을 작성하세요.'
        }
        
        max_length = length_map.get(length, 300)
        style_instruction = style_instructions.get(style, style_instructions['balanced'])
        
        # 텍스트가 너무 길면 청크로 나누기
        if len(text) > 10000:
            chunks = self.chunk_text(text)
            summaries = []
            
            for chunk in chunks:
                prompt = f"""
                {style_instruction}
                최대 {max_length}자로 요약하세요.
                
                텍스트:
                {chunk}
                """
                response = self.model.generate_content(prompt)
                summaries.append(response.text)
            
            # 청크 요약들을 다시 종합 요약
            combined = ' '.join(summaries)
            final_prompt = f"""
            다음은 긴 문서의 부분 요약들입니다.
            이를 하나의 일관된 요약문으로 통합하세요.
            {style_instruction}
            최대 {max_length}자로 작성하세요.
            
            부분 요약들:
            {combined}
            """
            final_response = self.model.generate_content(final_prompt)
            return final_response.text
        else:
            prompt = f"""
            {style_instruction}
            최대 {max_length}자로 요약하세요.
            
            텍스트:
            {text}
            """
            response = self.model.generate_content(prompt)
            return response.text

요약의 품질을 높이기 위한 몇 가지 고급 테크닉을 소개할게:

1. 컨텍스트 프리징 (Context Priming)
요약하기 전에 문서의 종류와 목적을 명확히 하면 더 좋은 결과를 얻을 수 있어:

def context_aware_summarize(text, document_type, audience):
    """
    컨텍스트를 고려한 요약
    
    Args:
        text: 요약할 텍스트
        document_type: 'news', 'research', 'business', 'technical'
        audience: 'general', 'expert', 'executive'
    """
    context_map = {
        'news': '뉴스 기사',
        'research': '연구 논문',
        'business': '비즈니스 문서',
        'technical': '기술 문서'
    }
    
    audience_map = {
        'general': '일반 대중',
        'expert': '해당 분야 전문가',
        'executive': '의사결정권자'
    }
    
    prompt = f"""
    당신은 {context_map.get(document_type, '문서')} 요약 전문가입니다.
    {audience_map.get(audience, '독자')}를 대상으로 요약문을 작성하세요.
    
    요약 시 다음을 고려하세요:
    - 대상 독자의 배경 지식 수준
    - 문서 유형에 맞는 핵심 정보
    - 적절한 전문 용어 사용 수준
    
    텍스트:
    {text}
    
    요약:
    """
    
    response = model.generate_content(prompt)
    return response.text

2. 멀티패스 요약 (Multi-pass Summarization)
한 번에 요약하는 대신 여러 단계를 거치면 더 정확한 요약을 얻을 수 있어:

def multipass_summarize(text):
    """
    3단계 요약 프로세스
    """
    # 1단계: 초벌 요약 (길게)
    first_pass = f"""
    다음 텍스트의 모든 중요한 정보를 포함하여 
    상세하게 요약하세요 (500자 내외):
    
    {text}
    """
    first_summary = model.generate_content(first_pass).text
    
    # 2단계: 핵심 추출
    second_pass = f"""
    다음 요약문에서 가장 핵심적인 3-5개의 포인트를 추출하세요:
    
    {first_summary}
    """
    key_points = model.generate_content(second_pass).text
    
    # 3단계: 최종 요약
    third_pass = f"""
    다음 핵심 포인트들을 바탕으로 
    간결하고 명확한 최종 요약문을 작성하세요 (200자 내외):
    
    {key_points}
    
    원문 참조:
    {text[:500]}...
    """
    final_summary = model.generate_content(third_pass).text
    
    return {
        'final_summary': final_summary,
        'key_points': key_points,
        'detailed_summary': first_summary
    }

이 멀티패스 방식은 특히 복잡한 기술 문서나 법률 문서를 요약할 때 효과적이야. 단계별로 정보를 정제하면서 중요한 내용이 누락되는 걸 방지할 수 있거든. 🎯

🗂️ 자동분류 에이전트 구축하기

이제 자동분류 시스템을 만들어보자! 자동분류는 텍스트를 미리 정의된 카테고리로 자동으로 분류하는 거야. 이메일 필터링, 고객 문의 라우팅, 콘텐츠 태깅 등 정말 다양하게 활용할 수 있어. 📊

먼저 간단한 분류 함수부터 만들어보자:

def classify_text(text, categories):
    """
    텍스트를 주어진 카테고리로 분류
    
    Args:
        text: 분류할 텍스트
        categories: 가능한 카테고리 리스트
    
    Returns:
        분류 결과 (카테고리, 신뢰도)
    """
    categories_str = ', '.join(categories)
    
    prompt = f"""
    다음 텍스트를 아래 카테고리 중 하나로 분류하세요.
    
    카테고리: {categories_str}
    
    텍스트:
    {text}
    
    응답 형식:
    카테고리: [선택된 카테고리]
    신뢰도: [0-100 사이의 숫자]
    이유: [분류 근거를 간단히 설명]
    """
    
    response = model.generate_content(prompt)
    return response.text

이건 기본적인 분류야. 하지만 실제로는 더 정교한 시스템이 필요해. 특히 다중 레이블 분류, 계층적 분류, 신뢰도 기반 필터링 같은 기능들이 중요하지.

이제 프로덕션 레벨의 분류 시스템을 만들어보자:

import json
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class ClassificationResult:
    """분류 결과를 담는 데이터 클래스"""
    primary_category: str
    confidence: float
    secondary_categories: List[Tuple[str, float]]
    reasoning: str
    tags: List[str]

class AdvancedClassifier:
    def __init__(self, model_name='gemini-pro'):
        self.model = genai.GenerativeModel(model_name)
        self.classification_history = []
        
    def classify(self, 
                text: str, 
                categories: List[str],
                multi_label: bool = False,
                min_confidence: float = 0.7) -> ClassificationResult:
        """
        고급 텍스트 분류
        
        Args:
            text: 분류할 텍스트
            categories: 카테고리 리스트
            multi_label: 다중 레이블 허용 여부
            min_confidence: 최소 신뢰도 임계값
        """
        if multi_label:
            prompt = self._create_multilabel_prompt(text, categories)
        else:
            prompt = self._create_singlelabel_prompt(text, categories)
        
        response = self.model.generate_content(prompt)
        result = self._parse_classification_response(response.text)
        
        # 신뢰도 필터링
        if result.confidence < min_confidence:
            result.primary_category = "UNCERTAIN"
            result.reasoning += f" (신뢰도 {result.confidence}가 임계값 {min_confidence}보다 낮음)"
        
        # 분류 이력 저장
        self.classification_history.append({
            'text': text[:100],
            'result': result,
            'timestamp': datetime.now()
        })
        
        return result
    
    def _create_singlelabel_prompt(self, text: str, categories: List[str]) -> str:
        """단일 레이블 분류 프롬프트 생성"""
        categories_desc = '\n'.join([f"- {cat}" for cat in categories])
        
        return f"""
        다음 텍스트를 가장 적합한 하나의 카테고리로 분류하세요.
        
        사용 가능한 카테고리:
        {categories_desc}
        
        텍스트:
        {text}
        
        다음 JSON 형식으로 응답하세요:
        {{
            "primary_category": "선택된 카테고리",
            "confidence": 0.95,
            "reasoning": "분류 근거",
            "tags": ["관련", "키워드", "리스트"]
        }}
        """
    
    def _create_multilabel_prompt(self, text: str, categories: List[str]) -> str:
        """다중 레이블 분류 프롬프트 생성"""
        categories_desc = '\n'.join([f"- {cat}" for cat in categories])
        
        return f"""
        다음 텍스트에 해당하는 모든 카테고리를 선택하세요.
        여러 개의 카테고리가 적용될 수 있습니다.
        
        사용 가능한 카테고리:
        {categories_desc}
        
        텍스트:
        {text}
        
        다음 JSON 형식으로 응답하세요:
        {{
            "primary_category": "가장 관련성 높은 카테고리",
            "confidence": 0.95,
            "secondary_categories": [
                ["카테고리2", 0.85],
                ["카테고리3", 0.70]
            ],
            "reasoning": "분류 근거",
            "tags": ["관련", "키워드", "리스트"]
        }}
        """
    
    def _parse_classification_response(self, response_text: str) -> ClassificationResult:
        """응답 텍스트를 파싱하여 ClassificationResult 객체 생성"""
        try:
            # JSON 추출 (응답에 다른 텍스트가 포함될 수 있음)
            json_start = response_text.find('{')
            json_end = response_text.rfind('}') + 1
            json_str = response_text[json_start:json_end]
            
            data = json.loads(json_str)
            
            return ClassificationResult(
                primary_category=data.get('primary_category', 'UNKNOWN'),
                confidence=float(data.get('confidence', 0.0)),
                secondary_categories=data.get('secondary_categories', []),
                reasoning=data.get('reasoning', ''),
                tags=data.get('tags', [])
            )
        except Exception as e:
            # 파싱 실패 시 기본값 반환
            return ClassificationResult(
                primary_category='PARSE_ERROR',
                confidence=0.0,
                secondary_categories=[],
                reasoning=f'응답 파싱 실패: {str(e)}',
                tags=[]
            )
    
    def batch_classify(self, texts: List[str], categories: List[str]) -> List[ClassificationResult]:
        """여러 텍스트를 일괄 분류"""
        results = []
        for text in texts:
            result = self.classify(text, categories)
            results.append(result)
        return results
    
    def get_classification_stats(self) -> Dict:
        """분류 통계 반환"""
        if not self.classification_history:
            return {}
        
        category_counts = {}
        total_confidence = 0
        
        for item in self.classification_history:
            cat = item['result'].primary_category
            category_counts[cat] = category_counts.get(cat, 0) + 1
            total_confidence += item['result'].confidence
        
        return {
            'total_classifications': len(self.classification_history),
            'category_distribution': category_counts,
            'average_confidence': total_confidence / len(self.classification_history)
        }

이제 실제로 사용할 수 있는 고객 문의 자동 분류 시스템을 만들어보자:

class CustomerInquiryClassifier:
    def __init__(self):
        self.classifier = AdvancedClassifier()
        
        # 고객 문의 카테고리 정의
        self.categories = {
            'URGENT': {
                'description': '긴급 문의 (서비스 장애, 결제 오류 등)',
                'priority': 1,
                'sla': '1시간 이내 응답'
            },
            'TECHNICAL': {
                'description': '기술 지원 문의',
                'priority': 2,
                'sla': '4시간 이내 응답'
            },
            'BILLING': {
                'description': '결제 및 청구 관련',
                'priority': 2,
                'sla': '4시간 이내 응답'
            },
            'GENERAL': {
                'description': '일반 문의',
                'priority': 3,
                'sla': '24시간 이내 응답'
            },
            'FEEDBACK': {
                'description': '피드백 및 제안',
                'priority': 4,
                'sla': '48시간 이내 응답'
            }
        }
        
    def classify_inquiry(self, inquiry_text: str, customer_info: Dict = None):
        """
        고객 문의를 분류하고 우선순위 할당
        
        Args:
            inquiry_text: 문의 내용
            customer_info: 고객 정보 (선택사항)
        """
        # 기본 분류
        categories_list = list(self.categories.keys())
        result = self.classifier.classify(
            inquiry_text, 
            categories_list,
            multi_label=True,
            min_confidence=0.6
        )
        
        # 우선순위 정보 추가
        category_info = self.categories.get(result.primary_category, {})
        
        # 감정 분석 추가
        sentiment = self._analyze_sentiment(inquiry_text)
        
        # 긴급도 조정 (부정적 감정이 강하면 우선순위 상향)
        adjusted_priority = category_info.get('priority', 3)
        if sentiment['score'] < -0.5:
            adjusted_priority = max(1, adjusted_priority - 1)
        
        return {
            'classification': result,
            'category_info': category_info,
            'sentiment': sentiment,
            'adjusted_priority': adjusted_priority,
            'recommended_action': self._get_recommended_action(result, sentiment),
            'assigned_team': self._assign_team(result.primary_category)
        }
    
    def _analyze_sentiment(self, text: str) -> Dict:
        """감정 분석"""
        prompt = f"""
        다음 텍스트의 감정을 분석하세요.
        
        텍스트: {text}
        
        JSON 형식으로 응답:
        {{
            "sentiment": "positive/neutral/negative",
            "score": -1.0에서 1.0 사이의 숫자,
            "emotion": "주요 감정 (예: 만족, 불만, 걱정 등)"
        }}
        """
        
        response = self.classifier.model.generate_content(prompt)
        try:
            json_start = response.text.find('{')
            json_end = response.text.rfind('}') + 1
            return json.loads(response.text[json_start:json_end])
        except:
            return {'sentiment': 'neutral', 'score': 0.0, 'emotion': 'unknown'}
    
    def _get_recommended_action(self, classification: ClassificationResult, sentiment: Dict) -> str:
        """추천 액션 생성"""
        actions = {
            'URGENT': '즉시 담당자에게 에스컬레이션',
            'TECHNICAL': '기술 지원팀에 티켓 생성',
            'BILLING': '결제팀에 전달 및 검토 요청',
            'GENERAL': '일반 상담팀 큐에 추가',
            'FEEDBACK': '제품팀에 피드백 전달'
        }
        
        action = actions.get(classification.primary_category, '검토 필요')
        
        if sentiment['sentiment'] == 'negative' and sentiment['score'] < -0.7:
            action = f"⚠️ 고객 불만 주의 - {action}"
        
        return action
    
    def _assign_team(self, category: str) -> str:
        """담당 팀 할당"""
        team_mapping = {
            'URGENT': 'Emergency Response Team',
            'TECHNICAL': 'Technical Support Team',
            'BILLING': 'Billing & Finance Team',
            'GENERAL': 'Customer Service Team',
            'FEEDBACK': 'Product Team'
        }
        return team_mapping.get(category, 'General Queue')

이 시스템은 실제 고객 서비스 센터에서 사용할 수 있는 수준이야. 자동 분류, 우선순위 할당, 감정 분석, 팀 배정까지 모두 자동화되어 있지! 😊

🔄 요약과 분류를 결합한 통합 에이전트

이제 요약과 분류를 하나로 합친 강력한 통합 에이전트를 만들어보자! 실제 업무에서는 이 두 기능을 함께 사용하는 경우가 많거든. 📱

통합 에이전트는 다음과 같은 워크플로우로 동작해:

class UnifiedAgent:
    """요약과 분류를 통합한 AI 에이전트"""
    
    def __init__(self):
        self.summarizer = AdvancedSummarizer()
        self.classifier = AdvancedClassifier()
        self.model = genai.GenerativeModel('gemini-pro')
        
    async def process_document(self, 
                              text: str, 
                              categories: List[str],
                              summarize: bool = True,
                              classify: bool = True,
                              extract_entities: bool = False) -> Dict:
        """
        문서를 종합적으로 처리
        
        Args:
            text: 처리할 텍스트
            categories: 분류 카테고리
            summarize: 요약 수행 여부
            classify: 분류 수행 여부
            extract_entities: 개체명 추출 여부
        
        Returns:
            처리 결과 딕셔너리
        """
        results = {
            'original_length': len(text),
            'processing_timestamp': datetime.now().isoformat()
        }
        
        # 병렬 처리를 위한 태스크 리스트
        tasks = []
        
        if summarize:
            summary_task = self._async_summarize(text)
            tasks.append(('summary', summary_task))
        
        if classify:
            classification_task = self._async_classify(text, categories)
            tasks.append(('classification', classification_task))
        
        if extract_entities:
            entities_task = self._async_extract_entities(text)
            tasks.append(('entities', entities_task))
        
        # 모든 태스크 실행
        for task_name, task in tasks:
            result = await task
            results[task_name] = result
        
        # 메타데이터 생성
        results['metadata'] = self._generate_metadata(results)
        
        return results
    
    async def _async_summarize(self, text: str) -> Dict:
        """비동기 요약"""
        try:
            summary = self.summarizer.summarize(text, style='balanced', length='medium')
            key_points = self._extract_key_points(text)
            
            return {
                'summary': summary,
                'key_points': key_points,
                'compression_ratio': len(summary) / len(text)
            }
        except Exception as e:
            return {'error': str(e)}
    
    async def _async_classify(self, text: str, categories: List[str]) -> Dict:
        """비동기 분류"""
        try:
            result = self.classifier.classify(text, categories, multi_label=True)
            return {
                'primary_category': result.primary_category,
                'confidence': result.confidence,
                'all_categories': result.secondary_categories,
                'tags': result.tags
            }
        except Exception as e:
            return {'error': str(e)}
    
    async def _async_extract_entities(self, text: str) -> Dict:
        """비동기 개체명 추출"""
        prompt = f"""
        다음 텍스트에서 주요 개체명을 추출하세요:
        - 인물 (PERSON)
        - 조직 (ORGANIZATION)
        - 장소 (LOCATION)
        - 날짜 (DATE)
        - 금액 (MONEY)
        
        텍스트:
        {text}
        
        JSON 형식으로 응답:
        {{
            "persons": ["이름1", "이름2"],
            "organizations": ["조직1", "조직2"],
            "locations": ["장소1", "장소2"],
            "dates": ["날짜1", "날짜2"],
            "money": ["금액1", "금액2"]
        }}
        """
        
        try:
            response = self.model.generate_content(prompt)
            json_start = response.text.find('{')
            json_end = response.text.rfind('}') + 1
            return json.loads(response.text[json_start:json_end])
        except Exception as e:
            return {'error': str(e)}
    
    def _extract_key_points(self, text: str) -> List[str]:
        """핵심 포인트 추출"""
        prompt = f"""
        다음 텍스트에서 가장 중요한 3-5개의 핵심 포인트를 추출하세요.
        각 포인트는 한 문장으로 작성하세요.
        
        텍스트:
        {text}
        
        핵심 포인트:
        """
        
        response = self.model.generate_content(prompt)
        points = [line.strip() for line in response.text.split('\n') if line.strip()]
        return points[:5]
    
    def _generate_metadata(self, results: Dict) -> Dict:
        """처리 결과 메타데이터 생성"""
        metadata = {
            'processing_time': datetime.now().isoformat(),
            'components_used': []
        }
        
        if 'summary' in results:
            metadata['components_used'].append('summarization')
            metadata['summary_length'] = len(results['summary'].get('summary', ''))
        
        if 'classification' in results:
            metadata['components_used'].append('classification')
            metadata['primary_category'] = results['classification'].get('primary_category')
        
        if 'entities' in results:
            metadata['components_used'].append('entity_extraction')
            metadata['entity_count'] = sum(len(v) for v in results['entities'].values() if isinstance(v, list))
        
        return metadata

이제 실제로 뉴스 기사를 자동으로 처리하는 시스템을 만들어보자:

class NewsProcessingAgent:
    """뉴스 기사 자동 처리 에이전트"""
    
    def __init__(self):
        self.agent = UnifiedAgent()
        self.news_categories = [
            '정치', '경제', '사회', '문화', '국제', 
            '과학기술', '스포츠', '연예'
        ]
        
    async def process_news_article(self, article: Dict) -> Dict:
        """
        뉴스 기사 종합 처리
        
        Args:
            article: {'title': str, 'content': str, 'source': str, 'date': str}
        """
        full_text = f"{article['title']}\n\n{article['content']}"
        
        # 통합 에이전트로 처리
        result = await self.agent.process_document(
            text=full_text,
            categories=self.news_categories,
            summarize=True,
            classify=True,
            extract_entities=True
        )
        
        # 뉴스 특화 분석 추가
        result['news_analysis'] = await self._analyze_news_specific(article)
        
        # 관련 기사 추천
        result['related_topics'] = self._generate_related_topics(result)
        
        # SEO 메타데이터 생성
        result['seo_metadata'] = self._generate_seo_metadata(article, result)
        
        return result
    
    async def _analyze_news_specific(self, article: Dict) -> Dict:
        """뉴스 특화 분석"""
        prompt = f"""
        다음 뉴스 기사를 분석하세요:
        
        제목: {article['title']}
        내용: {article['content'][:500]}...
        
        다음 항목을 JSON 형식으로 분석:
        {{
            "objectivity_score": 0-100 (객관성 점수),
            "controversy_level": "low/medium/high",
            "impact_assessment": "분석 내용",
            "fact_check_needed": true/false,
            "target_audience": "주요 독자층"
        }}
        """
        
        response = self.agent.model.generate_content(prompt)
        try:
            json_start = response.text.find('{')
            json_end = response.text.rfind('}') + 1
            return json.loads(response.text[json_start:json_end])
        except:
            return {}
    
    def _generate_related_topics(self, result: Dict) -> List[str]:
        """관련 토픽 생성"""
        topics = []
        
        # 분류 카테고리에서 토픽 추출
        if 'classification' in result:
            topics.append(result['classification'].get('primary_category'))
            topics.extend(result['classification'].get('tags', []))
        
        # 개체명에서 토픽 추출
        if 'entities' in result:
            entities = result['entities']
            topics.extend(entities.get('organizations', [])[:3])
            topics.extend(entities.get('persons', [])[:3])
        
        return list(set(topics))[:10]
    
    def _generate_seo_metadata(self, article: Dict, result: Dict) -> Dict:
        """SEO 메타데이터 생성"""
        summary = result.get('summary', {}).get('summary', '')
        tags = result.get('classification', {}).get('tags', [])
        
        return {
            'meta_description': summary[:160] if summary else article['title'],
            'keywords': ', '.join(tags[:10]),
            'og_title': article['title'],
            'og_description': summary[:200] if summary else '',
            'article_section': result.get('classification', {}).get('primary_category', '뉴스')
        }

⚡ 성능 최적화와 비용 관리

Gemini API를 실제 프로덕션 환경에서 사용하려면 성능 최적화와 비용 관리가 정말 중요해. 제대로 관리하지 않으면 예상치 못한 비용이 청구될 수 있거든! 💰

Gemini API는 토큰 기반으로 과금되기 때문에, 토큰 사용량을 줄이는 게 비용 절감의 핵심이야:

class TokenOptimizer:
    """토큰 사용량 최적화 유틸리티"""
    
    def __init__(self):
        self.token_limits = {
            'gemini-pro': 30720,  # 입력 토큰 제한
            'gemini-pro-vision': 12288
        }
        
    def estimate_tokens(self, text: str) -> int:
        """토큰 수 추정 (대략적)"""
        # 영어: 1 토큰 ≈ 4 글자
        # 한글: 1 토큰 ≈ 2-3 글자
        korean_chars = sum(1 for c in text if ord(c) >= 0xAC00 and ord(c) <= 0xD7A3)
        other_chars = len(text) - korean_chars
        
        estimated_tokens = (korean_chars / 2.5) + (other_chars / 4)
        return int(estimated_tokens)
    
    def optimize_prompt(self, prompt: str, max_tokens: int = 2000) -> str:
        """프롬프트 최적화"""
        current_tokens = self.estimate_tokens(prompt)
        
        if current_tokens <= max_tokens:
            return prompt
        
        # 토큰 수가 초과하면 텍스트 압축
        compression_ratio = max_tokens / current_tokens
        target_length = int(len(prompt) * compression_ratio * 0.9)  # 안전 마진
        
        # 중요한 부분 유지하면서 압축
        lines = prompt.split('\n')
        optimized_lines = []
        current_length = 0
        
        for line in lines:
            if current_length + len(line) <= target_length:
                optimized_lines.append(line)
                current_length += len(line)
            else:
                break
        
        return '\n'.join(optimized_lines)
    
    def chunk_large_text(self, text: str, chunk_size: int = 5000) -> List[str]:
        """큰 텍스트를 청크로 분할"""
        chunks = []
        words = text.split()
        current_chunk = []
        current_size = 0
        
        for word in words:
            word_tokens = self.estimate_tokens(word)
            
            if current_size + word_tokens > chunk_size:
                chunks.append(' '.join(current_chunk))
                current_chunk = [word]
                current_size = word_tokens
            else:
                current_chunk.append(word)
                current_size += word_tokens
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks

동일한 요청을 반복하지 않도록 캐싱 시스템을 구축하는 게 중요해:

import hashlib
import json
from typing import Optional
import redis

class CacheManager:
    """API 응답 캐싱 관리자"""
    
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis_client = redis.Redis(
            host=redis_host, 
            port=redis_port, 
            decode_responses=True
        )
        self.default_ttl = 3600 * 24  # 24시간
        
    def generate_cache_key(self, text: str, operation: str, params: Dict = None) -> str:
        """캐시 키 생성"""
        cache_data = {
            'text': text[:500],  # 텍스트 일부만 사용
            'operation': operation,
            'params': params or {}
        }
        
        cache_string = json.dumps(cache_data, sort_keys=True)
        return f"gemini:{operation}:{hashlib.md5(cache_string.encode()).hexdigest()}"
    
    def get_cached_result(self, cache_key: str) -> Optional[Dict]:
        """캐시된 결과 조회"""
        cached = self.redis_client.get(cache_key)
        if cached:
            return json.loads(cached)
        return None
    
    def cache_result(self, cache_key: str, result: Dict, ttl: int = None):
        """결과 캐싱"""
        ttl = ttl or self.default_ttl
        self.redis_client.setex(
            cache_key,
            ttl,
            json.dumps(result)
        )
    
    def invalidate_cache(self, pattern: str):
        """캐시 무효화"""
        keys = self.redis_client.keys(pattern)
        if keys:
            self.redis_client.delete(*keys)

class CachedAgent:
    """캐싱을 적용한 에이전트"""
    
    def __init__(self):
        self.agent = UnifiedAgent()
        self.cache = CacheManager()
        self.cache_hit_count = 0
        self.cache_miss_count = 0
        
    async def process_with_cache(self, text: str, categories: List[str]) -> Dict:
        """캐싱을 활용한 처리"""
        # 캐시 키 생성
        cache_key = self.cache.generate_cache_key(
            text, 
            'unified_process',
            {'categories': categories}
        )
        
        # 캐시 확인
        cached_result = self.cache.get_cached_result(cache_key)
        if cached_result:
            self.cache_hit_count += 1
            cached_result['from_cache'] = True
            return cached_result
        
        # 캐시 미스 - 실제 처리
        self.cache_miss_count += 1
        result = await self.agent.process_document(
            text, 
            categories,
            summarize=True,
            classify=True
        )
        
        # 결과 캐싱
        result['from_cache'] = False
        self.cache.cache_result(cache_key, result)
        
        return result
    
    def get_cache_stats(self) -> Dict:
        """캐시 통계"""
        total = self.cache_hit_count + self.cache_miss_count
        hit_rate = (self.cache_hit_count / total * 100) if total > 0 else 0
        
        return {
            'cache_hits': self.cache_hit_count,
            'cache_misses': self.cache_miss_count,
            'hit_rate': f"{hit_rate:.2f}%",
            'total_requests': total
        }

API 사용량과 비용을 실시간으로 모니터링하는 시스템도 중요해:

class CostMonitor:
    """API 비용 모니터링"""
    
    def __init__(self):
        self.pricing = {
            'gemini-pro': {
                'input': 0.00025,   # 1K 토큰당 $0.00025
                'output': 0.0005    # 1K 토큰당 $0.0005
            }
        }
        self.usage_log = []
        
    def log_api_call(self, 
                    model: str,
                    input_tokens: int,
                    output_tokens: int,
                    operation: str):
        """API 호출 로깅"""
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'model': model,
            'operation': operation,
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'cost': cost
        }
        
        self.usage_log.append(log_entry)
        
        # 비용 임계값 체크
        self._check_cost_threshold()
        
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """비용 계산"""
        pricing = self.pricing.get(model, self.pricing['gemini-pro'])
        
        input_cost = (input_tokens / 1000) * pricing['input']
        output_cost = (output_tokens / 1000) * pricing['output']
        
        return input_cost + output_cost
    
    def get_daily_stats(self) -> Dict:
        """일일 통계"""
        today = datetime.now().date()
        today_logs = [
            log for log in self.usage_log 
            if datetime.fromisoformat(log['timestamp']).date() == today
        ]
        
        if not today_logs:
            return {}
        
        total_cost = sum(log['cost'] for log in today_logs)
        total_calls = len(today_logs)
        total_input_tokens = sum(log['input_tokens'] for log in today_logs)
        total_output_tokens = sum(log['output_tokens'] for log in today_logs)
        
        return {
            'date': today.isoformat(),
            'total_calls': total_calls,
            'total_cost': f"${total_cost:.4f}",
            'total_input_tokens': total_input_tokens,
            'total_output_tokens': total_output_tokens,
            'average_cost_per_call': f"${total_cost/total_calls:.4f}" if total_calls > 0 else "$0"
        }
    
    def get_monthly_projection(self) -> Dict:
        """월간 비용 예측"""
        daily_stats = self.get_daily_stats()
        if not daily_stats:
            return {}
        
        daily_cost = float(daily_stats['total_cost'].replace('$', ''))
        monthly_projection = daily_cost * 30
        
        return {
            'daily_average': f"${daily_cost:.2f}",
            'monthly_projection': f"${monthly_projection:.2f}",
            'annual_projection': f"${monthly_projection * 12:.2f}"
        }
    
    def _check_cost_threshold(self):
        """비용 임계값 체크"""
        daily_stats = self.get_daily_stats()
        if not daily_stats:
            return
        
        daily_cost = float(daily_stats['total_cost'].replace('$', ''))
        
        # 일일 비용이 $10를 초과하면 경고
        if daily_cost > 10:
            print(f"⚠️ 경고: 오늘 API 비용이 ${daily_cost:.2f}입니다!")
        
        # 월간 예상 비용이 $200를 초과하면 경고
        monthly = self.get_monthly_projection()
        if monthly:
            monthly_cost = float(monthly['monthly_projection'].replace('$', ''))
            if monthly_cost > 200:
                print(f"⚠️ 경고: 월간 예상 비용이 ${monthly_cost:.2f}입니다!")

🔮 고급 기능: 멀티모달 처리

Gemini API의 진짜 강점은 멀티모달 처리야! 텍스트뿐만 아니라 이미지, 동영상까지 함께 처리할 수 있거든. 🎬

이미지가 포함된 문서를 처리하는 방법을 알아보자:

import PIL.Image
import io
import base64

class MultimodalAgent:
    """멀티모달 처리 에이전트"""
    
    def __init__(self):
        self.vision_model = genai.GenerativeModel('gemini-pro-vision')
        self.text_model = genai.GenerativeModel('gemini-pro')
        
    def analyze_image_with_text(self, image_path: str, text_context: str = None) -> Dict:
        """
        이미지와 텍스트를 함께 분석
        
        Args:
            image_path: 이미지 파일 경로
            text_context: 추가 텍스트 컨텍스트
        """
        # 이미지 로드
        image = PIL.Image.open(image_path)
        
        # 프롬프트 생성
        if text_context:
            prompt = f"""
            다음 이미지를 분석하고, 제공된 텍스트 컨텍스트와 연관지어 설명하세요.
            
            텍스트 컨텍스트:
            {text_context}
            
            다음 항목을 포함하여 분석하세요:
            1. 이미지에 무엇이 보이는가?
            2. 텍스트와 이미지의 관련성
            3. 주요 인사이트
            4. 추천 액션
            """
        else:
            prompt = """
            이 이미지를 상세히 분석하고 다음을 설명하세요:
            1. 주요 객체와 장면
            2. 이미지의 목적이나 메시지
            3. 주목할 만한 세부사항
            4. 가능한 활용 방안
            """
        
        # 이미지 분석
        response = self.vision_model.generate_content([prompt, image])
        
        return {
            'analysis': response.text,
            'image_path': image_path,
            'has_context': text_context is not None
        }
    
    def extract_text_from_image(self, image_path: str) -> str:
        """이미지에서 텍스트 추출 (OCR)"""
        image = PIL.Image.open(image_path)
        
        prompt = """
        이 이미지에 있는 모든 텍스트를 추출하세요.
        텍스트의 원래 형식과 구조를 최대한 유지하세요.
        """
        
        response = self.vision_model.generate_content([prompt, image])
        return response.text
    
    def analyze_document_with_images(self, 
                                    text: str, 
                                    image_paths: List[str]) -> Dict:
        """
        텍스트와 여러 이미지를 포함한 문서 종합 분석
        """
        results = {
            'text_analysis': {},
            'image_analyses': [],
            'integrated_summary': ''
        }
        
        # 1. 텍스트 분석
        text_prompt = f"""
        다음 문서를 분석하세요:
        
        {text}
        
        핵심 내용을 요약하고 주요 포인트를 추출하세요.
        """
        
        text_response = self.text_model.generate_content(text_prompt)
        results['text_analysis'] = {
            'summary': text_response.text,
            'length': len(text)
        }
        
        # 2. 각 이미지 분석
        for idx, image_path in enumerate(image_paths):
            image_analysis = self.analyze_image_with_text(image_path, text[:500])
            results['image_analyses'].append({
                'image_index': idx,
                'path': image_path,
                'analysis': image_analysis['analysis']
            })
        
        # 3. 통합 요약 생성
        integrated_prompt = f"""
        다음은 문서의 텍스트 분석과 이미지 분석 결과입니다.
        이를 통합하여 문서 전체에 대한 종합적인 요약을 작성하세요.
        
        텍스트 요약:
        {results['text_analysis']['summary']}
        
        이미지 분석들:
        {chr(10).join([f"이미지 {i+1}: {img['analysis'][:200]}..." for i, img in enumerate(results['image_analyses'])])}
        
        통합 요약을 작성하세요 (300자 이내):
        """
        
        integrated_response = self.text_model.generate_content(integrated_prompt)
        results['integrated_summary'] = integrated_response.text
        
        return results
    
    def classify_image(self, image_path: str, categories: List[str]) -> Dict:
        """이미지 분류"""
        image = PIL.Image.open(image_path)
        
        categories_str = ', '.join(categories)
        prompt = f"""
        이 이미지를 다음 카테고리 중 하나로 분류하세요: {categories_str}
        
        JSON 형식으로 응답:
        {{
            "category": "선택된 카테고리",
            "confidence": 0.95,
            "reasoning": "분류 근거"
        }}
        """
        
        response = self.vision_model.generate_content([prompt, image])
        
        try:
            json_start = response.text.find('{')
            json_end = response.text.rfind('}') + 1
            return json.loads(response.text[json_start:json_end])
        except:
            return {
                'category': 'Unknown',
                'confidence': 0.0,
                'reasoning': 'Parse error'
            }

소셜 미디어 포스트를 자동으로 분석하는 시스템을 만들어보자:

class SocialMediaAnalyzer:
    """소셜 미디어 콘텐츠 분석기"""
    
    def __init__(self):
        self.multimodal = MultimodalAgent()
        self.text_model = genai.GenerativeModel('gemini-pro')
        
    def analyze_post(self, post: Dict) -> Dict:
        """
        소셜 미디어 포스트 종합 분석
        
        Args:
            post: {
                'text': str,
                'images': List[str],  # 이미지 경로들
                'platform': str,  # 'instagram', 'twitter', 'facebook' 등
                'author': str,
                'timestamp': str
            }
        """
        analysis = {
            'post_id': post.get('id', 'unknown'),
            'platform': post['platform'],
            'timestamp': post['timestamp']
        }
        
        # 1. 텍스트 감정 분석
        if post.get('text'):
            analysis['sentiment'] = self._analyze_sentiment(post['text'])
            analysis['topics'] = self._extract_topics(post['text'])
            analysis['hashtags'] = self._extract_hashtags(post['text'])
        
        # 2. 이미지 분석
        if post.get('images'):
            analysis['image_analysis'] = []
            for img_path in post['images']:
                img_result = self.multimodal.analyze_image_with_text(
                    img_path,
                    post.get('text', '')
                )
                analysis['image_analysis'].append(img_result)
        
        # 3. 콘텐츠 품질 평가
        analysis['quality_metrics'] = self._evaluate_quality(post, analysis)
        
        # 4. 참여도 예측
        analysis['engagement_prediction'] = self._predict_engagement(analysis)
        
        # 5. 추천 사항
        analysis['recommendations'] = self._generate_recommendations(analysis)
        
        return analysis
    
    def _analyze_sentiment(self, text: str) -> Dict:
        """감정 분석"""
        prompt = f"""
        다음 소셜 미디어 포스트의 감정을 분석하세요:
        
        "{text}"
        
        JSON 형식으로 응답:
        {{
            "overall_sentiment": "positive/neutral/negative",
            "sentiment_score": -1.0에서 1.0 사이,
            "emotions": ["기쁨", "흥분" 등 감정 리스트],
            "tone": "캐주얼/전문적/유머러스 등"
        }}
        """
        
        response = self.text_model.generate_content(prompt)
        try:
            json_start = response.text.find('{')
            json_end = response.text.rfind('}') + 1
            return json.loads(response.text[json_start:json_end])
        except:
            return {'overall_sentiment': 'neutral', 'sentiment_score': 0.0}
    
    def _extract_topics(self, text: str) -> List[str]:
        """주제 추출"""
        prompt = f"""
        다음 텍스트에서 주요 주제/토픽을 3-5개 추출하세요:
        
        "{text}"
        
        주제만 쉼표로 구분하여 나열하세요.
        """
        
        response = self.text_model.generate_content(prompt)
        topics = [t.strip() for t in response.text.split(',')]
        return topics[:5]
    
    def _extract_hashtags(self, text: str) -> List[str]:
        """해시태그 추출"""
        import re
        hashtags = re.findall(r'#\w+', text)
        return hashtags
    
    def _evaluate_quality(self, post: Dict, analysis: Dict) -> Dict:
        """콘텐츠 품질 평가"""
        quality = {
            'text_quality': 0.5,
            'visual_quality': 0.5,
            'engagement_potential': 0.5,
            'overall_score': 0.5
        }
        
        # 텍스트 품질
        text = post.get('text', '')
        if text:
            # 길이 적절성
            if 50 < len(text) < 300:
                quality['text_quality'] += 0.2
            
            # 해시태그 사용
            if analysis.get('hashtags'):
                quality['text_quality'] += 0.1
            
            # 감정 명확성
            sentiment = analysis.get('sentiment', {})
            if abs(sentiment.get('sentiment_score', 0)) > 0.5:
                quality['text_quality'] += 0.2
        
        # 비주얼 품질
        if post.get('images'):
            quality['visual_quality'] += 0.3
            if len(post['images']) > 1:
                quality['visual_quality'] += 0.2
        
        # 전체 점수
        quality['overall_score'] = (
            quality['text_quality'] * 0.4 +
            quality['visual_quality'] * 0.3 +
            quality['engagement_potential'] * 0.3
        )
        
        return quality
    
    def _predict_engagement(self, analysis: Dict) -> Dict:
        """참여도 예측"""
        # 간단한 휴리스틱 기반 예측
        score = 0.5
        
        # 감정이 강할수록 참여도 높음
        sentiment = analysis.get('sentiment', {})
        sentiment_score = abs(sentiment.get('sentiment_score', 0))
        score += sentiment_score * 0.2
        
        # 이미지가 있으면 참여도 높음
        if analysis.get('image_analysis'):
            score += 0.2
        
        # 해시태그가 적절하면 참여도 높음
        hashtag_count = len(analysis.get('hashtags', []))
        if 2 <= hashtag_count <= 5:
            score += 0.1
        
        return {
            'predicted_score': min(score, 1.0),
            'confidence': 0.7,
            'factors': {
                'sentiment_impact': sentiment_score * 0.2,
                'visual_impact': 0.2 if analysis.get('image_analysis') else 0,
                'hashtag_impact': 0.1 if 2 <= hashtag_count <= 5 else 0
            }
        }
    
    def _generate_recommendations(self, analysis: Dict) -> List[str]:
        """개선 추천사항 생성"""
        recommendations = []
        
        # 텍스트 관련
        if not analysis.get('hashtags'):
            recommendations.append("관련 해시태그를 2-3개 추가하세요")
        
        sentiment = analysis.get('sentiment', {})
        if abs(sentiment.get('sentiment_score', 0)) < 0.3:
            recommendations.append("더 명확한 감정이나 의견을 표현하세요")
        
        # 이미지 관련
        if not analysis.get('image_analysis'):
            recommendations.append("시각적 요소(이미지/동영상)를 추가하세요")
        
        # 품질 관련
        quality = analysis.get('quality_metrics', {})
        if quality.get('overall_score', 0) < 0.6:
            recommendations.append("콘텐츠의 전반적인 품질을 개선하세요")
        
        return recommendations

📊 모니터링과 개선

에이전트를 배포한 후에는 지속적인 모니터링과 개선이 필요해. 실제 사용 데이터를 분석해서 시스템을 계속 발전시켜야 하거든! 📈

시스템의 성능을 추적하는 대시보드를 만들어보자:

class PerformanceMonitor:
    """성능 모니터링 시스템"""
    
    def __init__(self):
        self.metrics = {
            'api_calls': [],
            'response_times': [],
            'error_rates': [],
            'user_satisfaction': []
        }
        
    def log_api_call(self, 
                    operation: str,
                    response_time: float,
                    success: bool,
                    token_usage: Dict):
        """API 호출 로깅"""
        self.metrics['api_calls'].append({
            'timestamp': datetime.now(),
            'operation': operation,
            'response_time': response_time,
            'success': success,
            'token_usage': token_usage
        })
        
        self.metrics['response_times'].append(response_time)
        
        if not success:
            self.metrics['error_rates'].append({
                'timestamp': datetime.now(),
                'operation': operation
            })
    
    def get_performance_summary(self, time_window_hours: int = 24) -> Dict:
        """성능 요약 통계"""
        cutoff = datetime.now() - timedelta(hours=time_window_hours)
        
        recent_calls = [
            call for call in self.metrics['api_calls']
            if call['timestamp'] > cutoff
        ]
        
        if not recent_calls:
            return {}
        
        total_calls = len(recent_calls)
        successful_calls = sum(1 for call in recent_calls if call['success'])
        failed_calls = total_calls - successful_calls
        
        response_times = [call['response_time'] for call in recent_calls]
        avg_response_time = sum(response_times) / len(response_times)
        
        return {
            'time_window': f'{time_window_hours}시간',
            'total_calls': total_calls,
            'successful_calls': successful_calls,
            'failed_calls': failed_calls,
            'success_rate': f'{(successful_calls/total_calls*100):.2f}%',
            'average_response_time': f'{avg_response_time:.2f}초',
            'min_response_time': f'{min(response_times):.2f}초',
            'max_response_time': f'{max(response_times):.2f}초'
        }
    
    def get_operation_breakdown(self) -> Dict:
        """작업별 통계"""
        operation_stats = {}
        
        for call in self.metrics['api_calls']:
            op = call['operation']
            if op not in operation_stats:
                operation_stats[op] = {
                    'count': 0,
                    'total_time': 0,
                    'successes': 0,
                    'failures': 0
                }
            
            operation_stats[op]['count'] += 1
            operation_stats[op]['total_time'] += call['response_time']
            
            if call['success']:
                operation_stats[op]['successes'] += 1
            else:
                operation_stats[op]['failures'] += 1
        
        # 평균 계산
        for op in operation_stats:
            stats = operation_stats[op]
            stats['avg_time'] = stats['total_time'] / stats['count']
            stats['success_rate'] = (stats['successes'] / stats['count'] * 100)
        
        return operation_stats
    
    def detect_anomalies(self) -> List[Dict]:
        """이상 징후 감지"""
        anomalies = []
        
        # 최근 1시간 데이터
        recent_hour = datetime.now() - timedelta(hours=1)
        recent_calls = [
            call for call in self.metrics['api_calls']
            if call['timestamp'] > recent_hour
        ]
        
        if not recent_calls:
            return anomalies
        
        # 응답 시간 이상
        response_times = [call['response_time'] for call in recent_calls]
        avg_time = sum(response_times) / len(response_times)
        
        for call in recent_calls[-10:]:  # 최근 10개 체크
            if call['response_time'] > avg_time * 2:
                anomalies.append({
                    'type': 'slow_response',
                    'timestamp': call['timestamp'],
                    'operation': call['operation'],
                    'response_time': call['response_time'],
                    'threshold': avg_time * 2
                })
        
        # 에러율 이상
        recent_errors = [call for call in recent_calls if not call['success']]
        error_rate = len(recent_errors) / len(recent_calls)
        
        if error_rate > 0.1:  # 10% 이상
            anomalies.append({
                'type': 'high_error_rate',
                'timestamp': datetime.now(),
                'error_rate': f'{error_rate*100:.2f}%',
                'threshold': '10%'
            })
        
        return anomalies
    
    def generate_report(self) -> str:
        """종합 리포트 생성"""
        summary = self.get_performance_summary(24)
        operations = self.get_operation_breakdown()
        anomalies = self.detect_anomalies()
        
        report = f"""
=== 성능 모니터링 리포트 ===
생성 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

[전체 요약 (24시간)]
- 총 API 호출: {summary.get('total_calls', 0)}회
- 성공률: {summary.get('success_rate', 'N/A')}
- 평균 응답 시간: {summary.get('average_response_time', 'N/A')}

[작업별 통계]
"""
        
        for op, stats in operations.items():
            report += f"\n{op}:"
            report += f"\n  - 호출 횟수: {stats['count']}회"
            report += f"\n  - 평균 시간: {stats['avg_time']:.2f}초"
            report += f"\n  - 성공률: {stats['success_rate']:.2f}%"
        
        if anomalies:
            report += "\n\n[⚠️ 감지된 이상 징후]"
            for anomaly in anomalies:
                report += f"\n- {anomaly['type']}: {anomaly}"
        else:
            report += "\n\n[✅ 이상 징후 없음]"
        
        return report

사용자 피드백을 수집하고 분석하는 시스템도 중요해:

class FeedbackCollector:
    """사용자 피드백 수집 및 분석"""
    
    def __init__(self):
        self.feedback_db = []
        self.text_model = genai.GenerativeModel('gemini-pro')
        
    def collect_feedback(self, 
                        user_id: str,
                        operation: str,
                        result_id: str,
                        rating: int,
                        comment: str = None):
        """피드백 수집"""
        feedback = {
            'id': self._generate_id(),
            'user_id': user_id,
            'operation': operation,
            'result_id': result_id,
            'rating': rating,  # 1-5
            'comment': comment,
            'timestamp': datetime.now().isoformat()
        }
        
        self.feedback_db.append(feedback)
        
        # 부정적 피드백 즉시 알림
        if rating <= 2:
            self._alert_negative_feedback(feedback)
    
    def analyze_feedback_trends(self) -> Dict:
        """피드백 트렌드 분석"""
        if not self.feedback_db:
            return {}
        
        # 평균 평점
        avg_rating = sum(f['rating'] for f in self.feedback_db) / len(self.feedback_db)
        
        # 작업별 평점
        operation_ratings = {}
        for feedback in self.feedback_db:
            op = feedback['operation']
            if op not in operation_ratings:
                operation_ratings[op] = []
            operation_ratings[op].append(feedback['rating'])
        
        operation_avg = {
            op: sum(ratings) / len(ratings)
            for op, ratings in operation_ratings.items()
        }
        
        # 최근 트렌드 (최근 30일)
        recent_cutoff = datetime.now() - timedelta(days=30)
        recent_feedback = [
            f for f in self.feedback_db
            if datetime.fromisoformat(f['timestamp']) > recent_cutoff
        ]
        
        recent_avg = sum(f['rating'] for f in recent_feedback) / len(recent_feedback) if recent_feedback else 0
        
        return {
            'overall_average': f'{avg_rating:.2f}',
            'recent_average': f'{recent_avg:.2f}',
            'total_feedback': len(self.feedback_db),
            'operation_breakdown': operation_avg,
            'trend': 'improving' if recent_avg > avg_rating else 'declining'
        }
    
    def extract_improvement_suggestions(self) -> List[str]:
        """개선 제안 추출"""
        # 부정적 피드백의 코멘트 수집
        negative_feedback = [
            f for f in self.feedback_db
            if f['rating'] <= 2 and f.get('comment')
        ]
        
        if not negative_feedback:
            return []
        
        comments = '\n'.join([f['comment'] for f in negative_feedback])
        
        prompt = f"""
        다음은 사용자들의 부정적 피드백 코멘트들입니다.
        이를 분석하여 시스템 개선을 위한 구체적인 제안사항을 3-5개 추출하세요.
        
        피드백:
        {comments}
        
        개선 제안사항을 번호를 매겨 나열하세요.
        """
        
        response = self.text_model.generate_content(prompt)
        suggestions = [
            line.strip() 
            for line in response.text.split('\n') 
            if line.strip() and any(c.isdigit() for c in line[:3])
        ]
        
        return suggestions[:5]
    
    def _alert_negative_feedback(self, feedback: Dict):
        """부정적 피드백 알림"""
        print(f"""
        ⚠️ 부정적 피드백 감지!
        사용자: {feedback['user_id']}
        작업: {feedback['operation']}
        평점: {feedback['rating']}/5
        코멘트: {feedback.get('comment', 'N/A')}
        """)
    
    def _generate_id(self) -> str:
        import uuid
        return str(uuid.uuid4())