from __future__ import annotations import argparse import html import io import json import math import re import time import warnings import webbrowser import zipfile import xml.etree.ElementTree as ET from collections import Counter from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs import email SAMPLE_QUESTIONS = [ "Kasten öldürme suçunun cezası nedir?", "Hırsızlık suçunun yasal tanımı ve unsurları nelerdir?", "Adil yargılanma hakkı nasıl güvence altına alınır?", "Cumhurbaşkanı seçilebilmek için gerekli şartlar nelerdir?", "Olağanüstü hal (OHAL) hangi durumlarda ilan edilebilir?", "Haksız fiil sebebiyle tazminat sorumluluğunun şartları nelerdir?", "Evlilik birliği temelinden sarsılırsa ne olur?", ] @dataclass class Doc: id: str title: str text: str citation: str def tokenize(text: str) -> list[str]: return re.findall(r"\w+", text.lower(), flags=re.UNICODE) def iter_jsonl(path: Path): with path.open("r", encoding="utf-8") as f: for line in f: if line.strip(): yield json.loads(line) def resolve_corpus_file(data_dir: Path) -> Path: candidates = [ data_dir / "real_corpus.jsonl", data_dir / "corpus_index.jsonl", data_dir / "corpus.jsonl", Path("data") / "real_corpus.jsonl", Path("data") / "corpus_index.jsonl", Path("data") / "corpus.jsonl", ] for candidate in candidates: if candidate.exists(): return candidate raise FileNotFoundError("No corpus file found.") def load_docs(corpus_file: Path, limit: int | None = None) -> list[Doc]: docs: list[Doc] = [] for row in iter_jsonl(corpus_file): metadata = row.get("metadata") or {} doc_id = str(row.get("id") or metadata.get("chunk_id") or len(docs)) title = str(row.get("title") or metadata.get("category") or "Legal Source") text = str(row.get("text") or row.get("content") or "") if not text.strip(): continue citation = str(metadata.get("citation_label") or row.get("citation_label") or f"{title} - {doc_id}") docs.append(Doc(doc_id, title, text, citation)) if limit and len(docs) >= limit: break return docs def extract_docx_text(payload: bytes) -> str: with zipfile.ZipFile(io.BytesIO(payload)) as archive: xml_bytes = archive.read("word/document.xml") root = ET.fromstring(xml_bytes) namespace = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"} paragraphs = [] for paragraph in root.findall(".//w:p", namespace): texts = [node.text or "" for node in paragraph.findall(".//w:t", namespace)] text = "".join(texts).strip() if text: paragraphs.append(text) return "\n".join(paragraphs) def extract_pdf_text(payload: bytes) -> str: try: from pypdf import PdfReader except Exception as exc: raise RuntimeError("PDF destegi icin pypdf kurulmali: pip install pypdf") from exc reader = PdfReader(io.BytesIO(payload)) pages = [] for idx, page in enumerate(reader.pages, start=1): page_text = page.extract_text() or "" if page_text.strip(): pages.append(f"[Page {idx}]\n{page_text.strip()}") return "\n\n".join(pages) def extract_uploaded_text(filename: str, payload: bytes) -> str: suffix = Path(filename).suffix.lower() if suffix in {".txt", ".md", ".csv", ".json", ".jsonl"}: return payload.decode("utf-8", errors="ignore") if suffix == ".docx": return extract_docx_text(payload) if suffix == ".pdf": return extract_pdf_text(payload) raise ValueError("Desteklenen dosya tipleri: .txt, .md, .csv, .json, .jsonl, .docx, .pdf") def chunk_uploaded_text(text: str, filename: str, chunk_size: int = 900, overlap: int = 150) -> list[Doc]: text = re.sub(r"\r\n?", "\n", text) text = re.sub(r"\n{3,}", "\n\n", text).strip() if not text: return [] chunks: list[Doc] = [] start = 0 while start < len(text): end = min(start + chunk_size, len(text)) chunk = text[start:end].strip() if end < len(text): split_at = max(chunk.rfind("."), chunk.rfind("?"), chunk.rfind("!"), chunk.rfind("\n")) if split_at > int(chunk_size * 0.55): chunk = chunk[: split_at + 1].strip() end = start + split_at + 1 if chunk: number = len(chunks) + 1 chunks.append( Doc( id=f"UPLOAD_{number:03d}", title=f"{filename} - chunk {number}", text=chunk, citation=f"Uploaded file: {filename} | chunk {number}", ) ) next_start = end - overlap if next_start <= start: next_start = start + chunk_size start = next_start return chunks class SimpleBM25: def __init__(self, docs: list[Doc]) -> None: self.docs = docs self.doc_tokens = [tokenize(f"{doc.title} {doc.text}") for doc in docs] self.avgdl = sum(len(tokens) for tokens in self.doc_tokens) / max(len(self.doc_tokens), 1) df: Counter[str] = Counter() for tokens in self.doc_tokens: df.update(set(tokens)) n = len(docs) self.idf = {term: math.log(1 + (n - freq + 0.5) / (freq + 0.5)) for term, freq in df.items()} def search(self, query: str, top_k: int = 5) -> list[tuple[Doc, float]]: q_terms = tokenize(query) scores: list[tuple[int, float]] = [] k1 = 1.5 b = 0.75 for idx, tokens in enumerate(self.doc_tokens): tf = Counter(tokens) dl = len(tokens) or 1 score = 0.0 for term in q_terms: if term not in tf: continue numerator = tf[term] * (k1 + 1) denominator = tf[term] + k1 * (1 - b + b * dl / max(self.avgdl, 1)) score += self.idf.get(term, 0.0) * numerator / denominator if score > 0: scores.append((idx, score)) scores.sort(key=lambda item: item[1], reverse=True) return [(self.docs[idx], score) for idx, score in scores[:top_k]] class AnswerGenerator: def generate(self, question: str, results: list[tuple[Doc, float]]) -> str: raise NotImplementedError class ExtractiveGenerator(AnswerGenerator): def generate(self, question: str, results: list[tuple[Doc, float]]) -> str: if not results: return "Bu soru için kaynak bulunamadı." best = results[0][0] text = best.text.strip() title = best.title.strip() # Strip the title/question prefix from the beginning of the text if present if text.lower().startswith(title.lower()): text = text[len(title):].strip() # Strip any remaining leading punctuation or question marks text = re.sub(r"^[?\s,.:;!-]+", "", text).strip() # Ensure the first letter is capitalized if text: text = text[0].upper() + text[1:] return f"Kaynağa göre: {text}\n\nKaynak: {best.citation}" class LocalHFGenerator(AnswerGenerator): def __init__(self, model_name: str, max_new_tokens: int = 64) -> None: from transformers import AutoTokenizer, AutoModelForCausalLM, AutoModelForSeq2SeqLM import torch from pathlib import Path import json self.model_name = model_name self.max_new_tokens = max_new_tokens self.tokenizer = AutoTokenizer.from_pretrained(model_name) if self.tokenizer.pad_token is None: self.tokenizer.pad_token = self.tokenizer.eos_token peft_config_path = Path(model_name) / "adapter_config.json" is_peft = peft_config_path.exists() if is_peft: with peft_config_path.open("r", encoding="utf-8") as f: peft_data = json.load(f) base_model_name = peft_data.get("base_model_name_or_path") print(f"Loading PEFT adapter from {model_name} with base model {base_model_name}...") if torch.cuda.is_available(): from transformers import BitsAndBytesConfig quantization_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True, bnb_4bit_compute_dtype=torch.float16, ) base_model = AutoModelForCausalLM.from_pretrained( base_model_name, quantization_config=quantization_config, device_map="auto" ) else: base_model = AutoModelForCausalLM.from_pretrained( base_model_name, device_map={"": "cpu"}, torch_dtype=torch.float32, low_cpu_mem_usage=True, ) from peft import PeftModel self.model = PeftModel.from_pretrained(base_model, model_name) self.is_causal = True else: try: print(f"Loading base model {model_name}...") if torch.cuda.is_available(): from transformers import BitsAndBytesConfig quantization_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True, bnb_4bit_compute_dtype=torch.float16, ) self.model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=quantization_config, device_map="auto" ) else: self.model = AutoModelForCausalLM.from_pretrained( model_name, device_map={"": "cpu"}, torch_dtype=torch.float32, low_cpu_mem_usage=True, ) self.is_causal = True except Exception as e: print(f"Failed to load as causal LM: {e}") print(f"Loading seq2seq model {model_name}...") self.model = AutoModelForSeq2SeqLM.from_pretrained( model_name, device_map={"": "cpu"}, torch_dtype=torch.float32, low_cpu_mem_usage=True, ) self.is_causal = False self.model.eval() @staticmethod def build_prompt(question: str, results: list[tuple[Doc, float]]) -> str: context = "\n\n".join( f"[{rank}] Baslik: {doc.title}\nKaynak: {doc.citation}\nMetin: {doc.text[:600]}" for rank, (doc, _score) in enumerate(results[:3], start=1) ) return ( "Sen bir Turk hukuku RAG asistanisin. Yalnizca verilen kaynaklara dayanarak " "kisa ve dogru cevap ver. Kaynakta olmayan bilgiyi uretme.\n\n" f"Kaynaklar:\n{context}\n\n" f"Soru: {question}\n\n" "Cevap:" ) @staticmethod def clean_answer(answer: str, fallback_doc: Doc) -> str: answer = answer.strip() # Clean specific template structures answer = re.sub(r"(?i)Baslangic:\s*\[\d+\]", "", answer) answer = re.sub(r"(?i)Kendisi:\s*\[\d+\]", "", answer) answer = re.sub(r"(?i)Sonu[çc]:\s*\[\d+\]", "", answer) answer = re.sub(r"(?i)Kaynak:\s*\[\d+\]", "", answer) answer = re.sub(r"(?i)Sonu[çc]:", "", answer) answer = re.sub(r"(?i)İlk cihazı:", "", answer) # Remove prompt leakage/artifacts artifact_pattern = r"Bu metnin tümünü kontrol etmek için kaynaklarını seçerek, anlayışını gözden geçirerek ve sonuçları yazınız\.?" answer = re.sub(artifact_pattern, "", answer) # Filter out lines matching bracketed headers lines = [] for line in answer.splitlines(): line_strip = line.strip() if re.match(r"^\[(?:BaslangIC|Dışişler IC|Taahhût IC|Taahhüt IC)\]", line_strip, re.IGNORECASE): continue lines.append(line) answer = "\n".join(lines).strip() # Clean specific corrupted suffixes answer = re.sub(r"(?i)ülükdir\.?", "", answer) answer = re.sub(r"(?i)Kaynak:\s*Anaya\s*Sistemi\.?", "", answer) # Normal cleanup answer = re.sub(r"^\s*Soru:\s*", "", answer, flags=re.IGNORECASE) answer = re.split(r"\[\d+\]\s*Baslik:|\n\s*Baslik:|\n\s*Metin:", answer, maxsplit=1)[0].strip() answer = re.sub(r"\s+", " ", answer).strip() if len(answer.split()) < 5: answer = f"Verilen kaynaklara göre: {fallback_doc.text[:300]}...\n\nKaynak: {fallback_doc.citation}" if "Kaynak:" not in answer: answer = f"{answer}\n\nKaynak: {fallback_doc.citation}" return answer def generate(self, question: str, results: list[tuple[Doc, float]]) -> str: import torch if not results: return "Bu soru icin kaynak bulunamadi." # Format the context and question context = "\n\n".join( f"[{rank}] Baslik: {doc.title}\nKaynak: {doc.citation}\nMetin: {doc.text[:600]}" for rank, (doc, _score) in enumerate(results[:3], start=1) ) system_instruction = ( "Sen bir Türk hukuku RAG asistanısın. Yalnızca verilen kaynaklara dayanarak cevap ver. " "Kaynakta olmayan bilgiyi üretme ve cevabın sonunda kaynak belirt." ) user_content = ( f"[Kaynaklar]\n{context}\n\n" f"[Soru]\n{question}\n\n" "Cevabı yalnızca kaynaklara dayanarak Türkçe ver. " "Cevabın sonunda kullandığın citation bilgisini 'Kaynak:' satırıyla yaz." ) if self.is_causal and getattr(self.tokenizer, "chat_template", None): messages = [ {"role": "system", "content": system_instruction}, {"role": "user", "content": user_content} ] prompt = self.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) else: prompt = self.build_prompt(question, results) inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=768) device = "cuda" if torch.cuda.is_available() else "cpu" inputs = {k: v.to(device) for k, v in inputs.items()} with torch.inference_mode(): output_ids = self.model.generate( **inputs, max_new_tokens=self.max_new_tokens, do_sample=False, repetition_penalty=1.2, no_repeat_ngram_size=5, num_beams=1, use_cache=True, pad_token_id=self.tokenizer.pad_token_id, eos_token_id=self.tokenizer.eos_token_id, ) generated_ids = output_ids[0] if self.is_causal: input_len = inputs["input_ids"].shape[1] generated_ids = generated_ids[input_len:] answer = self.tokenizer.decode(generated_ids, skip_special_tokens=True).strip() # Clean answer fallback logic answer = self.clean_answer(answer, results[0][0]) return answer def build_generator(answer_mode: str, generation_model: str | None, max_new_tokens: int) -> AnswerGenerator: if answer_mode == "extractive": return ExtractiveGenerator() if not generation_model: raise ValueError("--generation-model is required when --answer-mode local_hf") try: return LocalHFGenerator(generation_model, max_new_tokens=max_new_tokens) except Exception as exc: raise RuntimeError(f"Local HF model could not be loaded: {exc}") from exc def first_submitted_question(payload: str, field_name: str) -> str: values = parse_qs(payload, keep_blank_values=True).get(field_name, []) for value in reversed(values): value = value.strip() if value: return value return "" def page( question: str = "", answer: str = "", results: list[tuple[Doc, float]] | None = None, answer_mode: str = "extractive", generation_model: str | None = None, upload_question: str = "", upload_answer: str = "", upload_results: list[tuple[Doc, float]] | None = None, upload_message: str = "", ) -> bytes: results = results or [] upload_results = upload_results or [] sample_buttons = "".join( f"" for q in SAMPLE_QUESTIONS ) source_cards = "".join( f"""

#{rank} | {html.escape(doc.title)}

Skor: {score:.3f}

{html.escape(doc.text[:900])}

{html.escape(doc.citation)}
""" for rank, (doc, score) in enumerate(results, start=1) ) upload_source_cards = "".join( f"""

#{rank} | {html.escape(doc.title)}

Skor: {score:.3f}

{html.escape(doc.text[:900])}

{html.escape(doc.citation)}
""" for rank, (doc, score) in enumerate(upload_results, start=1) ) body = f""" Türkçe Hukuk RAG Asistanı

Türkçe Hukuk RAG Asistanı

BM25 Arama + Kaynağa Dayalı Yanıt + Hukuki Kaynakça

Extractive BM25 Model Üretim (Base)

0.975 Recall@10
0.799 Token F1
0.908 Top-5 Hit
0.813 Kaynak Doğr.

Fine-Tuned QLoRA LLM Deneysel (SFT)

0.975 Recall@10
0.382 Token F1
0.908 Top-5 Hit
0.033 Kaynak Doğr.
Aktif Yanıt Modu: {html.escape(answer_mode)}
{f'
Aktif Model:QLoRA LLM | {html.escape(generation_model)}
' if generation_model else ''}

Hukuki Soru Sorun

{sample_buttons}
{f'

Cevap

{html.escape(answer)}
' if answer else ''} {f'

Getirilen Hukuki Kaynaklar (Top-5)

{source_cards}
' if results else ''}

Özel Doküman Testi

📂
Dokümanınızı buraya sürükleyin veya seçin
Desteklenen dosya türleri: .txt, .md, .docx, .pdf, .csv, .json, .jsonl
Dosya Seç
Seçili dosya yok
{f'
Bilgi: {html.escape(upload_message)}
' if upload_message else ''}
{f'

Doküman Cevabı

{html.escape(upload_answer)}
' if upload_answer else ''} {f'

Dokümandan Elde Edilen Parçalar

{upload_source_cards}
' if upload_results else ''}
""" return body.encode("utf-8") def build_handler(retriever: SimpleBM25, generator: AnswerGenerator, answer_mode: str, generation_model: str | None): class DemoHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(page(answer_mode=answer_mode, generation_model=generation_model)) def do_POST(self) -> None: if self.path == "/upload_ask": self.handle_upload_ask() return length = int(self.headers.get("Content-Length", "0")) payload = self.rfile.read(length).decode("utf-8") question = first_submitted_question(payload, "question") results = retriever.search(question, top_k=5) if question else [] answer = "" if question: try: started = time.perf_counter() answer = generator.generate(question, results) elapsed = time.perf_counter() - started answer = f"{answer}\n\nYanıt süresi: {elapsed:.1f} sn" except Exception as exc: answer = f"LLM yanıt üretirken hata oluştu: {exc}" self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(page(question, answer, results, answer_mode=answer_mode, generation_model=generation_model)) def handle_upload_ask(self) -> None: upload_question = "" upload_answer = "" upload_results: list[tuple[Doc, float]] = [] upload_message = "" try: content_length = int(self.headers.get("Content-Length", "0")) payload = self.rfile.read(content_length) content_type = self.headers.get("Content-Type", "") msg = email.message_from_bytes( b"Content-Type: " + content_type.encode("utf-8") + b"\r\n\r\n" + payload ) if not msg.is_multipart(): raise ValueError("Geçersiz istek: multipart form verisi bekleniyor.") upload_question = "" filename = "" file_payload = b"" for part in msg.get_payload(): name = part.get_param("name", header="Content-Disposition") if name == "upload_question": upload_question = part.get_payload(decode=True).decode("utf-8", errors="ignore").strip() elif name == "custom_file": filename = part.get_filename() file_payload = part.get_payload(decode=True) if not upload_question: raise ValueError("Lutfen yuklenen dokuman icin bir soru yazin.") if not filename: raise ValueError("Lutfen .txt, .md, .docx veya .pdf dosyasi secin.") filename = Path(filename).name text = extract_uploaded_text(filename, file_payload) upload_docs = chunk_uploaded_text(text, filename) if not upload_docs: raise ValueError("Yuklenen dosyadan okunabilir metin cikarilamadi.") custom_retriever = SimpleBM25(upload_docs) upload_results = custom_retriever.search(upload_question, top_k=5) try: started = time.perf_counter() upload_answer = generator.generate(upload_question, upload_results) elapsed = time.perf_counter() - started upload_answer = f"{upload_answer}\n\nYanıt süresi: {elapsed:.1f} sn" except Exception as exc: upload_answer = f"LLM yanıt üretirken hata oluştu: {exc}" upload_message = f"{filename} indexed with {len(upload_docs)} chunks." except Exception as exc: upload_message = f"Upload error: {exc}" self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write( page( answer_mode=answer_mode, generation_model=generation_model, upload_question=upload_question, upload_answer=upload_answer, upload_results=upload_results, upload_message=upload_message, ) ) return DemoHandler def main() -> None: parser = argparse.ArgumentParser(description="Browser demo for Turkish legal RAG") parser.add_argument("--data-dir", type=Path, default=Path("data")) parser.add_argument("--corpus-file", type=Path, default=None) parser.add_argument("--limit", type=int, default=None) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=7860) parser.add_argument("--answer-mode", choices=["extractive", "local_hf"], default="local_hf") parser.add_argument("--generation-model", default="outputs/models/turkish_llm_qlora") parser.add_argument("--max-new-tokens", type=int, default=64) parser.add_argument("--no-browser", action="store_true") args = parser.parse_args() corpus_file = args.corpus_file or resolve_corpus_file(args.data_dir) docs = load_docs(corpus_file, limit=args.limit) retriever = SimpleBM25(docs) generator = build_generator(args.answer_mode, args.generation_model, args.max_new_tokens) server = ThreadingHTTPServer( (args.host, args.port), build_handler(retriever, generator, args.answer_mode, args.generation_model), ) url = f"http://{args.host}:{args.port}" print(f"Loaded {len(docs)} documents from {corpus_file}") print(f"Answer mode: {args.answer_mode}") if args.generation_model: print(f"Generation model: {args.generation_model}") print(f"Demo running at {url}") if not args.no_browser: webbrowser.open(url) server.serve_forever() if __name__ == "__main__": main()