# 기본 RAG 루프 구현
class BasicRAG:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db

    def retrieve(self, query, top_k=5):
        query_embedding = self.embedding_model.encode([query])
        return self.vector_db.search(query_embedding, top_k)

    def generate(self, query, context):
        prompt = f"Context: {context}\n\nQuestion: {query}\n\nAnswer:"
        return self.llm.generate(prompt)

    def run(self, query):
        context = self.retrieve(query)
        return self.generate(query, context)

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticChunker:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)

    def chunk(self, text, max_chunk_size=512):
        sentences = text.split('.')
        embeddings = self.model.encode(sentences)

        # 문장 간 거리 계산 후 의미 있는 단위로 분할
        chunks = []
        current_chunk = []
        current_length = 0

        for sentence, embedding in zip(sentences, embeddings):
            if current_length + len(sentence.split()) > max_chunk_size:
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentence]
                current_length = len(sentence.split())
            else:
                current_chunk.append(sentence)
                current_length += len(sentence.split())

        if current_chunk:
            chunks.append(' '.join(current_chunk))
        return chunks

class RecursiveChunker:
    def __init__(self, chunk_size=512, overlap=50):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk(self, text):
        chunks = []
        start = 0

        while start < len(text):
            end = min(start + self.chunk_size, len(text))
            chunk = text[start:end]

            # 오버랩 추가
            if start > 0:
                overlap_start = max(0, start - self.overlap)
                chunk = text[overlap_start:end]

            chunks.append(chunk)
            start = end - self.overlap

        return chunks

import time
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import torch

class EmbeddingBenchmark:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
            'sentence-tiny': SentenceTransformer('sentence-tiny')
        }

    def benchmark(self, texts):
        results = {}
        for name, model in self.models.items():
            start_time = time.time()
            embeddings = model.encode(texts)
            end_time = time.time()

            results[name] = {
                'time': end_time - start_time,
                'memory': len(embeddings) * len(embeddings[0]) * 4  # float32
            }
        return results

# 사용 예시
benchmark = EmbeddingBenchmark()
test_texts = ["This is test text 1", "This is test text 2", "Another sample text"]
results = benchmark.benchmark(test_texts)

# Chroma 예제
import chromadb
from chromadb.utils import embedding_functions

class ChromaRAG:
    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(
            name="rag_collection",
            embedding_function=embedding_functions.DefaultEmbeddingFunction()
        )

    def add_documents(self, documents, ids):
        self.collection.add(documents=documents, ids=ids)

    def search(self, query, top_k=5):
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k
        )
        return results['documents'][0]

# Qdrant 예제
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

class QdrantRAG:
    def __init__(self):
        self.client = QdrantClient(host="localhost", port=6333)
        self.collection_name = "rag_collection"

    def create_collection(self):
        self.client.recreate_collection(
            collection_name=self.collection_name,
            vectors_config={"size": 384, "distance": "Cosine"}
        )

    def search(self, query, top_k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query,
            limit=top_k
        )
        return [hit.payload['text'] for hit in results]

# pgvector 예제
import psycopg2
from psycopg2.extras import RealDictCursor

class PGVectorRAG:
    def __init__(self, connection_string):
        self.conn = psycopg2.connect(connection_string)

    def search(self, query_embedding, top_k=5):
        with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute("""
                SELECT text, 1 - (embedding <-> %s) as similarity
                FROM documents
                ORDER BY similarity DESC
                LIMIT %s
            """, (query_embedding, top_k))
            return cursor.fetchall()

python
import numpy as np
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.utils import embedding_functions
import logging

class CompleteRAGPipeline:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.embedding_model = SentenceTransformer(model_name)
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(
            name="documents",
            embedding_function=embedding_functions.DefaultEmbeddingFunction()
        )
        self.logger = logging.getLogger(__name__)

    def setup(self, documents, ids):
        """문서 추가 및 임베딩"""
        try:
            embeddings = self.embedding_model.encode(documents)
            self.collection.add(
                documents=documents,
                embeddings=embeddings.tolist(),
                ids=ids
            )
            self.logger.info(f"Added {len(documents)} documents")
        except Exception as e:
            self.logger.error(f"Error adding documents: {e}")

    def retrieve_context(self, query, top_k=5):
        """문서 검색"""
        query_embedding = self.embedding_model.encode([query])
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            include=['documents', 'distances']
        )
        return results['documents'][0], results['distances'][0]

    def generate_answer(self, query, context):
        """답변 생성 (모델 사용)"""
        prompt = f"""
        사용자 질문: {query}


---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)