Spaces:
Sleeping
Sleeping
| import re | |
| import time | |
| import hashlib | |
| import logging | |
| from typing import List, Dict, Optional, Tuple | |
| from functools import lru_cache | |
| from groq import Groq | |
| from config import ( | |
| ENABLE_QUERY_REWRITING, | |
| MIN_HISTORY_FOR_REWRITE, | |
| MAX_HISTORY_TURNS_FOR_REWRITE, | |
| MAX_CONTENT_LENGTH_FOR_REWRITE, | |
| QUERY_REWRITE_MODEL, | |
| QUERY_REWRITE_TEMPERATURE, | |
| QUERY_REWRITE_MAX_TOKENS, | |
| QUERY_REWRITE_TIMEOUT, | |
| GROQ_API_KEY, | |
| QUERY_REWRITE_CACHE_SIZE, | |
| QUERY_REWRITE_CACHE_TTL_SECONDS | |
| ) | |
| from prompts import QUERY_REWRITE_PROMPT | |
| logger = logging.getLogger(__name__) | |
| _FOLLOW_UP_INDICATORS = frozenset([ | |
| "bu", "bunu", "bunun", "buna", "bunlar", | |
| "o", "onu", "onun", "ona", "onlar", | |
| "su", "sunu", "sunun", | |
| "boyle", "oyle", "soyle", | |
| "burada", "orada", "surada", | |
| "burdaki", "ordaki", | |
| "ayni", "diger", "sonraki", "onceki", | |
| "bahsettigin", "dedin", "soyledin", | |
| "yukardaki", "asagidaki", | |
| ]) | |
| _PRONOUN_REPLACEMENTS = { | |
| "bu": "{topic}", | |
| "bunu": "{topic}'nu", | |
| "bunun": "{topic}'nun", | |
| "buna": "{topic}'na", | |
| "bunlar": "{topic} ve ilgili konular", | |
| "o": "{topic}", | |
| "onu": "{topic}'nu", | |
| "onun": "{topic}'nun", | |
| "ona": "{topic}'na", | |
| "su": "{topic}", | |
| "sunu": "{topic}'nu", | |
| "sunun": "{topic}'nun", | |
| } | |
| class QueryRewriter: | |
| def __init__(self, cache_duration_seconds: int = QUERY_REWRITE_CACHE_TTL_SECONDS): | |
| self._indicators = _FOLLOW_UP_INDICATORS | |
| self._pattern = self._build_pattern() | |
| self._client = None | |
| self._cache: Dict[str, str] = {} | |
| self._cache_ttl: Dict[str, float] = {} | |
| self._max_cache_size: int = QUERY_REWRITE_CACHE_SIZE | |
| self._cache_duration: int = cache_duration_seconds | |
| self._cache_hits: int = 0 | |
| self._cache_misses: int = 0 | |
| def client(self) -> Groq: | |
| if self._client is None: | |
| self._client = Groq(api_key=GROQ_API_KEY, timeout=QUERY_REWRITE_TIMEOUT) | |
| return self._client | |
| def _build_pattern(self) -> re.Pattern: | |
| escaped = [re.escape(w) for w in self._indicators] | |
| return re.compile(r'\b(' + '|'.join(escaped) + r')\b', re.IGNORECASE) | |
| def is_follow_up(self, query: str, chat_history: List[Dict[str, str]]) -> bool: | |
| if not ENABLE_QUERY_REWRITING: | |
| return False | |
| if len(chat_history) < MIN_HISTORY_FOR_REWRITE * 2: | |
| return False | |
| query_lower = query.lower().strip() | |
| words = query_lower.split() | |
| subject_indicators = ['atlas', 'erp', 'sistem', 'modul', 'program', 'uygulama', 'yazılım', 'şirket', 'firma'] | |
| has_explicit_subject = any(subj in query_lower for subj in subject_indicators) | |
| if self._pattern.search(query_lower): | |
| return True | |
| if query_lower.startswith(('peki', 'ya ', 'ee', 'hmm', 'tamam', 'iyi', 'anladım')): | |
| return True | |
| if has_explicit_subject: | |
| return False | |
| if len(words) <= 5: | |
| return True | |
| question_starters = ['ne', 'nasıl', 'neden', 'niçin', 'kim', 'nerede', 'hangi', 'kaç', 'ne?', 'nasıl?'] | |
| if words and words[0].rstrip('?') in [q.rstrip('?') for q in question_starters]: | |
| return True | |
| if len(words) <= 8 and query_lower.endswith('?'): | |
| return True | |
| return False | |
| def format_history(self, chat_history: List[Dict[str, str]]) -> str: | |
| if not chat_history: | |
| return "" | |
| recent = chat_history[-(MAX_HISTORY_TURNS_FOR_REWRITE * 2):] | |
| parts = [] | |
| for msg in recent: | |
| role = "Kullanici" if msg["role"] == "user" else "Asistan" | |
| content = msg.get("content", "")[:MAX_CONTENT_LENGTH_FOR_REWRITE] | |
| parts.append(f"{role}: {content}") | |
| return "\n".join(parts) | |
| def get_last_topic(self, chat_history: List[Dict[str, str]]) -> Optional[str]: | |
| for msg in reversed(chat_history): | |
| if msg["role"] == "user": | |
| return msg.get("content", "")[:100] | |
| return None | |
| def _get_cache_key(self, query: str, chat_history: List[Dict[str, str]]) -> str: | |
| history_part = "|".join(m.get("content", "")[:50] for m in chat_history[-4:]) | |
| raw = f"{query}::{history_part}" | |
| return hashlib.md5(raw.encode("utf-8")).hexdigest() | |
| def _get_from_cache(self, cache_key: str) -> Optional[str]: | |
| if cache_key not in self._cache: | |
| self._cache_misses += 1 | |
| return None | |
| age = time.time() - self._cache_ttl.get(cache_key, 0) | |
| if age >= self._cache_duration: | |
| del self._cache[cache_key] | |
| del self._cache_ttl[cache_key] | |
| self._cache_misses += 1 | |
| logger.debug(f"Cache entry expired after {age:.1f}s") | |
| return None | |
| self._cache_hits += 1 | |
| return self._cache[cache_key] | |
| def _add_to_cache(self, cache_key: str, rewritten: str) -> None: | |
| if len(self._cache) >= self._max_cache_size and cache_key not in self._cache: | |
| oldest_key = min(self._cache_ttl, key=self._cache_ttl.get) | |
| del self._cache[oldest_key] | |
| del self._cache_ttl[oldest_key] | |
| logger.debug("Cache evicted oldest entry due to size limit") | |
| self._cache[cache_key] = rewritten | |
| self._cache_ttl[cache_key] = time.time() | |
| def rewrite(self, query: str, chat_history: List[Dict[str, str]]) -> Tuple[str, dict]: | |
| start_time = time.time() | |
| debug_info = { | |
| "original_query": query, | |
| "is_follow_up": False, | |
| "rewritten_query": query, | |
| "method": "none", | |
| "rewrite_time_ms": 0 | |
| } | |
| if not self.is_follow_up(query, chat_history): | |
| debug_info["rewrite_time_ms"] = int((time.time() - start_time) * 1000) | |
| return query, debug_info | |
| debug_info["is_follow_up"] = True | |
| cache_key = self._get_cache_key(query, chat_history) | |
| cached = self._get_from_cache(cache_key) | |
| if cached is not None: | |
| debug_info["rewritten_query"] = cached | |
| debug_info["method"] = "cache" | |
| debug_info["cache_hit"] = True | |
| debug_info["cache_size"] = len(self._cache) | |
| debug_info["cache_age_seconds"] = round(time.time() - self._cache_ttl.get(cache_key, time.time()), 2) | |
| debug_info["rewrite_time_ms"] = int((time.time() - start_time) * 1000) | |
| return cached, debug_info | |
| rewritten = self._safe_rewrite(query, chat_history, debug_info) | |
| self._add_to_cache(cache_key, rewritten) | |
| debug_info["rewritten_query"] = rewritten | |
| debug_info["cache_hit"] = False | |
| debug_info["cache_size"] = len(self._cache) | |
| debug_info["rewrite_time_ms"] = int((time.time() - start_time) * 1000) | |
| return rewritten, debug_info | |
| def _safe_rewrite(self, query: str, chat_history: List[Dict[str, str]], debug_info: dict) -> str: | |
| try: | |
| rewritten = self._llm_rewrite(query, chat_history) | |
| debug_info["method"] = "llm" | |
| return rewritten | |
| except Exception as e: | |
| logger.warning(f"LLM rewrite failed: {e}") | |
| debug_info["llm_error"] = str(e) | |
| try: | |
| rewritten = self._rule_based_rewrite(query, chat_history) | |
| if rewritten != query: | |
| debug_info["method"] = "rule" | |
| return rewritten | |
| except Exception as e: | |
| logger.warning(f"Rule-based rewrite failed: {e}") | |
| try: | |
| rewritten = self._append_context_fallback(query, chat_history) | |
| if rewritten != query: | |
| debug_info["method"] = "append" | |
| return rewritten | |
| except Exception as e: | |
| logger.warning(f"Append context failed: {e}") | |
| debug_info["method"] = "fallback_failed" | |
| return query | |
| def _llm_rewrite(self, query: str, chat_history: List[Dict[str, str]]) -> str: | |
| history_str = self.format_history(chat_history) | |
| prompt = QUERY_REWRITE_PROMPT.format( | |
| chat_history=history_str, | |
| original_query=query | |
| ) | |
| response = self.client.chat.completions.create( | |
| model=QUERY_REWRITE_MODEL, | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=QUERY_REWRITE_TEMPERATURE, | |
| max_tokens=QUERY_REWRITE_MAX_TOKENS | |
| ) | |
| rewritten = response.choices[0].message.content.strip() | |
| rewritten = self._clean_response(rewritten) | |
| if not rewritten or len(rewritten) < 3: | |
| return query | |
| return rewritten | |
| def _clean_response(self, response: str) -> str: | |
| lines = response.split('\n') | |
| result = lines[0].strip() | |
| for prefix in ['soru:', 'yeniden yazilmis soru:', 'cevap:']: | |
| if result.lower().startswith(prefix): | |
| result = result[len(prefix):].strip() | |
| result = result.strip('"\'') | |
| return result | |
| def _rule_based_rewrite(self, query: str, chat_history: List[Dict[str, str]]) -> str: | |
| topic = self._extract_topic(chat_history) | |
| if not topic: | |
| return query | |
| result = query | |
| query_lower = query.lower() | |
| for pronoun, replacement in _PRONOUN_REPLACEMENTS.items(): | |
| pattern = re.compile(r'\b' + re.escape(pronoun) + r'\b', re.IGNORECASE) | |
| if pattern.search(query_lower): | |
| filled = replacement.format(topic=topic) | |
| result = pattern.sub(filled, result, count=1) | |
| break | |
| return result | |
| def _append_context_fallback(self, query: str, chat_history: List[Dict[str, str]]) -> str: | |
| topic = self._extract_topic(chat_history) | |
| if not topic: | |
| return query | |
| return f"{topic} hakkinda: {query}" | |
| def _extract_topic(self, chat_history: List[Dict[str, str]]) -> Optional[str]: | |
| last_q = self.get_last_topic(chat_history) | |
| if not last_q: | |
| return None | |
| last_q = last_q.strip().rstrip('?') | |
| words = last_q.split() | |
| stop_words = {'nedir', 'nelerdir', 'nasıl', 'ne', 'kadar', 'mi', 'mu'} | |
| topic_words = [w for w in words if w.lower() not in stop_words] | |
| if topic_words: | |
| return ' '.join(topic_words[:4]) | |
| return last_q[:50] | |
| def clear_cache(self): | |
| self._cache.clear() | |
| self._cache_ttl.clear() | |
| def get_cache_stats(self) -> Dict[str, any]: | |
| now = time.time() | |
| ages = [now - ts for ts in self._cache_ttl.values()] if self._cache_ttl else [] | |
| total_requests = self._cache_hits + self._cache_misses | |
| return { | |
| "total_entries": len(self._cache), | |
| "max_size": self._max_cache_size, | |
| "ttl_seconds": self._cache_duration, | |
| "hit_rate": round(self._cache_hits / total_requests, 4) if total_requests > 0 else 0.0, | |
| "cache_hits": self._cache_hits, | |
| "cache_misses": self._cache_misses, | |
| "oldest_entry_age": round(max(ages), 2) if ages else 0.0, | |
| "newest_entry_age": round(min(ages), 2) if ages else 0.0, | |
| } | |
| def warm_cache(self, common_queries: List[Tuple[str, str]]) -> None: | |
| for query, rewrite in common_queries: | |
| cache_key = hashlib.md5(f"{query}::".encode("utf-8")).hexdigest() | |
| self._add_to_cache(cache_key, rewrite) | |
| def reset_metrics(self) -> None: | |
| self._cache_hits = 0 | |
| self._cache_misses = 0 | |
| _rewriter_instance: Optional[QueryRewriter] = None | |
| def get_query_rewriter() -> QueryRewriter: | |
| global _rewriter_instance | |
| if _rewriter_instance is None: | |
| _rewriter_instance = QueryRewriter() | |
| return _rewriter_instance | |
| def reset_query_rewriter(): | |
| global _rewriter_instance | |
| _rewriter_instance = None | |