import torch import spaces import os import re import gradio as gr import requests import time import json import traceback from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline # --- SSL ve Network Ayarları --- import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # --- 1. MODELLERİN HAZIRLANMASI --- ner_pipe = pipeline("ner", model="xlm-roberta-large-finetuned-conll03-english", aggregation_strategy="simple", device=-1) # Qwen modeli model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0" # Tokenizer'ı yükle tokenizer = AutoTokenizer.from_pretrained( model_id, trust_remote_code=True, padding_side="left" ) # Eğer tokenizer'da pad_token yoksa eos_token'u kullan if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token # LLM modelini yükle llm_model = None IS_LLM_ENABLED = False try: print("📦 Qwen LLM modeli yükleniyor...") print("⚠️ Bu işlem 7B model olduğu için biraz zaman alabilir...") # Qwen için özel yükleme llm_model = AutoModelForCausalLM.from_pretrained( model_id, torch_dtype=torch.float16, # float16 daha hızlı device_map="cpu", # CPU'da çalıştır trust_remote_code=True, low_cpu_mem_usage=True ) # Modeli evaluation moduna al llm_model.eval() # Device kontrolü device = next(llm_model.parameters()).device print(f"✅ Qwen modeli yüklendi! Device: {device}") print(f" Model dtype: {next(llm_model.parameters()).dtype}") IS_LLM_ENABLED = True except Exception as e: IS_LLM_ENABLED = False print(f"⚠️ Qwen yüklenemedi: {e}") print(traceback.format_exc()) # --- 2. WIKIPEDIA API --- def get_wiki_summary(term): """Türkçe Wikipedia'dan doğrudan API ile özet getirir""" print(f"\n{'='*50}") print(f"🔍 Türkçe Wikipedia sorgusu: '{term}'") # Özel terim dönüşümleri special_cases = { "Türk": "Türkler", "türk": "Türkler", "Altın Aslan": "Altın Aslan Ödülü", "Venedik Film Festivali": "Venedik Film Festivali", "Kış Uykusu": "Kış Uykusu (film)", "Inception": "Inception (film)", "Oscar": "Akademi Ödülü", "Nobel": "Nobel Ödülü", "Amerika": "Amerika", "1984": "1984 (roman)", "Apple": "Apple (şirket)", "Windows": "Microsoft Windows", "Linux": "Linux", "Python": "Python (programlama dili)", } # Arama terimlerini hazırla search_terms = [term] for key, value in special_cases.items(): if term.lower() == key.lower(): search_terms = [value, term] break if len(search_terms) == 1: if term.endswith("lar") or term.endswith("ler"): search_terms.append(term[:-3]) else: search_terms.append(term + "lar") search_terms.append(term + "ler") print(f"📝 Aranacak terimler: {search_terms}") headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': 'application/json', 'Accept-Language': 'tr,en;q=0.9' } for search_term in search_terms: print(f"🔎 '{search_term}' aranıyor...") try: url = "https://tr.wikipedia.org/w/api.php" # Doğrudan sayfa kontrolü params = { "action": "query", "format": "json", "titles": search_term, "prop": "extracts", "exintro": True, "explaintext": True, "exsentences": 3, "origin": "*" } response = requests.get(url, params=params, headers=headers, timeout=10, verify=False) if response.status_code == 200: try: data = response.json() pages = data.get('query', {}).get('pages', {}) for page_id, page_info in pages.items(): if page_id != "-1": extract = page_info.get('extract', '') if extract and len(extract) > 20: print(f"✅ Doğrudan sayfa bulundu: '{search_term}'") return extract except: pass # Arama yap search_params = { "action": "query", "format": "json", "list": "search", "srsearch": search_term, "srlimit": 3, "srprop": "snippet", "origin": "*" } search_response = requests.get(url, params=search_params, headers=headers, timeout=10, verify=False) if search_response.status_code == 200: try: search_data = search_response.json() search_results = search_data.get('query', {}).get('search', []) for result in search_results: title = result.get('title', '') if title: summary_params = { "action": "query", "format": "json", "titles": title, "prop": "extracts", "exintro": True, "explaintext": True, "exsentences": 3, "origin": "*" } summary_response = requests.get(url, params=summary_params, headers=headers, timeout=10, verify=False) if summary_response.status_code == 200: try: summary_data = summary_response.json() pages = summary_data.get('query', {}).get('pages', {}) for page_id, page_info in pages.items(): if page_id != "-1": extract = page_info.get('extract', '') if extract and len(extract) > 20: print(f"✅ '{search_term}' -> '{title}' bulundu") return extract except: pass except: pass except Exception as e: print(f"⚠️ Hata: {str(e)[:50]}") continue time.sleep(0.3) print(f"❌ '{term}' için Türkçe Wikipedia'da sonuç bulunamadı") print(f"{'='*50}\n") return None def test_wikipedia_connection(): """Wikipedia API bağlantısını test et""" logs = [] try: url = "https://tr.wikipedia.org/w/api.php" params = { "action": "query", "format": "json", "titles": "Türkiye", "prop": "extracts", "exintro": True, "explaintext": True, "origin": "*" } headers = {'User-Agent': 'Mozilla/5.0 (compatible; Test/1.0)'} response = requests.get(url, params=params, headers=headers, timeout=10, verify=False) logs.append(f"📡 API Status: {response.status_code}") if response.status_code == 200: try: data = response.json() pages = data.get('query', {}).get('pages', {}) for page_id, page_info in pages.items(): if page_id != "-1": extract = page_info.get('extract', '') if extract: logs.append(f"✅ Türkçe Wikipedia API çalışıyor!") logs.append(f"📄 Örnek: {extract[:100]}...") break else: logs.append(f"⚠️ Sayfa bulunamadı") except: logs.append(f"❌ JSON parse hatası") else: logs.append(f"❌ API hatası: {response.status_code}") except Exception as e: logs.append(f"❌ Bağlantı hatası: {str(e)}") return "\n".join(logs) # --- 3. LLM ANALİZ FONKSİYONU (QWEN İÇİN OPTİMİZE) --- @spaces.GPU def refine_label_with_llm(entity_text, full_sentence, wiki_context, custom_label_definitions): """Qwen ile etiket analizi yapar - RuntimeError çözümlü""" global llm_model, IS_LLM_ENABLED, tokenizer if not IS_LLM_ENABLED or llm_model is None: return "MISC", "LLM Aktif Değil" try: print(f"\n🧠 Qwen LLM analizi: '{entity_text}'") # Modeli CPU'da tut if next(llm_model.parameters()).device.type != "cpu": llm_model.to("cpu") if torch.cuda.is_available(): torch.cuda.empty_cache() # Etiket Tanımları label_rules = "" for k, v in custom_label_definitions.items(): anahtar_kelimeler = ', '.join(v['anahtar_kelimeler']) label_rules += f"### {k}\n- TANIM: {v['tanim']}\n- ANAHTARLAR: {anahtar_kelimeler}\n\n" # Qwen için özel prompt formatı prompt = f"""<|im_start|>system Sen bir NER (Named Entity Recognition) uzmanısın. Verilen varlığı analiz et ve doğru kategoriyi belirle. <|im_end|> <|im_start|>user Aşağıdaki varlığı analiz et ve hangi kategoriye ait olduğunu söyle. VARLIK: {entity_text} CÜMLE: {full_sentence} WIKIPEDIA BİLGİSİ: {wiki_context} KATEGORİLER VE TANIMLARI: {label_rules} Kurallar: 1. Cümledeki bağlama göre karar ver 2. Wikipedia bilgisiyle karşılaştır 3. Çelişki varsa cümledeki kullanıma öncelik ver 4. Sadece kategori adını yaz (MOVIE, BOOK, SOFTWARE, AWARD, PRODUCT, NORP, GPE) CEVAP: <|im_end|> <|im_start|>assistant """ print(f"🔄 Qwen generate başlatılıyor...") # Tokenize et inputs = tokenizer( prompt, return_tensors="pt", truncation=True, max_length=2048, padding=True ) inputs = {k: v.to("cpu") for k, v in inputs.items()} # Generate - CPU'da with torch.no_grad(): outputs = llm_model.generate( inputs.input_ids, max_new_tokens=200, do_sample=False, temperature=0.1, top_p=0.9, pad_token_id=tokenizer.eos_token_id, eos_token_id=tokenizer.eos_token_id, repetition_penalty=1.1 ) # Çıktıyı decode et full_output = tokenizer.decode(outputs[0], skip_special_tokens=True) print(f"✅ Qwen çıktısı alındı") # Assistant kısmını bul if "<|im_start|>assistant" in full_output: response = full_output.split("<|im_start|>assistant")[-1].strip() else: response = full_output[len(prompt):].strip() print(f"📝 Çıktı: {response[:200]}...") # Etiket Ayıklama final_label = "MISC" valid_keys = list(custom_label_definitions.keys()) # Regex ile etiket bul for label in valid_keys: if re.search(rf'\b{label}\b', response, re.IGNORECASE): final_label = label print(f"🏷️ Etiket bulundu: {final_label}") break # Eğer bulunamadıysa, tüm kelimeleri kontrol et if final_label == "MISC": words = re.findall(r'[A-Z_]+', response.upper()) for word in words: if word in valid_keys: final_label = word print(f"🏷️ Kelime eşleşmesi: {final_label}") break return final_label, response except RuntimeError as e: print(f"❌ RuntimeError: {str(e)}") # Modeli resetle try: llm_model.to("cpu") if torch.cuda.is_available(): torch.cuda.empty_cache() except: pass return "MISC", f"RuntimeError: {str(e)[:100]}" except Exception as e: print(f"❌ LLM hatası: {type(e).__name__} - {str(e)}") print(traceback.format_exc()) return "MISC", f"Hata: {str(e)[:100]}" # --- 4. ANA İŞLEM HATTI --- def process_ner(text): current_logs = ["🚀 Sistem Başlatıldı: Metin analiz ediliyor..."] final_results = [] yield "
".join(current_logs), [] custom_labels = { "MOVIE": {"tanim": "Sinematik eserler (film, dizi).", "anahtar_kelimeler": ["film", "dizi", "sinema", "yönetmen"]}, "BOOK": {"tanim": "Yazılı edebi eserler (kitap, roman).", "anahtar_kelimeler": ["kitap", "roman", "yazar", "öykü"]}, "SOFTWARE": {"tanim": "Dijital sistemler (yazılım, işletim sistemi).", "anahtar_kelimeler": ["yazılım", "kod", "program", "sistem"]}, "AWARD": {"tanim": "Resmi ödüller ve festival ödülleri.", "anahtar_kelimeler": ["ödül", "oscar", "altın aslan", "nobel", "festival"]}, "PRODUCT": {"tanim": "Ticari ürünler (marka, model).", "anahtar_kelimeler": ["marka", "model", "telefon", "bilgisayar"]}, "NORP": {"tanim": "Etnik ve dini gruplar, milliyetler.", "anahtar_kelimeler": ["türk", "fransız", "alman", "millet"]}, "GPE": {"tanim": "Siyasi bölgeler (ülke, şehir).", "anahtar_kelimeler": ["ülke", "şehir", "başkent", "eyalet"]} } try: initial_results = ner_pipe(text) current_logs.append(f"✅ NER işlemi tamamlandı: {len(initial_results)} varlık bulundu.") yield "
".join(current_logs), [] except Exception as e: current_logs.append(f"❌ NER hatası: {str(e)}") yield "
".join(current_logs), [] return for entity in initial_results: word = entity['word'].replace(" ", " ").strip() label = entity['entity_group'] if label in ["PER", "ORG", "LOC"]: res = {"Varlık": word, "İlk Etiket": label, "Nihai Etiket": label, "Kaynak": "Model"} current_logs.append(f"✅ {word}: {label} (Doğrudan Tespit)") final_results.append(res) table_display = [[r["Varlık"], r["İlk Etiket"], f"{r['Nihai Etiket']}", r["Kaynak"]] for r in final_results] yield "
".join(current_logs), table_display continue elif label == "MISC": current_logs.append(f"🔍 Analiz Başladı: '{word}'") yield "
".join(current_logs), [[r["Varlık"], r["İlk Etiket"], r["Nihai Etiket"], r["Kaynak"]] for r in final_results] start_time = time.time() current_logs.append(f"⏱️ Türkçe Wikipedia sorgusu: '{word}'") yield "
".join(current_logs), [[r["Varlık"], r["İlk Etiket"], r["Nihai Etiket"], r["Kaynak"]] for r in final_results] wiki_text = None try: wiki_text = get_wiki_summary(word) elapsed = time.time() - start_time if wiki_text: current_logs.append(f"✅ Wikipedia başarılı ({elapsed:.2f}s)") wiki_preview = wiki_text[:200] + "..." if len(wiki_text) > 200 else wiki_text current_logs.append(f"
🌐 Wiki: {wiki_preview}
") else: current_logs.append(f"❌ Wikipedia boş döndü ({elapsed:.2f}s)") except Exception as e: elapsed = time.time() - start_time current_logs.append(f"❌ Wikipedia hatası ({elapsed:.2f}s): {str(e)[:100]}") wiki_text = None yield "
".join(current_logs), [[r["Varlık"], r["İlk Etiket"], r["Nihai Etiket"], r["Kaynak"]] for r in final_results] if wiki_text and IS_LLM_ENABLED: try: current_logs.append(f"🧠 Qwen LLM analizi başlatılıyor: '{word}' (Bu işlem biraz zaman alabilir...)") yield "
".join(current_logs), [[r["Varlık"], r["İlk Etiket"], r["Nihai Etiket"], r["Kaynak"]] for r in final_results] new_label, cot_text = refine_label_with_llm(word, text, wiki_text, custom_labels) cot_preview = cot_text[:300] + "..." if len(cot_text) > 300 else cot_text current_logs.append(f"🧠 LLM Çıktısı: {cot_preview}") res = {"Varlık": word, "İlk Etiket": label, "Nihai Etiket": new_label, "Kaynak": "RAG+LLM"} current_logs.append(f"🏷️ Nihai Etiket: {new_label}") except Exception as e: error_msg = str(e)[:100] current_logs.append(f"❌ LLM analiz hatası: {error_msg}") res = {"Varlık": word, "İlk Etiket": label, "Nihai Etiket": "MISC", "Kaynak": f"Hata: {error_msg}"} else: if not wiki_text: res = {"Varlık": word, "İlk Etiket": label, "Nihai Etiket": "MISC", "Kaynak": "Model"} current_logs.append(f"⚠️ '{word}' için Wikipedia bilgisi olmadığından MISC olarak etiketlendi.") else: res = {"Varlık": word, "İlk Etiket": label, "Nihai Etiket": "MISC", "Kaynak": "Model"} current_logs.append(f"⚠️ LLM aktif değil, '{word}' MISC olarak etiketlendi.") final_results.append(res) table_display = [[r["Varlık"], r["İlk Etiket"], f"{r['Nihai Etiket']}", r["Kaynak"]] for r in final_results] yield "
".join(current_logs), table_display current_logs.append("✨ Analiz Tamamlandı.") yield "
".join(current_logs), [[r["Varlık"], r["İlk Etiket"], f"{r['Nihai Etiket']}", r["Kaynak"]] for r in final_results] # --- 5. DEBUG FONKSİYONLARI --- def debug_wikipedia(): result = test_wikipedia_connection() return f"
{result}
" def test_single_term(term): if not term: return "Lütfen bir terim girin." logs = [f"🔍 '{term}' için Türkçe Wikipedia testi"] logs.append("="*50) try: result = get_wiki_summary(term) if result: logs.append(f"✅ SONUÇ: {result}") else: logs.append(f"❌ Sonuç bulunamadı") except Exception as e: logs.append(f"❌ Hata: {str(e)}") return "\n".join(logs) def check_llm_status(): global IS_LLM_ENABLED, llm_model, model_id logs = ["=== Qwen LLM DURUM KONTROLÜ ==="] logs.append(f"LLM Aktif: {IS_LLM_ENABLED}") logs.append(f"Model: {model_id}") if llm_model is not None: try: device = next(llm_model.parameters()).device dtype = next(llm_model.parameters()).dtype logs.append(f"Device: {device}") logs.append(f"Dtype: {dtype}") logs.append(f"CUDA Kullanılabilir: {torch.cuda.is_available()}") logs.append(f"Model Boyutu: {sum(p.numel() for p in llm_model.parameters()):,} parametre") except Exception as e: logs.append(f"Device kontrol hatası: {str(e)}") else: logs.append("❌ Model yüklenmemiş!") return "\n".join(logs) # --- 6. GRADIO ARAYÜZÜ --- with gr.Blocks(title="RAG-NER Uzman Karar Destek Sistemi") as demo: gr.Markdown("# 🚀 RAG-NER Uzman Karar Destek Sistemi (Türkçe Wikipedia + Qwen)") gr.Markdown(f"**LLM Modeli:** {model_id}") with gr.Tab("Metin Analizi"): input_text = gr.Textbox( label="Analiz Edilecek Metin", placeholder="Analiz etmek istediğiniz metni buraya girin...", lines=3 ) examples_list = [ ["Türk yönetmen Nuri Bilge Ceylan, Venedik Film Festivali'nde Altın Aslan ödülünü kazandığı Kış Uykusu filmiyle uluslararası üne kavuştu."], ["Amerikalı yazar George Orwell'in ünlü kitabı 1984, Apple şirketinin Macintosh tanıtım reklamında ilham kaynağı olarak kullanılmış ve Windows işletim sistemine karşı bir metafor haline gelmişti."], ["Fransız oyuncu Marion Cotillard, Christopher Nolan'ın Inception filmiyle Oscar ödülüne aday gösterilmiş ve daha önce La Vie en Rose performansıyla En İyi Kadın Oyuncu Akademi Ödülü'nü kazanmıştı."] ] gr.Examples( examples=examples_list, inputs=input_text, label="Hazır Örnek Senaryolar" ) with gr.Row(): log_output = gr.HTML(label="Sistem Günlüğü") table_output = gr.Dataframe( headers=["Varlık", "İlk Etiket", "Nihai Etiket", "Kaynak"], datatype="markdown", wrap=True ) btn = gr.Button("Analiz Et", variant="primary") btn.click( process_ner, inputs=input_text, outputs=[log_output, table_output] ) with gr.Tab("Debug"): gr.Markdown("## 🐛 Debug Paneli") gr.Markdown("### 🌐 Türkçe Wikipedia Testi") with gr.Row(): debug_btn = gr.Button("Wikipedia Bağlantısını Test Et", variant="secondary") debug_output = gr.HTML(label="Debug Sonuçları") debug_btn.click(debug_wikipedia, outputs=debug_output) gr.Markdown("### 🔍 Tek Terim Testi") with gr.Row(): single_term_input = gr.Textbox( label="Test Edilecek Terim", placeholder="Örn: Türk, Venedik Film Festivali, Altın Aslan", scale=3 ) single_term_btn = gr.Button("Test Et", variant="secondary", scale=1) single_term_output = gr.Textbox( label="Sonuç", lines=15, interactive=False ) single_term_btn.click(test_single_term, inputs=single_term_input, outputs=single_term_output) gr.Markdown("### 🧠 Qwen LLM Durum Kontrolü") with gr.Row(): llm_status_btn = gr.Button("Qwen Durumunu Kontrol Et", variant="secondary") llm_status_output = gr.Textbox( label="Qwen Durumu", lines=8, interactive=False ) llm_status_btn.click(check_llm_status, outputs=llm_status_output) # --- 7. BAŞLATMA --- if __name__ == "__main__": print("\n" + "="*50) print("🚀 SİSTEM BAŞLATILIYOR...") print("="*50) print("\n📡 Türkçe Wikipedia bağlantısı test ediliyor...") test_result = test_wikipedia_connection() print(test_result) print("\n🧠 Qwen LLM durumu kontrol ediliyor...") print(check_llm_status()) print("\n" + "="*50) print("✅ Sistem hazır, Gradio başlatılıyor...") print("="*50 + "\n") demo.launch( auth=("admin", "1113"), auth_message="Lütfen uzman paneline erişmek için giriş yapın.", debug=True, share=True )