Spaces:
Paused
Paused
| """ | |
| HuggingFace Sentiment Analysis | |
| ───────────────────────────── | |
| เรียก HuggingFace Inference API เพื่อวิเคราะห์ความรู้สึกของข้อความภาษาไทย | |
| Model ที่แนะนำ: | |
| - airesearch/wangchanberta-base-att-spm-uncased (ภาษาไทย ทั่วไป) | |
| - pythainlp/thai-sentiment-analysis (ถ้ามี fine-tune) | |
| - อาจใช้ multilingual model: cardiffnlp/twitter-xlm-roberta-base-sentiment | |
| """ | |
| import httpx | |
| import logging | |
| from app.config import settings | |
| from app.models import SentimentResult | |
| logger = logging.getLogger(__name__) | |
| HF_API_URL = ( | |
| f"https://api-inference.huggingface.co/models/{settings.HF_MODEL_NAME}" | |
| ) | |
| # Label mapping จาก model output → ชื่อ standard | |
| LABEL_MAP = { | |
| # WangchanBERTa / Thai models | |
| "pos": "positive", | |
| "neu": "neutral", | |
| "neg": "negative", | |
| # cardiffnlp / XLM-RoBERTa | |
| "positive": "positive", | |
| "neutral": "neutral", | |
| "negative": "negative", | |
| # LABEL_0/1/2 fallback (ขึ้นกับ model) | |
| "LABEL_0": "negative", | |
| "LABEL_1": "neutral", | |
| "LABEL_2": "positive", | |
| } | |
| async def analyze_sentiment(text: str) -> SentimentResult: | |
| """ | |
| ส่งข้อความไปวิเคราะห์ที่ HuggingFace แล้วคืน SentimentResult | |
| HuggingFace Inference API คืนรูปแบบ: | |
| [[{"label": "pos", "score": 0.91}, {"label": "neg", "score": 0.05}, ...]] | |
| """ | |
| headers = {"Authorization": f"Bearer {settings.HF_API_TOKEN}"} | |
| payload = {"inputs": text, "options": {"wait_for_model": True}} | |
| try: | |
| async with httpx.AsyncClient(timeout=15.0) as client: | |
| response = await client.post(HF_API_URL, headers=headers, json=payload) | |
| response.raise_for_status() | |
| data = response.json() | |
| # Flatten nested list ที่ HF คืนมา | |
| scores = data[0] if isinstance(data[0], list) else data | |
| # แปลง label และสร้าง raw_scores dict | |
| raw_scores: dict[str, float] = {} | |
| for item in scores: | |
| normalized = LABEL_MAP.get(item["label"], item["label"].lower()) | |
| raw_scores[normalized] = round(item["score"], 4) | |
| # หา label ที่มี score สูงสุด | |
| best_label = max(raw_scores, key=raw_scores.get) | |
| return SentimentResult( | |
| label=best_label, | |
| score=raw_scores[best_label], | |
| raw_scores=raw_scores, | |
| ) | |
| except httpx.HTTPStatusError as e: | |
| logger.error(f"HuggingFace API error {e.response.status_code}: {e.response.text}") | |
| return _fallback_result() | |
| except httpx.TimeoutException: | |
| logger.error("HuggingFace API timeout") | |
| return _fallback_result() | |
| except Exception as e: | |
| logger.error(f"Unexpected error in sentiment analysis: {e}") | |
| return _fallback_result() | |
| def _fallback_result() -> SentimentResult: | |
| """คืน neutral เมื่อ API ล้มเหลว เพื่อไม่ให้ระบบหยุด""" | |
| return SentimentResult( | |
| label="neutral", | |
| score=0.0, | |
| raw_scores={"positive": 0.0, "neutral": 0.0, "negative": 0.0}, | |
| ) | |