Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """github-rag: GitHub kullanıcısının repolarını okuyup soruları yanıtlayan RAG. | |
| Kullanım: | |
| from github_rag import GithubRag | |
| rag = GithubRag.from_github("kadiirgok") # veya bir repo linki | |
| res = rag.ask("hangi dillerde yazılmış?") | |
| print(res["cevap"]) # res: {"cevap","kaynak","sure_saniye","dogrulandi"} | |
| """ | |
| import json | |
| import re | |
| import time | |
| from collections import Counter | |
| from pathlib import Path | |
| from .answer import answer_from_collection, answer_from_web | |
| from .config import Config, load_config | |
| from .github import fetch_metadata, ingest_code, ingest_github, resolve_target | |
| from .indexing import build_index, clear_collection, get_collection, reset_collection | |
| from .llm import make_llm | |
| from .prompts import CODE_PROMPT_TEMPLATE | |
| from .retrieval import _kompakt, _source_anahtari | |
| __version__ = "0.2.0" | |
| def _sanitize(name: str) -> str: | |
| """Hedefi güvenli bir koleksiyon adına dönüştürür.""" | |
| return re.sub(r"[^a-zA-Z0-9_\-]", "_", name) or "index" | |
| # Kod detayı isteyen soruları işaretleyen anahtar kelimeler. Bu kelimeler geçerse | |
| # README doğrulansa bile kod katmanına geçilir (kullanıcı implementasyon istiyor). | |
| _CODE_KELIMELERI = { | |
| "kod", "code", "fonksiyon", "function", "def ", "class ", "dosya", "file", | |
| "implement", "method", "metod", "module", "modül", "import", "değişken", | |
| "variable", "endpoint", "algoritma", "algorithm", "mimari", "yapısı", | |
| "nasıl çalış", "nasil calis", "hangi dosyada", "nerede tanımlı", | |
| } | |
| def _is_code_sorusu(soru: str) -> bool: | |
| """Soru kod/implementasyon detayı istiyor mu?""" | |
| s = soru.lower() | |
| return any(k in s for k in _CODE_KELIMELERI) | |
| class GithubRag: | |
| """Bir GitHub hedefini indeksleyip üzerinde soru-cevap yapan RAG aracı.""" | |
| def __init__(self, target: str | None = None, config: Config | None = None): | |
| self.config = config or load_config() | |
| self.target = target | |
| self.owner = None | |
| self.collection = None | |
| self.collection_name = None | |
| self.bilinen_anahtarlar: list[str] = [] | |
| self.repo_names: set[str] = set() | |
| self.repo_sayisi = 0 | |
| self.repo_metadata: list[dict] = [] | |
| self.son_repo: tuple[str, str] | None = None # takip soruları için son odak repo | |
| self._llm = None | |
| self._code_coll_cache: dict[str, object] = {} | |
| def llm(self): | |
| if self._llm is None: | |
| self._llm = make_llm(self.config) | |
| return self._llm | |
| # --- İndeksleme (Katman 1: README) --- | |
| def _prepare_collection(self, target: str): | |
| parsed = resolve_target(target) | |
| self.owner = parsed["owner"] | |
| key = parsed["repo"] or parsed["owner"] | |
| self.collection_name = f"repos_{_sanitize(key)}" | |
| self.collection = get_collection(self.config.chroma_dir, self.collection_name) | |
| def index(self, target: str | None = None, force: bool = False) -> int: | |
| """Hedefin repo+README'lerini çekip indeksler; repo sayısını döndürür.""" | |
| target = target or self.target | |
| if not target: | |
| raise ValueError("index() için bir GitHub hedefi gerekli.") | |
| self.target = target | |
| self._prepare_collection(target) | |
| if not force and self.collection.count() > 0: | |
| self._load_metadata() | |
| if not self.repo_metadata: | |
| self.repo_metadata = fetch_metadata(target, self.config.github_token) | |
| self._save_metadata() | |
| self._refresh_known_keys() | |
| print(f"'{target}' zaten indekslenmiş (cache). force=True ile yenilenebilir.") | |
| return self.repo_sayisi | |
| docs, metadata = ingest_github(target, self.config.github_token) | |
| if not docs: | |
| raise ValueError(f"'{target}' için okunabilir README bulunamadı.") | |
| # Force yeniden çekimde eski/öksüz chunk'ları temizle (build_index sadece upsert yapar). | |
| if force: | |
| clear_collection(self.config.chroma_dir, self.collection_name) | |
| self.repo_metadata = metadata | |
| self._save_metadata() | |
| build_index(docs, self.config.chroma_dir, self.collection_name, | |
| self.config.embedding_model) | |
| self._refresh_known_keys() | |
| return self.repo_sayisi | |
| def detect_changes(self, target: str | None = None) -> dict: | |
| """Cache'li metadata'yı GitHub'daki güncel durumla karşılaştırır. | |
| Hafiftir: embedding/README çekmez, yalnızca repo listesini (`fetch_metadata`) | |
| getirir. Karşılaştırma: repo adı seti + `updated_at` + `stars`. | |
| """ | |
| target = target or self.target | |
| if not target: | |
| raise ValueError("detect_changes için bir GitHub hedefi gerekli.") | |
| guncel = fetch_metadata(target, self.config.github_token) | |
| yeni = {m["name"]: m for m in guncel} | |
| eski = {m["name"]: m for m in (self.repo_metadata or [])} | |
| eklenen = sorted(set(yeni) - set(eski)) | |
| kaldirilan = sorted(set(eski) - set(yeni)) | |
| degisen = sorted( | |
| n for n in set(yeni) & set(eski) | |
| if yeni[n].get("updated_at") != eski[n].get("updated_at") | |
| or yeni[n].get("stars") != eski[n].get("stars") | |
| ) | |
| return {"hedef": target, "eklenen": eklenen, "kaldirilan": kaldirilan, | |
| "degisen": degisen, "guncel": not (eklenen or kaldirilan or degisen)} | |
| def _refresh_known_keys(self): | |
| """Bilinen repo adlarını metadata + koleksiyon kaynaklarından toplar. | |
| Metadata tüm repoları bilir (README'siz/şablon README'li olanlar dahil); | |
| koleksiyon kaynakları ise yerel klasör modu gibi durumları kapsar. İkisinin | |
| birleşimi, odağı doğru bulmak için tam repo listesini verir. | |
| """ | |
| try: | |
| names: set[str] = set() | |
| for m in self.repo_metadata: | |
| for key in ("name", "full_name"): | |
| val = (m or {}).get(key, "") | |
| if val: | |
| names.add(val.split("/")[-1]) | |
| metas = self.collection.get(include=["metadatas"]).get("metadatas", []) | |
| for m in metas: | |
| s = (m or {}).get("source", "") | |
| if s and s != "web": | |
| names.add(s) | |
| self.repo_names = names | |
| self.repo_sayisi = len(names) | |
| self.bilinen_anahtarlar = [_kompakt(s) for s in names] | |
| except Exception: | |
| self.bilinen_anahtarlar = [] | |
| self.repo_names = set() | |
| self.repo_sayisi = 0 | |
| # --- Metadata kalıcılığı (cache sonrası da erişilebilir) --- | |
| def _meta_path(self) -> Path: | |
| return Path(self.config.chroma_dir) / f"{self.collection_name}_meta.json" | |
| def _save_metadata(self): | |
| try: | |
| self._meta_path().write_text( | |
| json.dumps(self.repo_metadata, ensure_ascii=False, indent=2), | |
| encoding="utf-8") | |
| except Exception: | |
| pass | |
| def _load_metadata(self): | |
| p = self._meta_path() | |
| if not p.is_file(): | |
| self.repo_metadata = [] | |
| return | |
| try: | |
| self.repo_metadata = json.loads(p.read_text(encoding="utf-8")) | |
| except Exception: | |
| self.repo_metadata = [] | |
| # --- Kod okuma (Katman 2) --- | |
| def _find_focus(self, soru: str) -> tuple[str, str] | None: | |
| """Soruda adı geçen repoyu bulur; (owner, repo) döner, yoksa None.""" | |
| if not self.owner: | |
| return None | |
| soru_kompakt = _kompakt(soru) | |
| en_iyi, en_iyi_uzunluk = None, 0 | |
| for name in self.repo_names: | |
| key = _source_anahtari(name) | |
| if len(key) >= 4 and key in soru_kompakt and len(key) > en_iyi_uzunluk: | |
| en_iyi, en_iyi_uzunluk = name, len(key) | |
| return (self.owner, en_iyi) if en_iyi else None | |
| def read_code(self, owner: str, repo: str, force: bool = False): | |
| """Bir reponun kaynak kodunu çekip indeksler (koleksiyon döner, cache'li).""" | |
| collection_name = f"code_{_sanitize(owner)}_{_sanitize(repo)}" | |
| if collection_name in self._code_coll_cache: | |
| return self._code_coll_cache[collection_name] | |
| try: | |
| coll = get_collection(self.config.chroma_dir, collection_name) | |
| if not force and coll.count() > 0: | |
| self._code_coll_cache[collection_name] = coll | |
| return coll | |
| except Exception as exc: # noqa: BLE001 — bozuk HNSW indeksi vb. | |
| print(f"Kod koleksiyonu bozuk ({collection_name}), yeniden oluşturuluyor: {exc}") | |
| coll = reset_collection(self.config.chroma_dir, collection_name) | |
| try: | |
| docs = ingest_code(owner, repo, self.config.github_token, | |
| max_files=self.config.code_max_files, | |
| max_size=self.config.code_max_size) | |
| except Exception as exc: # noqa: BLE001 — kod okunamazsa sessizce atla | |
| print(f"Kod okunamadı ({owner}/{repo}): {exc}") | |
| self._code_coll_cache[collection_name] = coll | |
| return coll | |
| if docs: | |
| print(f"'{repo}' kodu okunuyor: {len(docs)} dosya...") | |
| # Force yeniden okumada eski chunk'ları temizle (upsert silmez). | |
| if force: | |
| clear_collection(self.config.chroma_dir, collection_name) | |
| # Kod küçük chunk'lara bölünür (token yoğunluğu yüksek -> bağlam taşmasını önler). | |
| build_index(docs, self.config.chroma_dir, collection_name, | |
| self.config.embedding_model, chunk_size=150, overlap=20) | |
| self._code_coll_cache[collection_name] = coll | |
| return coll | |
| # --- Sorgulama (iki katmanlı) --- | |
| # Repo adı/açıklamasından ilgi alanı çıkarmak için basit anahtar kelime eşlemesi. | |
| _ALAN_ANAHTARLARI = { | |
| "veri bilimi / makine öğrenmesi": [ | |
| "data-science", "data_science", "machine-learning", "ml", "churn", | |
| "telco", "deep-learning", "nlp", "dataset", "kaggle", "titanic", "veri", | |
| ], | |
| "mobil uygulama": ["flutter", "mobile", "dart", "android", "ios", "react-native"], | |
| "web / backend": ["web", "api", "backend", "frontend", "django", "flask", | |
| "fastapi", "react", "vue", "node", "spring", ".net"], | |
| "sohbet / asistan / RAG": ["chat", "sohbet", "bot", "asistan", "rag", "llm", "cv"], | |
| } | |
| def _meta_cevap(self, soru: str) -> str | None: | |
| """İndeks + metadata'dan kesin cevaplanabilen soruları doğrudan yanıtlar. | |
| Bu sorular LLM'e/web'e gitmemeli: cevap zaten elimizdeki metadata'da. | |
| """ | |
| s = soru.lower() | |
| # 0) Yıldız / popülerlik / en yeni-en eski (metadata'dan, deterministik) | |
| if any(k in s for k in ("yıldız", "yildiz", "star", "popüler", "populer", "fork")): | |
| return self._yildiz_cevabi() | |
| if any(k in s for k in ("en son", "en yeni", "son güncellen", "son guncellen")) and ("proj" in s or "repo" in s): | |
| return self._en_yeni_cevabi() | |
| # 0b) Gizli/özel repolar: yalnızca herkese açık repolar okunur | |
| if any(k in s for k in ("gizli", "özel repo", "ozel repo", "private")): | |
| return ("Yalnızca herkese açık (public) repolar okunabiliyor; gizli/özel repolar " | |
| f"görünmez. Şu an {self.repo_sayisi} herkese açık repo yüklü.") | |
| # 1) Dil / ilgi alanı özeti (repo metadata'sından, deterministik) | |
| if any(k in s for k in ("alan", "dil", "diller", "ilgilen", "konu", | |
| "neyi", "neyle", "neler yapmış", "neler yapmis")): | |
| return self._alan_ozeti() | |
| # 2) Repo sayısı / listesi | |
| if "proj" in s or "repo" in s: | |
| if "kaç" in s or "kac" in s: | |
| return f"Yüklenen GitHub hesabında {self.repo_sayisi} repo var." | |
| if any(k in s for k in ("hangi", "listele", "listesi", "neler")): | |
| return self._repo_listesi() | |
| return None | |
| def _yildiz_cevabi(self) -> str: | |
| """Yıldız sayısına göre sıralı ilk repolar.""" | |
| sirali = sorted(self.repo_metadata, key=lambda m: -(m.get("stars", 0) or 0)) | |
| if not sirali: | |
| return "Henüz repo bilgisi yüklenmedi." | |
| toplam = sum(m.get("stars", 0) or 0 for m in sirali) | |
| if toplam == 0: | |
| return "Hiçbir repo henüz yıldız almamış (toplam 0 yıldız)." | |
| satirlar = [f"En çok yıldız alan: **{sirali[0].get('name')}** " | |
| f"({sirali[0].get('stars', 0)} yıldız). Toplam {toplam} yıldız.", "", | |
| "İlk repolar:"] | |
| for m in sirali[:5]: | |
| if (m.get("stars", 0) or 0) > 0: | |
| satirlar.append(f"- {m.get('name')} · {m.get('stars')} yıldız") | |
| return "\n".join(satirlar) | |
| def _en_yeni_cevabi(self) -> str: | |
| """Son güncellenen repolar.""" | |
| sirali = sorted(self.repo_metadata, key=lambda m: m.get("updated_at", ""), reverse=True) | |
| if not sirali: | |
| return "Henüz repo bilgisi yüklenmedi." | |
| satirlar = ["Son güncellenen repolar:"] | |
| for m in sirali[:5]: | |
| satirlar.append(f"- {m.get('name')} · {(m.get('updated_at') or '')[:10]}") | |
| return "\n".join(satirlar) | |
| def _repo_listesi(self) -> str: | |
| """Repo adları + dil + kısa açıklama içeren zengin liste döndürür.""" | |
| if not self.repo_names: | |
| return "Henüz repo yüklenmedi." | |
| meta = {m.get("name"): m for m in self.repo_metadata} | |
| lines = [f"Bu hesapta {self.repo_sayisi} repo var:"] | |
| for name in sorted(self.repo_names): | |
| m = meta.get(name, {}) | |
| parca = f"- {name}" | |
| if m.get("language"): | |
| parca += f" · {m['language']}" | |
| if m.get("description"): | |
| parca += f" — {m['description'][:80]}" | |
| lines.append(parca) | |
| return "\n".join(lines) | |
| def _alan_ozeti(self) -> str: | |
| """Metadata'dan dil dağılımı + ilgi alanı çıkarır (LLM'siz, kesin).""" | |
| if not self.repo_metadata: | |
| return "Henüz repo bilgisi yüklenmedi." | |
| diller = Counter(m.get("language") for m in self.repo_metadata if m.get("language")) | |
| alanlar = Counter() | |
| for m in self.repo_metadata: | |
| metin = f"{m.get('name', '')} {m.get('description', '')}".lower() | |
| for alan, kelimeler in self._ALAN_ANAHTARLARI.items(): | |
| if any(k in metin for k in kelimeler): | |
| alanlar[alan] += 1 | |
| satirlar = [f"Toplam {self.repo_sayisi} repo."] | |
| if diller: | |
| satirlar.append("Kullanılan diller: " + | |
| ", ".join(f"{d} ({n})" for d, n in diller.most_common())) | |
| if alanlar: | |
| satirlar.append("İlgi alanları: " + | |
| ", ".join(f"{a} ({n} repo)" for a, n in alanlar.most_common())) | |
| return "\n".join(satirlar) | |
| def _repo_aciklama_cevabi(self, repo_name: str) -> str | None: | |
| """Belirli bir reponun metadata açıklamasını yapılandırılmış biçimde döndürür. | |
| README/kod içerik vermediğinde (örn. yalnızca notebook'tan oluşan repo), | |
| GitHub metadata'sındaki description/language/stars ile en azından ne olduğunu | |
| söyler; metadata da yoksa None döner (web fallback'e bırakır). | |
| """ | |
| meta = None | |
| for m in self.repo_metadata: | |
| if m.get("name") == repo_name or (m.get("full_name") or "").split("/")[-1] == repo_name: | |
| meta = m | |
| break | |
| if not meta: | |
| return None | |
| satirlar = [f"**{meta.get('name', repo_name)}**"] | |
| if meta.get("description"): | |
| satirlar.append(f"- Açıklama: {meta['description'].strip()}") | |
| if meta.get("language"): | |
| satirlar.append(f"- Dil: {meta['language']}") | |
| satirlar.append(f"- Yıldız: {meta.get('stars', 0)}") | |
| if meta.get("html_url"): | |
| satirlar.append(f"- Bağlantı: {meta['html_url']}") | |
| return "\n".join(satirlar) | |
| def profile(self) -> dict: | |
| """Yapısal özet: /profile uç noktası ve arayüz için.""" | |
| return { | |
| "repo_sayisi": self.repo_sayisi, | |
| "diller": {d: n for d, n in | |
| Counter(m.get("language") for m in self.repo_metadata | |
| if m.get("language")).most_common()}, | |
| "toplam_yildiz": sum(m.get("stars", 0) for m in self.repo_metadata), | |
| "repolar": sorted(self.repo_metadata, | |
| key=lambda m: -(m.get("stars", 0) or 0)), | |
| } | |
| def ask(self, soru: str) -> dict: | |
| """İki katmanlı RAG: önce README, yetmezse kod, en son web.""" | |
| if self.collection is None or self.collection.count() == 0: | |
| return {"cevap": "Önce bir GitHub hesabı yükleyin (index()).", | |
| "kaynak": "none", "sure_saniye": 0.0, "dogrulandi": False} | |
| # Soruda belirli bir repo adı geçiyor mu? (takip bağlamı değil, doğrudan sorudan) | |
| acik_focus = self._find_focus(soru) | |
| # Belirli bir repo adı yoksa, genel profil sorularını deterministik yanıtla | |
| # ("kaç repo var?", "hangi diller?" gibi). Repo adı geçiyorsa genel özete | |
| # atlamaz; soru o repoya dairdir, RAG'a gider. | |
| if acik_focus is None: | |
| meta = self._meta_cevap(soru) | |
| if meta is not None: | |
| return {"cevap": meta, "kaynak": "local", "sure_saniye": 0.0, "dogrulandi": True} | |
| # LLM anahtarı yoksa net bir yönlendirme döndür (uygulama yine de açılır). | |
| if not (self.config.deepseek_api_key or self.config.groq_api_key): | |
| return {"cevap": "Cevap üretmek için bir LLM anahtarı gerekli. `.env` dosyasına " | |
| "DEEPSEEK_API_KEY (ve/veya GROQ_API_KEY) ekleyin. Repo sayısı, " | |
| "dil dağılımı gibi özet sorular anahtarsız çalışır.", | |
| "kaynak": "none", "sure_saniye": 0.0, "dogrulandi": False} | |
| t0 = time.time() | |
| cfg = self.config | |
| # Odak repo: sorudaki ad, yoksa önceki sorunun reposu (takip sorusu). | |
| focus = acik_focus or self.son_repo | |
| if focus is not None: | |
| self.son_repo = focus | |
| source_filter = focus[1] if focus else None | |
| # 1) README katmanı (odak varsa o repoya kapsüllenir) | |
| res = answer_from_collection( | |
| self.llm, soru, self.collection, self.bilinen_anahtarlar, | |
| cfg.embedding_model, top_k=cfg.top_k, max_tokens=cfg.max_tokens, | |
| esik=cfg.distance_threshold, source_filter=source_filter) | |
| # Doğrulanmış ve kod detayı istemeyen README cevabı -> direkt dön. | |
| if res is not None and res[2] and not _is_code_sorusu(soru): | |
| return self._yanit(res, t0) | |
| # 2) Kod katmanı (yalnızca odaklı repo varsa) | |
| if focus: | |
| code_coll = self.read_code(*focus) | |
| if code_coll.count() > 0: | |
| code_res = answer_from_collection( | |
| self.llm, soru, code_coll, [], cfg.embedding_model, | |
| top_k=cfg.code_top_k, max_tokens=cfg.max_tokens, | |
| # Kod koleksiyonu zaten tek repoya (focus) kapsüllü olduğu için | |
| # mesafe eşiği uygulanmaz (esik=2.0 cosine uzaklığının üst sınırı). | |
| esik=2.0, majority_vote=False, | |
| dogrula_sayisal=False, prompt_template=CODE_PROMPT_TEMPLATE) | |
| if code_res is not None: | |
| return self._yanit(code_res, t0) | |
| # 3) README doğrulanamadıysa ama bir şey bulduysa onu döndür | |
| if res is not None: | |
| return self._yanit(res, t0) | |
| # 4) Repo adı biliniyor ama README/kod içerik vermediyse: metadata açıklamasına düş. | |
| if focus: | |
| meta_desc = self._repo_aciklama_cevabi(focus[1]) | |
| if meta_desc is not None: | |
| return {"cevap": meta_desc, "kaynak": "local", | |
| "sure_saniye": round(time.time() - t0, 3), "dogrulandi": True} | |
| # 5) Web fallback (yalnızca hesapla ilgisiz genel sorular için) | |
| if self._hesap_sorusu(soru): | |
| return {"cevap": "Yüklenen hesabın repolarında bu bilgiyi bulamadım. " | |
| "Bir repo adıyla (örn. \"<repo> ne işe yarıyor?\") sorabilirsin.", | |
| "kaynak": "none", "sure_saniye": round(time.time() - t0, 3), | |
| "dogrulandi": True} | |
| return self._yanit( | |
| answer_from_web(self.llm, soru, self.bilinen_anahtarlar, cfg.max_tokens), t0) | |
| def _hesap_sorusu(soru: str) -> bool: | |
| """Soru yüklü hesap/repo hakkında mı? (web araması anlamsız olur)""" | |
| s = soru.lower() | |
| return any(k in s for k in ("repo", "proje", "projeler", "hesab", "github", | |
| "kodları", "kodlari", "bu proje")) | |
| def _yanit(self, res: tuple[str, str, bool], t0: float) -> dict: | |
| cevap, kaynak, dogrulandi = res | |
| return {"cevap": cevap, "kaynak": kaynak, | |
| "sure_saniye": round(time.time() - t0, 3), "dogrulandi": dogrulandi} | |
| # --- Fabrika yardımcıları --- | |
| def from_github(cls, target: str, config: Config | None = None) -> "GithubRag": | |
| """Hedefi hemen indeksleyen kısayol.""" | |
| rag = cls(target=target, config=config) | |
| rag.index() | |
| return rag | |
| def from_local_folder(cls, folder_path: str, config: Config | None = None, | |
| collection_name: str = "repos_local") -> "GithubRag": | |
| """Yerel .txt klasörünü (örnek veri / test) indeksleyen kısayol.""" | |
| rag = cls(config=config) | |
| docs = [] | |
| for txt in sorted(Path(folder_path).glob("*.txt")): | |
| content = txt.read_text(encoding="utf-8").strip() | |
| if content: | |
| docs.append({"text": content, "source": txt.name}) | |
| if not docs: | |
| raise ValueError(f"'{folder_path}' içinde .txt dosyası bulunamadı.") | |
| rag.collection_name = collection_name | |
| rag.collection = get_collection(rag.config.chroma_dir, collection_name) | |
| build_index(docs, rag.config.chroma_dir, collection_name, | |
| rag.config.embedding_model) | |
| rag._refresh_known_keys() | |
| return rag | |
| __all__ = ["GithubRag", "Config", "load_config", "__version__"] | |