""" RAG Tourism Recommender System - MVP Backend FastAPI application with Neo4j and Vector Search integration """ from fastapi import FastAPI, HTTPException, BackgroundTasks, UploadFile, File, Form from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import List, Optional import logging import time from .services.kg_service import AdminKGService from .services.chroma_vector_service import ChromaVectorSearchService from .services.phobert_vector_service import PhoBERTVectorSearchService from .services.intent_service import get_intent_service from .services.phowhisper_stt_service import get_phowhisper_service from .services.ner_service import get_ner_service # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="RAG Tourism Recommender API", description="MVP version of tourism recommendation system with Neo4j and Vector Search", version="1.0.0" ) # Add CORS middleware. # CORS_ALLOW_ORIGINS (comma-separated) hỗ trợ: # - "*" -> cho tất cả (không kèm credentials) # - "https://abc.vercel.app" -> domain cụ thể (khớp chính xác) # - "https://*.vercel.app" -> WILDCARD subdomain (mọi preview/prod của # Vercel: tourism-git-xxx.vercel.app, ...) # Vercel sinh domain preview đổi liên tục nên wildcard rất tiện. import os as _os import re as _re2 _cors_origins_env = _os.getenv("CORS_ALLOW_ORIGINS", "*").strip() _allow_origins: list = [] _allow_origin_regex = None _allow_credentials = False if _cors_origins_env == "*": _allow_origins = ["*"] else: entries = [o.strip() for o in _cors_origins_env.split(",") if o.strip()] exacts = [e for e in entries if "*" not in e] wildcards = [e for e in entries if "*" in e] _allow_origins = exacts _allow_credentials = True if wildcards: parts = [] for w in wildcards: host = _re2.sub(r"^https?://", "", w).rstrip("/") # escape rồi thay '*' bằng nhóm ký tự subdomain hợp lệ esc = _re2.escape(host).replace(r"\*", r"[A-Za-z0-9-]+") parts.append(r"https?://" + esc) _allow_origin_regex = r"^(" + "|".join(parts) + r")$" app.add_middleware( CORSMiddleware, allow_origins=_allow_origins, allow_origin_regex=_allow_origin_regex, allow_credentials=_allow_credentials, allow_methods=["*"], allow_headers=["*"], ) def _clean_ward_term(term: str) -> str: """Làm sạch tên phường/xã lấy từ NER. NER (PhoBERT, slow tokenizer) đôi khi trả span tràn kèm ký hiệu BPE, vd "Phường An Bình GIỜ L@@". Loại '@@', cắt tại các từ nghi vấn/nối câu để chỉ giữ phần tên đơn vị hành chính. """ import re as _re term = (term or "").replace("@@", " ").strip() term = _re.split( r"(?i)\s+(giờ|gio|là|la|nay|hiện|hien|sau|trước|truoc|thuộc|thuoc|" r"được|duoc|có|co|tên|ten|bây|bay|đổi|doi|gộp|gop|còn|con)\b", term, )[0] term = _re.sub(r"\s+", " ", term).strip(" ,.?!") # Bỏ đuôi "mới"/"cũ" CHỈ khi phần TÊN (không tính từ loại) còn >= 2 từ — # vì có đơn vị tên thật chứa từ này (vd "Xã Đất Mới" ở Cà Mau). m = _re.match( r"(?i)^((?:phường|xã|thị trấn|thị xã)\s+)?(.+?)\s+(mới|moi|cũ|cu)$", term ) if m and len(m.group(2).split()) >= 2: term = (m.group(1) or "") + m.group(2) return term def _strip_diacritics(s: str) -> str: """Lowercase + remove Vietnamese diacritics (1:1) for fuzzy matching.""" import unicodedata s = (s or "").replace("đ", "d").replace("Đ", "D") nfd = unicodedata.normalize("NFD", s) return "".join(c for c in nfd if unicodedata.category(c) != "Mn").lower() def _is_hcm(province: str) -> bool: """True nếu tỉnh/thành là TP. Hồ Chí Minh (bỏ dấu, mọi biến thể).""" p = _strip_diacritics(province).replace(".", " ") p = " ".join(p.split()) # gộp khoảng trắng if not p: return False return ( "ho chi minh" in p or "sai gon" in p or p in {"hcm", "tphcm", "tp hcm", "hcmc"} or p.replace(" ", "") in {"hcm", "tphcm", "hochiminh", "saigon"} ) # Từ khóa nhận diện LOẠI hình POI trong câu hỏi (bỏ dấu để khớp linh hoạt). _CATEGORY_KEYWORDS = { "Khách sạn": [ "khach san", "hotel", "resort", "homestay", "nha nghi", "luu tru", "nghi duong", "khu nghi", ], "Nhà hàng": [ "nha hang", "quan an", "restaurant", "an uong", "am thuc", "quan nhau", "dac san", "quan com", "buffet", ], } def _detect_category(text: str) -> Optional[str]: """Trả về 'Khách sạn' / 'Nhà hàng' nếu câu hỏi nhắm tới loại đó, else None.""" low = _strip_diacritics(text) for cat, kws in _CATEGORY_KEYWORDS.items(): if any(kw in low for kw in kws): return cat return None def _same_category(value: str, target: str) -> bool: return _strip_diacritics(value) == _strip_diacritics(target) def _kg_location_count(kg) -> Optional[int]: """Số Location trong KG — để so với số trong vector index, tự rebuild khi lệch.""" try: with kg.driver.session() as s: return s.run("MATCH (l:Location) RETURN count(l) AS c").single()["c"] except Exception: return None # Cache of known old district/city names for fuzzy matching (filled lazily) _district_names_cache = None _province_names_cache = None def _detect_province(text: str, kg) -> Optional[str]: """Dò tên tỉnh (cũ/mới) xuất hiện trong câu, bỏ dấu/thường hóa. Cần thiết vì NER hay bỏ sót tỉnh viết thường ('cà mau'); dùng để lọc đúng địa bàn cho tìm kiếm theo từ khóa loại hình. """ global _province_names_cache if _province_names_cache is None: try: _province_names_cache = kg.get_all_province_names() except Exception: _province_names_cache = [] norm = _strip_diacritics(text) best = None for name in _province_names_cache: if _strip_diacritics(name) in norm: if best is None or len(name) > len(best): best = name return best # Global services kg_service = None vector_service = None phobert_service = None intent_service = None stt_service = None ner_service = None # Pydantic models class TourismQuery(BaseModel): text: str location: Optional[str] = None category: Optional[str] = None use_vector_search: bool = True class LocationResponse(BaseModel): name: str lat: Optional[float] lng: Optional[float] ward: str province: str confidence: float category: Optional[str] = "Điểm tham quan" source: str = "kg_search" class SearchResponse(BaseModel): query: str results: List[LocationResponse] total: int processing_time: float search_method: str intent_info: Optional[dict] = None ner_entities: Optional[list] = None @app.on_event("startup") async def startup_event(): """Initialize services on startup""" global kg_service, vector_service, phobert_service, intent_service, ner_service, stt_service logger.info("Initializing services...") # Initialize Knowledge Graph service try: kg_service = AdminKGService() logger.info("Knowledge Graph service initialized") except Exception as e: logger.error(f"Failed to initialize KG service: {e}") # Initialize ChromaDB Vector Search service try: vector_service = ChromaVectorSearchService() # Try to load existing index if vector_service.load_index(): logger.info("ChromaDB vector search service initialized and loaded") # Heuristic: if index exists but has no usable location metadata, rebuild from KG if available. # This prevents a common failure mode where the index was built from a minimal schema. try: sample = vector_service.collection.get(limit=5, include=["metadatas"]) if vector_service.collection else None metadatas = (sample or {}).get("metadatas") or [] has_location_fields = any( ((m.get("district_city") or "").strip() or (m.get("address") or "").strip()) and (m.get("ward") or "").strip() # schema mới phải có ward riêng for m in metadatas if isinstance(m, dict) ) # Rebuild khi thiếu metadata HOẶC số lượng trong index lệch với KG # (vd vừa thêm khách sạn/nhà hàng vào KG mà index chưa cập nhật). _idx_n = vector_service.collection.count() if vector_service.collection else 0 _kg_n = _kg_location_count(kg_service) if kg_service else None _stale = (_kg_n is not None and _idx_n < _kg_n) if (not has_location_fields or _stale) and kg_service: logger.warning(f"ChromaDB index cần rebuild (index={_idx_n}, KG={_kg_n}); rebuilding từ Neo4j KG...") kg_locations = kg_service.search_locations_by_name('', limit=5000) if kg_locations and vector_service.build_index(kg_locations): logger.info("ChromaDB index rebuilt successfully from KG data") except Exception as e: logger.warning(f"ChromaDB index inspection/rebuild skipped: {e}") else: logger.warning("ChromaDB index not found") # Try to build index from KG data if kg_service: try: logger.info("Building ChromaDB index from Knowledge Graph data...") kg_locations = kg_service.search_locations_by_name('', limit=5000) if kg_locations and vector_service.build_index(kg_locations): logger.info("ChromaDB index built successfully from KG data") else: logger.warning("Failed to build ChromaDB index") except Exception as e: logger.error(f"Failed to build ChromaDB index: {e}") except Exception as e: logger.error(f"Failed to initialize ChromaDB service: {e}") vector_service = None # Initialize PhoBERT Vector Search service try: phobert_service = PhoBERTVectorSearchService() # Try to load existing index if phobert_service.load_index(): logger.info("PhoBERT vector search service initialized and loaded") # Rebuild nếu index cũ thiếu metadata schema mới (ward/province riêng) try: sample = phobert_service.collection.get(limit=5, include=["metadatas"]) if phobert_service.collection else None metas = (sample or {}).get("metadatas") or [] ok = any(isinstance(m, dict) and (m.get("ward") or "").strip() for m in metas) _idx_n = phobert_service.collection.count() if phobert_service.collection else 0 _kg_n = _kg_location_count(kg_service) if kg_service else None _stale = (_kg_n is not None and _idx_n < _kg_n) if (not ok or _stale) and kg_service: logger.warning(f"PhoBERT index cần rebuild (index={_idx_n}, KG={_kg_n}); rebuilding từ KG...") kg_locations = kg_service.search_locations_by_name('', limit=5000) if kg_locations and phobert_service.build_index(kg_locations): logger.info("PhoBERT index rebuilt successfully") except Exception as e: logger.warning(f"PhoBERT index inspection/rebuild skipped: {e}") else: logger.warning("PhoBERT index not found") # Try to build index from KG data if kg_service: try: logger.info("Building PhoBERT index from Knowledge Graph data...") kg_locations = kg_service.search_locations_by_name('', limit=5000) if kg_locations and phobert_service.build_index(kg_locations): logger.info("PhoBERT index built successfully from KG data") else: logger.warning("Failed to build PhoBERT index") except Exception as e: logger.error(f"Failed to build PhoBERT index: {e}") except Exception as e: logger.error(f"Failed to initialize PhoBERT service: {e}") phobert_service = None # Initialize Intent Classification service try: intent_service = get_intent_service() service_info = intent_service.get_service_info() logger.info(f"Intent service initialized: {service_info['method']}") except Exception as e: logger.error(f"Failed to initialize Intent service: {e}") intent_service = None # Initialize STT (Speech-to-Text) service - PhoWhisper (Vietnamese optimized, no API key needed) try: stt_service = get_phowhisper_service("small") # Use 'small' for balance of speed/accuracy stt_info = stt_service.get_service_info() logger.info(f"STT service initialized: {stt_info['model']} ({stt_info['language']})") logger.info(f" Parameters: {stt_info['parameters']} (Vietnamese optimized)") logger.info(f" Type: {stt_info['type']} (API Required: {stt_info['api_required']})") logger.info(f" Offline Capable: {stt_info['offline_capable']}") logger.info(f" Tourism Optimized: {stt_info['features']['vietnamese_optimized']}") except Exception as e: logger.error(f"Failed to initialize PhoWhisper STT service: {e}") logger.warning("STT service will be unavailable. Check if transformers is installed.") stt_service = None # Initialize NER service try: ner_service = get_ner_service() if ner_service: service_info = ner_service.get_service_info() logger.info(f"NER service initialized: {service_info.get('status')}") else: raise Exception("NER service factory returned None") except Exception as e: logger.error(f"Failed to initialize NER service: {e}") ner_service = None @app.on_event("shutdown") async def shutdown_event(): """Clean up on shutdown""" global kg_service if kg_service: kg_service.close() @app.get("/") async def root(): return {"message": "RAG Tourism Recommender API v1.0", "status": "active"} @app.head("/") async def root_head(): """HF Space healthcheck dùng HEAD — trả 200 thay vì 405 spam log.""" return None @app.get("/health") async def health_check(): return { "status": "healthy", "version": "1.0.0", "kg_service": kg_service is not None, "vector_service": vector_service is not None and hasattr(vector_service, 'collection') and vector_service.collection is not None, "phobert_service": phobert_service is not None and hasattr(phobert_service, 'collection') and phobert_service.collection is not None, "intent_service": intent_service is not None, "stt_service": stt_service is not None, "ner_service": ner_service is not None } @app.get("/debug") async def debug_services(): """Debug endpoint to check service status""" debug_info = { "kg_service_status": kg_service is not None, "vector_service_status": vector_service is not None, "intent_service_status": intent_service is not None, } if vector_service: debug_info.update({ "vector_service_client": vector_service.client is not None, "vector_service_collection": vector_service.collection is not None, "vector_service_hasattr": hasattr(vector_service, 'collection'), "vector_service_condition": hasattr(vector_service, 'collection') and vector_service.collection is not None }) if intent_service: debug_info.update({ "intent_service_info": intent_service.get_service_info() }) # Try to get stats try: stats = vector_service.get_stats() debug_info["vector_service_stats"] = stats except Exception as e: debug_info["vector_service_stats_error"] = str(e) # PhoBERT (primary) index count — để kiểm tra index chính đã có data mới chưa try: if phobert_service and phobert_service.collection is not None: debug_info["phobert_index_count"] = phobert_service.collection.count() except Exception as e: debug_info["phobert_index_error"] = str(e) return debug_info @app.get("/stats") async def get_system_stats(): """Get system statistics""" if not kg_service: raise HTTPException(status_code=503, detail="Knowledge Graph service not available") try: stats = kg_service.get_database_stats() return {"database_stats": stats} except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get stats: {str(e)}") # Pydantic models for intent class IntentQuery(BaseModel): text: str class IntentResponse(BaseModel): text: str predicted_intent: str confidence: float method: str description: str all_probabilities: Optional[dict] = None @app.post("/classify-intent", response_model=IntentResponse) async def classify_intent(query: IntentQuery): """ Classify intent of Vietnamese tourism query """ if not intent_service: raise HTTPException(status_code=503, detail="Intent service not available") try: result = intent_service.predict_intent(query.text) return IntentResponse(**result) except Exception as e: raise HTTPException(status_code=500, detail=f"Intent classification failed: {str(e)}") @app.get("/intents") async def get_available_intents(): """Get available intent categories""" if not intent_service: raise HTTPException(status_code=503, detail="Intent service not available") try: categories = intent_service.get_intent_categories() service_info = intent_service.get_service_info() return { "categories": categories, "service_info": service_info } except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get intents: {str(e)}") # Pydantic models for NER class NERQuery(BaseModel): text: str class NEREntity(BaseModel): entity_group: str score: float word: str start: Optional[int] = None end: Optional[int] = None class NERResponse(BaseModel): text: str entities: List[NEREntity] service_info: dict @app.post("/extract-entities", response_model=NERResponse) async def extract_entities(query: NERQuery): """ Extract named entities from Vietnamese tourism query """ if not ner_service: raise HTTPException(status_code=503, detail="NER service not available") try: entities = ner_service.extract_entities(query.text) service_info = ner_service.get_service_info() # Pydantic validation is strict, so we need to ensure all fields are present validated_entities = [ NEREntity( entity_group=e.get('entity_group'), score=e.get('score'), word=e.get('word'), start=e.get('start'), end=e.get('end') ) for e in entities ] return NERResponse( text=query.text, entities=validated_entities, service_info=service_info ) except Exception as e: logger.error(f"NER extraction failed: {e}") raise HTTPException(status_code=500, detail=f"NER extraction failed: {str(e)}") @app.post("/search", response_model=SearchResponse) async def search_locations(query: TourismQuery, background_tasks: BackgroundTasks = None): """ Search for tourism locations using hybrid approach (KG + Vector Search + Intent Classification) """ start_time = time.time() results = [] search_method = "hybrid" ner_entities = [] # Step -1: Text Normalization (khối chuẩn hóa trong Fig. 2 của bài báo) # Bung viết tắt (tp.hcm, q.1...) + phục hồi dấu địa danh (quan binh thanh # -> Quận Bình Thạnh) TRƯỚC khi vào NER/Intent. from .services.text_normalizer import get_text_normalizer norm = get_text_normalizer().normalize(query.text) processed_text = norm.text if norm.changed: logger.info(f"Normalized query: {query.text!r} -> {processed_text!r} | {norm.changes}") # Step 0: Extract Named Entities (NER) main_search_term = processed_text location_filter = query.location if ner_service: try: extracted_entities = ner_service.extract_entities(processed_text) # Convert numpy types in NER entities to Python native types for Pydantic serialization ner_entities = [ { 'entity_group': e.get('entity_group'), 'score': float(e.get('score')) if e.get('score') is not None else None, 'word': e.get('word'), 'start': int(e.get('start')) if e.get('start') is not None else None, 'end': int(e.get('end')) if e.get('end') is not None else None } for e in extracted_entities ] logger.info(f"NER extracted: {extracted_entities}") # Prioritize TOURISM entity for search term — nhưng CHỈ khi entity # đủ dài để mang nghĩa. NER đôi khi cắt cụt ("Nhà thờ cổ kính" -> # chỉ "Nhà"): đem 1 từ đi vector search sẽ trả 0 kết quả; khi đó # giữ nguyên câu gốc tốt hơn. tourism_entities = [e['word'] for e in extracted_entities if e['entity_group'] == 'TOURISM'] if tourism_entities: candidate = " ".join(tourism_entities).strip() if len(candidate.split()) >= 2 and len(candidate) >= 6: main_search_term = candidate logger.info(f"Using NER TOURISM entities as main search term: '{main_search_term}'") else: logger.info( f"TOURISM entity '{candidate}' quá ngắn -> giữ nguyên câu gốc làm search term" ) # For administrative queries, extract ward names ward_entities = [e['word'] for e in extracted_entities if e['entity_group'] in ['OLD_WARD', 'NEW_WARD']] if ward_entities: # Use the first ward name found ward_search_term = _clean_ward_term(ward_entities[0]) logger.info(f"Found ward entity for admin query: '{ward_search_term}'") # Use location entities for filtering location_entities = [e['word'] for e in extracted_entities if e['entity_group'] in ['PROVINCE', 'DISTRICT', 'CITY']] if location_entities and not location_filter: # Only override if user didn't specify a location location_filter = " ".join(location_entities) logger.info(f"Using NER LOCATION entities as filter: '{location_filter}'") except Exception as e: logger.warning(f"NER extraction failed during search: {e}") try: # Step 1: Classify intent (optional but helpful for routing) intent_info = None if intent_service: try: intent_info = intent_service.predict_intent(processed_text) logger.info(f"Intent classified: {intent_info['predicted_intent']} (confidence: {intent_info['confidence']:.3f})") # Adjust search strategy based on intent # Ngưỡng tin cậy: intent ~0% là argmax chọn bừa khi không có # tín hiệu nào — KHÔNG được tin để rẽ nhánh admin (vd câu # "Quán cà phê view đẹp Biên Hòa" từng bị gán ask_admin_change 0%). MIN_INTENT_CONFIDENCE = 0.15 if intent_info['confidence'] < MIN_INTENT_CONFIDENCE: logger.info( f"Intent confidence {intent_info['confidence']:.2f} < " f"{MIN_INTENT_CONFIDENCE} -> bỏ qua routing admin, mặc định tourism" ) elif not intent_service.is_tourism_intent(intent_info['predicted_intent']): logger.info("Non-tourism intent detected - focusing on administrative data") search_method = "administrative_focused" except Exception as e: logger.warning(f"Intent classification failed: {e}") # Step 2: Handle administrative queries (non-tourism intents) ward_search_term = locals().get('ward_search_term', main_search_term) # Fallback to main if not found # Normalize ward name by removing common prefixes import re ward_name_clean = re.sub(r'^(Phường|Xã|Thị trấn|Thị xã)\s+', '', ward_search_term, flags=re.IGNORECASE).strip() if (intent_info and intent_info.get('confidence', 0) >= 0.15 and not intent_service.is_tourism_intent(intent_info['predicted_intent']) and kg_service): logger.info(f"Processing administrative query with intent: {intent_info['predicted_intent']}") logger.info(f"Ward search term: '{ward_search_term}' -> normalized: '{ward_name_clean}'") try: # Cấp TỈNH trước (vd "Bình Phước thuộc tỉnh nào sau sáp nhập?"). # CHỈ chạy khi câu thực sự hỏi hành chính và KHÔNG chứa từ khóa # du lịch — tránh "thác nước đẹp ở Đồng Nai" (lỡ bị định tuyến # admin) trả nhầm ánh xạ tỉnh thay vì gợi ý thác. _q_low = processed_text.lower() _has_admin_cue = re.search( r"(?i)(sáp nhập|sap nhap|thuộc tỉnh|thuoc tinh|đổi tên|tên mới|" r"tên cũ|hợp nhất|sắp xếp|trực thuộc|tỉnh nào|còn tỉnh)", _q_low) _has_tourism_kw = any(k in _q_low for k in [ "thác", "suối", "hồ", "chùa", "đền", "miếu", "núi", "biển", "đảo", "chợ", "công viên", "bảo tàng", "nhà thờ", "du lịch", "tham quan", "chơi", "ăn", "quán", "địa điểm", "khách sạn", "nhà hàng", "resort", "homestay", "nhà nghỉ", "lưu trú", "ẩm thực", "cà phê", "cafe", "quán ăn"]) province_term = next( (e.get('word') for e in (ner_entities or []) if e.get('entity_group') == 'PROVINCE'), None ) if province_term and _has_admin_cue and not _has_tourism_kw: pm = kg_service.get_province_mapping(province_term) if pm: if pm['unchanged']: summary = (f"Tỉnh {pm['old_name']} giữ nguyên tên sau sáp nhập " f"(không đổi).") else: summary = (f"Tỉnh {pm['old_name']} nay thuộc tỉnh/thành " f"{pm['new_name']} sau sáp nhập 2025.") results.append(LocationResponse( name=summary, lat=None, lng=None, ward='', province=pm['new_name'], confidence=0.97, source="administrative_kg", )) others = [o for o in pm['merged_from'] if o.strip().lower() != pm['old_name'].strip().lower()] if others: results.append(LocationResponse( name=f"Tỉnh {pm['new_name']} mới gồm: " + ", ".join(pm['merged_from']), lat=None, lng=None, ward='', province=pm['new_name'], confidence=0.9, source="administrative_kg", )) return SearchResponse( query=query.text, results=results, total=len(results), processing_time=time.time() - start_time, search_method=search_method, intent_info=intent_info, ner_entities=ner_entities, ) # "Phường X gồm/gộp từ (những) phường/xã nào" — hỏi X (tên MỚI) # được gộp từ các đơn vị cũ nào. Ép về nhánh ask_admin_change để # tra thành phần, tránh bị coi là ánh xạ cũ->mới rồi khớp nhầm. if re.search(r"(?i)\b(gồm|gộp|bao gồm|hợp nhất từ|sáp nhập từ)\b", processed_text): intent_info['predicted_intent'] = 'ask_admin_change' if intent_info['predicted_intent'] == 'ask_old_name': # Query: "Phường X cũ tên là gì?" old_names_result = kg_service.get_old_ward_names(ward_name_clean) if old_names_result: for ward_info in old_names_result: # Create a summary response old_names_list = [f"{u['type']} {u['name']}" for u in ward_info['old_names']] summary = f"Trước đây gồm: {', '.join(old_names_list)}" results.append(LocationResponse( name=summary, lat=None, lng=None, ward=ward_info['new_ward_name'], province='', confidence=0.95, source="administrative_kg" )) elif intent_info['predicted_intent'] == 'ask_new_name': # Query: "Phường X mới tên là gì?" new_name_result = kg_service.get_new_ward_name(ward_name_clean) if new_name_result: summary = f"{new_name_result['old_type']} {new_name_result['old_name']} → {new_name_result['new_type']} {new_name_result['new_name']}" if new_name_result.get('merged_into'): summary += " (đã sáp nhập)" results.append(LocationResponse( name=summary, lat=None, lng=None, ward=new_name_result['district'], province=new_name_result['province'], confidence=0.95, source="administrative_kg" )) elif intent_info['predicted_intent'] == 'ask_location_mapping': # "Phường X giờ là phường nào / nay thuộc đâu?" — bản chất # là hỏi ánh xạ cũ -> mới, tái dùng get_new_ward_name. mapping = kg_service.get_new_ward_name(ward_name_clean) if mapping: summary = ( f"{mapping['old_type']} {mapping['old_name']} nay là " f"{mapping['new_type']} {mapping['new_name']}" ) if mapping.get('province'): summary += f", thuộc {mapping['province']}" results.append(LocationResponse( name=summary, lat=None, lng=None, ward=mapping.get('district') or '', province=mapping.get('province') or '', confidence=0.95, source="administrative_kg" )) elif intent_info['predicted_intent'] == 'ask_admin_change': # Query: "Phường X có bao nhiêu xã phường?" or "Phường X được gộp từ những phường nào?" ward_details = kg_service.get_ward_merger_details_with_locations(ward_name_clean) if ward_details: # Main ward summary old_names_summary = ", ".join([f"{w['old_type']} {w['old_name']}" for w in ward_details['old_wards'][:3]]) if ward_details['total_old_wards'] > 3: old_names_summary += f" và {ward_details['total_old_wards'] - 3} đơn vị khác" main_summary = f"{ward_details['new_ward_type']} {ward_details['new_ward_name']}" main_summary += f" (được gộp từ: {old_names_summary})" results.append(LocationResponse( name=main_summary, lat=None, lng=None, ward=ward_details['district'], province=ward_details['province'], confidence=0.95, source="administrative_kg" )) # For each old ward for old_ward in ward_details['old_wards']: # Old ward header old_ward_header = f"├─ {old_ward['old_type']} {old_ward['old_name']}" if old_ward['tourism_locations']: old_ward_header += f" ({len(old_ward['tourism_locations'])} địa điểm)" else: old_ward_header += " (không có địa điểm du lịch)" results.append(LocationResponse( name=old_ward_header, lat=None, lng=None, ward=ward_details['district'], province=ward_details['province'], confidence=0.9, source="administrative_old_ward" )) # Tourism locations in this old ward (limit to 3 per ward) for i, loc in enumerate(old_ward['tourism_locations'][:3]): is_last_location = (i == len(old_ward['tourism_locations'][:3]) - 1) tree_char = " └─" if is_last_location else " ├─" results.append(LocationResponse( name=f"{tree_char} 🏛️ {loc['name']}", lat=float(loc['lat']) if loc['lat'] else None, lng=float(loc['lng']) if loc['lng'] else None, ward=ward_details['district'], province=ward_details['province'], confidence=0.85, source="administrative_tourism" )) # Show count if more locations exist if len(old_ward['tourism_locations']) > 3: remaining = len(old_ward['tourism_locations']) - 3 results.append(LocationResponse( name=f" └─ ... và {remaining} địa điểm khác", lat=None, lng=None, ward=ward_details['district'], province='', confidence=0.8, source="administrative_tourism_summary" )) # District/city-level fallback: cấp quận/huyện đã bị BÃI BỎ sau # sáp nhập 01/07/2025 nên không có "tên mới" cho quận — câu trả # lời đúng là danh sách phường mới mà các phường cũ thuộc quận # đó được gộp vào (vd: "Quận Bình Thạnh cũ tên mới là gì?"). # Chỉ kích hoạt fallback quận khi câu có TÍN HIỆU hành chính rõ # ràng — tránh câu du lịch chứa tên địa danh ("cà phê Biên Hòa") # bị trả nhầm bảng sáp nhập. _admin_cues = re.search( r'(?i)(sáp nhập|sap nhap|hợp nhất|chia tách|đổi tên|tên mới|tên cũ|' r'gộp|địa giới|hành chính|phường|xã\b|quận|huyện|thị xã|trực thuộc)', processed_text, ) if not results and _admin_cues: district_term = None # 1) NER entity for e in (ner_entities or []): if e.get('entity_group') in ('DISTRICT', 'CITY'): district_term = e.get('word') break # 2) Fuzzy match bỏ dấu/thường hóa với danh sách quận/huyện # trong KG — bắt được cả câu gõ thiếu dấu/sai hoa thường # (vd: "quân binh thanh" ≈ "Quận Bình Thạnh"). if not district_term: global _district_names_cache if _district_names_cache is None: try: _district_names_cache = kg_service.get_all_district_names() except Exception: _district_names_cache = [] norm_q = _strip_diacritics(processed_text) def _wb(needle: str, hay: str) -> bool: # khớp theo RANH GIỚI TỪ, tránh 'an bien' ⊂ 'tran bien' return re.search( r'(?= 6 and _wb(nd_bare, norm_q)): if best is None or len(nd) > len(_strip_diacritics(best)): best = dname district_term = best # 3) Regex thô cuối cùng if not district_term: m = re.search( r'(?i)\b((?:quận|huyện|thị xã)\s+[\wÀ-ỹ]+(?:\s+[\wÀ-ỹ]+)?)', processed_text, ) if m: district_term = m.group(1) if district_term: mappings = kg_service.get_district_mapping(district_term) if mappings: district_label = mappings[0].get('district_city') or district_term province = mappings[0].get('province') or '' results.append(LocationResponse( name=(f"{district_label} (cũ): cấp quận/huyện đã được bãi bỏ " f"sau sáp nhập 01/07/2025. {len(mappings)} phường/xã cũ " f"nay thuộc các đơn vị hành chính mới sau:"), lat=None, lng=None, ward=district_label, province=province, confidence=0.95, source="administrative_kg", )) def _unit_label(utype, uname): # tránh "Phường Phường 1" khi tên đã chứa sẵn loại đơn vị utype = (utype or '').strip() uname = (uname or '').strip() if utype and not uname.lower().startswith(utype.lower()): return f"{utype.capitalize()} {uname}" return uname grouped = {} for mp in mappings: key = _unit_label(mp.get('new_type'), mp.get('new_name')) old = _unit_label(mp.get('old_type'), mp.get('old_name')) grouped.setdefault(key, []).append(old) for new_ward, olds in grouped.items(): shown = ", ".join(olds[:4]) if len(olds) > 4: shown += f" và {len(olds) - 4} đơn vị khác" results.append(LocationResponse( name=f"├─ {new_ward} ← gộp từ: {shown}", lat=None, lng=None, ward=district_label, province=province, confidence=0.9, source="administrative_district", )) # If we got administrative results, skip tourism search if results: logger.info(f"Returned {len(results)} administrative results") return SearchResponse( query=query.text, results=results, total=len(results), processing_time=time.time() - start_time, search_method=search_method, intent_info=intent_info, ner_entities=ner_entities ) except Exception as e: logger.error(f"Administrative query failed: {e}") # Continue to tourism search as fallback # Nhánh admin không ra kết quả -> rơi xuống tìm du lịch; phải đổi # search_method để frontend không render POI vào panel hành chính. if not results and search_method == "administrative_focused": search_method = "hybrid_fallback" # Step 3a: Truy vấn theo ĐỊA GIỚI — nếu câu nhắc tới phường/xã cụ thể # ("Tân Triều có gì chơi"), lấy POI nằm trong phường đó từ KG trước, # và dùng tên phường làm location_filter cho vector search phía dưới. ward_entities_t = [ e.get('word') for e in (ner_entities or []) if e.get('entity_group') in ('OLD_WARD', 'NEW_WARD') ] if ward_entities_t and kg_service: ward_bare = re.sub( r'(?i)^(phường|xã|thị trấn|thị xã)\s+', '', _clean_ward_term(ward_entities_t[0]) ).strip() if ward_bare: if not location_filter: location_filter = ward_bare try: for loc in kg_service.get_locations_in_ward(ward_bare, limit=8): if loc.get('name'): results.append(LocationResponse( name=loc['name'], lat=float(loc['lat']) if loc.get('lat') is not None else None, lng=float(loc['lng']) if loc.get('lng') is not None else None, ward=loc.get('ward') or '', province=loc.get('province') or '', confidence=0.92, source="kg_ward_match", )) if results: logger.info( f"KG ward match: {len(results)} POI trong phường/xã '{ward_bare}'" ) except Exception as e: logger.warning(f"KG ward lookup failed: {e}") # Step 3b: Tìm theo TỪ KHÓA LOẠI HÌNH + tỉnh ("suối/thác/chùa... ở Đồng Nai"). # Bắt từ khóa từ NER TOURISM hoặc từ danh sách head-noun trong câu. if kg_service and len(results) < 5: TOURISM_KEYWORDS = [ "thác", "suối", "hồ", "chùa", "đền", "miếu", "đình", "lăng", "núi", "biển", "đảo", "chợ", "công viên", "bảo tàng", "nhà thờ", "khu du lịch", "vườn quốc gia", "thiền viện", "căn cứ", "di tích", "khách sạn", "nhà hàng", "quán ăn", "resort", "homestay", ] low = processed_text.lower() kw = next((k for k in sorted(TOURISM_KEYWORDS, key=len, reverse=True) if k in low), None) if kw: # Ưu tiên tỉnh từ NER (location_filter); nếu rỗng -> dò trong câu prov = location_filter or _detect_province(processed_text, kg_service) if prov and not location_filter: location_filter = prov # để vector search phía dưới cũng lọc theo tỉnh try: existing = {r.name.lower() for r in results} for loc in kg_service.search_by_keyword(kw, province=prov, limit=10): if loc.get('name') and loc['name'].lower() not in existing: results.append(LocationResponse( name=loc['name'], lat=float(loc['lat']) if loc.get('lat') is not None else None, lng=float(loc['lng']) if loc.get('lng') is not None else None, ward=loc.get('ward') or '', province=loc.get('province') or '', confidence=0.9, category=loc.get('category') or 'Điểm tham quan', source="kg_keyword_match", )) if results: logger.info(f"KG keyword match: '{kw}' (tỉnh={prov}) -> {len(results)} kết quả") except Exception as e: logger.warning(f"KG keyword search failed: {e}") # Step 2.5: Nếu câu hỏi nêu rõ LOẠI (khách sạn/nhà hàng), tra thẳng KG # theo category + địa điểm — không phụ thuộc ngưỡng vector (embedding # yếu với câu ngắn), bảo đảm ra đúng kết quả. _cat_wanted = _detect_category(processed_text) if _cat_wanted and kg_service: try: _loc = location_filter or _detect_province(processed_text, kg_service) or '' existing = {r.name.lower() for r in results} cat_rows = kg_service.search_by_category(_cat_wanted, _loc, limit=15) # Nếu lọc theo địa điểm ra rỗng, thử lại không kèm địa điểm if not cat_rows and _loc: cat_rows = kg_service.search_by_category(_cat_wanted, '', limit=15) for loc in cat_rows: if loc.get('name') and loc['name'].lower() not in existing: results.append(LocationResponse( name=loc['name'], lat=float(loc['lat']) if loc.get('lat') is not None else None, lng=float(loc['lng']) if loc.get('lng') is not None else None, ward=loc.get('ward') or loc.get('district_city') or '', province=loc.get('province') or '', confidence=0.9, category=loc.get('category') or _cat_wanted, source="kg_category", )) logger.info(f"KG category '{_cat_wanted}' (loc={_loc}) -> {len(cat_rows)} kết quả") except Exception as e: logger.warning(f"KG category search failed: {e}") # Step 3: Knowledge Graph search (for tourism locations) if kg_service: kg_results = kg_service.search_locations_by_name(main_search_term, limit=10) for loc in kg_results: if loc.get('name'): results.append(LocationResponse( name=loc['name'], lat=loc.get('lat'), lng=loc.get('lng'), ward=loc.get('ward', ''), province=loc.get('province', ''), confidence=0.9, # High confidence for exact matches category=loc.get('category') or 'Điểm tham quan', source="kg_search" )) # Method 2: PhoBERT Vector search (PRIMARY - 768 dimensions) if query.use_vector_search and phobert_service and hasattr(phobert_service, 'collection') and phobert_service.collection is not None: phobert_results = phobert_service.get_recommendations( main_search_term, top_k=8, # Increased for better coverage location_filter=location_filter ) # Add PhoBERT results that aren't already in KG results existing_names = {r.name.lower() for r in results} for loc in phobert_results: if loc['name'] and loc['name'].lower() not in existing_names: # Boost PhoBERT confidence for primary usage # Convert numpy.float32 to Python float to fix Pydantic serialization similarity_score = float(loc.get('similarity_score', 0.5)) # Ngưỡng liên quan: chỉ loại kết quả vector RẤT xa nghĩa # (rác thật sự). Embedding này có thang điểm nén ~0.25-0.45 # nên để thấp (0.2); việc phân loại "chính xác/gợi ý" do # frontend đảm nhiệm. Chỉnh qua env MIN_VECTOR_SIMILARITY. if similarity_score < float(_os.getenv("MIN_VECTOR_SIMILARITY", "0.2")): continue boosted_confidence = float(min(0.95, similarity_score + 0.1)) results.append(LocationResponse( name=loc['name'], lat=float(loc.get('lat')) if loc.get('lat') is not None else None, lng=float(loc.get('lng')) if loc.get('lng') is not None else None, ward=loc.get('ward') or loc.get('district_city', ''), province=loc.get('province', ''), confidence=boosted_confidence, category=loc.get('category') or 'Điểm tham quan', source="phobert_768d_primary" )) # Method 3: ChromaDB Vector search (fallback - 384 dimensions) elif query.use_vector_search and vector_service and hasattr(vector_service, 'collection') and vector_service.collection is not None: vector_results = vector_service.get_recommendations( main_search_term, top_k=5, location_filter=location_filter ) # Add vector results that aren't already in KG results existing_names = {r.name.lower() for r in results} for loc in vector_results: # ChromaDB already limits results if loc['name'] and loc['name'].lower() not in existing_names: # Ngưỡng liên quan (đồng bộ với nhánh PhoBERT) if float(loc.get('similarity_score', 0.5)) < float(_os.getenv("MIN_VECTOR_SIMILARITY", "0.2")): continue # Convert numpy types to Python types for Pydantic serialization results.append(LocationResponse( name=loc['name'], lat=float(loc.get('lat')) if loc.get('lat') is not None else None, lng=float(loc.get('lng')) if loc.get('lng') is not None else None, ward=loc.get('ward') or loc.get('district_city', ''), province=loc.get('province', ''), confidence=float(loc.get('similarity_score', 0.5)), category=loc.get('category') or 'Điểm tham quan', source="chroma_384d_fallback" )) # If no results, try broader search if not results and kg_service and query.location: province_results = kg_service.search_locations_by_province(query.location, limit=5) for loc in province_results: results.append(LocationResponse( name=loc['location_name'], lat=loc.get('latitude'), lng=loc.get('longitude'), ward=loc.get('ward_name', ''), province=loc.get('province_name', ''), confidence=0.7, category='Điểm tham quan', source="province_search" )) # Lọc theo LOẠI nếu câu hỏi nhắm rõ (vd "khách sạn ở quận 1" -> chỉ # giữ Khách sạn). Nếu lọc ra rỗng thì giữ nguyên để không trả 0 kết quả. _wanted_cat = _detect_category(processed_text) if _wanted_cat: filtered = [r for r in results if _same_category(r.category or '', _wanted_cat)] if filtered: results = filtered logger.info(f"Lọc theo loại '{_wanted_cat}' -> {len(results)} kết quả") # Sort: ưu tiên MẠNH các điểm ở TP. Hồ Chí Minh lên đầu, # sau đó mới đến độ liên quan (confidence). results.sort(key=lambda x: (_is_hcm(x.province), x.confidence), reverse=True) except Exception as e: logger.error(f"Search error: {e}") raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}") processing_time = time.time() - start_time return SearchResponse( query=query.text, results=results[:10], # Limit to top 10 total=len(results), processing_time=processing_time, search_method=search_method, intent_info=intent_info, ner_entities=ner_entities ) # STT (Speech-to-Text) Endpoints from .models.stt_models import STTTranscriptionResponse, STTServiceInfo, STTHealthCheck @app.post("/stt/transcribe", response_model=STTTranscriptionResponse) async def transcribe_audio( audio_file: UploadFile = File(..., description="Audio file to transcribe"), optimize_for_tourism: bool = Form(True, description="Apply Vietnamese tourism optimization"), include_segments: bool = Form(True, description="Include word-level segments"), noise_reduction: bool = Form(True, description="Apply noise reduction preprocessing") ): """ Transcribe audio file to text using Whisper Large v3 Optimized for Vietnamese tourism queries """ if not stt_service: raise HTTPException( status_code=503, detail="STT service not available. Check OPENAI_API_KEY configuration." ) try: # Validate file type if not audio_file.content_type or not any( fmt in audio_file.content_type.lower() for fmt in ['audio', 'video'] # Accept audio/* and video/* MIME types ): raise HTTPException( status_code=400, detail=f"Invalid file type: {audio_file.content_type}. Please upload an audio file." ) # Read audio data audio_data = await audio_file.read() if len(audio_data) == 0: raise HTTPException(status_code=400, detail="Empty audio file") # Transcribe audio result = await stt_service.transcribe_audio_data( audio_data=audio_data, filename=audio_file.filename or "audio.wav" ) if not result["success"]: raise HTTPException( status_code=500, detail=f"Transcription failed: {result.get('error', 'Unknown error')}" ) # Add file info and model info to response file_info = { "filename": audio_file.filename, "content_type": audio_file.content_type, "size_bytes": len(audio_data) } model_info = stt_service.get_service_info() return STTTranscriptionResponse( success=result["success"], transcription=result.get("transcription"), language=result.get("language"), confidence=result.get("confidence"), duration=result.get("duration"), processing_time=result["processing_time"], segments=[ { "text": seg["text"], "start": seg["start"], "end": seg["end"], "confidence": seg.get("confidence") } for seg in result.get("segments", []) ], error=result.get("error"), file_info=file_info, model_info=model_info ) except HTTPException: raise except Exception as e: logger.error(f"STT endpoint error: {e}") raise HTTPException(status_code=500, detail=f"STT processing failed: {str(e)}") @app.get("/stt/info", response_model=STTServiceInfo) async def get_stt_service_info(): """Get STT service information and capabilities""" if not stt_service: raise HTTPException( status_code=503, detail="STT service not available" ) try: info = stt_service.get_service_info() return STTServiceInfo( service=info["service"], model=info["model"], language=info["language"], max_file_size_mb=info["max_file_size_mb"], max_duration_minutes=info["max_duration_minutes"], supported_formats=info.get("supported_formats", []), # PhoWhisper is a local model: get_service_info() exposes "api_required", # not "api_available". Derive availability defensively to avoid a KeyError. api_available=info.get("api_available", not info.get("api_required", False)), features={ "noise_reduction": True, "vietnamese_optimization": True, "segment_timestamps": True, "confidence_scores": True, "multiple_formats": True } ) except Exception as e: logger.error(f"STT info error: {e}") raise HTTPException(status_code=500, detail=f"Failed to get STT info: {str(e)}") @app.get("/stt/health", response_model=STTHealthCheck) async def check_stt_health(): """Check STT service health and configuration""" import os from datetime import datetime try: # PhoWhisper runs locally and does NOT require an OpenAI API key. # We still surface whether one is configured (for the optional Whisper-API # fallback) but it must NOT affect health status. api_key_configured = bool(os.getenv('OPENAI_API_KEY')) service_available = stt_service is not None # Test temp directory temp_writable = True if stt_service: try: test_file = os.path.join(stt_service.temp_dir, "test_write") with open(test_file, 'w') as f: f.write("test") os.remove(test_file) except Exception: temp_writable = False status = "healthy" if (service_available and temp_writable) else "unhealthy" error_message = None if not service_available: error_message = "STT service not initialized" elif not temp_writable: error_message = "Temporary directory not writable" return STTHealthCheck( service_name="PhoWhisperSTTService", status=status, api_key_configured=api_key_configured, model_accessible=service_available, temp_directory_writable=temp_writable, last_check=datetime.now().isoformat(), error_message=error_message ) except Exception as e: logger.error(f"STT health check error: {e}") return STTHealthCheck( service_name="WhisperSTTService", status="unhealthy", api_key_configured=False, model_accessible=False, temp_directory_writable=False, last_check=datetime.now().isoformat(), error_message=str(e) ) @app.post("/stt/search", response_model=SearchResponse) async def transcribe_and_search( audio_file: UploadFile = File(..., description="Audio file with tourism query"), location: Optional[str] = Form(None, description="Optional location filter"), use_vector_search: bool = Form(True, description="Enable vector search") ): """ Transcribe audio and immediately search for tourism locations Combines STT + Tourism Search in one endpoint """ if not stt_service: raise HTTPException( status_code=503, detail="STT service not available" ) try: # Step 1: Transcribe audio audio_data = await audio_file.read() stt_result = await stt_service.transcribe_audio_data( audio_data=audio_data, filename=audio_file.filename or "audio.wav" ) if not stt_result["success"]: raise HTTPException( status_code=500, detail=f"Transcription failed: {stt_result.get('error', 'Unknown error')}" ) transcribed_text = stt_result.get("transcription", "") if not transcribed_text.strip(): raise HTTPException( status_code=400, detail="No speech detected in audio file" ) # Step 2: Search using transcribed text query = TourismQuery( text=transcribed_text, location=location, use_vector_search=use_vector_search ) # Use the existing search logic search_response = await search_locations(query) # Add STT metadata to response search_response.query = f"🎤 {transcribed_text}" # Indicate voice input search_response.processing_time += stt_result["processing_time"] return search_response except HTTPException: raise except Exception as e: logger.error(f"STT search error: {e}") raise HTTPException(status_code=500, detail=f"STT search failed: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)