Spaces:
Running
Running
| import gradio as gr | |
| import os | |
| import json | |
| import requests | |
| from datetime import datetime | |
| import time | |
| from typing import List, Dict, Any, Generator, Tuple | |
| import logging | |
| import re | |
| # 로깅 설정 | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # 추가 임포트 | |
| from bs4 import BeautifulSoup | |
| from urllib.parse import urlparse | |
| import urllib.request | |
| # Gemini API 임포트 | |
| try: | |
| from google import genai | |
| from google.genai import types | |
| GEMINI_AVAILABLE = True | |
| except ImportError: | |
| GEMINI_AVAILABLE = False | |
| logger.warning("Google Gemini API가 설치되지 않았습니다. pip install google-genai로 설치하세요.") | |
| # 환경 변수에서 토큰 가져오기 | |
| FRIENDLI_TOKEN = os.getenv("FRIENDLI_TOKEN", "YOUR_FRIENDLI_TOKEN") | |
| BAPI_TOKEN = os.getenv("BAPI_TOKEN", "YOUR_BRAVE_API_TOKEN") | |
| GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "YOUR_GEMINI_API_KEY") | |
| API_URL = "https://api.friendli.ai/dedicated/v1/chat/completions" | |
| BRAVE_SEARCH_URL = "https://api.search.brave.com/res/v1/web/search" | |
| MODEL_ID = "dep89a2fld32mcm" | |
| TEST_MODE = os.getenv("TEST_MODE", "false").lower() == "true" | |
| # 전역 변수 | |
| conversation_history = [] | |
| class LLMCollaborativeSystem: | |
| def __init__(self): | |
| self.token = FRIENDLI_TOKEN | |
| self.bapi_token = BAPI_TOKEN | |
| self.gemini_api_key = GEMINI_API_KEY | |
| self.api_url = API_URL | |
| self.brave_url = BRAVE_SEARCH_URL | |
| self.model_id = MODEL_ID | |
| self.test_mode = TEST_MODE or (self.token == "YOUR_FRIENDLI_TOKEN") | |
| self.use_gemini = False | |
| self.gemini_client = None | |
| if self.test_mode: | |
| logger.warning("테스트 모드로 실행됩니다.") | |
| if self.bapi_token == "YOUR_BRAVE_API_TOKEN": | |
| logger.warning("Brave API 토큰이 설정되지 않았습니다.") | |
| if self.gemini_api_key == "YOUR_GEMINI_API_KEY": | |
| logger.warning("Gemini API 토큰이 설정되지 않았습니다.") | |
| def set_llm_mode(self, mode: str): | |
| """LLM 모드 설정 (default 또는 commercial)""" | |
| if mode == "commercial" and GEMINI_AVAILABLE and self.gemini_api_key != "YOUR_GEMINI_API_KEY": | |
| self.use_gemini = True | |
| if not self.gemini_client: | |
| self.gemini_client = genai.Client(api_key=self.gemini_api_key) | |
| logger.info("Gemini 2.5 Pro 모드로 전환되었습니다.") | |
| else: | |
| self.use_gemini = False | |
| logger.info("기본 LLM 모드로 전환되었습니다.") | |
| def create_headers(self): | |
| """API 헤더 생성""" | |
| return { | |
| "Authorization": f"Bearer {self.token}", | |
| "Content-Type": "application/json" | |
| } | |
| def create_brave_headers(self): | |
| """Brave API 헤더 생성""" | |
| return { | |
| "Accept": "application/json", | |
| "Accept-Encoding": "gzip", | |
| "X-Subscription-Token": self.bapi_token | |
| } | |
| def create_supervisor_initial_prompt(self, user_query: str) -> str: | |
| """감독자 AI 초기 프롬프트 생성""" | |
| return f"""당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 이 질문에 대해: | |
| 1. 전체적인 접근 방향과 프레임워크를 제시하세요 | |
| 2. 핵심 요소와 고려사항을 구조화하여 설명하세요 | |
| 3. 이 주제에 대해 조사가 필요한 5-7개의 구체적인 키워드나 검색어를 제시하세요 | |
| 키워드는 다음 형식으로 제시하세요: | |
| [검색 키워드]: 키워드1, 키워드2, 키워드3, 키워드4, 키워드5""" | |
| def create_researcher_prompt(self, user_query: str, supervisor_guidance: str, search_results: Dict[str, List[Dict]]) -> str: | |
| """조사자 AI 프롬프트 생성""" | |
| search_summary = "" | |
| all_results = [] | |
| for keyword, results in search_results.items(): | |
| search_summary += f"\n\n**{keyword}에 대한 검색 결과:**\n" | |
| for i, result in enumerate(results[:10], 1): # 상위 10개만 표시 | |
| search_summary += f"{i}. {result.get('title', 'N/A')} (신뢰도: {result.get('credibility_score', 0):.2f})\n" | |
| search_summary += f" - {result.get('description', 'N/A')}\n" | |
| search_summary += f" - 출처: {result.get('url', 'N/A')}\n" | |
| if result.get('published'): | |
| search_summary += f" - 게시일: {result.get('published')}\n" | |
| all_results.extend(results) | |
| # 모순 감지 | |
| contradictions = self.detect_contradictions(all_results) | |
| contradiction_text = "" | |
| if contradictions: | |
| contradiction_text = "\n\n**발견된 정보 모순:**\n" | |
| for cont in contradictions[:3]: # 최대 3개만 표시 | |
| contradiction_text += f"- {cont['type']}: {cont['source1']} vs {cont['source2']}\n" | |
| return f"""당신은 정보를 조사하고 정리하는 조사자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 감독자 AI의 지침: | |
| {supervisor_guidance} | |
| 브레이브 검색 결과 (신뢰도 점수 포함): | |
| {search_summary} | |
| {contradiction_text} | |
| 위 검색 결과를 바탕으로: | |
| 1. 각 키워드별로 중요한 정보를 정리하세요 | |
| 2. 신뢰할 수 있는 출처(신뢰도 0.7 이상)를 우선적으로 참고하세요 | |
| 3. 출처를 명확히 표기하여 실행자 AI가 검증할 수 있도록 하세요 | |
| 4. 정보의 모순이 있다면 양쪽 관점을 모두 제시하세요 | |
| 5. 최신 트렌드나 중요한 통계가 있다면 강조하세요 | |
| 6. 신뢰도가 낮은 정보는 주의 표시와 함께 포함하세요""" | |
| def create_supervisor_execution_prompt(self, user_query: str, research_summary: str) -> str: | |
| """감독자 AI의 실행 지시 프롬프트""" | |
| return f"""당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 조사자 AI가 정리한 조사 내용: | |
| {research_summary} | |
| 위 조사 내용을 기반으로 실행자 AI에게 아주 구체적인 지시를 내려주세요: | |
| 1. 조사된 정보를 어떻게 활용할지 명확히 지시하세요 | |
| 2. 실행 가능한 단계별 작업을 구체적으로 제시하세요 | |
| 3. 각 단계에서 참고해야 할 조사 내용을 명시하세요 | |
| 4. 예상되는 결과물의 형태를 구체적으로 설명하세요""" | |
| def create_executor_prompt(self, user_query: str, supervisor_guidance: str, research_summary: str) -> str: | |
| """실행자 AI 프롬프트 생성""" | |
| return f"""당신은 세부적인 내용을 구현하는 실행자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 조사자 AI가 정리한 조사 내용: | |
| {research_summary} | |
| 감독자 AI의 구체적인 지시: | |
| {supervisor_guidance} | |
| 위 조사 내용과 지시사항을 바탕으로: | |
| 1. 조사된 정보를 적극 활용하여 구체적인 실행 계획을 작성하세요 | |
| 2. 각 단계별로 참고한 조사 내용을 명시하세요 | |
| 3. 실제로 적용 가능한 구체적인 방법론을 제시하세요 | |
| 4. 예상되는 성과와 측정 방법을 포함하세요""" | |
| def create_executor_final_prompt(self, user_query: str, initial_response: str, supervisor_feedback: str, research_summary: str) -> str: | |
| """실행자 AI 최종 보고서 프롬프트""" | |
| return f"""당신은 세부적인 내용을 구현하는 실행자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 조사자 AI의 조사 내용: | |
| {research_summary} | |
| 당신의 초기 답변: | |
| {initial_response} | |
| 감독자 AI의 피드백 및 개선사항: | |
| {supervisor_feedback} | |
| 위 피드백을 완전히 반영하여 최종 보고서를 작성하세요: | |
| 1. 감독자의 모든 개선사항을 반영하세요 | |
| 2. 조사 내용을 더욱 구체적으로 활용하세요 | |
| 3. 실행 가능성을 높이는 세부 계획을 포함하세요 | |
| 4. 명확한 결론과 다음 단계를 제시하세요 | |
| 5. 전문적이고 완성도 높은 최종 보고서 형식으로 작성하세요""" | |
| def create_evaluator_prompt(self, user_query: str, supervisor_responses: List[str], researcher_response: str, executor_responses: List[str], evaluator_responses: List[str] = None) -> str: | |
| """평가자 AI 프롬프트 생성""" | |
| evaluator_history = "" | |
| if evaluator_responses and len(evaluator_responses) > 0: | |
| evaluator_history = f""" | |
| 평가자 AI의 이전 평가들: | |
| - 조사 결과 평가: {evaluator_responses[0] if len(evaluator_responses) > 0 else 'N/A'} | |
| - 초기 구현 평가: {evaluator_responses[1] if len(evaluator_responses) > 1 else 'N/A'} | |
| """ | |
| return f"""당신은 전체 협력 과정과 결과를 평가하는 평가자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 감독자 AI의 분석 및 지시: | |
| - 초기 분석: {supervisor_responses[0]} | |
| - 실행 지시: {supervisor_responses[1] if len(supervisor_responses) > 1 else 'N/A'} | |
| - 개선 지시: {supervisor_responses[2] if len(supervisor_responses) > 2 else 'N/A'} | |
| 조사자 AI의 조사 결과: | |
| {researcher_response} | |
| 실행자 AI의 구현: | |
| - 초기 구현: {executor_responses[0]} | |
| - 최종 보고서: {executor_responses[1] if len(executor_responses) > 1 else 'N/A'} | |
| {evaluator_history} | |
| 위 전체 과정을 평가하여: | |
| 1. **품질 평가**: 각 AI의 답변 품질과 역할 수행도를 평가하세요 (10점 만점) | |
| 2. **협력 효과성**: AI 간 협력이 얼마나 효과적이었는지 평가하세요 | |
| 3. **정보 활용도**: 웹 검색 정보가 얼마나 잘 활용되었는지 평가하세요 | |
| 4. **개선점**: 향후 개선이 필요한 부분을 구체적으로 제시하세요 | |
| 5. **최종 평점**: 전체 프로세스에 대한 종합 평가를 제시하세요 | |
| 평가는 구체적이고 건설적으로 작성하세요.""" | |
| def extract_keywords(self, supervisor_response: str) -> List[str]: | |
| """감독자 응답에서 키워드 추출""" | |
| keywords = [] | |
| # [검색 키워드]: 형식으로 키워드 찾기 | |
| keyword_match = re.search(r'\[검색 키워드\]:\s*(.+)', supervisor_response, re.IGNORECASE) | |
| if keyword_match: | |
| keyword_str = keyword_match.group(1) | |
| keywords = [k.strip() for k in keyword_str.split(',') if k.strip()] | |
| # 키워드가 없으면 기본 키워드 생성 | |
| if not keywords: | |
| keywords = ["best practices", "implementation guide", "case studies", "latest trends", "success factors"] | |
| return keywords[:7] # 최대 7개로 제한 | |
| def generate_synonyms(self, keyword: str) -> List[str]: | |
| """키워드의 동의어/유사어 생성""" | |
| synonyms = { | |
| "optimization": ["improvement", "enhancement", "efficiency", "tuning"], | |
| "performance": ["speed", "efficiency", "throughput", "latency"], | |
| "strategy": ["approach", "method", "technique", "plan"], | |
| "implementation": ["deployment", "execution", "development", "integration"], | |
| "analysis": ["evaluation", "assessment", "study", "research"], | |
| "management": ["administration", "governance", "control", "supervision"], | |
| "best practices": ["proven methods", "industry standards", "guidelines", "recommendations"], | |
| "trends": ["developments", "innovations", "emerging", "future"], | |
| "machine learning": ["ML", "AI", "deep learning", "neural networks"], | |
| "프로젝트": ["project", "사업", "업무", "작업"] | |
| } | |
| # 키워드 정규화 | |
| keyword_lower = keyword.lower() | |
| # 직접 매칭되는 동의어가 있으면 반환 | |
| if keyword_lower in synonyms: | |
| return synonyms[keyword_lower][:2] # 최대 2개 | |
| # 부분 매칭 확인 | |
| for key, values in synonyms.items(): | |
| if key in keyword_lower or keyword_lower in key: | |
| return values[:2] | |
| # 동의어가 없으면 빈 리스트 | |
| return [] | |
| def calculate_credibility_score(self, result: Dict) -> float: | |
| """검색 결과의 신뢰도 점수 계산 (0-1)""" | |
| score = 0.5 # 기본 점수 | |
| url = result.get('url', '') | |
| title = result.get('title', '') | |
| description = result.get('description', '') | |
| # URL 기반 점수 | |
| trusted_domains = [ | |
| '.edu', '.gov', '.org', 'wikipedia.org', 'nature.com', | |
| 'sciencedirect.com', 'ieee.org', 'acm.org', 'springer.com', | |
| 'harvard.edu', 'mit.edu', 'stanford.edu', 'github.com' | |
| ] | |
| for domain in trusted_domains: | |
| if domain in url: | |
| score += 0.2 | |
| break | |
| # HTTPS 사용 여부 | |
| if url.startswith('https://'): | |
| score += 0.1 | |
| # 제목과 설명의 길이 (너무 짧으면 신뢰도 감소) | |
| if len(title) > 20: | |
| score += 0.05 | |
| if len(description) > 50: | |
| score += 0.05 | |
| # 광고/스팸 키워드 체크 | |
| spam_keywords = ['buy now', 'sale', 'discount', 'click here', '100% free'] | |
| if any(spam in (title + description).lower() for spam in spam_keywords): | |
| score -= 0.3 | |
| # 날짜 정보가 있으면 가산점 | |
| if any(year in description for year in ['2024', '2023', '2022']): | |
| score += 0.1 | |
| return max(0, min(1, score)) # 0-1 범위로 제한 | |
| def fetch_url_content(self, url: str, max_length: int = 2000) -> str: | |
| """URL에서 콘텐츠 추출""" | |
| try: | |
| # User-Agent 설정 | |
| headers = { | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' | |
| } | |
| req = urllib.request.Request(url, headers=headers) | |
| with urllib.request.urlopen(req, timeout=5) as response: | |
| html = response.read().decode('utf-8', errors='ignore') | |
| soup = BeautifulSoup(html, 'html.parser') | |
| # 스크립트와 스타일 제거 | |
| for script in soup(["script", "style"]): | |
| script.decompose() | |
| # 본문 텍스트 추출 | |
| text = soup.get_text() | |
| # 공백 정리 | |
| lines = (line.strip() for line in text.splitlines()) | |
| chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) | |
| text = ' '.join(chunk for chunk in chunks if chunk) | |
| # 길이 제한 | |
| if len(text) > max_length: | |
| text = text[:max_length] + "..." | |
| return text | |
| except Exception as e: | |
| logger.error(f"URL 콘텐츠 가져오기 실패 {url}: {str(e)}") | |
| return "" | |
| def detect_contradictions(self, results: List[Dict]) -> List[Dict]: | |
| """검색 결과 간 모순 감지""" | |
| contradictions = [] | |
| # 간단한 모순 감지 패턴 | |
| opposite_pairs = [ | |
| ("increase", "decrease"), | |
| ("improve", "worsen"), | |
| ("effective", "ineffective"), | |
| ("success", "failure"), | |
| ("benefit", "harm"), | |
| ("positive", "negative"), | |
| ("growth", "decline") | |
| ] | |
| # 결과들을 비교 | |
| for i in range(len(results)): | |
| for j in range(i + 1, len(results)): | |
| desc1 = results[i].get('description', '').lower() | |
| desc2 = results[j].get('description', '').lower() | |
| # 반대 개념이 포함되어 있는지 확인 | |
| for word1, word2 in opposite_pairs: | |
| if (word1 in desc1 and word2 in desc2) or (word2 in desc1 and word1 in desc2): | |
| # 같은 주제에 대해 반대 의견인지 확인 | |
| common_words = set(desc1.split()) & set(desc2.split()) | |
| if len(common_words) > 5: # 공통 단어가 5개 이상이면 같은 주제로 간주 | |
| contradictions.append({ | |
| 'source1': results[i]['url'], | |
| 'source2': results[j]['url'], | |
| 'type': f"{word1} vs {word2}", | |
| 'desc1': results[i]['description'][:100], | |
| 'desc2': results[j]['description'][:100] | |
| }) | |
| return contradictions | |
| def brave_search(self, query: str) -> List[Dict]: | |
| """Brave Search API 호출""" | |
| if self.test_mode or self.bapi_token == "YOUR_BRAVE_API_TOKEN": | |
| # 테스트 모드에서는 시뮬레이션된 결과 반환 | |
| test_results = [] | |
| for i in range(5): | |
| test_results.append({ | |
| "title": f"Best Practices for {query} - Source {i+1}", | |
| "description": f"Comprehensive guide on implementing {query} with proven methodologies and real-world examples from industry leaders.", | |
| "url": f"https://example{i+1}.com/{query.replace(' ', '-')}", | |
| "credibility_score": 0.7 + (i * 0.05) | |
| }) | |
| return test_results | |
| try: | |
| params = { | |
| "q": query, | |
| "count": 20, # 20개로 증가 | |
| "safesearch": "moderate", | |
| "freshness": "pw" # Past week for recent results | |
| } | |
| response = requests.get( | |
| self.brave_url, | |
| headers=self.create_brave_headers(), | |
| params=params, | |
| timeout=10 | |
| ) | |
| if response.status_code == 200: | |
| data = response.json() | |
| results = [] | |
| for item in data.get("web", {}).get("results", [])[:20]: | |
| result = { | |
| "title": item.get("title", ""), | |
| "description": item.get("description", ""), | |
| "url": item.get("url", ""), | |
| "published": item.get("published", "") | |
| } | |
| # 신뢰도 점수 계산 | |
| result["credibility_score"] = self.calculate_credibility_score(result) | |
| results.append(result) | |
| # 신뢰도 점수 기준으로 정렬 | |
| results.sort(key=lambda x: x['credibility_score'], reverse=True) | |
| return results | |
| else: | |
| logger.error(f"Brave API 오류: {response.status_code}") | |
| return [] | |
| except Exception as e: | |
| logger.error(f"Brave 검색 중 오류: {str(e)}") | |
| return [] | |
| def simulate_streaming(self, text: str, role: str) -> Generator[str, None, None]: | |
| """테스트 모드에서 스트리밍 시뮬레이션""" | |
| words = text.split() | |
| for i in range(0, len(words), 3): | |
| chunk = " ".join(words[i:i+3]) | |
| yield chunk + " " | |
| time.sleep(0.05) | |
| def call_gemini_streaming(self, messages: List[Dict[str, str]], role: str) -> Generator[str, None, None]: | |
| """Gemini API 스트리밍 호출""" | |
| if not self.gemini_client: | |
| yield "❌ Gemini API 클라이언트가 초기화되지 않았습니다." | |
| return | |
| try: | |
| # 시스템 프롬프트 설정 | |
| system_prompts = { | |
| "supervisor": "당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다.", | |
| "researcher": "당신은 정보를 조사하고 체계적으로 정리하는 조사자 AI입니다.", | |
| "executor": "당신은 세부적인 내용을 구현하는 실행자 AI입니다.", | |
| "evaluator": "당신은 전체 협력 과정과 결과를 평가하는 평가자 AI입니다." | |
| } | |
| # Gemini 형식의 contents 구성 | |
| contents = [] | |
| # 시스템 프롬프트를 첫 번째 사용자 메시지로 추가 | |
| contents.append(types.Content( | |
| role="user", | |
| parts=[types.Part.from_text(text=system_prompts.get(role, ""))] | |
| )) | |
| contents.append(types.Content( | |
| role="model", | |
| parts=[types.Part.from_text(text="네, 이해했습니다. 제 역할을 수행하겠습니다.")] | |
| )) | |
| # 사용자 메시지 추가 | |
| for msg in messages: | |
| if msg["role"] == "user": | |
| contents.append(types.Content( | |
| role="user", | |
| parts=[types.Part.from_text(text=msg["content"])] | |
| )) | |
| # GenerateContentConfig 설정 | |
| generate_content_config = types.GenerateContentConfig( | |
| temperature=0.7, | |
| top_p=0.8, | |
| max_output_tokens=4096, | |
| response_mime_type="text/plain" | |
| ) | |
| # 스트리밍 생성 | |
| for chunk in self.gemini_client.models.generate_content_stream( | |
| model="gemini-2.5-pro", | |
| contents=contents, | |
| config=generate_content_config, | |
| ): | |
| if chunk.text: | |
| yield chunk.text | |
| except Exception as e: | |
| logger.error(f"Gemini API 오류: {str(e)}") | |
| yield f"❌ Gemini API 오류: {str(e)}" | |
| def call_llm_streaming(self, messages: List[Dict[str, str]], role: str) -> Generator[str, None, None]: | |
| """스트리밍 LLM API 호출""" | |
| # Gemini 모드인 경우 | |
| if self.use_gemini: | |
| yield from self.call_gemini_streaming(messages, role) | |
| return | |
| # 테스트 모드 | |
| if self.test_mode: | |
| logger.info(f"테스트 모드 스트리밍 - Role: {role}") | |
| test_responses = { | |
| "supervisor_initial": """이 질문에 대한 거시적 분석을 제시하겠습니다. | |
| 1. **핵심 개념 파악** | |
| - 질문의 본질적 요소를 심층 분석합니다 | |
| - 관련된 주요 이론과 원칙을 검토합니다 | |
| - 다양한 관점에서의 접근 방법을 고려합니다 | |
| 2. **전략적 접근 방향** | |
| - 체계적이고 단계별 해결 방안을 수립합니다 | |
| - 장단기 목표를 명확히 설정합니다 | |
| - 리스크 요인과 대응 방안을 마련합니다 | |
| 3. **기대 효과와 과제** | |
| - 예상되는 긍정적 성과를 분석합니다 | |
| - 잠재적 도전 과제를 식별합니다 | |
| - 지속가능한 발전 방향을 제시합니다 | |
| [검색 키워드]: machine learning optimization, performance improvement strategies, model efficiency techniques, hyperparameter tuning best practices, latest ML trends 2024""", | |
| "researcher": """조사 결과를 종합하여 다음과 같이 정리했습니다. | |
| **1. Machine Learning Optimization (신뢰도 높음)** | |
| - 최신 연구에 따르면 모델 최적화의 핵심은 아키텍처 설계와 훈련 전략의 균형입니다 (신뢰도: 0.85) | |
| - AutoML 도구들이 하이퍼파라미터 튜닝을 자동화하여 효율성을 크게 향상시킵니다 (신뢰도: 0.82) | |
| - 출처: ML Conference 2024 (https://mlconf2024.org), Google Research (https://research.google) | |
| **2. Performance Improvement Strategies (신뢰도 높음)** | |
| - 데이터 품질 개선이 모델 성능 향상의 80%를 차지한다는 연구 결과 (신뢰도: 0.90) | |
| - 앙상블 기법과 전이학습이 주요 성능 개선 방법으로 입증됨 (신뢰도: 0.78) | |
| - 출처: Stanford AI Lab (https://ai.stanford.edu), MIT CSAIL (https://csail.mit.edu) | |
| **3. Model Efficiency Techniques (신뢰도 중간)** | |
| - 모델 경량화(Pruning, Quantization)로 추론 속도 10배 향상 가능 (신뢰도: 0.75) | |
| - Knowledge Distillation으로 모델 크기 90% 감소, 성능 유지 (신뢰도: 0.72) | |
| - 출처: ArXiv 논문 (https://arxiv.org/abs/2023.xxxxx) | |
| **4. 실제 적용 사례 (신뢰도 높음)** | |
| - Netflix: 추천 시스템 개선으로 사용자 만족도 35% 향상 (신뢰도: 0.88) | |
| - Tesla: 실시간 객체 인식 속도 50% 개선 (신뢰도: 0.80) | |
| - OpenAI: GPT 모델 효율성 개선으로 비용 70% 절감 (신뢰도: 0.85) | |
| **핵심 인사이트:** | |
| - 최신 트렌드는 효율성과 성능의 균형에 초점 | |
| - 2024년 들어 Sparse Models와 MoE(Mixture of Experts) 기법이 부상 | |
| - 실무 적용 시 단계별 검증이 성공의 핵심""", | |
| "supervisor_execution": """조사 내용을 바탕으로 실행자 AI에게 다음과 같이 구체적으로 지시합니다. | |
| **1단계: 현재 모델 진단 (1주차)** | |
| - 조사된 벤치마크 기준으로 현재 모델 성능 평가 | |
| - Netflix 사례를 참고하여 주요 병목 지점 식별 | |
| - AutoML 도구를 활용한 초기 최적화 가능성 탐색 | |
| **2단계: 데이터 품질 개선 (2-3주차)** | |
| - 조사 결과의 "80% 규칙"에 따라 데이터 정제 우선 실행 | |
| - 데이터 증강 기법 적용 (조사된 최신 기법 활용) | |
| - A/B 테스트로 개선 효과 측정 | |
| **3단계: 모델 최적화 구현 (4-6주차)** | |
| - Knowledge Distillation 적용하여 모델 경량화 | |
| - 조사된 Pruning 기법으로 추론 속도 개선 | |
| - Tesla 사례의 실시간 처리 최적화 기법 벤치마킹 | |
| **4단계: 성과 검증 및 배포 (7-8주차)** | |
| - OpenAI 사례의 비용 절감 지표 적용 | |
| - 조사된 성능 지표로 개선율 측정 | |
| - 단계적 배포 전략 수립""", | |
| "executor": """감독자의 지시와 조사 내용을 기반으로 구체적인 실행 계획을 수립합니다. | |
| **1단계: 현재 모델 진단 (1주차)** | |
| - 월요일-화요일: MLflow를 사용한 현재 모델 메트릭 수집 | |
| * 조사 결과 참고: Netflix가 사용한 핵심 지표 (정확도, 지연시간, 처리량) | |
| - 수요일-목요일: AutoML 도구 (Optuna, Ray Tune) 설정 및 초기 실행 | |
| * 조사된 best practice에 따라 search space 정의 | |
| - 금요일: 진단 보고서 작성 및 개선 우선순위 결정 | |
| **2단계: 데이터 품질 개선 (2-3주차)** | |
| - 데이터 정제 파이프라인 구축 | |
| * 조사 결과의 "80% 규칙" 적용: 누락값, 이상치, 레이블 오류 처리 | |
| * 코드 예시: `data_quality_pipeline.py` 구현 | |
| - 데이터 증강 구현 | |
| * 최신 기법 적용: MixUp, CutMix, AutoAugment | |
| * 검증 데이터셋으로 효과 측정 (목표: 15% 성능 향상) | |
| **3단계: 모델 최적화 구현 (4-6주차)** | |
| - Knowledge Distillation 구현 | |
| * Teacher 모델: 현재 대규모 모델 | |
| * Student 모델: 90% 작은 크기 목표 (조사 결과 기반) | |
| * 구현 프레임워크: PyTorch/TensorFlow""", | |
| "supervisor_review": """실행자 AI의 계획을 검토한 결과, 조사 내용이 잘 반영되었습니다. 다음과 같은 개선사항을 제안합니다. | |
| **강점** | |
| - 조사된 사례들(Netflix, Tesla, OpenAI)이 각 단계에 적절히 활용됨 | |
| - 구체적인 도구와 기법이 명시되어 실행 가능성이 높음 | |
| - 측정 가능한 목표가 조사 결과를 기반으로 설정됨 | |
| **개선 필요사항** | |
| 1. **리스크 관리 강화** | |
| - 각 단계별 실패 시나리오와 대응 방안 추가 필요 | |
| - 기술적 문제 발생 시 백업 계획 수립 | |
| 2. **비용 분석 구체화** | |
| - OpenAI 사례의 70% 절감을 위한 구체적인 비용 계산 | |
| - ROI 분석 및 투자 대비 효과 측정 방법 | |
| **추가 권장사항** | |
| - 최신 연구 동향 모니터링 체계 구축 | |
| - 경쟁사 벤치마킹을 위한 정기적인 조사 프로세스""", | |
| "executor_final": """감독자 AI의 피드백을 완전히 반영하여 최종 실행 보고서를 작성합니다. | |
| # 🎯 기계학습 모델 성능 향상 최종 실행 보고서 | |
| ## 📋 Executive Summary | |
| 본 보고서는 웹 검색을 통해 수집된 최신 사례와 감독자 AI의 전략적 지침을 바탕으로, 8주간의 체계적인 모델 최적화 프로젝트를 제시합니다. | |
| ### 🎯 목표 달성 지표 | |
| | 지표 | 현재 | 목표 | 개선율 | | |
| |------|------|------|--------| | |
| | 모델 크기 | 2.5GB | 250MB | 90% 감소 | | |
| | 추론 속도 | 45ms | 4.5ms | 10배 향상 | | |
| | 운영 비용 | $2,000/월 | $600/월 | 70% 절감 | | |
| | 정확도 | 92% | 90.5% | 1.5% 손실 | | |
| ## 📊 1단계: 현재 모델 진단 및 베이스라인 설정 (1주차) | |
| ### 실행 계획 | |
| **월-화요일: 성능 메트릭 수집** | |
| - MLflow를 통한 현재 모델 전체 분석 | |
| - Netflix 사례 기반 핵심 지표: | |
| - 정확도: 92% | |
| - 지연시간: 45ms | |
| - 처리량: 1,000 req/s | |
| - GPU 메모리: 8GB | |
| **수-목요일: AutoML 초기 탐색** | |
| ```python | |
| # Optuna 하이퍼파라미터 최적화 설정 | |
| study = optuna.create_study(direction="maximize") | |
| study.optimize(objective, n_trials=200) | |
| # Ray Tune 분산 학습 설정 | |
| analysis = tune.run( | |
| train_model, | |
| config=search_space, | |
| num_samples=50, | |
| resources_per_trial={"gpu": 1} | |
| ) | |
| ``` | |
| ### 예상 산출물 | |
| - ✅ 상세 성능 베이스라인 문서 | |
| - ✅ 개선 기회 우선순위 매트릭스 | |
| - ✅ 리스크 레지스터 | |
| ## 📊 2단계: 데이터 품질 개선 (2-3주차) | |
| ### 실행 계획 | |
| **데이터 정제 파이프라인 구축** | |
| > 조사 결과의 "80% 규칙" 적용: 데이터 품질이 성능의 80%를 결정 | |
| ```python | |
| class DataQualityPipeline: | |
| def __init__(self): | |
| self.validators = [ | |
| MissingValueHandler(threshold=0.05), | |
| OutlierDetector(method='isolation_forest'), | |
| LabelConsistencyChecker(), | |
| DataDriftMonitor() | |
| ] | |
| def process(self, data): | |
| for validator in self.validators: | |
| data = validator.transform(data) | |
| self.log_metrics(validator.get_stats()) | |
| return data | |
| ``` | |
| **고급 데이터 증강 기법** | |
| - **MixUp**: 15% 정확도 향상 예상 | |
| - **CutMix**: 경계 검출 성능 20% 개선 | |
| - **AutoAugment**: 자동 최적 증강 정책 탐색 | |
| ### 리스크 대응 전략 | |
| | 리스크 | 확률 | 영향도 | 대응 방안 | | |
| |--------|------|--------|-----------| | |
| | 데이터 품질 저하 | 중간 | 높음 | 롤백 메커니즘 구현 | | |
| | 증강 과적합 | 낮음 | 중간 | 검증셋 분리 및 교차 검증 | | |
| | 처리 시간 증가 | 높음 | 낮음 | 병렬 처리 파이프라인 | | |
| ## 📊 3단계: 모델 최적화 구현 (4-6주차) | |
| ### Knowledge Distillation 상세 계획 | |
| **Teacher-Student 아키텍처** | |
| - Teacher 모델: 현재 2.5GB 모델 | |
| - Student 모델 스펙: | |
| - 파라미터: 250M → 25M (90% 감소) | |
| - 레이어: 24 → 6 | |
| - Hidden dimension: 1024 → 256 | |
| **훈련 전략** | |
| ```python | |
| distillation_config = { | |
| "temperature": 5.0, | |
| "alpha": 0.7, # KD loss weight | |
| "beta": 0.3, # Original loss weight | |
| "epochs": 50, | |
| "learning_rate": 1e-4, | |
| "batch_size": 128 | |
| } | |
| ``` | |
| ### Pruning & Quantization | |
| **구조적 Pruning 계획** | |
| 1. Magnitude 기반 중요도 평가 | |
| 2. 50% 채널 제거 | |
| 3. Fine-tuning: 10 에폭 | |
| 4. 성능 검증 및 반복 | |
| **INT8 Quantization** | |
| - Post-training quantization 적용 | |
| - Calibration dataset: 1,000 샘플 | |
| - 예상 속도 향상: 4배 | |
| ## 📊 4단계: 성과 검증 및 프로덕션 배포 (7-8주차) | |
| ### 종합 성능 검증 | |
| **성능 지표 달성도 검증** | |
| | 테스트 항목 | 방법 | 성공 기준 | 결과 | | |
| |-------------|------|-----------|------| | |
| | 추론 속도 | A/B 테스트 | <5ms | ✅ 4.5ms | | |
| | 정확도 | 홀드아웃 검증 | >90% | ✅ 90.5% | | |
| | 메모리 사용량 | 프로파일링 | <300MB | ✅ 250MB | | |
| | 처리량 | 부하 테스트 | >5000 req/s | ✅ 6000 req/s | | |
| ### 단계적 배포 전략 | |
| ```mermaid | |
| graph LR | |
| A[1% 트래픽] --> B[10% 트래픽] | |
| B --> C[50% 트래픽] | |
| C --> D[100% 전환] | |
| A -->|Day 1-3| B | |
| B -->|Day 4-7| C | |
| C -->|Day 8-14| D | |
| ``` | |
| ### 모니터링 대시보드 | |
| **핵심 메트릭** | |
| - 🔴 P99 지연시간: < 10ms | |
| - 🟡 오류율: < 0.1% | |
| - 🟢 CPU/GPU 사용률: < 80% | |
| ## 💰 ROI 분석 | |
| ### 비용-효익 분석 | |
| | 항목 | 비용/효익 | 세부 내역 | | |
| |------|-----------|-----------| | |
| | **초기 투자** | $50,000 | 인건비 + 인프라 | | |
| | **월간 절감액** | $14,000 | 서버 + GPU 비용 | | |
| | **투자 회수 기간** | 3.6개월 | - | | |
| | **1년 순이익** | $118,000 | 절감액 - 투자비 | | |
| ### 장기 효과 | |
| - 🚀 확장성 10배 향상 | |
| - 💡 신규 서비스 출시 가능 | |
| - 🌍 탄소 배출 70% 감소 | |
| ## 📈 지속적 개선 계획 | |
| ### 월간 모니터링 | |
| - 성능 지표 리뷰 | |
| - 사용자 피드백 분석 | |
| - 기술 부채 관리 | |
| ### 분기별 업데이트 | |
| - 모델 재훈련 | |
| - 새로운 최적화 기법 도입 | |
| - 벤치마크 업데이트 | |
| ### 차기 프로젝트 로드맵 | |
| | 분기 | 프로젝트 | 예상 효과 | | |
| |------|----------|-----------| | |
| | Q2 2025 | 엣지 디바이스 배포 | 지연시간 90% 감소 | | |
| | Q3 2025 | 연합 학습 도입 | 프라이버시 강화 | | |
| | Q4 2025 | AutoML 플랫폼 구축 | 개발 속도 5배 향상 | | |
| ## 📝 결론 및 권고사항 | |
| ### 핵심 성과 | |
| - ✅ 모든 목표 지표 달성 | |
| - ✅ 예산 내 프로젝트 완료 | |
| - ✅ 리스크 성공적 관리 | |
| ### 향후 권고사항 | |
| 1. **즉시 실행**: 1-2단계 즉시 착수 | |
| 2. **팀 구성**: ML엔지니어 2명, DevOps 1명 필수 | |
| 3. **인프라 준비**: GPU 서버 사전 확보 | |
| 4. **변경 관리**: 이해관계자 사전 교육 | |
| > 본 프로젝트는 최신 연구 결과와 업계 베스트 프랙티스를 적용하여, 8주 만에 모델 성능을 획기적으로 개선하고 운영 비용을 70% 절감하는 성과를 달성할 것으로 확신합니다. | |
| --- | |
| *작성일: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}* | |
| *작성자: 협력적 AI 시스템 (감독자, 조사자, 실행자, 평가자 AI)*""", | |
| "evaluator": """## 📊 전체 협력 과정 평가 보고서 | |
| ### 1️⃣ 품질 평가 (10점 만점) | |
| | AI 역할 | 점수 | 평가 내용 | | |
| |---------|------|-----------| | |
| | **감독자 AI** | 9.5/10 | 거시적 관점에서 체계적인 분석과 방향 제시 | | |
| | **조사자 AI** | 9.0/10 | 웹 검색을 통한 최신 정보 수집 우수 | | |
| | **실행자 AI** | 8.5/10 | 조사 내용을 잘 활용한 구체적 계획 수립 | | |
| **상세 평가:** | |
| - ✅ 감독자 AI: 단계별 구체적인 지시사항 제공이 탁월함 | |
| - ✅ 조사자 AI: 신뢰도 평가와 모순 감지 기능이 효과적 | |
| - ✅ 실행자 AI: 실행 가능한 단계별 접근법 제시 우수 | |
| ### 2️⃣ 협력 효과성 평가 | |
| **강점:** | |
| - 🔄 AI 간 역할 분담이 명확하고 상호보완적 | |
| - 📊 정보 흐름이 체계적이고 일관성 있음 | |
| - ✨ 피드백 반영이 효과적으로 이루어짐 | |
| **개선점:** | |
| - ⚡ 실시간 상호작용 메커니즘 추가 고려 | |
| - 📋 중간 점검 단계 도입 필요 | |
| ### 3️⃣ 정보 활용도 평가 | |
| | 평가 항목 | 달성도 | 세부 내용 | | |
| |-----------|--------|-----------| | |
| | 검색 범위 | 95% | 20개 이상의 웹 소스에서 정보 수집 | | |
| | 신뢰도 평가 | 90% | 0.7 이상의 신뢰도 소스 우선 활용 | | |
| | 정보 통합 | 85% | 다양한 관점의 정보를 균형있게 통합 | | |
| **우수한 점:** | |
| - ✅ 신뢰도 기반 정보 선별 효과적 | |
| - ✅ 실제 기업 사례 적절히 활용 | |
| - ✅ 최신 트렌드 반영 우수 | |
| **보완 필요:** | |
| - 📚 학술 논문 등 더 깊이 있는 자료 활용 | |
| - 🌍 지역별/산업별 특성 고려 필요 | |
| ### 4️⃣ 향후 개선 방향 | |
| #### 1. **실시간 협업 강화** | |
| - AI 간 중간 체크포인트 추가 | |
| - 동적 역할 조정 메커니즘 도입 | |
| #### 2. **정보 검증 강화** | |
| - 교차 검증 프로세스 추가 | |
| - 전문가 검토 단계 고려 | |
| #### 3. **맞춤화 강화** | |
| - 사용자 컨텍스트 더 깊이 반영 | |
| - 산업별/규모별 맞춤 전략 제공 | |
| ### 5️⃣ 최종 평점: ⭐⭐⭐⭐⭐ **9.0/10** | |
| > **종합 평가:** | |
| > | |
| > 본 협력 시스템은 각 AI의 전문성을 효과적으로 활용하여 사용자 질문에 대한 종합적이고 실행 가능한 답변을 제공했습니다. 특히 웹 검색을 통한 최신 정보 활용과 단계적 피드백 반영이 우수했습니다. | |
| > | |
| > 향후 실시간 협업과 맞춤화를 더욱 강화한다면 더욱 뛰어난 성과를 달성할 수 있을 것입니다. | |
| --- | |
| **평가 완료 시각**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}""" | |
| } | |
| # 프롬프트 내용에 따라 적절한 응답 선택 | |
| if role == "supervisor" and "조사자 AI가 정리한" in messages[0]["content"]: | |
| response = test_responses["supervisor_execution"] | |
| elif role == "supervisor" and messages[0]["content"].find("실행자 AI의 답변") > -1: | |
| response = test_responses["supervisor_review"] | |
| elif role == "supervisor": | |
| response = test_responses["supervisor_initial"] | |
| elif role == "researcher": | |
| response = test_responses["researcher"] | |
| elif role == "executor" and "최종 보고서" in messages[0]["content"]: | |
| response = test_responses["executor_final"] | |
| elif role == "evaluator": | |
| response = test_responses["evaluator"] | |
| else: | |
| response = test_responses["executor"] | |
| yield from self.simulate_streaming(response, role) | |
| return | |
| # 실제 API 호출 | |
| try: | |
| system_prompts = { | |
| "supervisor": "당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다.", | |
| "researcher": "당신은 정보를 조사하고 체계적으로 정리하는 조사자 AI입니다.", | |
| "executor": "당신은 세부적인 내용을 구현하는 실행자 AI입니다.", | |
| "evaluator": "당신은 전체 협력 과정과 결과를 평가하는 평가자 AI입니다." | |
| } | |
| full_messages = [ | |
| {"role": "system", "content": system_prompts.get(role, "")}, | |
| *messages | |
| ] | |
| payload = { | |
| "model": self.model_id, | |
| "messages": full_messages, | |
| "max_tokens": 4096, | |
| "temperature": 0.7, | |
| "top_p": 0.8, | |
| "stream": True, | |
| "stream_options": {"include_usage": True} | |
| } | |
| logger.info(f"API 스트리밍 호출 시작 - Role: {role}") | |
| response = requests.post( | |
| self.api_url, | |
| headers=self.create_headers(), | |
| json=payload, | |
| stream=True, | |
| timeout=10 | |
| ) | |
| if response.status_code != 200: | |
| logger.error(f"API 오류: {response.status_code}") | |
| yield f"❌ API 오류 ({response.status_code}): {response.text[:200]}" | |
| return | |
| for line in response.iter_lines(): | |
| if line: | |
| line = line.decode('utf-8') | |
| if line.startswith("data: "): | |
| data = line[6:] | |
| if data == "[DONE]": | |
| break | |
| try: | |
| chunk = json.loads(data) | |
| if "choices" in chunk and chunk["choices"]: | |
| content = chunk["choices"][0].get("delta", {}).get("content", "") | |
| if content: | |
| yield content | |
| except json.JSONDecodeError: | |
| continue | |
| except requests.exceptions.Timeout: | |
| yield "⏱️ API 호출 시간이 초과되었습니다. 다시 시도해주세요." | |
| except requests.exceptions.ConnectionError: | |
| yield "🔌 API 서버에 연결할 수 없습니다. 인터넷 연결을 확인해주세요." | |
| except Exception as e: | |
| logger.error(f"스트리밍 중 오류: {str(e)}") | |
| yield f"❌ 오류 발생: {str(e)}" | |
| # 시스템 인스턴스 생성 | |
| llm_system = LLMCollaborativeSystem() | |
| # 내부 히스토리 관리 (UI에는 표시하지 않음) | |
| internal_history = [] | |
| def process_query_streaming(user_query: str, llm_mode: str): | |
| """스트리밍을 지원하는 쿼리 처리""" | |
| global internal_history | |
| if not user_query: | |
| return "", "", "", "", "", "❌ 질문을 입력해주세요." | |
| # LLM 모드 설정 | |
| llm_system.set_llm_mode(llm_mode) | |
| conversation_log = [] | |
| all_responses = {"supervisor": [], "researcher": [], "executor": [], "evaluator": []} | |
| try: | |
| # 1단계: 감독자 AI 초기 분석 및 키워드 추출 | |
| supervisor_prompt = llm_system.create_supervisor_initial_prompt(user_query) | |
| supervisor_initial_response = "" | |
| supervisor_text = "[초기 분석] 🔄 생성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": supervisor_prompt}], | |
| "supervisor" | |
| ): | |
| supervisor_initial_response += chunk | |
| supervisor_text = f"[초기 분석] - {datetime.now().strftime('%H:%M:%S')}\n{supervisor_initial_response}" | |
| yield supervisor_text, "", "", "", "", "🔄 감독자 AI가 분석 중..." | |
| all_responses["supervisor"].append(supervisor_initial_response) | |
| # 키워드 추출 | |
| keywords = llm_system.extract_keywords(supervisor_initial_response) | |
| logger.info(f"추출된 키워드: {keywords}") | |
| # 2단계: 브레이브 검색 수행 | |
| researcher_text = "[웹 검색] 🔍 검색 중...\n" | |
| yield supervisor_text, researcher_text, "", "", "", "🔍 웹 검색 수행 중..." | |
| search_results = {} | |
| total_search_count = 0 | |
| # 원래 키워드로 검색 | |
| for keyword in keywords: | |
| results = llm_system.brave_search(keyword) | |
| if results: | |
| search_results[keyword] = results | |
| total_search_count += len(results) | |
| researcher_text += f"✓ '{keyword}' 검색 완료 ({len(results)}개 결과)\n" | |
| yield supervisor_text, researcher_text, "", "", "", f"🔍 '{keyword}' 검색 중..." | |
| # 동의어로 추가 검색 | |
| synonyms = llm_system.generate_synonyms(keyword) | |
| for synonym in synonyms: | |
| syn_results = llm_system.brave_search(f"{keyword} {synonym}") | |
| if syn_results: | |
| search_results[f"{keyword} ({synonym})"] = syn_results | |
| total_search_count += len(syn_results) | |
| researcher_text += f"✓ 동의어 '{synonym}' 검색 완료 ({len(syn_results)}개 결과)\n" | |
| yield supervisor_text, researcher_text, "", "", "", f"🔍 동의어 '{synonym}' 검색 중..." | |
| researcher_text += f"\n📊 총 {total_search_count}개의 검색 결과 수집 완료\n" | |
| # URL 콘텐츠 가져오기 (상위 3개) | |
| researcher_text += "\n[콘텐츠 분석] 📖 주요 웹페이지 내용 분석 중...\n" | |
| yield supervisor_text, researcher_text, "", "", "", "📖 웹페이지 내용 분석 중..." | |
| content_analyzed = 0 | |
| for keyword, results in search_results.items(): | |
| for result in results[:2]: # 각 키워드당 상위 2개만 | |
| if content_analyzed >= 5: # 총 5개까지만 | |
| break | |
| url = result.get('url', '') | |
| if url and result.get('credibility_score', 0) >= 0.7: | |
| content = llm_system.fetch_url_content(url) | |
| if content: | |
| result['content_preview'] = content[:500] # 미리보기 저장 | |
| content_analyzed += 1 | |
| researcher_text += f"✓ 콘텐츠 분석 완료: {url[:50]}...\n" | |
| yield supervisor_text, researcher_text, "", "", "", f"📖 분석 중: {url[:30]}..." | |
| # 3단계: 조사자 AI가 검색 결과 정리 | |
| researcher_prompt = llm_system.create_researcher_prompt(user_query, supervisor_initial_response, search_results) | |
| researcher_response = "" | |
| researcher_text = "[조사 결과 정리] 🔄 생성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": researcher_prompt}], | |
| "researcher" | |
| ): | |
| researcher_response += chunk | |
| researcher_text = f"[조사 결과 정리] - {datetime.now().strftime('%H:%M:%S')}\n{researcher_response}" | |
| yield supervisor_text, researcher_text, "", "", "", "📝 조사자 AI가 정리 중..." | |
| all_responses["researcher"].append(researcher_response) | |
| # 4단계: 평가자 AI가 조사 결과 평가 | |
| evaluator_research_prompt = f"""당신은 전체 협력 과정과 결과를 평가하는 평가자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 감독자 AI의 초기 분석: | |
| {supervisor_initial_response} | |
| 조사자 AI의 조사 결과: | |
| {researcher_response} | |
| 위 조사 결과를 평가하여: | |
| 1. 조사의 충실도와 신뢰성을 평가하세요 | |
| 2. 누락된 중요 정보가 있는지 확인하세요 | |
| 3. 조사 결과의 활용 가능성을 평가하세요 | |
| 4. 개선이 필요한 부분을 구체적으로 제시하세요""" | |
| evaluator_research_response = "" | |
| evaluator_text = "[조사 결과 평가] 🔄 평가 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": evaluator_research_prompt}], | |
| "evaluator" | |
| ): | |
| evaluator_research_response += chunk | |
| evaluator_text = f"[조사 결과 평가] - {datetime.now().strftime('%H:%M:%S')}\n{evaluator_research_response}" | |
| yield supervisor_text, researcher_text, "", evaluator_text, "", "📊 평가자 AI가 조사 결과 평가 중..." | |
| all_responses["evaluator"].append(evaluator_research_response) | |
| # 5단계: 감독자 AI가 평가를 반영한 실행 지시 | |
| supervisor_execution_prompt = f"""당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 조사자 AI가 정리한 조사 내용: | |
| {researcher_response} | |
| 평가자 AI의 조사 결과 평가: | |
| {evaluator_research_response} | |
| 위 조사 내용과 평가를 기반으로 실행자 AI에게 아주 구체적인 지시를 내려주세요: | |
| 1. 평가자의 피드백을 반영하여 지시를 개선하세요 | |
| 2. 조사된 정보를 어떻게 활용할지 명확히 지시하세요 | |
| 3. 실행 가능한 단계별 작업을 구체적으로 제시하세요 | |
| 4. 예상되는 결과물의 형태를 구체적으로 설명하세요""" | |
| supervisor_execution_response = "" | |
| supervisor_text += "\n\n---\n\n[실행 지시] 🔄 생성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": supervisor_execution_prompt}], | |
| "supervisor" | |
| ): | |
| supervisor_execution_response += chunk | |
| temp_text = f"{all_responses['supervisor'][0]}\n\n---\n\n[실행 지시] - {datetime.now().strftime('%H:%M:%S')}\n{supervisor_execution_response}" | |
| supervisor_text = f"[초기 분석] - {datetime.now().strftime('%H:%M:%S')}\n{temp_text}" | |
| yield supervisor_text, researcher_text, "", evaluator_text, "", "🎯 감독자 AI가 지시 중..." | |
| all_responses["supervisor"].append(supervisor_execution_response) | |
| # 6단계: 실행자 AI가 조사 내용과 지시를 기반으로 초기 구현 | |
| executor_prompt = llm_system.create_executor_prompt(user_query, supervisor_execution_response, researcher_response) | |
| executor_response = "" | |
| executor_text = "[초기 구현] 🔄 생성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": executor_prompt}], | |
| "executor" | |
| ): | |
| executor_response += chunk | |
| executor_text = f"[초기 구현] - {datetime.now().strftime('%H:%M:%S')}\n{executor_response}" | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, "", "🔧 실행자 AI가 구현 중..." | |
| all_responses["executor"].append(executor_response) | |
| # 7단계: 평가자 AI가 초기 구현 평가 | |
| evaluator_execution_prompt = f"""당신은 전체 협력 과정과 결과를 평가하는 평가자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 실행자 AI의 초기 구현: | |
| {executor_response} | |
| 감독자 AI의 지시사항: | |
| {supervisor_execution_response} | |
| 위 초기 구현을 평가하여: | |
| 1. 지시사항이 얼마나 잘 반영되었는지 평가하세요 | |
| 2. 구현의 실행 가능성과 구체성을 평가하세요 | |
| 3. 누락된 중요 요소가 있는지 확인하세요 | |
| 4. 개선이 필요한 부분을 구체적으로 제시하세요""" | |
| evaluator_execution_response = "" | |
| evaluator_text += "\n\n---\n\n[초기 구현 평가] 🔄 평가 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": evaluator_execution_prompt}], | |
| "evaluator" | |
| ): | |
| evaluator_execution_response += chunk | |
| temp_text = f"{all_responses['evaluator'][0]}\n\n---\n\n[초기 구현 평가] - {datetime.now().strftime('%H:%M:%S')}\n{evaluator_execution_response}" | |
| evaluator_text = f"[조사 결과 평가] - {datetime.now().strftime('%H:%M:%S')}\n{temp_text}" | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, "", "📊 평가자 AI가 구현 평가 중..." | |
| all_responses["evaluator"].append(evaluator_execution_response) | |
| # 8단계: 감독자 AI가 평가를 반영한 개선 지시 | |
| supervisor_improvement_prompt = f"""당신은 거시적 관점에서 분석하고 지도하는 감독자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 실행자 AI의 초기 구현: | |
| {executor_response} | |
| 평가자 AI의 구현 평가: | |
| {evaluator_execution_response} | |
| 위 평가를 반영하여 최종 보고서 작성을 위한 개선 지시를 내려주세요: | |
| 1. 평가자가 지적한 모든 개선사항을 구체적으로 반영하세요 | |
| 2. 추가로 필요한 구체적인 내용을 지시하세요 | |
| 3. 최종 보고서의 구조와 포함해야 할 요소를 명확히 제시하세요""" | |
| supervisor_improvement_response = "" | |
| supervisor_text += "\n\n---\n\n[개선 지시] 🔄 생성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": supervisor_improvement_prompt}], | |
| "supervisor" | |
| ): | |
| supervisor_improvement_response += chunk | |
| temp_text = f"{all_responses['supervisor'][0]}\n\n---\n\n[실행 지시] - {datetime.now().strftime('%H:%M:%S')}\n{all_responses['supervisor'][1]}\n\n---\n\n[개선 지시] - {datetime.now().strftime('%H:%M:%S')}\n{supervisor_improvement_response}" | |
| supervisor_text = f"[초기 분석] - {datetime.now().strftime('%H:%M:%S')}\n{temp_text}" | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, "", "🔄 감독자 AI가 개선 지시 중..." | |
| all_responses["supervisor"].append(supervisor_improvement_response) | |
| # 9단계: 평가자 AI가 전체 과정 최종 평가 | |
| evaluator_final_prompt = llm_system.create_evaluator_prompt( | |
| user_query, | |
| all_responses["supervisor"], | |
| all_responses["researcher"][0], | |
| all_responses["executor"], | |
| all_responses["evaluator"] # 이전 평가들도 전달 | |
| ) | |
| evaluator_final_response = "" | |
| evaluator_text += "\n\n---\n\n[전체 과정 최종 평가] 🔄 평가 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": evaluator_final_prompt}], | |
| "evaluator" | |
| ): | |
| evaluator_final_response += chunk | |
| temp_text = f"{all_responses['evaluator'][0]}\n\n---\n\n[초기 구현 평가] - {datetime.now().strftime('%H:%M:%S')}\n{all_responses['evaluator'][1]}\n\n---\n\n[전체 과정 최종 평가] - {datetime.now().strftime('%H:%M:%S')}\n{evaluator_final_response}" | |
| evaluator_text = f"[조사 결과 평가] - {datetime.now().strftime('%H:%M:%S')}\n{temp_text}" | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, "", "📊 평가자 AI가 최종 평가 중..." | |
| all_responses["evaluator"].append(evaluator_final_response) | |
| # 10단계: 실행자 AI 최종 보고서 (모든 피드백 반영) | |
| final_executor_prompt = f"""당신은 세부적인 내용을 구현하는 실행자 AI입니다. | |
| 사용자 질문: {user_query} | |
| 조사자 AI의 조사 내용: | |
| {researcher_response} | |
| 당신의 초기 구현: | |
| {executor_response} | |
| 감독자 AI의 개선 지시: | |
| {supervisor_improvement_response} | |
| 평가자 AI의 전체 평가: | |
| {evaluator_final_response} | |
| 위 모든 피드백을 완전히 반영하여 최종 보고서를 작성하세요: | |
| 1. 모든 개선사항과 지시사항을 빠짐없이 반영하세요 | |
| 2. 조사 내용을 최대한 구체적으로 활용하세요 | |
| 3. 실행 가능성을 높이는 세부 계획을 포함하세요 | |
| 4. 명확한 결론과 다음 단계를 제시하세요 | |
| 5. 전문적이고 완성도 높은 최종 보고서 형식으로 작성하세요 | |
| **중요: 마크다운 형식을 적극 활용하세요** | |
| - 제목은 #, ##, ### 을 사용하여 계층적으로 구성 | |
| - 중요한 내용은 **굵게** 표시 | |
| - 리스트는 -, * 또는 1. 2. 3. 형식 사용 | |
| - 표가 필요한 경우 마크다운 표 형식 사용: | |
| | 항목 | 내용 | 비고 | | |
| |------|------|------| | |
| | 예시1 | 설명1 | 참고1 | | |
| - 코드는 ``` 로 감싸서 표시 | |
| - 인용구는 > 를 사용 | |
| - 구분선은 --- 사용""" | |
| final_executor_response = "" | |
| executor_text += "\n\n---\n\n[최종 보고서] 🔄 작성 중...\n" | |
| for chunk in llm_system.call_llm_streaming( | |
| [{"role": "user", "content": final_executor_prompt}], | |
| "executor" | |
| ): | |
| final_executor_response += chunk | |
| temp_text = f"[초기 구현] - {datetime.now().strftime('%H:%M:%S')}\n{all_responses['executor'][0]}\n\n---\n\n[최종 보고서] - {datetime.now().strftime('%H:%M:%S')}\n{final_executor_response}" | |
| executor_text = temp_text | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, "", "📄 최종 보고서 작성 중..." | |
| all_responses["executor"].append(final_executor_response) | |
| # 최종 결과 생성 (최종 보고서를 메인으로) | |
| final_summary = f"""# 🎯 최종 종합 보고서 | |
| ## 📌 사용자 질문 | |
| **{user_query}** | |
| --- | |
| ## 📄 최종 보고서 (실행자 AI - 모든 피드백 반영) | |
| {final_executor_response} | |
| --- | |
| ## 📊 전체 프로세스 평가 (평가자 AI) | |
| {evaluator_final_response} | |
| --- | |
| ## 🔍 핵심 조사 결과 요약 (조사자 AI) | |
| {researcher_response[:800]}... | |
| --- | |
| ## 📋 프로세스 완료 | |
| | 항목 | 내용 | | |
| |------|------| | |
| | **사용 모델** | {'Gemini 2.5 Pro' if llm_system.use_gemini else '기본 LLM'} | | |
| | **프로세스** | 감독→조사→평가→감독→실행→평가→감독→평가→실행 | | |
| | **총 단계** | 9단계 협력 완료 | | |
| | **생성 시간** | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | | |
| --- | |
| > 💡 **참고**: 이 보고서는 4개 AI의 협력을 통해 생성되었으며, 다단계 평가 프로세스를 거쳐 품질이 검증되었습니다.""" | |
| # 내부 히스토리 업데이트 (UI에는 표시하지 않음) | |
| internal_history.append((user_query, final_summary)) | |
| yield supervisor_text, researcher_text, executor_text, evaluator_text, final_summary, "✅ 최종 보고서 완성!" | |
| except Exception as e: | |
| error_msg = f"❌ 처리 중 오류: {str(e)}" | |
| yield "", "", "", "", error_msg, error_msg | |
| def clear_all(): | |
| """모든 내용 초기화""" | |
| global internal_history | |
| internal_history = [] | |
| return "", "", "", "", "", "🔄 초기화되었습니다." | |
| # Gradio 인터페이스 | |
| css = """ | |
| .gradio-container { | |
| font-family: 'Arial', sans-serif; | |
| } | |
| .supervisor-box textarea { | |
| border-left: 4px solid #667eea !important; | |
| padding-left: 10px !important; | |
| background-color: #f8f9ff !important; | |
| } | |
| .researcher-box textarea { | |
| border-left: 4px solid #10b981 !important; | |
| padding-left: 10px !important; | |
| background-color: #f0fdf4 !important; | |
| } | |
| .executor-box textarea { | |
| border-left: 4px solid #764ba2 !important; | |
| padding-left: 10px !important; | |
| background-color: #faf5ff !important; | |
| } | |
| .evaluator-box textarea { | |
| border-left: 4px solid #f59e0b !important; | |
| padding-left: 10px !important; | |
| background-color: #fffbeb !important; | |
| } | |
| .final-report-box { | |
| border: 2px solid #3b82f6 !important; | |
| border-radius: 8px !important; | |
| padding: 16px !important; | |
| background-color: #eff6ff !important; | |
| margin-top: 10px !important; | |
| font-size: 14px !important; | |
| max-height: 700px !important; | |
| overflow-y: auto !important; | |
| line-height: 1.6 !important; | |
| } | |
| .final-report-box h1 { | |
| color: #1e40af !important; | |
| font-size: 24px !important; | |
| margin-bottom: 12px !important; | |
| } | |
| .final-report-box h2 { | |
| color: #2563eb !important; | |
| font-size: 20px !important; | |
| margin-top: 16px !important; | |
| margin-bottom: 10px !important; | |
| } | |
| .final-report-box h3 { | |
| color: #3b82f6 !important; | |
| font-size: 18px !important; | |
| margin-top: 12px !important; | |
| margin-bottom: 8px !important; | |
| } | |
| .final-report-box table { | |
| border-collapse: collapse !important; | |
| width: 100% !important; | |
| margin: 16px 0 !important; | |
| } | |
| .final-report-box th, .final-report-box td { | |
| border: 1px solid #cbd5e1 !important; | |
| padding: 8px 10px !important; | |
| text-align: left !important; | |
| } | |
| .final-report-box th { | |
| background-color: #e0e7ff !important; | |
| font-weight: bold !important; | |
| color: #1e40af !important; | |
| } | |
| .final-report-box tr:nth-child(even) { | |
| background-color: #f8fafc !important; | |
| } | |
| .final-report-box tr:hover { | |
| background-color: #f0f4f8 !important; | |
| } | |
| .final-report-box code { | |
| background-color: #f1f5f9 !important; | |
| padding: 2px 6px !important; | |
| border-radius: 4px !important; | |
| font-family: 'Consolas', 'Monaco', monospace !important; | |
| color: #dc2626 !important; | |
| } | |
| .final-report-box pre { | |
| background-color: #1e293b !important; | |
| color: #e2e8f0 !important; | |
| padding: 12px !important; | |
| border-radius: 6px !important; | |
| overflow-x: auto !important; | |
| margin: 12px 0 !important; | |
| font-size: 13px !important; | |
| } | |
| .final-report-box pre code { | |
| background-color: transparent !important; | |
| color: #e2e8f0 !important; | |
| padding: 0 !important; | |
| } | |
| .final-report-box blockquote { | |
| border-left: 4px solid #3b82f6 !important; | |
| padding-left: 12px !important; | |
| margin-left: 0 !important; | |
| margin: 12px 0 !important; | |
| color: #475569 !important; | |
| font-style: italic !important; | |
| background-color: #f0f9ff !important; | |
| padding: 10px 12px !important; | |
| border-radius: 0 6px 6px 0 !important; | |
| } | |
| .final-report-box ul, .final-report-box ol { | |
| margin-left: 20px !important; | |
| margin-bottom: 12px !important; | |
| } | |
| .final-report-box li { | |
| margin-bottom: 6px !important; | |
| line-height: 1.6 !important; | |
| } | |
| .final-report-box strong { | |
| color: #1e40af !important; | |
| font-weight: 600 !important; | |
| } | |
| .final-report-box em { | |
| color: #3730a3 !important; | |
| } | |
| .final-report-box hr { | |
| border: none !important; | |
| border-top: 2px solid #cbd5e1 !important; | |
| margin: 24px 0 !important; | |
| } | |
| .final-report-box a { | |
| color: #2563eb !important; | |
| text-decoration: underline !important; | |
| } | |
| .final-report-box a:hover { | |
| color: #1d4ed8 !important; | |
| } | |
| """ | |
| with gr.Blocks(title="협력적 LLM 시스템 - 다단계 평가", theme=gr.themes.Soft(), css=css) as app: | |
| gr.Markdown( | |
| """ | |
| # 🤝 협력적 LLM 시스템 (다단계 평가 프로세스) | |
| ### 📋 프로세스 플로우 | |
| ``` | |
| 감독(분석) → 조사(검색) → 평가(조사) → 감독(지시) → 실행(초안) | |
| → 평가(초안) → 감독(개선) → 평가(최종) → 실행(완성) | |
| ``` | |
| **4개 AI의 협력을 통한 최고 품질의 답변 생성** | |
| """ | |
| ) | |
| # 입력 섹션 | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown(""" | |
| ## 🚀 4개 AI의 협력 시스템 | |
| - **감독자 AI**: 거시적 분석과 전략 수립 | |
| - **조사자 AI**: 웹 검색과 정보 수집/정리 | |
| - **실행자 AI**: 구체적 계획 수립과 실행 | |
| - **평가자 AI**: 전체 과정 평가와 개선점 제시 | |
| ### 🌟 주요 기능 | |
| - 최대 4096 토큰 지원 (긴 응답 가능) | |
| - 20개 검색 결과와 동의어 검색 | |
| - 신뢰도 기반 정보 평가 | |
| - 다단계 평가와 피드백 반영 | |
| ### 📋 프로세스 | |
| 감독 → 조사 → 평가 → 감독 → 실행 → 평가 → 감독 → 평가 → 실행 | |
| """) | |
| # LLM 선택 옵션 | |
| llm_mode = gr.Radio( | |
| choices=["default", "commercial"], | |
| value="default", | |
| label="LLM 모드 선택", | |
| info="commercial을 선택하면 Gemini 2.5 Pro를 사용합니다" | |
| ) | |
| user_input = gr.Textbox( | |
| label="질문 입력", | |
| placeholder="예: 기계학습 모델의 성능을 향상시키는 방법은?", | |
| lines=3 | |
| ) | |
| with gr.Row(): | |
| submit_btn = gr.Button("🚀 분석 시작", variant="primary", scale=2) | |
| clear_btn = gr.Button("🗑️ 초기화", scale=1) | |
| status_text = gr.Textbox( | |
| label="상태", | |
| interactive=False, | |
| value="대기 중...", | |
| max_lines=2 | |
| ) | |
| # 최종 결과 섹션 추가 | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("### 📊 최종 종합 보고서") | |
| final_report = gr.Markdown( | |
| value="*최종 보고서가 여기에 표시됩니다.*", | |
| elem_classes=["final-report-box"] | |
| ) | |
| # AI 출력들 - 2x2 그리드 | |
| with gr.Row(): | |
| # 상단 행 | |
| with gr.Column(): | |
| gr.Markdown("### 🧠 감독자 AI (거시적 분석)") | |
| supervisor_output = gr.Textbox( | |
| label="", | |
| lines=12, | |
| max_lines=18, | |
| interactive=False, | |
| elem_classes=["supervisor-box"] | |
| ) | |
| with gr.Column(): | |
| gr.Markdown("### 🔍 조사자 AI (웹 검색 & 정리)") | |
| researcher_output = gr.Textbox( | |
| label="", | |
| lines=12, | |
| max_lines=18, | |
| interactive=False, | |
| elem_classes=["researcher-box"] | |
| ) | |
| with gr.Row(): | |
| # 하단 행 | |
| with gr.Column(): | |
| gr.Markdown("### 👁️ 실행자 AI (미시적 구현)") | |
| executor_output = gr.Textbox( | |
| label="", | |
| lines=12, | |
| max_lines=18, | |
| interactive=False, | |
| elem_classes=["executor-box"] | |
| ) | |
| with gr.Column(): | |
| gr.Markdown("### 📊 평가자 AI (전체 평가)") | |
| evaluator_output = gr.Textbox( | |
| label="", | |
| lines=12, | |
| max_lines=18, | |
| interactive=False, | |
| elem_classes=["evaluator-box"] | |
| ) | |
| # 예제 | |
| gr.Examples( | |
| examples=[ | |
| "기계학습 모델의 성능을 향상시키는 최신 방법은?", | |
| "2025년 효과적인 프로젝트 관리 도구와 전략은?", | |
| "지속 가능한 비즈니스 모델의 최신 트렌드는?", | |
| "최신 데이터 시각화 도구와 기법은?", | |
| "원격 팀의 생산성을 높이는 검증된 방법은?", | |
| "스타트업을 위한 효과적인 마케팅 전략은?", | |
| "AI 윤리와 규제의 최신 동향은?", | |
| "클라우드 네이티브 애플리케이션 개발 모범 사례는?" | |
| ], | |
| inputs=user_input, | |
| label="💡 예제 질문" | |
| ) | |
| # 이벤트 핸들러 | |
| submit_btn.click( | |
| fn=process_query_streaming, | |
| inputs=[user_input, llm_mode], | |
| outputs=[supervisor_output, researcher_output, executor_output, evaluator_output, final_report, status_text] | |
| ).then( | |
| fn=lambda: "", | |
| outputs=[user_input] | |
| ) | |
| user_input.submit( | |
| fn=process_query_streaming, | |
| inputs=[user_input, llm_mode], | |
| outputs=[supervisor_output, researcher_output, executor_output, evaluator_output, final_report, status_text] | |
| ).then( | |
| fn=lambda: "", | |
| outputs=[user_input] | |
| ) | |
| clear_btn.click( | |
| fn=clear_all, | |
| outputs=[supervisor_output, researcher_output, executor_output, evaluator_output, final_report, status_text] | |
| ) | |
| if __name__ == "__main__": | |
| app.queue() # 스트리밍을 위한 큐 활성화 | |
| app.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=True, | |
| show_error=True | |
| ) | |