diff --git a/app/services/langgraph_enhanced/agents/analysis_agent.py b/app/services/langgraph_enhanced/agents/analysis_agent.py
index a2f1230..6c08de7 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 = self.llm.invoke(prompt)
-            strategy = self.parse_analysis_strategy(response.content.strip())
+            response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="analysis_strategy")
+            strategy = self.parse_analysis_strategy(response_text.strip())
             
             # 종목명 추출
             stock_symbol = self._extract_stock_symbol(user_query)
@@ -413,8 +413,7 @@ async def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict
 ✅ **균형성**: 호재와 악재의 **영향력을 비교 분석**하여 종합적인 판단 제시
 ✅ **실용성**: 실제 투자에 바로 활용 가능한 구체적 전략 제시"""
                 
-                analysis_response = self.llm.invoke(analysis_prompt)
-                analysis_result = analysis_response.content
+                analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="analysis", log_label="integrated_investment_analysis")
                 
                 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 54a1ce2..acf2928 100644
--- a/app/services/langgraph_enhanced/agents/base_agent.py
+++ b/app/services/langgraph_enhanced/agents/base_agent.py
@@ -5,15 +5,22 @@
 
 from abc import ABC, abstractmethod
 from typing import Dict, Any
-from ..llm_manager import LLMManager
+import time
+try:
+    # 실제 실행 환경에서는 전역 llm_manager를 사용
+    from ..llm_manager import llm_manager as _global_llm_manager
+except Exception:
+    # 테스트/데모 환경에서 외부 의존성 없이 임시 매니저를 주입 가능
+    _global_llm_manager = None
 
 
 class BaseAgent(ABC):
     """기본 에이전트 클래스"""
     
     def __init__(self, purpose: str = "general"):
-        self.llm_manager = LLMManager()
-        self.llm = self.llm_manager.get_llm(purpose=purpose)
+        # 전역 LLM 매니저 공유 (캐시 공유). 테스트 환경에서는 외부 의존성 없이 주입 가능
+        self.llm_manager = _global_llm_manager
+        self.llm = self.llm_manager.get_llm(purpose=purpose) if self.llm_manager else None
         self.purpose = purpose
         self.agent_name = ""
     
@@ -31,3 +38,22 @@ 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 4a91f7e..23835e9 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 = self.llm.invoke(prompt)
+            response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="confidence_evaluation")
             
             # 응답 파싱
-            evaluation = self.parse_response(response.content)
+            evaluation = self.parse_response(response_text)
             
             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 ab326e9..3a3b341 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 = self.llm.invoke(prompt)
-            strategy = self.parse_data_strategy(response.content.strip())
+            response_text = self.invoke_llm_with_cache(prompt, purpose="analysis", log_label="data_strategy")
+            strategy = self.parse_data_strategy(response_text.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 358b102..ceacc17 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 = self.llm.invoke(prompt)
-            result = self.parse_response(response.content.strip())
+            response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="intent_detection")
+            result = self.parse_response(response_text.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 f5542b7..619a13c 100644
--- a/app/services/langgraph_enhanced/agents/knowledge_agent.py
+++ b/app/services/langgraph_enhanced/agents/knowledge_agent.py
@@ -422,8 +422,8 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str,
                 required_services=query_analysis.get('required_services', [])
             )
             
-            response = self.llm.invoke(prompt)
-            strategy = self.parse_education_strategy(response.content.strip())
+            response_text = self.invoke_llm_with_cache(prompt, purpose="knowledge", log_label="knowledge_strategy")
+            strategy = self.parse_education_strategy(response_text.strip())
             
             # 4. 설명 생성
             if rag_context:
@@ -450,8 +450,7 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str,
 
 명확하고 구체적으로 설명해주세요."""
                 
-                explanation_response = self.llm.invoke(explanation_prompt)
-                explanation_result = explanation_response.content
+                explanation_result = self.invoke_llm_with_cache(explanation_prompt, purpose="knowledge", log_label="knowledge_explanation_rag")
                 
                 self.log(f"RAG 기반 지식 교육 완료")
             else:
@@ -469,8 +468,7 @@ def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str,
 
 명확하고 친절하게 설명해주세요."""
                 
-                explanation_response = self.llm.invoke(explanation_prompt)
-                explanation_result = explanation_response.content
+                explanation_result = self.invoke_llm_with_cache(explanation_prompt, purpose="knowledge", log_label="knowledge_explanation_basic")
             
             return {
                 'success': True,
diff --git a/app/services/langgraph_enhanced/agents/news_agent.py b/app/services/langgraph_enhanced/agents/news_agent.py
index 0ee8bf4..ad729d5 100644
--- a/app/services/langgraph_enhanced/agents/news_agent.py
+++ b/app/services/langgraph_enhanced/agents/news_agent.py
@@ -194,87 +194,79 @@ def _format_news_data(self, news_data: List[Dict[str, Any]]) -> str:
         return "\n".join(formatted)
     
     async def process(self, user_query: str, query_analysis: Dict[str, Any]) -> Dict[str, Any]:
-        """뉴스 에이전트 처리 (async)"""
+        """뉴스 에이전트 처리 (async)
+        Fast-path: 단순 뉴스 질의는 전략 LLM/분석 LLM 생략하고 news_service 직접 호출(10s 타임박스) + 간단 요약 반환
+        """
         try:
             self.log(f"뉴스 수집 시작: {user_query}")
-            
-            # LLM이 뉴스 수집 전략 결정
+            primary = 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] = {}
+
+            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 미사용)
+                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
+                }
+
+            # 일반 경로: 기존 전략 + 분석(단, 캐시/로깅 적용)
             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 = 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:
+            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:
                 analysis_prompt = self.generate_news_analysis_prompt(news_data, strategy, user_query)
-                
-                # 매일경제 컨텍스트 추가
-                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 [])}건")
+                analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="news", log_label="news_analysis_short")
             else:
                 analysis_result = "관련 뉴스를 찾을 수 없습니다. 다른 키워드로 검색해보세요."
-                self.log("뉴스를 찾을 수 없음")
-            
+
             return {
                 'success': True,
                 'news_data': news_data,
                 'analysis_result': analysis_result,
-                'strategy': strategy,
-                'mk_context': mk_context
+                'strategy': strategy
             }
             
         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 f568976..f4b566f 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 = self.llm.invoke(prompt)
-        analysis_result = self.parse_response(response.content.strip())
+        response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="query_analysis")
+        analysis_result = self.parse_response(response_text.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 = self.llm.invoke(prompt)
-        analysis_result = self.parse_response(response.content.strip())
+        response_text = self.invoke_llm_with_cache(prompt, purpose="classification", log_label="query_analysis_dup")
+        analysis_result = self.parse_response(response_text.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 8971cc9..da95547 100644
--- a/app/services/langgraph_enhanced/agents/response_agent.py
+++ b/app/services/langgraph_enhanced/agents/response_agent.py
@@ -127,8 +127,7 @@ def process(self, user_query: str, query_analysis: Dict[str, Any], collected_dat
                 collected_information=collected_info
             )
             
-            response = self.llm.invoke(prompt)
-            final_response = response.content
+            final_response = self.invoke_llm_with_cache(prompt, purpose="response", log_label="final_response_generation")
             
             self.log("최종 응답 생성 완료")
             
diff --git a/app/services/langgraph_enhanced/agents/result_combiner.py b/app/services/langgraph_enhanced/agents/result_combiner.py
index 24a7532..9694d02 100644
--- a/app/services/langgraph_enhanced/agents/result_combiner.py
+++ b/app/services/langgraph_enhanced/agents/result_combiner.py
@@ -202,8 +202,7 @@ def process(
             )
             
             # LLM 호출
-            response = self.llm.invoke(prompt)
-            combined_response = response.content
+            combined_response = self.invoke_llm_with_cache(prompt, purpose="response", log_label="result_combination")
             
             # 신뢰도 추출
             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 2b00f17..5c6e49e 100644
--- a/app/services/langgraph_enhanced/agents/visualization_agent.py
+++ b/app/services/langgraph_enhanced/agents/visualization_agent.py
@@ -280,8 +280,7 @@ 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_response = self.llm.invoke(analysis_prompt)
-                analysis_result = analysis_response.content
+                analysis_result = self.invoke_llm_with_cache(analysis_prompt, purpose="analysis", log_label="chart_analysis")
                 
                 self.log("차트 분석 완료")
             else:
diff --git a/app/services/langgraph_enhanced/llm_manager.py b/app/services/langgraph_enhanced/llm_manager.py
index e9e098d..09b50ee 100644
--- a/app/services/langgraph_enhanced/llm_manager.py
+++ b/app/services/langgraph_enhanced/llm_manager.py
@@ -3,10 +3,19 @@
 깔끔하게 Gemini만 사용하도록 단순화
 """
 
-from typing import Optional
+from typing import Optional, List, Dict, Any
 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:
@@ -17,6 +26,20 @@ 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, 
@@ -26,6 +49,7 @@ def get_llm(self,
         """
         Gemini LLM 인스턴스 반환 (용도별 최적화된 파라미터)
         """
+        t0 = time.time()
         # 모델명이 없으면 기본 모델 사용
         if model_name is None:
             model_name = self.default_model
@@ -36,8 +60,9 @@ def get_llm(self,
         # 캐시에서 확인
         cache_key = f"{model_name}_{purpose}_{optimized_params['temperature']}_{hash(str(optimized_params))}"
         if cache_key in self.llm_cache:
-            return self.llm_cache[cache_key]
-
+            llm_cached = self.llm_cache[cache_key]
+            print(f"📦 get_llm HIT: {model_name} ({purpose}) - {(time.time()-t0)*1000:.1f}ms")
+            return llm_cached
         # API 키 확인
         google_api_key = settings.google_api_key
         if not google_api_key:
@@ -53,7 +78,7 @@ def get_llm(self,
         # 캐시에 저장
         self.llm_cache[cache_key] = llm
         
-        print(f"🤖 Gemini LLM 로드: {model_name} ({purpose}, temperature: {optimized_params['temperature']})")
+        print(f"🤖 Gemini LLM 로드: {model_name} ({purpose}, temperature: {optimized_params['temperature']}) - {(time.time()-t0)*1000:.1f}ms")
         return llm
     
     def _get_optimized_params(self, purpose: str, temperature: float, **kwargs) -> dict:
@@ -110,26 +135,97 @@ 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 호출 시 캐싱 적용"""
-        import hashlib
-        
-        # 캐시 키 생성 (프롬프트 + 목적 해시)
+        """LLM 호출 시 캐싱 적용 + 타이밍 로그"""
+        t0 = time.time()
         cache_key = hashlib.md5(f"{prompt}_{purpose}".encode()).hexdigest()
-        
-        # 캐시에서 확인
         cached_response = self.response_cache.get(cache_key)
         if cached_response:
-            print(f"📦 캐시에서 LLM 응답 반환: {purpose}")
+            print(f"📦 LLM 응답 캐시 HIT: {purpose} - {(time.time()-t0)*1000:.1f}ms")
             return cached_response
-        
-        # LLM 호출
+        print(f"⏳ LLM 호출 시작: {purpose}")
         response = llm.invoke(prompt)
         response_text = response.content if hasattr(response, 'content') else str(response)
-        
-        # 캐시에 저장
-        self.response_cache.set(cache_key, response_text)
-        print(f"💾 LLM 응답 캐시 저장: {purpose}")
-        
+        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 호출 후 저장
+        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}")
         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 dba7631..7a7d0b0 100644
--- a/app/services/langgraph_enhanced/workflow_router.py
+++ b/app/services/langgraph_enhanced/workflow_router.py
@@ -137,7 +137,15 @@ def _build_workflow(self) -> StateGraph:
         
         # 다른 전문 에이전트들 → 결과 통합
         workflow.add_edge("analysis_agent", "result_combiner")
-        workflow.add_edge("news_agent", "result_combiner")
+        # 뉴스 에이전트는 Fast-path 지원: 조건부로 바로 응답 생성으로 이동
+        workflow.add_conditional_edges(
+            "news_agent",
+            self._route_after_news,
+            {
+                "response_agent": "response_agent",
+                "result_combiner": "result_combiner"
+            }
+        )
         workflow.add_edge("knowledge_agent", "result_combiner")
         workflow.add_edge("visualization_agent", "result_combiner")
         
@@ -157,6 +165,8 @@ 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"]
             
@@ -167,6 +177,7 @@ 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}")
@@ -179,6 +190,8 @@ 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"]
             
@@ -215,6 +228,7 @@ 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}")
@@ -286,6 +300,8 @@ 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"]
@@ -321,6 +337,7 @@ 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}")
@@ -335,6 +352,8 @@ 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"]
             
             # 에이전트별로 결과 구조화
@@ -393,6 +412,7 @@ 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}")
@@ -407,6 +427,8 @@ 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", {})
             
@@ -423,6 +445,7 @@ 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}")
@@ -474,6 +497,8 @@ 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
             
@@ -508,6 +533,7 @@ 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 실패')
                 
@@ -524,7 +550,12 @@ 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:
@@ -553,6 +584,7 @@ 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}")
@@ -561,6 +593,15 @@ 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:
         """지식 에이전트 노드"""
@@ -583,6 +624,8 @@ 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}")
@@ -598,6 +641,7 @@ 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
             
             # 통합 결과가 없으면 기존 방식으로 응답 생성
@@ -625,6 +669,7 @@ 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', '응답 생성 실패')
                 
@@ -683,6 +728,11 @@ def _route_after_query_analysis(self, state: WorkflowState) -> str:
             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"]):
diff --git a/app/services/portfolio/enhanced_portfolio_service.py b/app/services/portfolio/enhanced_portfolio_service.py
index 22ddec9..189748f 100644
--- a/app/services/portfolio/enhanced_portfolio_service.py
+++ b/app/services/portfolio/enhanced_portfolio_service.py
@@ -127,6 +127,12 @@ 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,
@@ -135,6 +141,11 @@ 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 492c6ad..e36bc95 100644
--- a/app/services/workflow_components/news_service.py
+++ b/app/services/workflow_components/news_service.py
@@ -1,12 +1,19 @@
-"""뉴스 조회 서비스 (동적 프롬프팅 지원 + 매일경제 RSS + Google RSS 번역 통합)"""
+"""뉴스 조회 서비스 (yfinance 우선 + 캐싱 + Google RSS 풀백 + 번역/정규화/중복 제거)"""
 
 import asyncio
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
+import re
+from datetime import datetime, timezone
+import yfinance as yf
+import asyncio
 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/에서 개별 관리
 
@@ -15,9 +22,9 @@ class NewsService:
     """금융 뉴스 조회를 담당하는 서비스 (통합 뉴스 서비스)
     
     뉴스 소스:
-    1. 매일경제 RSS + Neo4j (수동 업데이트, 임베딩 검색)
-    2. Google RSS (실시간, 자동 번역)
-    3. 기존 RSS (Naver, Daum - 폴백용)
+    1. yfinance (우선)
+    2. Google RSS (실시간, 자동 번역) - 풀백
+    3. 매일경제 RSS + Neo4j (임베딩 컨텍스트, 분석용)
     """
     
     def __init__(self):
@@ -25,6 +32,11 @@ 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 초기화"""
@@ -230,8 +242,16 @@ async def get_mk_news_with_embedding(self, query: str, category: str = None, lim
         try:
             print(f"📚 매일경제 KG 컨텍스트 검색 (분석용): {query}")
             
-            # 매일경제 지식그래프에서 검색
-            mk_results = await self.mk_kg_service.search_news(query, category, limit)
+            # 매일경제 지식그래프에서 검색 (타임아웃 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 = []
             
             # 결과 포맷팅
             formatted_results = []
@@ -347,67 +367,270 @@ async def get_comprehensive_news(self,
                                     use_google_rss: bool = True,
                                     translate: bool = True,
                                     korean_query: str = None) -> List[Dict[str, Any]]:
-        """✨ 종합 뉴스 검색 (매일경제 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]]: 통합된 뉴스 리스트
+        """✨ 종합 뉴스 검색 (병렬 수집 + .KS는 KG/RSS 우선 + 무가정)
+        - yfinance / Neo4j KG / Google RSS를 병렬 실행하고 제한시간 내 결과 선택
+        - 한국(.KS) 종목은 KG→RSS→yfinance 우선, 그 외는 yfinance→RSS→KG
+        - 뉴스가 없으면 가정/추정 생성 없이 빈 결과 반환
         """
         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)
             
-            # FallbackAgent를 통한 자동 풀백 실행
-            fallback_helper = get_news_source_fallback()
-            
-            # 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)}개")
+            # 심볼 추정으로 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:
-                print(f"  ⚠️ 모든 뉴스 소스 실패")
-                all_news = []
+                candidates = [yf_news, rss_news, kg_news]
             
-            # 중복 제거 (URL 기준 + 제목 유사도)
-            unique_news = self._remove_duplicates(all_news)
+            # 첫 비어있지 않은 후보 선택
+            for cand in candidates:
+                if cand:
+                    selected = cand
+                    break
+            else:
+                selected = []
             
-            # 관련도 + 최신순 정렬
+            # 중복 제거 및 정렬
+            unique_news = self._remove_duplicates(selected)
             sorted_news = self._sort_news_by_relevance(unique_news, query)
             
-            print(f"✅ 실시간 뉴스 검색 결과: {len(sorted_news)}개 (중복 제거 후)")
-            return sorted_news[:10]  # 최대 10개 반환
-            
+            elapsed = (datetime.now() - overall_start).total_seconds() * 1000
+            print(f"✅ 실시간 뉴스 검색 결과: {len(sorted_news)}개 (중복 제거 후) | {elapsed:.1f}ms | KR={is_kr}")
+            return sorted_news[: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: <TICKER> 한 줄만
+        """
+        prompt = f"""
+당신은 금융 데이터 전문가입니다. 아래 질의를 Yahoo Finance에서 사용하는 정확한 티커로 변환하세요.
+
+규칙:
+- 한국 주식: 6자리 코드 + .KS (삼성전자→005930.KS, 네이버→035420.KS)
+- 미국 주식: 표준 티커 (테슬라→TSLA, 애플→AAPL, 디즈니→DIS)
+- 유럽/기타: 거래소 접미사 (LVMH→MC.PA, BMW→BMW.DE)
+- 불명확하면 가장 가능성 높은 단일 티커를 제시
+
+질의: "{query_text}"
+
+정확히 아래 한 줄만 반환:
+data_query: <TICKER>
+"""
+        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 71a34c3..64d7efc 100755
--- a/chat_terminal.py
+++ b/chat_terminal.py
@@ -12,11 +12,12 @@
 from datetime import datetime
 
 class ChatTerminal:
-    def __init__(self, server_url="http://localhost:8001", user_id=1):
+    def __init__(self, server_url="http://localhost:8001", user_id=1, debug=False):
         self.server_url = server_url
         self.user_id = user_id
         self.session_id = f"terminal_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
         self.chat_history = []
+        self.debug = debug
         
         print("🤖 금융 챗봇 터미널 시작")
         print("=" * 50)
@@ -38,7 +39,8 @@ def send_message(self, message):
             payload = {
                 "message": message,
                 "user_id": str(self.user_id),  # user_id를 문자열로 변환
-                "session_id": self.session_id
+                "session_id": self.session_id,
+                "debug": self.debug
             }
             
             # 시작 시간 기록
@@ -97,6 +99,25 @@ def display_response(self, response):
         if "response_time" in response:
             response_time = response['response_time']
             print(f"\n응답 시간: {response_time:.2f}초\n")
+
+        # 디버그 메타 표시
+        if self.debug:
+            action = response.get("action_data") or {}
+            if action:
+                print("🔎 디버그 메타")
+                print("-" * 50)
+                qa = action.get("query_analysis", {})
+                sp = action.get("service_plan", {})
+                cf = action.get("confidence_evaluation", {})
+                print(f"• intent={qa.get('primary_intent')}, complexity={qa.get('complexity_level') or qa.get('complexity')}")
+                if qa.get('is_investment_question') is not None:
+                    print(f"• is_investment_question={qa.get('is_investment_question')}")
+                if sp:
+                    print(f"• exec_mode={sp.get('execution_mode')}, next={sp.get('next_agent')}, agents={sp.get('agents_to_execute')}")
+                if cf:
+                    print(f"• overall_conf={cf.get('overall_confidence')}")
+                print(f"• workflow_type={action.get('workflow_type')}")
+                print("-" * 50)
     
     def show_help(self):
         """도움말 표시"""
@@ -211,6 +232,7 @@ 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='디버그 메타 정보 출력')
     
     args = parser.parse_args()
     
@@ -219,7 +241,7 @@ def main():
     os.environ['LANGCHAIN_TRACING_V2'] = 'true'
     
     try:
-        chat = ChatTerminal(args.server, args.user_id)
+        chat = ChatTerminal(args.server, args.user_id, debug=args.debug)
         chat.run()
     except Exception as e:
         print(f"❌ 프로그램 실행 오류: {e}")
diff --git a/tests/cache_logging_demo.py b/tests/cache_logging_demo.py
new file mode 100644
index 0000000..448f7b7
--- /dev/null
+++ b/tests/cache_logging_demo.py
@@ -0,0 +1,60 @@
+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
new file mode 100644
index 0000000..3ab97bc
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,10 @@
+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
new file mode 100644
index 0000000..7762140
--- /dev/null
+++ b/tests/test_e2e_cache_timing_real.py
@@ -0,0 +1,69 @@
+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
new file mode 100644
index 0000000..0c131a4
--- /dev/null
+++ b/tests/test_news_yf_sources.py
@@ -0,0 +1,60 @@
+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())
+
+