RAG Veri Hazırlama: Yapay Zeka Sisteminin Başarısını Belirleyen Temel

from docling.document_converter import DocumentConverterfrom docling.datamodel.base_models import InputFormat
converter = DocumentConverter()
# Layout analizi ile PDF parse etresult = converter.convert("technical-manual.pdf")
# Yapılandırılmış içeriğe erişfor element in result.document.body:    if element.type == "table":        # Tablo yapısı korunarak çıkarıldı        markdown_table = element.export_to_markdown()    elif element.type == "text":        # Bölüm bağlamıyla metin        text = element.text        section = element.section_header
# RAG ingestion için markdown olarak export etmarkdown_output = result.document.export_to_markdown()

from bs4 import BeautifulSoupfrom readability import Document
def extract_html_content(html: str) -> dict:    """Çeşitli yapıları handle ederek HTML'den anlamlı içerik çıkar."""
    # Ana içerik çıkarma için readability kullan    doc = Document(html)    main_content = doc.summary()    title = doc.title()
    # Yapı için BeautifulSoup ile parse et    soup = BeautifulSoup(main_content, 'html.parser')
    # Navigasyon, reklamlar, footer'ları kaldır    for element in soup.find_all(['nav', 'footer', 'aside', 'script', 'style']):        element.decompose()
    # Yapıyı koruyarak metin çıkar    text_blocks = []    for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'li', 'td']):        text = element.get_text(strip=True)        if text:            text_blocks.append({                'type': element.name,                'text': text,                'level': int(element.name[1]) if element.name.startswith('h') else 0            })
    return {        'title': title,        'blocks': text_blocks,        'full_text': soup.get_text(separator='\n', strip=True)    }

import refrom typing import Listimport unicodedata
class TextPreprocessor:    def __init__(self):        self.pii_patterns = {            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',            'phone': r'\b\d{3}[-.\s]?\d{3}[-.\s]?\d{4}\b',            'ssn': r'\b\d{3}-\d{2}-\d{4}\b',            'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',        }
    def normalize_whitespace(self, text: str) -> str:        """Tüm boşlukları tek boşluğa normalize et."""        text = re.sub(r'\s+', ' ', text)        return text.strip()
    def normalize_unicode(self, text: str) -> str:        """Unicode karakterleri tutarlı forma normalize et."""        return unicodedata.normalize('NFKC', text)
    def redact_pii(self, text: str) -> str:        """PII pattern'lerini tespit et ve maskele."""        for pii_type, pattern in self.pii_patterns.items():            text = re.sub(pattern, f'[MASKELENDI_{pii_type.upper()}]', text)        return text
    def remove_boilerplate(self, text: str, patterns: List[str] = None) -> str:        """Bilinen şablon metinleri kaldır."""        default_patterns = [            r'Page \d+ of \d+',            r'Copyright \d{4}.*?(?=\n|$)',            r'All rights reserved\.?',        ]        patterns = patterns or default_patterns        for pattern in patterns:            text = re.sub(pattern, '', text, flags=re.IGNORECASE)        return text
    def process(self, text: str, redact_pii: bool = True) -> str:        """Tam ön-işleme pipeline'ını çalıştır."""        text = self.normalize_unicode(text)        text = self.remove_boilerplate(text)        text = self.normalize_whitespace(text)        if redact_pii:            text = self.redact_pii(text)        return text

from datasketch import MinHash, MinHashLSHimport hashlibfrom typing import List, Set
class Deduplicator:    def __init__(self, threshold: float = 0.8, num_perm: int = 128):        self.threshold = threshold        self.num_perm = num_perm        self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)        self.exact_hashes: Set[str] = set()        self.doc_id = 0
    def _compute_minhash(self, text: str) -> MinHash:        """Metin için MinHash imzası hesapla."""        minhash = MinHash(num_perm=self.num_perm)        words = text.lower().split()        for i in range(len(words) - 2):            shingle = ' '.join(words[i:i+3])            minhash.update(shingle.encode('utf-8'))        return minhash
    def is_duplicate(self, text: str) -> bool:        """Tam veya yakın kopyaları kontrol et."""        # Tam kopya kontrolü        text_hash = hashlib.md5(text.encode()).hexdigest()        if text_hash in self.exact_hashes:            return True        self.exact_hashes.add(text_hash)
        # Yakın-kopya kontrolü        minhash = self._compute_minhash(text)        if self.lsh.query(minhash):            return True
        self.lsh.insert(f"doc_{self.doc_id}", minhash)        self.doc_id += 1        return False
    def deduplicate(self, documents: List[str]) -> List[str]:        """Doküman listesinden kopyaları kaldır."""        return [doc for doc in documents if not self.is_duplicate(doc)]

from langchain_text_splitters import RecursiveCharacterTextSplitter
def recursive_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> list:    """    Ayırıcı hiyerarşisi kullanarak metni recursive olarak böl.    Önce paragrafları, sonra cümleleri, sonra kelimeleri bir arada tutmaya çalışır.    """    splitter = RecursiveCharacterTextSplitter(        separators=[            "\n\n",     # Önce paragraflar            "\n",       # Sonra satır sonları            ". ",       # Sonra cümleler            ", ",       # Sonra cümlecikler            " ",        # Son olarak kelimeler        ],        chunk_size=chunk_size,        chunk_overlap=overlap,        length_function=len,    )    return splitter.split_text(text)

from langchain_experimental.text_splitter import SemanticChunkerfrom langchain_openai.embeddings import OpenAIEmbeddings
def semantic_chunking(text: str) -> list:    """    Cümleler arası semantik benzerliğe göre böl.    Semantik olarak ilişkili içeriği gruplar.    """    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    splitter = SemanticChunker(        embeddings=embeddings,        breakpoint_threshold_type="percentile",        breakpoint_threshold_amount=95,  # En üst %5 farklılıkta böl        min_chunk_size=100    )
    return splitter.split_text(text)
# Performans: Retrieval'da %70'e kadar doğruluk iyileştirmesi (içerik tipine göre değişir)# Trade-off: Chunking sırasında embedding çağrıları gerektirir

from langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain.retrievers import ParentDocumentRetrieverfrom langchain.storage import InMemoryStorefrom langchain_chroma import Chroma
def setup_hierarchical_chunking(documents: list, embeddings):    """    Hassasiyet + bağlam için parent-child hiyerarşisi oluştur.    Küçük child chunk'larda ara, büyük parent chunk'ları döndür.    """    # Parent splitter: bağlam için büyük chunk'lar    parent_splitter = RecursiveCharacterTextSplitter(        chunk_size=2000,        chunk_overlap=200    )
    # Child splitter: hassas retrieval için küçük chunk'lar    child_splitter = RecursiveCharacterTextSplitter(        chunk_size=400,        chunk_overlap=50    )
    # Parent dokümanlar için depolama    docstore = InMemoryStore()
    # Vektör deposu child chunk'ları indeksler    vectorstore = Chroma(        collection_name="child_chunks",        embedding_function=embeddings    )
    # Retriever child'larda arar, parent'ları döndürür    retriever = ParentDocumentRetriever(        vectorstore=vectorstore,        docstore=docstore,        child_splitter=child_splitter,        parent_splitter=parent_splitter    )
    retriever.add_documents(documents)    return retriever
# Performans: Bağlamı koruyarak yapılandırılmış dokümanlarda iyileştirilmiş alaka# Trade-off: 2-3x depolama overhead'i

from anthropic import Anthropicfrom typing import List
def add_contextual_headers(    document: str,    chunks: List[str],    model: str = "claude-3-5-haiku-latest") -> List[str]:    """    Claude kullanarak chunk'a özel bağlam ekle.    Contextual embedding'ler tek başına retrieval başarısızlıklarını %35 azaltır,    BM25 hybrid search ile birlikte %49, reranking eklendiğinde ise %67.    """    client = Anthropic()    contextualized_chunks = []
    context_prompt = """İşte tam doküman:<document>{document}</document>
İşte o dokümandan bir chunk:<chunk>{chunk}</chunk>
Bu chunk'ı doküman içinde konumlandırmak için kısa bir bağlam (2-3 cümle) sağla. Şunlara odaklan:1. Bu chunk hangi bölüme/konuya ait2. Tartışılan ana varlıklar veya kavramlar3. Dokümanın ana konusuyla nasıl ilişkili
Bağlam:"""
    for chunk in chunks:        response = client.messages.create(            model=model,            max_tokens=200,            messages=[{                "role": "user",                "content": context_prompt.format(document=document, chunk=chunk)            }]        )
        context = response.content[0].text        contextualized_chunks.append(f"{context}\n\n{chunk}")
    return contextualized_chunks
# Prompt caching ile maliyet: ~milyon doküman token başına $1.02

def add_structural_context(chunks: List[dict]) -> List[dict]:    """    LLM çağrısı olmadan doküman yapısına dayalı bağlam ekle.    Yapı-duyarlı chunking'den metadata kullanır.    """    contextualized = []
    for chunk in chunks:        metadata = chunk.get('metadata', {})        content = chunk['content']
        context_parts = []        if 'document_title' in metadata:            context_parts.append(f"Kaynak: {metadata['document_title']}")        if 'header_1' in metadata:            context_parts.append(f"Bölüm: {metadata['header_1']}")        if 'header_2' in metadata:            context_parts.append(f"Alt Bölüm: {metadata['header_2']}")
        context = " | ".join(context_parts)        contextualized.append({            'content': f"{context}\n\n{content}" if context else content,            'metadata': metadata        })
    return contextualized

from typing import Listimport numpy as npfrom openai import OpenAI
client = OpenAI()
def get_embeddings(    texts: List[str],    dimensions: int = 1024) -> List[List[float]]:    """    Boyut azaltma ile OpenAI embedding'leri al.    256-dim text-embedding-3-large tam ada-002'yi geçer.    """    response = client.embeddings.create(        model="text-embedding-3-large",        input=texts,        dimensions=dimensions  # Matryoshka truncation    )    return [item.embedding for item in response.data]

def batch_embed_with_normalization(    texts: List[str],    batch_size: int = 100,    dimensions: int = 1024) -> np.ndarray:    """    L2 normalizasyonuyla metinleri batch'ler halinde embed et.    Normalizasyon dot product ile cosine similarity sağlar.    """    all_embeddings = []
    for i in range(0, len(texts), batch_size):        batch = texts[i:i + batch_size]        embeddings = get_embeddings(batch, dimensions)        all_embeddings.extend(embeddings)
    embeddings_array = np.array(all_embeddings)
    # Cosine similarity için L2 normalize et    norms = np.linalg.norm(embeddings_array, axis=1, keepdims=True)    return embeddings_array / norms

from dataclasses import dataclassfrom typing import List, Optionalfrom datetime import datetime
@dataclassclass ChunkMetadata:    # İçerik tabanlı    keywords: List[str]    entities: List[str]    content_type: str
    # Yapısal    document_title: str    section_header: Optional[str]    chunk_index: int
    # Bağlamsal    source_url: Optional[str]    ingestion_date: datetime    language: str
    # Teknik    word_count: int    has_code: bool    has_table: bool

import spacyfrom collections import Counterfrom typing import Dict, List
class MetadataExtractor:    def __init__(self):        self.nlp = spacy.load("en_core_web_sm")
    def extract_entities(self, text: str) -> Dict[str, List[str]]:        """spaCy kullanarak adlandırılmış varlıkları çıkar."""        doc = self.nlp(text)        entities = {}        for ent in doc.ents:            if ent.label_ not in entities:                entities[ent.label_] = []            entities[ent.label_].append(ent.text)        return entities
    def extract_keywords(self, text: str, top_n: int = 10) -> List[str]:        """İsim öbeklerini kullanarak anahtar kelimeler çıkar."""        doc = self.nlp(text)        noun_chunks = [chunk.text.lower() for chunk in doc.noun_chunks]        chunk_counts = Counter(noun_chunks)        return [word for word, _ in chunk_counts.most_common(top_n)]
    def detect_content_type(self, text: str) -> str:        """Sezgisel içerik tipi tespiti."""        code_patterns = ['def ', 'function ', 'class ', 'import ', '```']        if any(pattern in text for pattern in code_patterns):            return 'code'
        tech_indicators = ['API', 'database', 'server', 'deployment']        if sum(1 for ind in tech_indicators if ind.lower() in text.lower()) >= 2:            return 'technical'
        return 'general'

from qdrant_client import QdrantClientfrom qdrant_client.models import PointStruct, VectorParams, Distance, Filter, FieldCondition, MatchValue
def store_chunks_with_metadata(    chunks: List[str],    embeddings: List[List[float]],    metadata_list: List[dict],    collection_name: str = "documents"):    """Zengin metadata ile chunk'ları Qdrant'ta sakla."""    client = QdrantClient(host="localhost", port=6333)
    client.recreate_collection(        collection_name=collection_name,        vectors_config=VectorParams(            size=len(embeddings[0]),            distance=Distance.COSINE        )    )
    points = [        PointStruct(            id=idx,            vector=embedding,            payload={"text": chunk, **metadata}        )        for idx, (chunk, embedding, metadata)        in enumerate(zip(chunks, embeddings, metadata_list))    ]
    client.upsert(collection_name=collection_name, points=points)

def search_with_filter(    query_embedding: List[float],    collection_name: str,    content_type: str = None,    top_k: int = 10) -> List[dict]:    """Opsiyonel metadata filtresi ile ara."""    client = QdrantClient(host="localhost", port=6333)
    query_filter = None    if content_type:        query_filter = Filter(            must=[FieldCondition(                key="content_type",                match=MatchValue(value=content_type)            )]        )
    results = client.search(        collection_name=collection_name,        query_vector=query_embedding,        query_filter=query_filter,        limit=top_k    )
    return [{"text": hit.payload["text"], "score": hit.score} for hit in results]

import numpy as npfrom sklearn.metrics.pairwise import cosine_similarityfrom typing import List
def evaluate_chunk_coherence(chunks: List[str], embedding_model) -> dict:    """    Chunk'lar içindeki semantik tutarlılığı ölç.    Yüksek tutarlılık = chunk tek konu tartışıyor.    """    coherence_scores = []
    for chunk in chunks:        sentences = chunk.split('. ')        if len(sentences) < 2:            coherence_scores.append(1.0)            continue
        embeddings = np.array(embedding_model.embed(sentences))        similarities = cosine_similarity(embeddings)
        n = len(sentences)        coherence = (similarities.sum() - n) / (n * (n - 1)) if n > 1 else 1.0        coherence_scores.append(coherence)
    return {        'mean_coherence': np.mean(coherence_scores),        'min_coherence': np.min(coherence_scores),        'low_coherence_count': sum(1 for s in coherence_scores if s < 0.5)    }

def evaluate_boundary_quality(chunks: List[str]) -> dict:    """Chunk'ların temiz sınırları olup olmadığını kontrol et."""    bad_starts = 0    bad_ends = 0
    lowercase_starters = ['and', 'but', 'or', 'so', 'because', 'however']
    for chunk in chunks:        first_word = chunk.split()[0].lower() if chunk.split() else ''        if first_word in lowercase_starters:            bad_starts += 1
        if chunk and chunk.rstrip()[-1] not in '.!?:':            bad_ends += 1
    return {        'bad_start_ratio': bad_starts / len(chunks),        'bad_end_ratio': bad_ends / len(chunks),        'clean_boundary_ratio': 1 - (bad_starts + bad_ends) / (2 * len(chunks))    }

def evaluate_retrieval_quality(    embeddings: np.ndarray,    test_queries: List[str],    relevant_chunk_ids: List[List[int]],    embedding_model) -> dict:    """Retrieval testleri kullanarak embedding kalitesini değerlendir."""    query_embeddings = np.array(embedding_model.embed(test_queries))    similarities = cosine_similarity(query_embeddings, embeddings)
    hits_at_k = {1: 0, 5: 0, 10: 0}    mrr_sum = 0
    for i, relevant_ids in enumerate(relevant_chunk_ids):        ranked_indices = np.argsort(similarities[i])[::-1]
        for rank, idx in enumerate(ranked_indices):            if idx in relevant_ids:                mrr_sum += 1 / (rank + 1)                for k in hits_at_k:                    if rank < k:                        hits_at_k[k] += 1                break
    n_queries = len(test_queries)    return {        'mrr': mrr_sum / n_queries,        'hit_rate@1': hits_at_k[1] / n_queries,        'hit_rate@5': hits_at_k[5] / n_queries,        'hit_rate@10': hits_at_k[10] / n_queries    }