입력 질의 → 검색기 → 문서 조각 → 증강된 프롬프트 → LLM 생성 → 출력

from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans
import numpy as np

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

    def chunk_by_semantic(self, text, threshold=0.7):
        sentences = text.split('. ')
        embeddings = self.model.encode(sentences)

        # K-means 클러스터링으로 의미적 그룹 생성
        kmeans = KMeans(n_clusters=max(1, len(sentences)//3))
        kmeans.fit(embeddings)

        clusters = {}
        for i, cluster_id in enumerate(kmeans.labels_):
            if cluster_id not in clusters:
                clusters[cluster_id] = []
            clusters[cluster_id].append(sentences[i])

        return [' '.join(cluster) for cluster in clusters.values()]

import re

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

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

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

            # 중간에 문장 분리점 찾기
            if end < len(text):
                separator = chunk.rfind('. ')
                if separator > self.chunk_size * 0.3:
                    end = start + separator + 1
                    chunk = text[start:end]

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

        return chunks

from sentence_transformers import SentenceTransformer
import time

def compare_embedding_models():
    models = {
        'all-MiniLM-L6-v2': 'fast',
        'all-mpnet-base-v2': 'medium',
        'multi-qa-MiniLM-L6-v2': 'fast',
        'paraphrase-multilingual-MiniLM-v2': 'fast'
    }

    test_sentences = [
        "The quick brown fox jumps over the lazy dog.",
        "Machine learning is a subset of artificial intelligence.",
        "Natural language processing enables computers to understand text."
    ]

    results = {}
    for model_name, speed in models.items():
        model = SentenceTransformer(model_name)

        # 성능 측정
        start_time = time.time()
        embeddings = model.encode(test_sentences)
        end_time = time.time()

        results[model_name] = {
            'speed': end_time - start_time,
            'dimensions': len(embeddings[0]),
            'speed_category': speed
        }

    return results

# 사용 예시
model_results = compare_embedding_models()
print("모델 성능 비교:")
for name, metrics in model_results.items():
    print(f"{name}: {metrics['speed']:.2f}s, {metrics['dimensions']}차원")

import chromadb
from qdrant_client import QdrantClient
import psycopg2
import numpy as np

class VectorDBComparison:
    def __init__(self):
        # Chroma
        self.chroma_client = chromadb.Client()
        self.chroma_collection = self.chroma_client.get_or_create_collection(
            name="rag_test"
        )

        # Qdrant
        self.qdrant_client = QdrantClient(host='localhost', port=6333)

        # pgvector (PostgreSQL)
        self.pg_conn = psycopg2.connect(
            host="localhost", 
            database="rag_db", 
            user="user", 
            password="password"
        )

    def chroma_insert(self, documents, embeddings):
        self.chroma_collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=[f"doc_{i}" for i in range(len(documents))]
        )

    def qdrant_insert(self, documents, embeddings):
        self.qdrant_client.upsert(
            collection_name="rag_test",
            points=[
                {
                    "id": i,
                    "vector": emb.tolist(),
                    "payload": {"text": doc}
                }
                for i, (doc, emb) in enumerate(zip(documents, embeddings))
            ]
        )

    def pgvector_insert(self, documents, embeddings):
        with self.pg_conn.cursor() as cursor:
            for i, (doc, emb) in enumerate(zip(documents, embeddings)):
                cursor.execute(
                    "INSERT INTO documents (id, content, embedding) VALUES (%s, %s, %s)",
                    (i, doc, np.array(emb).tobytes())
                )
            self.pg_conn.commit()

import chromadb
from sentence_transformers import SentenceTransformer
from typing import List, Dict
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SimpleRAGPipeline:
    def __init__(self, chunk_size: int = 512):
        self.chunker = RecursiveChunker(chunk_size=chunk_size)
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        self.chroma_client = chromadb.Client()
        self.collection = self.chroma_client.get_or_create_collection(
            name="documents"
        )

    def add_documents(self, documents: List[str]):
        """문서 추가 및 임베딩"""
        all_chunks = []
        all_texts = []

        # 문서 청킹
        for doc in documents:
            chunks = self.chunker.chunk_recursive(doc)
            all_chunks.extend(chunks)
            all_texts.extend([doc] * len(chunks))

        # 임베딩 생성
        embeddings = self.embedder.encode(all_chunks)

        # Chroma에 저장
        self.collection.add(
            documents=all_chunks,
            embeddings=embeddings.tolist(),
            ids=[f"chunk_{i}" for i in range(len(all_chunks))]
        )

    def retrieve(self, query: str, top_k: int = 3) -> List[Dict]:
        """검색"""
        query_embedding = self.embedder.encode([query])
        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=top_k
        )

        return [
            {
                'text': text,
                'score': score
            }
            for text, score in zip(results['documents'][0], results['distances'][0])
        ]

    def generate_response(self, query: str, retrieved_docs: List[Dict]) -> str:
        """응답 생성"""
        context = "\n".join([doc['text'] for doc in retrieved_docs])
        prompt = f"""
        주어진 컨텍스트를 기반으로 질문에 답변해주세요.

        컨텍스트:
        {context}

        질문: {query}
        답변:
        """

        # 간단한 LLM 호출 예시 (실제 구현에서는 실제 LLM 사용)
        return f"검색된 컨텍스트 기반 답변: {query}에 대한 정보는 {len(retrieved_docs)}개 문서에 포함되어 있습니다."

# 사용 예시
rag = SimpleRAGPipeline()

# 문서 추가
documents = [
    "Python은 고급 프로그래밍 언어로, 간결하고 읽기 쉬운 문법을 갖추고 있습니다.",
    "Django는 파이썬 기반 웹 프레임워크로, 빠르고 안전한 웹 개발을 지원합니다.",
    "React는 JavaScript 라이브러리로, 사용자 인터페이스 개발에 최적화되어 있습니다."
]

rag.add_documents(documents)

# 검색 및 응답 생성
query = "Python 프로그래밍 언어의 특징은?"
retrieved = rag.retrieve(query)
response = rag.generate_response(query, retrieved)
print(f"질의: {query}")
print(f"응답: {response}")

python
class QueryTransformer:
    def __init__(self):
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')

    def transform_query(self, original_query: str) -> List[str]:
        """질의 변환

---

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