from typing import Optional, List, Dict, Generator from groq import Groq from config import ( GROQ_API_KEY, LLM_MODELS, DEFAULT_LLM_MODEL, ) from prompts import SYSTEM_PROMPT_TR class LLMGenerator: def __init__(self, api_key: str = None): self._api_key = api_key or GROQ_API_KEY if not self._api_key: raise ValueError("API key required.") self._client = None self.available_models = LLM_MODELS @property def client(self) -> Groq: if self._client is None: self._client = Groq(api_key=self._api_key) return self._client def _format_chat_history(self, chat_history: List[Dict[str, str]]) -> str: if not chat_history: return "Henüz konuşma geçmişi yok." parts = [] for msg in chat_history: role = "Kullanici" if msg["role"] == "user" else "Asistan" content = msg["content"][:500] parts.append(f"{role}: {content}") return "\n".join(parts) def _build_prompt( self, question: str, context: str, chat_history: Optional[List[Dict[str, str]]] = None ) -> str: history_str = self._format_chat_history(chat_history or []) return SYSTEM_PROMPT_TR.format( context=context, question=question, chat_history=history_str ) def generate( self, question: str, context: str, chat_history: Optional[List[Dict[str, str]]] = None, model_id: str = DEFAULT_LLM_MODEL, temperature: float = 0.1, max_tokens: int = 1024 ) -> str: prompt = self._build_prompt(question, context, chat_history) try: response = self.client.chat.completions.create( model=model_id, messages=[ {"role": "system", "content": "Sen Türkçe yanıt veren bir uzman asistansın."}, {"role": "user", "content": prompt} ], temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as e: return f"Hata oluştu: {str(e)}" def generate_stream( self, question: str, context: str, chat_history: Optional[List[Dict[str, str]]] = None, model_id: str = DEFAULT_LLM_MODEL, temperature: float = 0.1, max_tokens: int = 1024 ) -> Generator[str, None, None]: prompt = self._build_prompt(question, context, chat_history) try: stream = self.client.chat.completions.create( model=model_id, messages=[ {"role": "system", "content": "Sen Türkçe yanıt veren bir uzman asistansın."}, {"role": "user", "content": prompt} ], temperature=temperature, max_tokens=max_tokens, stream=True ) for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content except Exception as e: yield f"Hata oluştu: {str(e)}" @staticmethod def get_model_display_name(model_id: str) -> str: for name, mid in LLM_MODELS.items(): if mid == model_id: return name return model_id @staticmethod def get_model_id(display_name: str) -> str: return LLM_MODELS.get(display_name, DEFAULT_LLM_MODEL) _llm_generator_instance = None def get_llm_generator() -> LLMGenerator: global _llm_generator_instance if _llm_generator_instance is None: _llm_generator_instance = LLMGenerator() return _llm_generator_instance def reset_llm_generator(): global _llm_generator_instance _llm_generator_instance = None if __name__ == "__main__": generator = LLMGenerator() context = "Atlas ERP sistemi muhasebe ve finans modülleri içerir." question = "Atlas nedir?" response = generator.generate(question, context) print(f"Response: {response}")