Spaces:
Sleeping
Sleeping
File size: 4,746 Bytes
79d4fd5 0ff62f0 79d4fd5 92925b6 79d4fd5 0ff62f0 79d4fd5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | import hashlib
import logging
import time
from collections import OrderedDict
from typing import Any, Dict, List, Optional
from groq import Groq
from config import GROQ_API_KEY
from prompts import SUMMARIZATION_PROMPT
logger = logging.getLogger(__name__)
class ConversationSummarizer:
MAX_HISTORY_TURNS = 10
SUMMARY_MAX_TOKENS = 300
RECENT_MESSAGES_TO_KEEP = 8
MAX_INPUT_TEXT_LENGTH = 2000
MAX_CACHE_SIZE = 100
def __init__(
self,
api_key: Optional[str] = None,
model: str = "llama-3.1-8b-instant",
max_history_turns: int = 10,
):
self._api_key = api_key or GROQ_API_KEY
self._model = model
self._max_turns = max_history_turns
self._client: Optional[Groq] = None
self._summary_cache: OrderedDict[str, str] = OrderedDict()
@property
def client(self) -> Groq:
if self._client is None:
if not self._api_key:
raise ValueError("GROQ_API_KEY not configured")
self._client = Groq(api_key=self._api_key, timeout=5.0)
return self._client
def _generate_cache_key(self, messages: List[Dict[str, str]]) -> str:
parts = []
for msg in messages:
content = str(msg.get("content", ""))[:50]
parts.append(content)
raw = "||".join(parts)
return hashlib.md5(raw.encode("utf-8")).hexdigest()
def summarize_if_needed(self, chat_history: List[Dict[str, str]]) -> List[Dict[str, str]]:
if len(chat_history) <= self._max_turns * 2:
return chat_history
try:
recent_messages = chat_history[-self.RECENT_MESSAGES_TO_KEEP:]
older_messages = chat_history[:-self.RECENT_MESSAGES_TO_KEEP]
db_messages = [m for m in older_messages if m.get("is_database")]
non_db_older = [m for m in older_messages if not m.get("is_database")]
summary = self._summarize_conversation(non_db_older) if non_db_older else ""
result = []
if summary:
result.append({
"role": "system",
"content": f"Önceki konuşma özeti: {summary}",
})
result.extend(db_messages)
result.extend(recent_messages)
return result
except Exception as e:
logger.warning(f"Summarization failed, returning original history: {e}")
return chat_history
def _summarize_conversation(self, messages: List[Dict[str, str]]) -> str:
cache_key = self._generate_cache_key(messages)
if cache_key in self._summary_cache:
self._summary_cache.move_to_end(cache_key)
logger.debug("Conversation summary cache hit")
return self._summary_cache[cache_key]
conversation_parts = []
for msg in messages:
role = msg.get("role", "unknown")
content = str(msg.get("content", ""))
conversation_parts.append(f"{role}: {content}")
conversation_text = "\n".join(conversation_parts)
if len(conversation_text) > self.MAX_INPUT_TEXT_LENGTH:
conversation_text = conversation_text[-self.MAX_INPUT_TEXT_LENGTH:]
prompt = SUMMARIZATION_PROMPT.format(conversation_text=conversation_text)
try:
response = self.client.chat.completions.create(
model=self._model,
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=self.SUMMARY_MAX_TOKENS,
)
summary = response.choices[0].message.content.strip()
if not summary:
summary = "Konuşma geçmişi mevcut ama özeti oluşturulamadı."
except Exception as e:
logger.warning(f"Summary generation failed: {e}")
summary = "Konuşma geçmişi mevcut ama özeti oluşturulamadı."
while len(self._summary_cache) >= self.MAX_CACHE_SIZE:
self._summary_cache.popitem(last=False)
self._summary_cache[cache_key] = summary
return summary
def clear_cache(self) -> None:
self._summary_cache.clear()
_summarizer_instance: Optional[ConversationSummarizer] = None
def get_conversation_summarizer(
api_key: Optional[str] = None,
model: Optional[str] = None,
) -> ConversationSummarizer:
global _summarizer_instance
if _summarizer_instance is None:
_summarizer_instance = ConversationSummarizer(
api_key=api_key,
model=model or "llama-3.1-8b-instant",
)
return _summarizer_instance
def reset_conversation_summarizer() -> None:
global _summarizer_instance
_summarizer_instance = None
|