diff --git a/app/services/langgraph_enhanced/agents/analysis_agent.py b/app/services/langgraph_enhanced/agents/analysis_agent.py index 6c08de7..a2f1230 100644 --- a/app/services/langgraph_enhanced/agents/analysis_agent.py +++ b/app/services/langgraph_enhanced/agents/analysis_agent.py @@ -225,8 +225,8 @@ async def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict required_services=query_analysis.get('required_services', []) ) - response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="analysis_strategy") - strategy = self.parse_analysis_strategy(response_text.strip()) + response = self.llm.invoke(prompt) + strategy = self.parse_analysis_strategy(response.content.strip()) # 종목명 추출 stock_symbol = self._extract_stock_symbol(user_query) @@ -413,7 +413,8 @@ async def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict ✅ **균형성**: 호재와 악재의 **영향력을 비교 분석**하여 종합적인 판단 제시 ✅ **실용성**: 실제 투자에 바로 활용 가능한 구체적 전략 제시""" - analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="analysis", log_label="integrated_investment_analysis") + analysis_response = self.llm.invoke(analysis_prompt) + analysis_result = analysis_response.content self.log(f"통합 투자 분석 완료: {stock_symbol or stock_name}") else: diff --git a/app/services/langgraph_enhanced/agents/base_agent.py b/app/services/langgraph_enhanced/agents/base_agent.py index acf2928..54a1ce2 100644 --- a/app/services/langgraph_enhanced/agents/base_agent.py +++ b/app/services/langgraph_enhanced/agents/base_agent.py @@ -5,22 +5,15 @@ from abc import ABC, abstractmethod from typing import Dict, Any -import time -try: - # 실제 실행 환경에서는 전역 llm_manager를 사용 - from ..llm_manager import llm_manager as _global_llm_manager -except Exception: - # 테스트/데모 환경에서 외부 의존성 없이 임시 매니저를 주입 가능 - _global_llm_manager = None +from ..llm_manager import LLMManager class BaseAgent(ABC): """기본 에이전트 클래스""" def __init__(self, purpose: str = "general"): - # 전역 LLM 매니저 공유 (캐시 공유). 테스트 환경에서는 외부 의존성 없이 주입 가능 - self.llm_manager = _global_llm_manager - self.llm = self.llm_manager.get_llm(purpose=purpose) if self.llm_manager else None + self.llm_manager = LLMManager() + self.llm = self.llm_manager.get_llm(purpose=purpose) self.purpose = purpose self.agent_name = "" @@ -38,22 +31,3 @@ def log(self, message: str): """로그 출력""" print(f"🤖 {self.agent_name}: {message}") - def invoke_llm_with_cache(self, prompt: str, purpose: str = None, log_label: str = None) -> str: - """LLM 호출(캐시 적용) + 실행 시간 로깅 공통 헬퍼""" - label = log_label or "llm_invoke" - start = time.time() - print(f"⏳ [{self.agent_name}] {label} 시작") - try: - response_text = self.llm_manager.invoke_with_cache( - self.llm, - prompt, - purpose=(purpose or self.purpose) - ) - elapsed = (time.time() - start) * 1000 - print(f"✅ [{self.agent_name}] {label} 완료 - {elapsed:.1f}ms") - return response_text - except Exception as e: - elapsed = (time.time() - start) * 1000 - print(f"❌ [{self.agent_name}] {label} 실패 - {elapsed:.1f}ms - {e}") - raise - diff --git a/app/services/langgraph_enhanced/agents/confidence_calculator.py b/app/services/langgraph_enhanced/agents/confidence_calculator.py index 23835e9..4a91f7e 100644 --- a/app/services/langgraph_enhanced/agents/confidence_calculator.py +++ b/app/services/langgraph_enhanced/agents/confidence_calculator.py @@ -209,10 +209,10 @@ def process( ) # LLM 호출 - response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="confidence_evaluation") + response = self.llm.invoke(prompt) # 응답 파싱 - evaluation = self.parse_response(response_text) + evaluation = self.parse_response(response.content) print(f"📊 신뢰도 평가 완료:") print(f" 전체 신뢰도: {evaluation['overall_confidence']:.2f}") diff --git a/app/services/langgraph_enhanced/agents/data_agent.py b/app/services/langgraph_enhanced/agents/data_agent.py index 3a3b341..ab326e9 100644 --- a/app/services/langgraph_enhanced/agents/data_agent.py +++ b/app/services/langgraph_enhanced/agents/data_agent.py @@ -32,22 +32,22 @@ def get_prompt_template(self) -> str: **Yahoo Finance에서 사용하는 정확한 심볼**을 data_query에 입력하세요. ### 변환 규칙: -1. 한국 주식: 6자리 코드 + `.KS` +1. **한국 주식**: 6자리 코드 + `.KS` - 예: 삼성전자 → 005930.KS, 네이버 → 035420.KS -2. 미국 주식: 표준 티커 심볼 (1~5자 알파벳) +2. **미국 주식**: 표준 티커 심볼 (1~5자 알파벳) - 예: 테슬라 → TSLA, 애플 → AAPL, 디즈니 → DIS, 스타벅스 → SBUX, 나이키 → NKE - **당신의 금융 지식을 활용하여 모든 회사명을 정확한 티커 심볼로 변환하세요** -3. 유럽 주식: 티커 + 거래소 접미사 +3. **유럽 주식**: 티커 + 거래소 접미사 - 프랑스 (파리): `.PA` (예: LVMH → MC.PA, 에르메스 → RMS.PA) - 영국 (런던): `.L` (예: BP → BP.L) - 독일 (프랑크푸르트): `.DE` (예: BMW → BMW.DE) -4. 이미 심볼 형태인 경우: 그대로 사용 +4. **이미 심볼 형태**인 경우: 그대로 사용 - 예: "TSLA 주가" → TSLA, "DIS 차트" → DIS -중요: +**중요**: - 회사명(한글/영어)을 받으면 반드시 Yahoo Finance 티커 심볼로 변환하세요 - 개별 상장되지 않은 브랜드(예: 구찌)는 모기업 심볼(Kering)을 사용하거나 "상장되지 않음" 안내 @@ -199,8 +199,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, required_services=query_analysis.get('required_services', []) ) - response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="data_strategy") - strategy = self.parse_data_strategy(response_text.strip()) + response = self.llm.invoke(prompt) + strategy = self.parse_data_strategy(response.content.strip()) # 실제 데이터 조회 data = financial_data_service.get_financial_data(strategy['data_query']) diff --git a/app/services/langgraph_enhanced/agents/investment_intent_detector.py b/app/services/langgraph_enhanced/agents/investment_intent_detector.py index ceacc17..358b102 100644 --- a/app/services/langgraph_enhanced/agents/investment_intent_detector.py +++ b/app/services/langgraph_enhanced/agents/investment_intent_detector.py @@ -173,8 +173,8 @@ def detect(self, user_query: str) -> Dict[str, Any]: """투자 의도 감지""" try: prompt = self.get_prompt_template().format(user_query=user_query) - response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="intent_detection") - result = self.parse_response(response_text.strip()) + response = self.llm.invoke(prompt) + result = self.parse_response(response.content.strip()) self.log(f"투자 의도 감지: {result['is_investment_question']} (신뢰도: {result['confidence']:.2f})") self.log(f" 근거: {result['reasoning']}") diff --git a/app/services/langgraph_enhanced/agents/knowledge_agent.py b/app/services/langgraph_enhanced/agents/knowledge_agent.py index 619a13c..994b456 100644 --- a/app/services/langgraph_enhanced/agents/knowledge_agent.py +++ b/app/services/langgraph_enhanced/agents/knowledge_agent.py @@ -403,12 +403,105 @@ def _get_rag_context(self, user_query: str, namespace: str, top_k: int = 5) -> s self.log(f"RAG 검색 오류: {e}") return "" + def _get_simple_knowledge_response(self, user_query: str) -> str: + """Fast-path 지식 응답: 간단한 질문에 대한 빠른 답변""" + try: + # 1. 내장 지식 DB에서 직접 검색 + knowledge_db = self._load_knowledge_database() + + # 키워드 기반 매칭 + query_lower = user_query.lower() + matched_concept = None + + for concept, data in knowledge_db.items(): + concept_name = data.get('name', '').lower() + if concept.lower() in query_lower or concept_name in query_lower: + matched_concept = concept + break + + if matched_concept: + knowledge_data = knowledge_db[matched_concept] + print(f"⚡ Fast-path 지식 매칭: {matched_concept}") + + # 간단한 설명 생성 + response_parts = [f"📚 **{knowledge_data.get('name', matched_concept)}**\n"] + response_parts.append(f"💡 **정의**: {knowledge_data.get('definition', '정의 없음')}") + + if 'formula' in knowledge_data: + response_parts.append(f"📊 **공식**: {knowledge_data['formula']}") + + if 'examples' in knowledge_data and knowledge_data['examples']: + response_parts.append(f"📝 **예시**: {knowledge_data['examples'][0]}") + + if 'usage' in knowledge_data: + response_parts.append(f"🎯 **활용**: {knowledge_data['usage']}") + + return "\n".join(response_parts) + + # 2. RAG로 빠른 검색 + namespace = self._determine_namespace_simple(user_query) + rag_context = self._get_rag_context(user_query, namespace, top_k=3) + + if rag_context: + print(f"⚡ Fast-path RAG 검색: {namespace}") + # 간단한 LLM 호출로 요약 + summary_prompt = f"""다음 지식 정보를 바탕으로 사용자 질문에 간단히 답변해주세요. + +질문: {user_query} + +지식 정보: +{rag_context} + +간단하고 명확한 답변을 3-4문장으로 작성해주세요.""" + + response = self.llm.invoke(summary_prompt) + return response.content.strip() + + return "해당 질문에 대한 정보를 찾을 수 없습니다." + + except Exception as e: + print(f"⚠️ Fast-path 지식 검색 실패: {e}") + return "" + + def _determine_namespace_simple(self, user_query: str) -> str: + """간단한 네임스페이스 결정 (키워드 기반)""" + query_lower = user_query.lower() + + # 키워드 기반 분류 + if any(word in query_lower for word in ['뭐야', '무엇', '의미', '정의', '개념', '란', '이란']): + return 'terminology' + elif any(word in query_lower for word in ['분석', '재무', '경제', '실적', '지표']): + return 'financial_analysis' + elif any(word in query_lower for word in ['청년', '정책', '지원', '혜택']): + return 'youth_policy' + elif any(word in query_lower for word in ['투자', '전략', '포트폴리오', '분산']): + return 'investment_strategy' + else: + return 'terminology' # 기본값 + def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, Any]: - """지식 에이전트 처리 (네임스페이스 라우팅)""" + """지식 에이전트 처리 (네임스페이스 라우팅) - Fast-path 지원""" try: self.log(f"지식 교육 시작: {user_query}") - # 1. 네임스페이스 결정 + # Fast-path 판정: 단순 지식 질의 + primary_intent = query_analysis.get('primary_intent', 'knowledge') + complexity = query_analysis.get('complexity_level', 'simple') + is_simple_knowledge = (primary_intent == 'knowledge' and complexity == 'simple') + + if is_simple_knowledge: + print("⚡ Knowledge Fast-path: 단순 지식 질의 감지 - 전략 LLM 생략") + # Fast-path: 바로 지식 검색 및 간단한 설명 + simple_response = self._get_simple_knowledge_response(user_query) + if simple_response: + return { + 'success': True, + 'explanation_result': simple_response, + 'fast_path': True, + 'skip_result_combiner': True # 결과 통합 건너뛰기 플래그 + } + + # 일반 경로: 1. 네임스페이스 결정 namespace = self._determine_namespace(user_query, query_analysis) # 2. RAG 컨텍스트 가져오기 @@ -422,8 +515,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, required_services=query_analysis.get('required_services', []) ) - response_text = self.invoke_llm_with_cache(prompt, purpose="knowledge", log_label="knowledge_strategy") - strategy = self.parse_education_strategy(response_text.strip()) + response = self.llm.invoke(prompt) + strategy = self.parse_education_strategy(response.content.strip()) # 4. 설명 생성 if rag_context: @@ -450,7 +543,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, 명확하고 구체적으로 설명해주세요.""" - explanation_result = self.invoke_llm_with_cache(explanation_prompt, purpose="knowledge", log_label="knowledge_explanation_rag") + explanation_response = self.llm.invoke(explanation_prompt) + explanation_result = explanation_response.content self.log(f"RAG 기반 지식 교육 완료") else: @@ -468,7 +562,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, 명확하고 친절하게 설명해주세요.""" - explanation_result = self.invoke_llm_with_cache(explanation_prompt, purpose="knowledge", log_label="knowledge_explanation_basic") + explanation_response = self.llm.invoke(explanation_prompt) + explanation_result = explanation_response.content return { 'success': True, diff --git a/app/services/langgraph_enhanced/agents/news_agent.py b/app/services/langgraph_enhanced/agents/news_agent.py index ad729d5..81007c1 100644 --- a/app/services/langgraph_enhanced/agents/news_agent.py +++ b/app/services/langgraph_enhanced/agents/news_agent.py @@ -193,80 +193,147 @@ def _format_news_data(self, news_data: List[Dict[str, Any]]) -> str: return "\n".join(formatted) + async def _collect_news_fast_path(self, user_query: str) -> List[Dict[str, Any]]: + """Fast-path 뉴스 수집: news_service 직접 호출""" + try: + from ...workflow_components.news_service import NewsService + news_service = NewsService() + + # 종합 뉴스 서비스 직접 호출 + news_data = await news_service.get_comprehensive_news( + query=user_query, + use_google_rss=True, + translate=True + ) + + print(f"⚡ Fast-path 뉴스 수집 완료: {len(news_data)}개") + return news_data + + except Exception as e: + print(f"⚠️ Fast-path 뉴스 수집 실패: {e}") + return [] + + def _format_simple_news_response(self, news_data: List[Dict[str, Any]]) -> str: + """Fast-path용 간단한 뉴스 응답 포맷""" + if not news_data: + return "최근 관련 뉴스를 찾을 수 없습니다." + + response_parts = [f"📰 최근 뉴스 {len(news_data)}건을 찾았습니다:\n"] + + for i, news in enumerate(news_data[:5], 1): # 최대 5개만 + title = news.get('title', '제목 없음') + summary = news.get('summary', '요약 없음') + source = news.get('source', '출처 없음') + published = news.get('published', '날짜 없음') + + response_parts.append(f"{i}. **{title}**") + response_parts.append(f" 📅 {published} | 📰 {source}") + response_parts.append(f" 📝 {summary[:100]}{'...' if len(summary) > 100 else ''}") + response_parts.append("") + + return "\n".join(response_parts) + async def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, Any]: - """뉴스 에이전트 처리 (async) - Fast-path: 단순 뉴스 질의는 전략 LLM/분석 LLM 생략하고 news_service 직접 호출(10s 타임박스) + 간단 요약 반환 - """ + """뉴스 에이전트 처리 (async) - Fast-path 지원""" try: self.log(f"뉴스 수집 시작: {user_query}") - primary = query_analysis.get('primary_intent', 'news') + + # Fast-path 판정: 단순 뉴스 질의 + primary_intent = query_analysis.get('primary_intent', 'news') complexity = query_analysis.get('complexity_level', 'simple') - is_simple_news = (primary == 'news' and complexity == 'simple') - - import asyncio - news_data: List[Dict[str, Any]] = [] - mk_context = "" - strategy: Dict[str, Any] = {} - + is_simple_news = (primary_intent == 'news' and complexity == 'simple') + if is_simple_news: - # Fast-path: news_service 직접 호출 + 타임박스 10s - try: - news_data = await asyncio.wait_for( - news_service.get_comprehensive_news(query=user_query, translate=True), - timeout=10.0 - ) - except asyncio.TimeoutError: - self.log("뉴스 수집 타임아웃(10s)") - news_data = [] - - # 간단 요약(LLM 미사용) + print("⚡ News Fast-path: 단순 뉴스 질의 감지 - 전략 LLM 생략") + # Fast-path: 바로 뉴스 수집 + news_data = await self._collect_news_fast_path(user_query) if news_data: - lines = [] - for i, n in enumerate(news_data[:3], 1): - title = n.get('title', '제목 없음') - src = n.get('source', 'N/A') - pub = n.get('published', '')[:10] - lines.append(f"{i}. {title} (출처: {src}, {pub})") - analysis_result = "\n".join(lines) - else: - analysis_result = "관련 뉴스를 찾을 수 없습니다." - - return { - 'success': True, - 'news_data': news_data, - 'analysis_result': analysis_result, - 'strategy': {}, - 'fast_path': True - } - - # 일반 경로: 기존 전략 + 분석(단, 캐시/로깅 적용) + simple_response = self._format_simple_news_response(news_data) + return { + 'success': True, + 'news_data': news_data, + 'analysis_result': simple_response, + 'fast_path': True, + 'skip_result_combiner': True # 결과 통합 건너뛰기 플래그 + } + + # 일반 경로: LLM이 뉴스 수집 전략 결정 prompt = self.get_prompt_template().format( user_query=user_query, primary_intent=query_analysis.get('primary_intent', 'news'), complexity_level=query_analysis.get('complexity_level', 'simple'), required_services=query_analysis.get('required_services', []) ) - response_text = self.invoke_llm_with_cache(prompt, purpose="news", log_label="news_strategy") - strategy = self.parse_news_strategy(response_text.strip()) - - # 실제 뉴스 수집 - news_data = await news_service.get_comprehensive_news( - query=strategy.get('search_query') or user_query, - translate=True - ) - - # 간단 분석(짧은 요약)으로 토큰 최소화 - if news_data: + + response = self.llm.invoke(prompt) + strategy = self.parse_news_strategy(response.content.strip()) + + print(f"🔍 [NewsAgent] 생성된 전략:") + print(f" - search_strategy: {strategy.get('search_strategy')}") + print(f" - search_query: {strategy.get('search_query')}") + print(f" - news_sources: {strategy.get('news_sources')}") + + # 실제 뉴스 수집 (async) + news_data = [] + mk_context = "" # 매일경제 컨텍스트는 별도로 저장 + + try: + if strategy['news_sources'] in ['google', 'both']: + print(f"📰 [NewsAgent] Google RSS에서 뉴스 수집 시작: {strategy['search_query']}") + # async 함수 직접 호출 - 리스트 반환 + google_news = await news_service.get_comprehensive_news( + query=strategy['search_query'] + ) + + print(f" ✅ [NewsAgent] Google RSS 결과: {len(google_news) if google_news else 0}개") + + if google_news and isinstance(google_news, list): + news_data.extend(google_news) + + if strategy['news_sources'] in ['mk', 'both']: + # 매일경제 KG 컨텍스트는 한국어 핵심 키워드 사용 + # 예: "금리 뉴스 분석해줘" → "금리" + korean_keyword = self._extract_korean_keyword(user_query) + print(f" 📚 [NewsAgent] 매일경제 KG 검색 키워드: {korean_keyword}") + + # async 함수 호출 - 문자열 반환 + mk_context = await news_service.get_analysis_context_from_kg( + query=korean_keyword, + limit=5 + ) + + # 중복 제거 및 정렬 + news_data = self._deduplicate_news(news_data) + + except Exception as e: + self.log(f"뉴스 수집 오류: {e}") + import traceback + traceback.print_exc() + news_data = [] + mk_context = "" + + # 뉴스 분석 + if news_data or mk_context: analysis_prompt = self.generate_news_analysis_prompt(news_data, strategy, user_query) - analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="news", log_label="news_analysis_short") + + # 매일경제 컨텍스트 추가 + if mk_context: + analysis_prompt += f"\n\n{mk_context}" + + analysis_response = self.llm.invoke(analysis_prompt) + analysis_result = analysis_response.content + + self.log(f"뉴스 분석 완료: {len(news_data or [])}건") else: analysis_result = "관련 뉴스를 찾을 수 없습니다. 다른 키워드로 검색해보세요." - + self.log("뉴스를 찾을 수 없음") + return { 'success': True, 'news_data': news_data, 'analysis_result': analysis_result, - 'strategy': strategy + 'strategy': strategy, + 'mk_context': mk_context } except Exception as e: diff --git a/app/services/langgraph_enhanced/agents/query_analyzer.py b/app/services/langgraph_enhanced/agents/query_analyzer.py index f4b566f..f568976 100644 --- a/app/services/langgraph_enhanced/agents/query_analyzer.py +++ b/app/services/langgraph_enhanced/agents/query_analyzer.py @@ -217,8 +217,8 @@ def process(self, user_query: str) -> Dict[str, Any]: # 2. 일반 쿼리 분석 prompt = self.get_prompt_template().format(user_query=user_query) - response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="query_analysis") - analysis_result = self.parse_response(response_text.strip()) + response = self.llm.invoke(prompt) + analysis_result = self.parse_response(response.content.strip()) # 3. 투자 의도 정보 통합 analysis_result['is_investment_question'] = is_investment_question @@ -251,8 +251,8 @@ def process(self, user_query: str) -> Dict[str, Any]: # 2. 일반 쿼리 분석 prompt = self.get_prompt_template().format(user_query=user_query) - response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="query_analysis_dup") - analysis_result = self.parse_response(response_text.strip()) + response = self.llm.invoke(prompt) + analysis_result = self.parse_response(response.content.strip()) # 3. 투자 의도 정보 통합 analysis_result['is_investment_question'] = is_investment_question diff --git a/app/services/langgraph_enhanced/agents/response_agent.py b/app/services/langgraph_enhanced/agents/response_agent.py index da95547..8971cc9 100644 --- a/app/services/langgraph_enhanced/agents/response_agent.py +++ b/app/services/langgraph_enhanced/agents/response_agent.py @@ -127,7 +127,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any], collected_dat collected_information=collected_info ) - final_response = self.invoke_llm_with_cache(prompt, purpose="response", log_label="final_response_generation") + response = self.llm.invoke(prompt) + final_response = response.content self.log("최종 응답 생성 완료") diff --git a/app/services/langgraph_enhanced/agents/result_combiner.py b/app/services/langgraph_enhanced/agents/result_combiner.py index 9694d02..24a7532 100644 --- a/app/services/langgraph_enhanced/agents/result_combiner.py +++ b/app/services/langgraph_enhanced/agents/result_combiner.py @@ -202,7 +202,8 @@ def process( ) # LLM 호출 - combined_response = self.invoke_llm_with_cache(prompt, purpose="response", log_label="result_combination") + response = self.llm.invoke(prompt) + combined_response = response.content # 신뢰도 추출 confidence = self._extract_confidence(combined_response) diff --git a/app/services/langgraph_enhanced/agents/visualization_agent.py b/app/services/langgraph_enhanced/agents/visualization_agent.py index 5c6e49e..2b00f17 100644 --- a/app/services/langgraph_enhanced/agents/visualization_agent.py +++ b/app/services/langgraph_enhanced/agents/visualization_agent.py @@ -280,7 +280,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, # 차트 분석 if chart_data and 'error' not in chart_data and strategy.get('include_analysis', True): analysis_prompt = self.generate_chart_analysis_prompt(chart_data, strategy, user_query) - analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="analysis", log_label="chart_analysis") + analysis_response = self.llm.invoke(analysis_prompt) + analysis_result = analysis_response.content self.log("차트 분석 완료") else: diff --git a/app/services/langgraph_enhanced/llm_manager.py b/app/services/langgraph_enhanced/llm_manager.py index 09b50ee..e9e098d 100644 --- a/app/services/langgraph_enhanced/llm_manager.py +++ b/app/services/langgraph_enhanced/llm_manager.py @@ -3,19 +3,10 @@ 깔끔하게 Gemini만 사용하도록 단순화 """ -from typing import Optional, List, Dict, Any +from typing import Optional from langchain_google_genai import ChatGoogleGenerativeAI from app.config import settings from app.utils.common_utils import CacheManager -import hashlib -import time - -try: - import numpy as _np - from sentence_transformers import SentenceTransformer -except Exception: - _np = None - SentenceTransformer = None class LLMManager: @@ -26,20 +17,6 @@ def __init__(self): self.default_model = "gemini-2.0-flash" # 정식 2.0 버전, 높은 할당량 # LLM 응답 캐싱 (5분 TTL) self.response_cache = CacheManager(default_ttl=300) - # 목적별 TTL 테이블 - self.purpose_ttl: Dict[str, int] = { - "classification": 90, - "analysis": 120, - "news": 300, - "knowledge": 3600, - "response": 600, - "general": 300, - } - # 의미(임베딩) 기반 캐시 - self.semantic_cache_enabled = _np is not None and SentenceTransformer is not None - self._semantic_model = None - self.semantic_cache: Dict[str, List[Dict[str, Any]]] = {} # purpose -> entries - self.semantic_capacity_per_purpose = 500 def get_llm(self, model_name: Optional[str] = None, @@ -49,7 +26,6 @@ def get_llm(self, """ Gemini LLM 인스턴스 반환 (용도별 최적화된 파라미터) """ - t0 = time.time() # 모델명이 없으면 기본 모델 사용 if model_name is None: model_name = self.default_model @@ -60,9 +36,8 @@ def get_llm(self, # 캐시에서 확인 cache_key = f"{model_name}_{purpose}_{optimized_params['temperature']}_{hash(str(optimized_params))}" if cache_key in self.llm_cache: - llm_cached = self.llm_cache[cache_key] - print(f"📦 get_llm HIT: {model_name} ({purpose}) - {(time.time()-t0)*1000:.1f}ms") - return llm_cached + return self.llm_cache[cache_key] + # API 키 확인 google_api_key = settings.google_api_key if not google_api_key: @@ -78,7 +53,7 @@ def get_llm(self, # 캐시에 저장 self.llm_cache[cache_key] = llm - print(f"🤖 Gemini LLM 로드: {model_name} ({purpose}, temperature: {optimized_params['temperature']}) - {(time.time()-t0)*1000:.1f}ms") + print(f"🤖 Gemini LLM 로드: {model_name} ({purpose}, temperature: {optimized_params['temperature']})") return llm def _get_optimized_params(self, purpose: str, temperature: float, **kwargs) -> dict: @@ -135,97 +110,26 @@ def get_default_llm(self, temperature: float = 0.7, purpose: str = "general", ** return self.get_llm(model_name=None, temperature=temperature, purpose=purpose, **kwargs) def invoke_with_cache(self, llm: ChatGoogleGenerativeAI, prompt: str, purpose: str = "general") -> str: - """LLM 호출 시 캐싱 적용 + 타이밍 로그""" - t0 = time.time() + """LLM 호출 시 캐싱 적용""" + import hashlib + + # 캐시 키 생성 (프롬프트 + 목적 해시) cache_key = hashlib.md5(f"{prompt}_{purpose}".encode()).hexdigest() + + # 캐시에서 확인 cached_response = self.response_cache.get(cache_key) if cached_response: - print(f"📦 LLM 응답 캐시 HIT: {purpose} - {(time.time()-t0)*1000:.1f}ms") + print(f"📦 캐시에서 LLM 응답 반환: {purpose}") return cached_response - print(f"⏳ LLM 호출 시작: {purpose}") - response = llm.invoke(prompt) - response_text = response.content if hasattr(response, 'content') else str(response) - ttl = self.purpose_ttl.get(purpose, 300) - self.response_cache.set(cache_key, response_text, ttl=ttl) - print(f"💾 LLM 응답 캐시 저장: {purpose} - {(time.time()-t0)*1000:.1f}ms (ttl={ttl}s)") - return response_text - - # === 의미(임베딩) 기반 캐시 === - def _get_embedding_model(self): - if not self.semantic_cache_enabled: - return None - if self._semantic_model is None: - try: - # 경량 모델 기본 사용 - self._semantic_model = SentenceTransformer("all-MiniLM-L6-v2") - except Exception: - self.semantic_cache_enabled = False - self._semantic_model = None - return self._semantic_model - - def _semantic_lookup(self, prompt: str, purpose: str, threshold: float = 0.92) -> Optional[str]: - if not self.semantic_cache_enabled: - return None - model = self._get_embedding_model() - if model is None: - return None - entries = self.semantic_cache.get(purpose, []) - if not entries: - return None - try: - qv = model.encode([prompt])[0] - best = None - best_sim = -1.0 - for e in entries: - sim = float(_np.dot(qv, e["vec"]) / ( _np.linalg.norm(qv) * _np.linalg.norm(e["vec"]) + 1e-9 )) - if sim > best_sim: - best_sim = sim - best = e - if best and best_sim >= threshold: - print(f"📦🔎 의미 캐시 HIT: {purpose} (sim={best_sim:.3f})") - return best["response"] - except Exception: - return None - return None - - def _semantic_store(self, prompt: str, purpose: str, response_text: str): - if not self.semantic_cache_enabled: - return - model = self._get_embedding_model() - if model is None: - return - try: - vec = model.encode([prompt])[0] - entries = self.semantic_cache.setdefault(purpose, []) - entries.append({"vec": vec, "response": response_text}) - # 용량 제한 - if len(entries) > self.semantic_capacity_per_purpose: - self.semantic_cache[purpose] = entries[-self.semantic_capacity_per_purpose:] - except Exception: - pass - - def invoke_with_semantic_cache(self, llm: ChatGoogleGenerativeAI, prompt: str, purpose: str = "general", threshold: float = 0.92) -> str: - # 1) 정확 캐시 조회 - exact = self.invoke_with_cache(llm, prompt, purpose) - if exact is not None: - # invoke_with_cache는 miss 시 LLM을 호출하므로, 의미 캐시만 별도로 제공하려면 분리 필요. - # 여기서는 의미 캐시 우선 시나리오를 위해 정확 캐시 조회만 선행 체크 후 직접 처리 - cache_key = hashlib.md5(f"{prompt}_{purpose}".encode()).hexdigest() - cached_only = self.response_cache.get(cache_key) - if cached_only is not None: - return cached_only - # 2) 의미 캐시 조회 - sem = self._semantic_lookup(prompt, purpose, threshold) - if sem is not None: - return sem - # 3) LLM 호출 후 저장 + + # LLM 호출 response = llm.invoke(prompt) response_text = response.content if hasattr(response, 'content') else str(response) - ttl = self.purpose_ttl.get(purpose, 300) - cache_key = hashlib.md5(f"{prompt}_{purpose}".encode()).hexdigest() - self.response_cache.set(cache_key, response_text, ttl=ttl) - self._semantic_store(prompt, purpose, response_text) - print(f"💾 의미 캐시에 저장: {purpose}") + + # 캐시에 저장 + self.response_cache.set(cache_key, response_text) + print(f"💾 LLM 응답 캐시 저장: {purpose}") + return response_text def clear_cache(self): diff --git a/app/services/langgraph_enhanced/workflow_router.py b/app/services/langgraph_enhanced/workflow_router.py index 7a7d0b0..469c177 100644 --- a/app/services/langgraph_enhanced/workflow_router.py +++ b/app/services/langgraph_enhanced/workflow_router.py @@ -93,12 +93,14 @@ def _build_workflow(self) -> StateGraph: # 시작점 설정 workflow.set_entry_point("query_analyzer") - # 쿼리 분석 → 조건부 라우팅 (단순 쿼리 최적화) + # 쿼리 분석 → 조건부 라우팅 (Fast-path 지원) workflow.add_conditional_edges( "query_analyzer", self._route_after_query_analysis, { "data_agent": "data_agent", # 단순 주가 조회 + "news_agent": "news_agent", # Fast-path: 단순 뉴스 질의 + "knowledge_agent": "knowledge_agent", # Fast-path: 단순 지식 질의 "service_planner": "service_planner", # 복잡한 쿼리 "response_agent": "response_agent" # 일반 인사 } @@ -135,18 +137,24 @@ def _build_workflow(self) -> StateGraph: } ) - # 다른 전문 에이전트들 → 결과 통합 - workflow.add_edge("analysis_agent", "result_combiner") - # 뉴스 에이전트는 Fast-path 지원: 조건부로 바로 응답 생성으로 이동 + # 다른 전문 에이전트들 → 조건부 라우팅 (Fast-path 지원) workflow.add_conditional_edges( "news_agent", self._route_after_news, { - "response_agent": "response_agent", - "result_combiner": "result_combiner" + "result_combiner": "result_combiner", # 일반 경로 + "response_agent": "response_agent" # Fast-path + } + ) + workflow.add_conditional_edges( + "knowledge_agent", + self._route_after_knowledge, + { + "result_combiner": "result_combiner", # 일반 경로 + "response_agent": "response_agent" # Fast-path } ) - workflow.add_edge("knowledge_agent", "result_combiner") + workflow.add_edge("analysis_agent", "result_combiner") workflow.add_edge("visualization_agent", "result_combiner") # 결과 통합 → 신뢰도 계산 @@ -165,8 +173,6 @@ def _build_workflow(self) -> StateGraph: def _query_analyzer_node(self, state: WorkflowState) -> WorkflowState: """쿼리 분석 노드""" try: - import time as _t - _ts = _t.time() user_query = state["user_query"] analyzer = self.agents["query_analyzer"] @@ -177,7 +183,6 @@ def _query_analyzer_node(self, state: WorkflowState) -> WorkflowState: print(f"🔍 쿼리 분석 완료: {query_analysis['primary_intent']} (신뢰도: {query_analysis['confidence']:.2f})") print(f" 근거: {query_analysis['reasoning']}") print(f" 다음 에이전트: {state['next_agent']}") - print(f"⏱ query_analyzer 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ 쿼리 분석 에이전트 오류: {e}") @@ -190,8 +195,6 @@ def _query_analyzer_node(self, state: WorkflowState) -> WorkflowState: def _service_planner_node(self, state: WorkflowState) -> WorkflowState: """서비스 계획 노드 - 복잡도 분석 및 실행 전략 수립""" try: - import time as _t - _ts = _t.time() user_query = state["user_query"] query_analysis = state["query_analysis"] @@ -228,7 +231,6 @@ def _service_planner_node(self, state: WorkflowState) -> WorkflowState: agents_list = service_plan.get('agents_to_execute', []) if agents_list: print(f" 병렬 실행 에이전트: {', '.join(agents_list)}") - print(f"⏱ service_planner 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ 서비스 플래너 오류: {e}") @@ -300,8 +302,6 @@ def _determine_next_agent(self, strategy: Dict[str, Any], query_analysis: Dict[s def _parallel_executor_node(self, state: WorkflowState) -> WorkflowState: """병렬 실행 노드 - 여러 에이전트 동시 실행""" try: - import time as _t - _ts = _t.time() user_query = state["user_query"] query_analysis = state["query_analysis"] service_plan = state["service_plan"] @@ -337,7 +337,6 @@ def _parallel_executor_node(self, state: WorkflowState) -> WorkflowState: state["chart_data"] = result.get('chart_data', {}) print(f"✅ 병렬 실행 완료: {len(parallel_results)}개 에이전트") - print(f"⏱ parallel_executor 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ 병렬 실행 오류: {e}") @@ -352,8 +351,6 @@ def _parallel_executor_node(self, state: WorkflowState) -> WorkflowState: def _result_combiner_node(self, state: WorkflowState) -> WorkflowState: """결과 통합 노드 - LLM 기반 지능형 결과 통합""" try: - import time as _t - _ts = _t.time() user_query = state["user_query"] # 에이전트별로 결과 구조화 @@ -412,7 +409,6 @@ def _result_combiner_node(self, state: WorkflowState) -> WorkflowState: state["combined_result"] = combined_result print(f"✅ 결과 통합 완료") - print(f"⏱ result_combiner 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ 결과 통합 오류: {e}") @@ -427,8 +423,6 @@ def _result_combiner_node(self, state: WorkflowState) -> WorkflowState: def _confidence_calculator_node(self, state: WorkflowState) -> WorkflowState: """신뢰도 계산 노드 - 응답 품질 평가""" try: - import time as _t - _ts = _t.time() user_query = state["user_query"] combined_result = state.get("combined_result", {}) @@ -445,7 +439,6 @@ def _confidence_calculator_node(self, state: WorkflowState) -> WorkflowState: print(f" 전체 신뢰도: {confidence_evaluation.get('overall_confidence', 0):.2f}") print(f" 데이터 품질: {confidence_evaluation.get('data_quality', 0):.2f}") print(f" 응답 완성도: {confidence_evaluation.get('response_completeness', 0):.2f}") - print(f"⏱ confidence_calculator 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ 신뢰도 계산 오류: {e}") @@ -497,8 +490,6 @@ def handle_success(s, r): def _analysis_agent_node(self, state: WorkflowState) -> WorkflowState: """분석 에이전트 노드 (async 처리 - RAG + 뉴스 통합)""" try: - import time as _t - _ts = _t.time() import asyncio import concurrent.futures @@ -533,7 +524,6 @@ def run_async_in_thread(): print(f"📈 통합 투자 분석 완료: {result.get('stock_symbol', '일반')}") print(f" - RAG 컨텍스트: {result.get('rag_context_length', 0)} 글자") print(f" - 뉴스: {result.get('news_count', 0)}건") - print(f"⏱ analysis_agent 소요: {(_t.time()-_ts)*1000:.1f}ms") else: state["error"] = result.get('error', 'analysis_agent 실패') @@ -550,12 +540,7 @@ def _news_agent_node(self, state: WorkflowState) -> WorkflowState: def handle_success(s, r): s["news_data"] = r['news_data'] s["news_analysis"] = r['analysis_result'] - # Fast-path 여부 기록(있으면 바로 응답으로 보낼 수 있음) - if r.get('fast_path'): - s["news_fast_path"] = True print(f"📰 뉴스 수집 및 분석 완료: {len(r['news_data'])}건") - import time as _t - _ts = _t.time() # NewsAgent가 async이므로 동기적으로 실행 try: @@ -584,7 +569,6 @@ def handle_success(s, r): handle_success(state, result) else: state["error"] = result.get('error', 'news_agent 실패') - print(f"⏱ news_agent 소요: {(_t.time()-_ts)*1000:.1f}ms") except Exception as e: print(f"❌ news_agent 오류: {e}") @@ -593,15 +577,6 @@ def handle_success(s, r): state["error"] = f"news_agent 오류: {str(e)}" return state - - def _route_after_news(self, state: WorkflowState) -> str: - """뉴스 에이전트 후 라우팅 - Fast-path면 바로 응답""" - # Fast-path 플래그가 있고 다른 데이터가 섞이지 않았다면 바로 응답으로 - if state.get("news_fast_path"): - has_other = bool(state.get("financial_data")) or bool(state.get("knowledge_context")) or bool(state.get("chart_data")) or bool(state.get("analysis_result")) - if not has_other: - return "response_agent" - return "result_combiner" def _knowledge_agent_node(self, state: WorkflowState) -> WorkflowState: """지식 에이전트 노드""" @@ -624,8 +599,6 @@ def handle_success(s, r): def _response_agent_node(self, state: WorkflowState) -> WorkflowState: """응답 에이전트 노드""" try: - import time as _t - _ts = _t.time() # 디버그: state 키 확인 print(f"🔍 response_agent_node state 키: {list(state.keys())}") print(f" financial_data 있음: {'financial_data' in state}") @@ -641,7 +614,6 @@ def _response_agent_node(self, state: WorkflowState) -> WorkflowState: if combined_result.get("combined_response"): state["final_response"] = combined_result["combined_response"] print(f"💬 메타 에이전트 통합 응답 사용") - print(f"⏱ response_agent 소요: {(_t.time()-_ts)*1000:.1f}ms") return state # 통합 결과가 없으면 기존 방식으로 응답 생성 @@ -669,7 +641,6 @@ def _response_agent_node(self, state: WorkflowState) -> WorkflowState: if result['success']: state["final_response"] = result['final_response'] print(f"💬 기본 응답 생성 완료") - print(f"⏱ response_agent 소요: {(_t.time()-_ts)*1000:.1f}ms") else: state["error"] = result.get('error', '응답 생성 실패') @@ -716,23 +687,30 @@ def _route_after_planning(self, state: WorkflowState) -> str: return next_agent def _route_after_query_analysis(self, state: WorkflowState) -> str: - """쿼리 분석 후 라우팅 - 단순 쿼리 최적화""" + """쿼리 분석 후 라우팅 - Fast-path 지원""" query_analysis = state.get("query_analysis", {}) primary_intent = query_analysis.get("primary_intent", "general") complexity = query_analysis.get("complexity_level", "simple") user_query = state["user_query"].lower() + print(f"🔍 라우팅 디버그: intent={primary_intent}, complexity={complexity}, query='{user_query}'") + + # Fast-path: 단순 뉴스 질의 (조건 완화) + if (primary_intent == "news" and complexity == "simple"): + print(f"⚡ News Fast-path: 단순 뉴스 질의 감지 - 메타 에이전트 건너뛰기") + return "news_agent" + + # Fast-path: 단순 지식 질의 (조건 완화) + if (primary_intent == "knowledge" and complexity == "simple"): + print(f"⚡ Knowledge Fast-path: 단순 지식 질의 감지 - 메타 에이전트 건너뛰기") + return "knowledge_agent" + # 단순 주가 조회는 바로 data_agent로 (메타 에이전트 건너뛰기) if (primary_intent == "data" and complexity == "simple" and any(keyword in user_query for keyword in ["주가", "가격", "시세", "현재가", "stock", "price"])): print(f"⚡ 단순 주가 조회 감지 - 메타 에이전트 건너뛰기") return "data_agent" - - # 단순 지식 질문은 바로 knowledge_agent로 (메타 에이전트 건너뛰기) - if (primary_intent == "knowledge" and complexity == "simple"): - print(f"⚡ 단순 지식 질문 감지 - 메타 에이전트 건너뛰기") - return "knowledge_agent" # 일반 인사는 바로 response_agent로 if primary_intent == "general" and any(keyword in user_query for keyword in ["안녕", "hello", "hi"]): @@ -742,6 +720,39 @@ def _route_after_query_analysis(self, state: WorkflowState) -> str: # 복잡한 쿼리는 서비스 플래너로 return "service_planner" + def _route_after_news(self, state: WorkflowState) -> str: + """뉴스 에이전트 후 라우팅 - Fast-path 지원""" + # Fast-path 플래그 확인 + news_data = state.get("news_data", []) + analysis_result = state.get("analysis_result", "") + + print(f"🔍 News 라우팅 디버그: news_data={len(news_data)}, analysis_result={bool(analysis_result)}") + + # Fast-path 조건: 뉴스 데이터와 분석 결과가 모두 있으면 + if news_data and analysis_result: + print(f"⚡ News Fast-path 완료 - 결과 통합 건너뛰기") + return "response_agent" + + # 일반 경로는 결과 통합으로 + print(f"📋 News 일반 경로 - 결과 통합으로") + return "result_combiner" + + def _route_after_knowledge(self, state: WorkflowState) -> str: + """지식 에이전트 후 라우팅 - Fast-path 지원""" + # Fast-path 결과 확인 + knowledge_context = state.get("knowledge_context", "") + + print(f"🔍 Knowledge 라우팅 디버그: knowledge_context={bool(knowledge_context)}") + + # Fast-path 조건: 지식 컨텍스트가 있으면 + if knowledge_context: + print(f"⚡ Knowledge Fast-path 완료 - 결과 통합 건너뛰기") + return "response_agent" + + # 일반 경로는 결과 통합으로 + print(f"📋 Knowledge 일반 경로 - 결과 통합으로") + return "result_combiner" + def _route_after_data(self, state: WorkflowState) -> str: """데이터 에이전트 후 라우팅 (투자 질문 감지)""" service_plan = state.get("service_plan", {}) diff --git a/app/services/portfolio/enhanced_portfolio_service.py b/app/services/portfolio/enhanced_portfolio_service.py index 189748f..22ddec9 100644 --- a/app/services/portfolio/enhanced_portfolio_service.py +++ b/app/services/portfolio/enhanced_portfolio_service.py @@ -127,12 +127,6 @@ async def recommend_enhanced_portfolio( step6_start = time.time() now = now_utc_z() - # 디버깅: recommended_stocks 상태 확인 - print(f"🔍 [디버깅] recommended_stocks 타입: {type(recommended_stocks)}") - print(f"🔍 [디버깅] recommended_stocks 개수: {len(recommended_stocks) if recommended_stocks else 0}") - if recommended_stocks: - print(f"🔍 [디버깅] 첫 번째 종목: {recommended_stocks[0].stockName if recommended_stocks[0] else 'None'}") - result = PortfolioRecommendationResult( portfolioId=profile.profileId, userId=profile.userId, @@ -141,11 +135,6 @@ async def recommend_enhanced_portfolio( createdAt=now, updatedAt=now ) - - # 디버깅: 결과 객체 상태 확인 - print(f"🔍 [디버깅] result.recommendedStocks 개수: {len(result.recommendedStocks) if result.recommendedStocks else 0}") - print(f"🔍 [디버깅] result.allocationSavings: {result.allocationSavings}") - step6_time = time.time() - step6_start print(f"⏱️ [단계 6] 결과 생성: {step6_time:.3f}초") diff --git a/app/services/workflow_components/news_service.py b/app/services/workflow_components/news_service.py index e36bc95..492c6ad 100644 --- a/app/services/workflow_components/news_service.py +++ b/app/services/workflow_components/news_service.py @@ -1,19 +1,12 @@ -"""뉴스 조회 서비스 (yfinance 우선 + 캐싱 + Google RSS 풀백 + 번역/정규화/중복 제거)""" +"""뉴스 조회 서비스 (동적 프롬프팅 지원 + 매일경제 RSS + Google RSS 번역 통합)""" import asyncio -from typing import List, Dict, Any, Optional -import re -from datetime import datetime, timezone -import yfinance as yf -import asyncio +from typing import List, Dict, Any from langchain_google_genai import ChatGoogleGenerativeAI from app.config import settings from app.services.workflow_components.data_agent_service import NewsCollector from app.services.workflow_components.mk_rss_scraper import MKKnowledgeGraphService, search_mk_news from app.services.workflow_components.google_rss_translator import google_rss_translator, search_google_news -from app.utils.common_utils import CacheManager -from deep_translator import GoogleTranslator -from app.services.langgraph_enhanced.llm_manager import llm_manager from app.utils.stock_utils import get_company_name_from_symbol # prompt_manager는 agents/에서 개별 관리 @@ -22,9 +15,9 @@ class NewsService: """금융 뉴스 조회를 담당하는 서비스 (통합 뉴스 서비스) 뉴스 소스: - 1. yfinance (우선) - 2. Google RSS (실시간, 자동 번역) - 풀백 - 3. 매일경제 RSS + Neo4j (임베딩 컨텍스트, 분석용) + 1. 매일경제 RSS + Neo4j (수동 업데이트, 임베딩 검색) + 2. Google RSS (실시간, 자동 번역) + 3. 기존 RSS (Naver, Daum - 폴백용) """ def __init__(self): @@ -32,11 +25,6 @@ def __init__(self): self.mk_kg_service = MKKnowledgeGraphService() # 매일경제 지식그래프 self.google_translator = google_rss_translator # Google RSS 번역 self.llm = self._initialize_llm() - # 뉴스 캐시(10분), 네거티브 캐시(30초) - self.news_cache = CacheManager(default_ttl=600) - self.negative_cache_ttl = 30 - # 번역기 (필요 시만 사용) - self._translator: Optional[GoogleTranslator] = None def _initialize_llm(self): """LLM 초기화""" @@ -242,16 +230,8 @@ async def get_mk_news_with_embedding(self, query: str, category: str = None, lim try: print(f"📚 매일경제 KG 컨텍스트 검색 (분석용): {query}") - # 매일경제 지식그래프에서 검색 (타임아웃 6s) - import asyncio - try: - mk_results = await asyncio.wait_for( - self.mk_kg_service.search_news(query, category, limit), - timeout=6.0 - ) - except asyncio.TimeoutError: - print("⏱️ 매일경제 KG 검색 타임아웃(6s)") - mk_results = [] + # 매일경제 지식그래프에서 검색 + mk_results = await self.mk_kg_service.search_news(query, category, limit) # 결과 포맷팅 formatted_results = [] @@ -367,270 +347,67 @@ async def get_comprehensive_news(self, use_google_rss: bool = True, translate: bool = True, korean_query: str = None) -> List[Dict[str, Any]]: - """✨ 종합 뉴스 검색 (병렬 수집 + .KS는 KG/RSS 우선 + 무가정) - - yfinance / Neo4j KG / Google RSS를 병렬 실행하고 제한시간 내 결과 선택 - - 한국(.KS) 종목은 KG→RSS→yfinance 우선, 그 외는 yfinance→RSS→KG - - 뉴스가 없으면 가정/추정 생성 없이 빈 결과 반환 + """✨ 종합 뉴스 검색 (매일경제 RSS + Google RSS) + ✨ FallbackAgent 사용 + + 전략: + - 매일경제 RSS (한국어) → korean_query 사용 + - Google RSS (영어) → query 사용 + + Args: + query: 검색 쿼리 (영어) + use_google_rss: Google RSS 실시간 검색 사용 여부 + translate: Google RSS 뉴스 번역 여부 + korean_query: 한국어 검색 쿼리 (매일경제용) + + Returns: + List[Dict[str, Any]]: 통합된 뉴스 리스트 """ try: from app.services.langgraph_enhanced.agents import get_news_source_fallback - print(f"📰 종합 뉴스 검색 시작: {query}") - overall_start = datetime.now() - # 특별 케이스 + print(f"📰 실시간 뉴스 검색 (FallbackAgent): {query}") + + all_news = [] + + # 특별한 케이스: 오늘 하루 시장 뉴스 요청 if query == "오늘 하루 시장 뉴스": return await self.get_today_market_news(limit=10) - # 심볼 추정으로 KR 여부 판단 - symbol_hint = self._maybe_extract_symbol(query) - is_kr = bool(symbol_hint and symbol_hint.endswith('.KS')) - - # 준비: 작업 정의 - async def run_yf(): - yf_key = self._make_cache_key("yf", query) - cached = self.news_cache.get(yf_key) - if cached is not None: - print(f"📦 yfinance 캐시 HIT: {len(cached)}개") - return cached - _t0 = datetime.now() - data = await self._try_yfinance_news(query, limit=8, translate=translate) - print(f"⏱ yfinance 소요: {(datetime.now()-_t0).total_seconds()*1000:.1f}ms") - if data: - self.news_cache.set(yf_key, data, ttl=600) - else: - self.news_cache.set(yf_key, [], ttl=self.negative_cache_ttl) - return data - - async def run_kg(): - try: - _t0 = datetime.now() - _data = await asyncio.wait_for(self.get_mk_news_with_embedding(query, limit=8), timeout=6.0) - print(f"⏱ KG 소요: {(datetime.now()-_t0).total_seconds()*1000:.1f}ms") - return _data - except Exception as _e: - print(f"⚠️ KG 실패/타임아웃: {_e}") - return [] - - async def run_rss(): - if not use_google_rss: - return [] - try: - fallback_helper = get_news_source_fallback() - _t0 = datetime.now() - result = await asyncio.wait_for( - fallback_helper.get_news_with_fallback(query=query, primary_source="google_rss", limit=8), - timeout=6.0 - ) - print(f"⏱ RSS 소요: {(datetime.now()-_t0).total_seconds()*1000:.1f}ms") - return result['data'] if result.get('success') else [] - except Exception as _e: - print(f"⚠️ RSS 실패/타임아웃: {_e}") - return [] - - # 병렬 실행 - tasks = [run_yf(), run_kg(), run_rss()] - yf_news, kg_news, rss_news = await asyncio.gather(*tasks, return_exceptions=False) - - # 선택 우선순위 - candidates = [] - if is_kr: - candidates = [kg_news, rss_news, yf_news] - else: - candidates = [yf_news, rss_news, kg_news] + # FallbackAgent를 통한 자동 풀백 실행 + fallback_helper = get_news_source_fallback() - # 첫 비어있지 않은 후보 선택 - for cand in candidates: - if cand: - selected = cand - break + # Primary 소스 결정 + primary_source = "google_rss" if use_google_rss else "mk_rss" + + # 뉴스 수집 with 자동 풀백 + result = await fallback_helper.get_news_with_fallback( + query=query, + primary_source=primary_source, + limit=5 + ) + + if result['success']: + all_news = result['data'] + print(f" ✅ 뉴스 수집 성공 (소스: {result['source']}): {len(all_news)}개") else: - selected = [] + print(f" ⚠️ 모든 뉴스 소스 실패") + all_news = [] - # 중복 제거 및 정렬 - unique_news = self._remove_duplicates(selected) + # 중복 제거 (URL 기준 + 제목 유사도) + unique_news = self._remove_duplicates(all_news) + + # 관련도 + 최신순 정렬 sorted_news = self._sort_news_by_relevance(unique_news, query) - elapsed = (datetime.now() - overall_start).total_seconds() * 1000 - print(f"✅ 실시간 뉴스 검색 결과: {len(sorted_news)}개 (중복 제거 후) | {elapsed:.1f}ms | KR={is_kr}") - return sorted_news[:10] + print(f"✅ 실시간 뉴스 검색 결과: {len(sorted_news)}개 (중복 제거 후)") + return sorted_news[:10] # 최대 10개 반환 + except Exception as e: print(f"❌ 뉴스 검색 중 오류: {e}") import traceback traceback.print_exc() return [] - - async def _try_yfinance_news(self, query: str, limit: int = 8, translate: bool = True) -> List[Dict[str, Any]]: - """yfinance 뉴스 시도(티커 추정 → 뉴스 수집 → 정규화/번역/정렬)""" - symbol = self._maybe_extract_symbol(query) - if not symbol: - return [] - try: - print(f"🔎 yfinance 뉴스 시도: symbol={symbol}") - ticker = yf.Ticker(symbol) - - # 네트워크 지연 방지를 위해 executor + 타임아웃 적용 - loop = asyncio.get_event_loop() - try: - items = await asyncio.wait_for( - loop.run_in_executor(None, lambda: getattr(ticker, "news", None)), - timeout=5.0 - ) - except asyncio.TimeoutError: - print("⏱️ yfinance 뉴스 조회 타임아웃(5s)") - items = None - - items = items or [] - if not items: - return [] - normalized: List[Dict[str, Any]] = [] - for it in items[:limit]: - title = it.get("title", "").strip() - link = it.get("link") or it.get("url") or "" - pub_ts = it.get("providerPublishTime") or it.get("provider_publish_time") - if isinstance(pub_ts, (int, float)): - published = datetime.fromtimestamp(pub_ts, tz=timezone.utc).isoformat() - else: - published = datetime.now(timezone.utc).isoformat() - summary = it.get("summary", "") - news_item = { - "title": title, - "summary": summary, - "url": link, - "published": published, - "source": "yfinance", - "language": "en", - "translated": False, - "symbol": symbol - } - normalized.append(news_item) - # 번역(옵션) - if translate and normalized: - await self._ensure_translator() - # 타이틀은 반드시 번역, 요약은 선택적(시간 단축) - async def _tr_title(n: Dict[str, Any]): - n["title_en"] = n.get("title", "") - if self._translator and n["title_en"]: - try: - n["title"] = await asyncio.wait_for( - self._translate_text(n["title_en"]), timeout=3.0 - ) - except Exception: - n["title"] = n["title_en"] - else: - n["title"] = n["title_en"] - n["translated"] = True - n["language"] = "ko" - - async def _tr_summary(n: Dict[str, Any]): - n["summary_en"] = n.get("summary", "") - if self._translator and n["summary_en"]: - try: - n["summary"] = await asyncio.wait_for( - self._translate_text(n["summary_en"][:400]), timeout=3.0 - ) - except Exception: - n["summary"] = n["summary_en"] - - # 병렬 번역(타이틀 필수, 요약은 베스트Effort) - await asyncio.gather(*[_tr_title(n) for n in normalized], return_exceptions=True) - await asyncio.gather(*[_tr_summary(n) for n in normalized], return_exceptions=True) - return normalized - except Exception as e: - print(f"❌ yfinance 뉴스 오류: {e}") - return [] - - async def _ensure_translator(self): - if self._translator is None: - try: - self._translator = GoogleTranslator(source='auto', target='ko') - except Exception: - self._translator = None - - async def _translate_text(self, text: str) -> str: - """번역 비동기 헬퍼(run_in_executor)""" - if not text: - return text - loop = asyncio.get_event_loop() - try: - return await loop.run_in_executor(None, self._translator.translate, text) - except Exception: - return text - - def _maybe_extract_symbol(self, text: str) -> Optional[str]: - """심볼 추정: 규칙 기반 → 캐시 → LLM 폴백(데이터 에이전트 방식 차용)""" - t = (text or "").strip() - # 한국: 6자리.KS - if re.match(r"^\d{6}\.KS$", t): - return t - # 미국: 대문자/숫자/.-/^ 최대 10자 - if re.match(r"^[A-Z0-9\.^\-]{1,10}$", t): - return t - # 텍스트에서 심볼 추출 시도 - try: - from app.utils.stock_utils import extract_symbol_from_query - symbol = extract_symbol_from_query(t) - if symbol: - return symbol - except Exception: - return None - # 캐시 확인 - key = self._make_cache_key("symres", t) - cached = self.news_cache.get(key) - if cached is not None: - return cached or None - # LLM 폴백으로 심볼 해석(데이터 에이전트 방식의 규칙을 프롬프트에 포함) - try: - resolved = self._resolve_symbol_with_llm(t) - # 결과 캐시(양/음) - self.news_cache.set(key, resolved or "", ttl=24 * 3600 if resolved else 300) - return resolved - except Exception: - # 음수 캐시(5분) - self.news_cache.set(key, "", ttl=300) - return None - - def _resolve_symbol_with_llm(self, query_text: str) -> Optional[str]: - """LLM을 사용해 회사명/자유 질의를 Yahoo Finance 티커로 매핑 - - 한국: 6자리 + .KS (예: 삼성전자 → 005930.KS) - - 미국: 표준 티커 (AAPL, TSLA 등) - - 유럽: 거래소 접미사 (예: MC.PA, BMW.DE) - - 출력 형식: data_query: 한 줄만 - """ - prompt = f""" -당신은 금융 데이터 전문가입니다. 아래 질의를 Yahoo Finance에서 사용하는 정확한 티커로 변환하세요. - -규칙: -- 한국 주식: 6자리 코드 + .KS (삼성전자→005930.KS, 네이버→035420.KS) -- 미국 주식: 표준 티커 (테슬라→TSLA, 애플→AAPL, 디즈니→DIS) -- 유럽/기타: 거래소 접미사 (LVMH→MC.PA, BMW→BMW.DE) -- 불명확하면 가장 가능성 높은 단일 티커를 제시 - -질의: "{query_text}" - -정확히 아래 한 줄만 반환: -data_query: -""" - llm = llm_manager.get_llm(purpose="classification") - text = llm_manager.invoke_with_cache(llm, prompt, purpose="classification") - # 파싱 - try: - for line in (text or "").splitlines(): - if ":" in line: - k, v = line.split(":", 1) - if k.strip().lower() == "data_query": - ticker = v.strip() - # 간단 유효성 검사 - if re.match(r"^\d{6}\.KS$", ticker) or re.match(r"^[A-Z][A-Z0-9\.^\-]{0,9}$", ticker): - return ticker - return ticker # 느슨 허용(추가 검증은 downstream) - except Exception: - pass - return None - - def _make_cache_key(self, source: str, query: str) -> str: - q = (query or "").strip().lower() - q = re.sub(r"\s+", " ", q) - return f"news:{source}:{q}" def _remove_duplicates(self, news_list: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """중복 뉴스 제거 (URL + 제목 유사도 기반)""" diff --git a/chat_terminal.py b/chat_terminal.py index 64d7efc..8bf3da6 100755 --- a/chat_terminal.py +++ b/chat_terminal.py @@ -15,9 +15,9 @@ class ChatTerminal: def __init__(self, server_url="http://localhost:8001", user_id=1, debug=False): self.server_url = server_url self.user_id = user_id + self.debug = debug self.session_id = f"terminal_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" self.chat_history = [] - self.debug = debug print("🤖 금융 챗봇 터미널 시작") print("=" * 50) @@ -98,13 +98,13 @@ def display_response(self, response): # 응답 시간 표시 if "response_time" in response: response_time = response['response_time'] - print(f"\n응답 시간: {response_time:.2f}초\n") - - # 디버그 메타 표시 + print(f"\n응답 시간: {response_time:.2f}초") + + # 디버그 정보 표시 if self.debug: action = response.get("action_data") or {} if action: - print("🔎 디버그 메타") + print("\n🔎 디버그 메타") print("-" * 50) qa = action.get("query_analysis", {}) sp = action.get("service_plan", {}) @@ -118,6 +118,8 @@ def display_response(self, response): print(f"• overall_conf={cf.get('overall_confidence')}") print(f"• workflow_type={action.get('workflow_type')}") print("-" * 50) + + print() def show_help(self): """도움말 표시""" @@ -232,7 +234,8 @@ def main(): help='서버 URL (기본값: http://localhost:8001)') parser.add_argument('--user-id', '-u', type=int, default=1, help='사용자 ID (기본값: 1)') - parser.add_argument('--debug', '-d', action='store_true', help='디버그 메타 정보 출력') + parser.add_argument('--debug', '-d', action='store_true', + help='디버그 메타 정보 출력') args = parser.parse_args() diff --git a/tests/cache_logging_demo.py b/tests/cache_logging_demo.py deleted file mode 100644 index 448f7b7..0000000 --- a/tests/cache_logging_demo.py +++ /dev/null @@ -1,60 +0,0 @@ -import time - - -from app.services.langgraph_enhanced.agents.base_agent import BaseAgent - - -class FakeLLMManager: - """LLM 호출 캐시/지연을 모사하는 페이크 매니저""" - def __init__(self): - self._cache = {} - - def get_llm(self, *args, **kwargs): - return object() # 더미 객체 - - def invoke_with_cache(self, llm, prompt: str, purpose: str = "general") -> str: - key = (prompt, purpose) - if key in self._cache: - print("📦 [FAKE] cache hit") - return self._cache[key] - # 첫 호출에는 인위적 지연 - time.sleep(0.2) - resp = f"FAKE_RESPONSE[{purpose}] for {prompt[:30]}" - self._cache[key] = resp - print("🆕 [FAKE] cache set") - return resp - - -class MinimalAgent(BaseAgent): - def __init__(self): - # super().__init__ 호출을 피하고 직접 주입(실제 LLM 사용 방지) - self.llm_manager = FakeLLMManager() - self.llm = self.llm_manager.get_llm(purpose="analysis") - self.purpose = "analysis" - self.agent_name = "minimal_agent" - - def get_prompt_template(self) -> str: - return "테스트 프롬프트: {msg}" - - def process(self, msg: str): - prompt = self.get_prompt_template().format(msg=msg) - # 1차 호출(캐시 미스 → 지연) - r1 = self.invoke_llm_with_cache(prompt, purpose=self.purpose, log_label="demo_call") - # 2차 호출(캐시 히트 → 즉시) - r2 = self.invoke_llm_with_cache(prompt, purpose=self.purpose, log_label="demo_call") - return r1, r2 - - -def main(): - agent = MinimalAgent() - t0 = time.time() - r1, r2 = agent.process("안녕하세요") - dt = (time.time() - t0) * 1000 - print(f"\n⏱ 전체 실행 시간: {dt:.1f}ms") - print(f"응답 동일성 확인: {r1 == r2}") - - -if __name__ == "__main__": - main() - - diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index 3ab97bc..0000000 --- a/tests/conftest.py +++ /dev/null @@ -1,10 +0,0 @@ -import os -import sys - - -# 프로젝트 루트를 PYTHONPATH에 추가하여 `app` 모듈 임포트 가능하게 설정 -PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - - diff --git a/tests/test_e2e_cache_timing_real.py b/tests/test_e2e_cache_timing_real.py deleted file mode 100644 index 7762140..0000000 --- a/tests/test_e2e_cache_timing_real.py +++ /dev/null @@ -1,69 +0,0 @@ -import os -import time -import pytest - - -def _check_env_or_skip(): - """실제 실행에 필요한 환경이 갖춰졌는지 점검하고 없으면 스킵""" - from app.config import settings - missing = [] - if not getattr(settings, "google_api_key", None): - missing.append("GOOGLE_API_KEY") - # Pinecone는 일부 시나리오에서만 쓰이므로 필수는 아님. 필요시 아래 주석 해제 - # try: - # from app.services.pinecone_config import PINECONE_API_KEY - # if not PINECONE_API_KEY: - # missing.append("PINECONE_API_KEY") - # except Exception: - # pass - if missing: - pytest.skip(f"실제 E2E 실행 스킵: 필수 환경변수 없음: {', '.join(missing)}") - - -def _measure(router, prompt): - t0 = time.perf_counter() - result = router.process_query(prompt) - dt = (time.perf_counter() - t0) * 1000 - print(f"\n⏱ '{prompt[:20]}...' 총 소요: {dt:.1f}ms | success={result.get('success')}") - return dt, result - - -def test_e2e_cache_timing_real(): - _check_env_or_skip() - - # 전역 LLM 캐시 초기화 - from app.services.langgraph_enhanced.llm_manager import llm_manager - try: - llm_manager.clear_cache() - except Exception: - pass - - from app.services.langgraph_enhanced.workflow_router import WorkflowRouter - router = WorkflowRouter() - - scenarios = [ - "안녕하세요", # 일반 인사 - "삼성전자 주가 알려줘", # 단순 주가 - "삼성전자 투자 분석해줘", # 투자 분석(복합) - "PER이 뭐야?", # 지식(RAG) - ] - - strict = os.getenv("STRICT_CACHE_ASSERT", "0") == "1" - - for q in scenarios: - print(f"\n=== 시나리오: {q} ===") - # 1차: 캐시 미스 - t1, _ = _measure(router, q) - # 2차: 캐시 히트 기대 - t2, _ = _measure(router, q) - - # 요약 출력 - delta = t1 - t2 - ratio = (t2 / t1) if t1 > 0 else 0 - print(f"📊 캐시 효과: 1차={t1:.1f}ms → 2차={t2:.1f}ms | Δ={delta:.1f}ms | ratio={ratio:.2f}") - - # 엄격 모드일 때만 성능 단언 - if strict: - assert t2 <= t1 * 0.9 or delta >= 60.0 - - diff --git a/tests/test_news_yf_sources.py b/tests/test_news_yf_sources.py deleted file mode 100644 index 0c131a4..0000000 --- a/tests/test_news_yf_sources.py +++ /dev/null @@ -1,60 +0,0 @@ -import asyncio -import time - - -def _print_sample(news, max_items=3): - n = len(news or []) - print(f"총 {n}건") - for i, item in enumerate((news or [])[:max_items], 1): - t = item.get("title", "N/A") - src = item.get("source", "?") - lang = item.get("language", "?") - pub = item.get("published", "") - print(f" {i}. [{src}|{lang}] {t} ({pub[:19]})") - - -async def _run_once(query: str): - from app.services.workflow_components.news_service import news_service - t0 = time.perf_counter() - news = await news_service.get_comprehensive_news(query=query, translate=True) - dt = (time.perf_counter() - t0) * 1000 - print(f"⏱ '{query}' 1차: {dt:.1f}ms", end=" | ") - _print_sample(news) - t1 = time.perf_counter() - news2 = await news_service.get_comprehensive_news(query=query, translate=True) - dt2 = (time.perf_counter() - t1) * 1000 - print(f"⏱ '{query}' 2차(캐시 기대): {dt2:.1f}ms", end=" | ") - _print_sample(news2) - - -def test_news_yfinance_multiple_symbols(): - # 다양한 종목/표현 테스트 - queries = [ - "TSLA", # 미국 - "AAPL", # 미국 - "MSFT", # 미국 - "NVDA", # 미국 - "005930.KS", # 한국(삼성전자) - "035420.KS", # 한국(네이버) - "Samsung Electronics", # 영문 회사명 - "Naver", # 영문 회사명 - "삼성전자", # 한글 회사명(심볼 추출 경로) - "네이버", # 한글 회사명(심볼 추출 경로) - ] - - async def main(): - # 캐시 초기화 (매 실행 시 동일 조건) - from app.services.workflow_components.news_service import news_service - try: - news_service.news_cache.clear() - print("🧹 뉴스 캐시 초기화 완료") - except Exception: - pass - - for q in queries: - print(f"\n=== 테스트: {q} ===") - await _run_once(q) - - asyncio.run(main()) - -