import os import torch import google.generativeai as genai from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain_google_genai import ChatGoogleGenerativeAI from langchain.prompts import PromptTemplate # Vektör veritabanı yolu VECTOR_STORE_PATH = os.path.join(os.path.dirname(__file__), "vector_store.faiss") # Doküman klasörü DOCS_FOLDER = os.path.join(os.path.dirname(__file__), "docs") os.makedirs(DOCS_FOLDER, exist_ok=True) # Cache klasörü CACHE_DIR = os.path.join(os.path.dirname(__file__), "cache") os.makedirs(CACHE_DIR, exist_ok=True) class RAGChatbot: def __init__(self, api_key=None): # API anahtarını kaydet self.api_key = api_key # GPU bellek optimizasyonu if torch.cuda.is_available(): torch.cuda.empty_cache() # Vektör veritabanını yükle veya oluştur self.vectorstore = self._load_or_create_vectorstore() # LLM modeli yapılandır (eğer API anahtarı varsa) if self.api_key: self.llm = self._setup_gemini() # LangChain RAG zincirini oluştur self.qa_chain = self._create_qa_chain() print("RAG chatbot initialized with Gemini") else: self.llm = None self.qa_chain = None print("RAG chatbot initialized without API key") def set_api_key(self, api_key): """API anahtarını güncelle ve modeli yeniden yapılandır""" self.api_key = api_key if self.api_key: self.llm = self._setup_gemini() self.qa_chain = self._create_qa_chain() return True return False def _setup_gemini(self): """Gemini API'yi yapılandır""" if not self.api_key: return None try: # Gemini API'yi yapılandır genai.configure(api_key=self.api_key) # Kullanılabilir modelleri kontrol et models = genai.list_models() gemini_models = [m.name for m in models if "gemini" in m.name.lower()] print(f"Available Gemini models: {gemini_models}") # Gemini 2.0 Flash modelini seç model_name = "models/gemini-1.5-flash" # Gemini 2.0 Flash için güncellenebilir # LLM oluştur llm = ChatGoogleGenerativeAI( model=model_name, temperature=0.3, top_p=0.95, top_k=40, google_api_key=self.api_key, convert_system_message_to_human=True ) print(f"Using Gemini model: {model_name}") return llm except Exception as e: print(f"Error setting up Gemini API: {e}") return None def _load_documents(self): """Dokümanları yükle ve işle""" documents = [] # Klasördeki tüm dosyaları kontrol et for filename in os.listdir(DOCS_FOLDER): file_path = os.path.join(DOCS_FOLDER, filename) try: if filename.endswith(".pdf"): loader = PyPDFLoader(file_path) documents.extend(loader.load()) elif filename.endswith(".txt"): loader = TextLoader(file_path, encoding="utf-8") documents.extend(loader.load()) except Exception as e: print(f"Error loading file {file_path}: {e}") # Doküman yoksa örnek doküman oluştur if not documents: print("No documents found, creating sample document...") sample_text = """ # Deprem Öncesi Hazırlık ## Deprem Çantası Hazırlama Deprem çantanızda bulunması gerekenler: - Su (kişi başı günlük 2 litre, en az 3 günlük) - Bozulmayan yiyecekler (konserve, kuru gıda, vs.) - El feneri ve yedek piller - İlk yardım çantası - Düdük (yardım çağırmak için) - Toz maskesi - Islak mendil ve çöp torbaları - Önemli belgelerin kopyaları (kimlik, sigorta, vs.) - Şarj edilebilir powerbank - Battaniye - Yedek kıyafet - Kişisel hijyen malzemeleri ## Ev İçi Güvenlik - Ağır eşyaları alt raflara yerleştirin - Dolapları ve kitaplıkları duvara sabitleyin - Aynalar ve tablolar gibi asılı eşyaları güvenli şekilde monte edin - Kimyasal maddeleri güvenli ve kapalı dolaplarda saklayın - Gaz, su ve elektrik tesisatını kontrol ettirin # Deprem Anında Yapılması Gerekenler - Çök-Kapan-Tutun hareketini uygulayın - Sağlam bir masa altına girin veya iç duvar köşesine çökün - Pencere ve dış duvarlardan uzak durun - Asansör kullanmayın - Merdivenlerde durmayın # Deprem Sonrası - Önce kendi güvenliğinizi sağlayın - Yaralılara ilk yardım uygulayın - Gaz, su ve elektriği kapatın - Hasarlı binalardan uzak durun - Yetkililerin talimatlarını dinleyin - Telefonunuzu acil durumlar dışında kullanmayın # Acil Durum İletişim Bilgileri - AFAD: 122 - Ambulans: 112 - İtfaiye: 110 - Polis: 155 """ # Örnek dokümanı kaydet sample_file_path = os.path.join(DOCS_FOLDER, "deprem_bilgileri.txt") with open(sample_file_path, "w", encoding="utf-8") as f: f.write(sample_text) # Dokümanı yükle loader = TextLoader(sample_file_path, encoding="utf-8") documents.extend(loader.load()) # Dokümanları parçalara ayır text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len ) chunks = text_splitter.split_documents(documents) return chunks def _load_or_create_vectorstore(self): """Vektör veritabanını yükle veya oluştur""" # Embedding modeli embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"} ) # Vektör veritabanı var mı kontrol et if os.path.exists(VECTOR_STORE_PATH) and os.path.isdir(VECTOR_STORE_PATH): print(f"Loading vector store: {VECTOR_STORE_PATH}") vectorstore = FAISS.load_local(VECTOR_STORE_PATH, embeddings) else: print("Creating vector store...") # Dokümanları yükle chunks = self._load_documents() # Vektör veritabanı oluştur vectorstore = FAISS.from_documents(chunks, embeddings) # Vektör veritabanını kaydet os.makedirs(os.path.dirname(VECTOR_STORE_PATH), exist_ok=True) vectorstore.save_local(VECTOR_STORE_PATH) print(f"Vector store created and saved: {VECTOR_STORE_PATH}") return vectorstore def _create_qa_chain(self): """LangChain QA zincirini oluştur""" if not self.llm: return None # Prompt şablonu template = """ Sen bir deprem güvenliği uzmanısın. Aşağıdaki bilgilere dayanarak soruyu doğru, kapsamlı ve yardımcı bir şekilde yanıtla. Bağlam: {context} Soru: {question} Yanıt: """ prompt = PromptTemplate( template=template, input_variables=["context", "question"] ) # RetrievalQA zinciri qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}), chain_type_kwargs={"prompt": prompt}, return_source_documents=True ) return qa_chain def answer(self, question): """Soruya yanıt ver""" if not self.api_key or not self.llm or not self.qa_chain: return "Lütfen önce Gemini API anahtarınızı girin. Hazırlık Sohbeti sekmesinin üst kısmındaki API Anahtarı alanına geçerli bir Gemini API anahtarı girmeniz gerekmektedir." try: # Yanıt al result = self.qa_chain({"query": question}) # Yanıtı ve kaynakları formatla answer = result["result"] # Kaynakları ekle sources = [] for doc in result["source_documents"]: if hasattr(doc, "metadata") and "source" in doc.metadata: sources.append(doc.metadata["source"]) if sources: answer += "\n\nKaynaklar:\n" + "\n".join(set(sources)) return answer except Exception as e: print(f"Error generating answer: {e}") return f"Üzgünüm, sorunuza yanıt verirken bir hata oluştu. Lütfen API anahtarınızın doğru olduğundan emin olun ve tekrar deneyin." # Singleton örneği chatbot = None def get_chatbot(api_key=None): """Chatbot singleton örneğini döndür""" global chatbot if chatbot is None: chatbot = RAGChatbot(api_key) elif api_key: chatbot.set_api_key(api_key) return chatbot def answer(question, api_key=None): """Gradio arayüzü için wrapper fonksiyon""" chatbot = get_chatbot(api_key) return chatbot.answer(question) # Test if __name__ == "__main__": # Test API anahtarı test_api_key = "YOUR_GEMINI_API_KEY" # Test sorusu test_question = "Deprem çantasında neler bulundurmalıyım?" # Chatbot'u başlat chatbot = RAGChatbot(test_api_key) # Yanıt al response = chatbot.answer(test_question) print(f"Soru: {test_question}") print(f"Yanıt: {response}")