-- 일반 DB에서 검색
SELECT * FROM documents 
WHERE title LIKE '%파이썬%' 
   OR content LIKE '%파이썬%';
 
-- 결과: "파이썬"이라는 단어가 있는 문서만 나옴
-- "Python", "프로그래밍", "코딩"은 검색되지 않음

# 벡터 DB에서 검색
query = "파이썬 학습 방법"
# 내부적으로 벡터로 변환: [0.2, -0.1, 0.8, 0.3, ...]
 
results = vectorstore.similarity_search(query, k=5)
# 결과: 의미적으로 유사한 문서들이 점수순으로 나옴
# - "Python 프로그래밍 입문"
# - "파이썬으로 시작하는 프로그래밍" 
# - "초보자를 위한 Python 가이드"
# - "프로그래밍 언어 학습법"

# 텍스트를 벡터로 변환하는 예시
from sentence_transformers import SentenceTransformer
 
model = SentenceTransformer('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2')
 
texts = [
    "파이썬은 프로그래밍 언어입니다",
    "Python is a programming language", 
    "자바스크립트로 웹 개발을 합니다",
    "고양이가 소파에서 잠을 잡니다"
]
 
embeddings = model.encode(texts)
print(f"벡터 차원: {embeddings.shape}")
# 출력: (4, 384) - 4개 문장, 각각 384차원 벡터
 
# 유사도 계산
from sklearn.metrics.pairwise import cosine_similarity
 
similarity = cosine_similarity(embeddings)
print("유사도 매트릭스:")
print(similarity)
 
# 결과 예시 (값은 설명용 가상 수치):
#           문장1  문장2  문장3  문장4
# 문장1    1.00   0.85   0.23   0.12
# 문장2    0.85   1.00   0.21   0.09  
# 문장3    0.23   0.21   1.00   0.15
# 문장4    0.12   0.09   0.15   1.00

king_vector - man_vector + woman_vector ≈ queen_vector

from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
 
embeddings = HuggingFaceEmbeddings(model_name="jhgan/ko-sroberta-multitask")
 
# 로컬에 저장
vectorstore = Chroma(
    collection_name="my_documents",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

from langchain.vectorstores import FAISS
 
# 인덱스를 메모리에 올려 검색
vectorstore = FAISS.from_documents(
    documents=doc_chunks,
    embedding=embeddings
)
 
# 로컬 파일로 저장
vectorstore.save_local("faiss_index")
 
# 불러오기
vectorstore = FAISS.load_local("faiss_index", embeddings)

import pinecone
from langchain.vectorstores import Pinecone
 
# Pinecone 초기화
pinecone.init(
    api_key="your-api-key",
    environment="your-env"
)
 
# 인덱스 생성
index = pinecone.Index("my-index")
 
vectorstore = Pinecone(
    index=index,
    embedding_function=embeddings.embed_query,
    text_key="text"
)

import weaviate
from langchain.vectorstores import Weaviate
 
client = weaviate.Client("http://localhost:8080")
 
vectorstore = Weaviate(
    client=client,
    index_name="Document",
    text_key="content",
    embedding=embeddings
)

import os
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
 
class VectorDBBuilder:
    def __init__(self, model_name="jhgan/ko-sroberta-multitask"):
        print("🔧 벡터 DB 빌더 초기화 중...")
        self.embeddings = HuggingFaceEmbeddings(
            model_name=model_name,
            model_kwargs={'device': 'cpu'},  # GPU 사용시 'cuda'
            encode_kwargs={'normalize_embeddings': True}  # 벡터 길이를 1로 정규화 (코사인 유사도 계산에 적합)
        )
        
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
        )
        
    def load_documents(self, directory_path):
        """다양한 형식의 문서 로딩"""
        print(f"📁 {directory_path}에서 문서 로딩 중...")
        
        loader = DirectoryLoader(
            directory_path,
            glob="**/*.txt",  # .txt, .md, .py 등 확장
            loader_cls=TextLoader,
            loader_kwargs={'encoding': 'utf-8'}  # 한글 인코딩
        )
        
        documents = loader.load()
        print(f"✅ {len(documents)}개 문서 로딩 완료")
        
        return documents
    
    def split_documents(self, documents):
        """문서를 청크로 분할"""
        print("✂️ 문서 분할 중...")
        
        doc_chunks = self.text_splitter.split_documents(documents)
        
        # 메타데이터 추가
        for i, chunk in enumerate(doc_chunks):
            chunk.metadata['chunk_id'] = i
            chunk.metadata['chunk_size'] = len(chunk.page_content)
            
        print(f"✅ {len(doc_chunks)}개 청크 생성 완료")
        return doc_chunks
    
    def build_vectorstore(self, doc_chunks, persist_directory="./chroma_db"):
        """벡터 스토어 구축"""
        print("🔢 벡터화 및 저장 중...")
        
        # 기존 DB가 있으면 로드, 없으면 새로 생성
        if os.path.exists(persist_directory):
            print("📂 기존 벡터 DB 발견, 업데이트 모드...")
            vectorstore = Chroma(
                persist_directory=persist_directory,
                embedding_function=self.embeddings
            )
            # 새 문서 추가
            vectorstore.add_documents(doc_chunks)
        else:
            print("🆕 새로운 벡터 DB 생성 중...")
            vectorstore = Chroma.from_documents(
                documents=doc_chunks,
                embedding=self.embeddings,
                persist_directory=persist_directory
            )
        
        # 디스크에 저장
        vectorstore.persist()
        print(f"✅ 벡터 DB 저장 완료: {persist_directory}")
        
        return vectorstore
 
# 사용 예시
def main():
    # 1. 빌더 초기화
    builder = VectorDBBuilder()
    
    # 2. 문서 로딩
    documents = builder.load_documents("./knowledge_base")
    
    # 3. 문서 분할
    chunks = builder.split_documents(documents)
    
    # 4. 벡터 스토어 구축
    vectorstore = builder.build_vectorstore(chunks)
    
    # 5. 검색 테스트
    print("\n🔍 검색 테스트:")
    test_queries = [
        "파이썬 설치 방법",
        "에러 해결 방법",
        "라이브러리 사용법"
    ]
    
    for query in test_queries:
        print(f"\n질문: {query}")
        results = vectorstore.similarity_search_with_score(query, k=3)
        
        for i, (doc, score) in enumerate(results, 1):
            # Chroma 기본 점수는 거리라서 작을수록 더 유사
            print(f"{i}. 점수: {score:.3f}")
            print(f"   출처: {doc.metadata.get('source', '알 수 없음')}")
            print(f"   내용: {doc.page_content[:100]}...")
 
if __name__ == "__main__":
    main()

def advanced_search_demo(vectorstore):
    """다양한 검색 방법 시연"""
    
    query = "파이썬으로 웹 크롤링 하는 방법"
    
    print("1️⃣ 기본 유사도 검색")
    basic_results = vectorstore.similarity_search(query, k=3)
    for doc in basic_results:
        print(f"- {doc.page_content[:80]}...")
    
    print("\n2️⃣ 점수와 함께 검색")
    scored_results = vectorstore.similarity_search_with_score(query, k=3)
    for doc, score in scored_results:
        print(f"- 점수: {score:.3f} | {doc.page_content[:60]}...")
    
    print("\n3️⃣ MMR(최대 주변 관련성) 검색")
    # 다양성을 고려한 검색 - 비슷한 문서들 대신 다양한 관점의 문서 선택
    mmr_results = vectorstore.max_marginal_relevance_search(
        query, 
        k=3,
        lambda_mult=0.7  # 1에 가까울수록 관련성 우선, 0에 가까울수록 다양성 우선
    )
    for doc in mmr_results:
        print(f"- {doc.page_content[:80]}...")
    
    print("\n4️⃣ 메타데이터 필터 검색")
    # 특정 조건의 문서만 검색
    filtered_results = vectorstore.similarity_search(
        query,
        k=3,
        filter={"source": "python_tutorial.txt"}  # 특정 파일만
    )
    for doc in filtered_results:
        print(f"- {doc.metadata['source']} | {doc.page_content[:60]}...")
 
# 실행
advanced_search_demo(vectorstore)

import time
import psutil
from typing import List
 
class VectorDBAnalyzer:
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
    
    def measure_search_performance(self, queries: List[str], k: int = 5):
        """검색 성능 측정"""
        print("⚡ 검색 성능 분석 중...")
        
        total_time = 0
        memory_before = psutil.virtual_memory().used / 1024 / 1024  # MB
        
        for query in queries:
            start_time = time.time()
            
            results = self.vectorstore.similarity_search(query, k=k)
            
            end_time = time.time()
            query_time = end_time - start_time
            total_time += query_time
            
            print(f"📊 '{query[:30]}...' - {query_time:.3f}초, {len(results)}개 결과")
        
        memory_after = psutil.virtual_memory().used / 1024 / 1024  # MB
        avg_time = total_time / len(queries)
        memory_used = memory_after - memory_before
        
        print(f"\n📈 성능 요약:")
        print(f"   평균 검색 시간: {avg_time:.3f}초")
        print(f"   총 검색 시간: {total_time:.3f}초") 
        print(f"   메모리 사용량: {memory_used:.1f}MB")
        
        return {
            "avg_time": avg_time,
            "total_time": total_time,
            "memory_used": memory_used
        }
    
    def analyze_embedding_quality(self, sample_queries: List[str]):
        """임베딩 품질 분석"""
        print("🎯 임베딩 품질 분석 중...")
        
        for query in sample_queries:
            print(f"\n🔍 '{query}' 검색 결과:")
            
            results = self.vectorstore.similarity_search_with_score(query, k=5)
            
            for i, (doc, score) in enumerate(results, 1):
                # 거리 기준(작을수록 유사). 임계값은 예시이며 모델·데이터에 맞게 조정
                relevance = "🔥 매우 관련" if score < 0.3 else "👍 관련" if score < 0.6 else "🤔 약간 관련"
                print(f"   {i}. {relevance} (점수: {score:.3f})")
                print(f"      {doc.page_content[:70]}...")
 
# 사용 예시
analyzer = VectorDBAnalyzer(vectorstore)
 
test_queries = [
    "파이썬 설치하는 법",
    "크롤링 라이브러리 추천",
    "데이터베이스 연결 방법",
    "API 사용법 알려줘"
]
 
# 성능 측정
performance = analyzer.measure_search_performance(test_queries)
 
# 품질 분석
analyzer.analyze_embedding_quality(test_queries[:2])

# 다양한 임베딩 모델 비교
embedding_models = {
    "한국어 특화": "jhgan/ko-sroberta-multitask",
    "다국어": "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    "영어 고성능": "sentence-transformers/all-mpnet-base-v2",
    "경량화": "sentence-transformers/all-MiniLM-L6-v2"
}
 
def compare_embedding_models(text_samples):
    """임베딩 모델 성능 비교"""
    results = {}
    
    for model_name, model_path in embedding_models.items():
        print(f"🧪 {model_name} 모델 테스트 중...")
        
        start_time = time.time()
        embeddings = HuggingFaceEmbeddings(model_name=model_path)
        
        # 샘플 텍스트 임베딩
        vectors = embeddings.embed_documents(text_samples)
        
        end_time = time.time()
        
        results[model_name] = {
            "time": end_time - start_time,
            "dimension": len(vectors[0]),
            "model_path": model_path
        }
        
        print(f"   ⏱️ 시간: {results[model_name]['time']:.2f}초")
        print(f"   📐 차원: {results[model_name]['dimension']}차원")
    
    return results
 
# 테스트 실행
sample_texts = [
    "파이썬은 간단하고 읽기 쉬운 프로그래밍 언어입니다",
    "머신러닝과 데이터 분석에 널리 사용됩니다",
    "Django와 Flask는 인기 있는 웹 프레임워크입니다"
]
 
model_comparison = compare_embedding_models(sample_texts)

def optimize_chunking_strategy(documents):
    """다양한 청킹 전략 비교"""
    
    strategies = {
        "기본": RecursiveCharacterTextSplitter(
            chunk_size=1000, 
            chunk_overlap=200
        ),
        "작은 청크": RecursiveCharacterTextSplitter(
            chunk_size=500, 
            chunk_overlap=100
        ),
        "큰 청크": RecursiveCharacterTextSplitter(
            chunk_size=2000, 
            chunk_overlap=400
        ),
        "문장 단위 우선": RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ".", "!", "?"]  # 문장 단위 우선
        )
    }
    
    results = {}
    
    for strategy_name, splitter in strategies.items():
        chunks = splitter.split_documents(documents)
        
        # 청크 크기 분포 분석
        chunk_sizes = [len(chunk.page_content) for chunk in chunks]
        
        results[strategy_name] = {
            "total_chunks": len(chunks),
            "avg_size": sum(chunk_sizes) / len(chunk_sizes),
            "min_size": min(chunk_sizes),
            "max_size": max(chunk_sizes)
        }
        
        print(f"📊 {strategy_name} 전략:")
        print(f"   청크 수: {results[strategy_name]['total_chunks']}")
        print(f"   평균 크기: {results[strategy_name]['avg_size']:.0f}자")
        print(f"   크기 범위: {results[strategy_name]['min_size']} ~ {results[strategy_name]['max_size']}")
        print()
    
    return results
 
# 실행
chunking_analysis = optimize_chunking_strategy(documents)

# Chroma 인덱스 설정 최적화
def create_optimized_vectorstore(documents, embeddings):
    """최적화된 벡터 스토어 생성"""
    
    # HNSW 알고리즘 설정 (Hierarchical Navigable Small World)
    collection_metadata = {
        "hnsw:space": "cosine",           # 코사인 유사도 사용
        "hnsw:M": 16,                     # 노드당 연결 수 (클수록 정확도↑, 메모리↑)
        "hnsw:construction_ef": 200,      # 구축 시 탐색 범위
        "hnsw:search_ef": 100,            # 검색 시 탐색 범위
    }
    
    vectorstore = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory="./optimized_chroma_db",
        collection_metadata=collection_metadata
    )
    
    return vectorstore
 
# FAISS 인덱스 최적화
def create_faiss_index(documents, embeddings):
    """FAISS를 이용한 고성능 인덱스"""
    import faiss
    
    # 문서를 벡터로 변환
    texts = [doc.page_content for doc in documents]
    import numpy as np
    doc_embeddings = np.array(embeddings.embed_documents(texts), dtype="float32")  # FAISS는 float32 배열 필요
    
    # FAISS 인덱스 생성
    dimension = doc_embeddings.shape[1]
    
    # IVF (Inverted File) 인덱스 - 대용량 데이터용
    nlist = max(1, min(int(len(documents) / 10), 1000))  # 클러스터 수
    index = faiss.IndexIVFFlat(
        faiss.IndexFlatIP(dimension),  # Inner Product
        dimension, 
        nlist,
        faiss.METRIC_INNER_PRODUCT  # 지정하지 않으면 L2 거리로 동작
    )
    
    # 훈련 및 추가
    index.train(doc_embeddings)
    index.add(doc_embeddings)
    
    return index, texts
 
# 사용
optimized_vectorstore = create_optimized_vectorstore(chunks, embeddings)

class CompanyDocumentSearcher:
    """회사 문서 전용 검색 시스템"""
    
    def __init__(self, company_name="우리회사"):
        self.company_name = company_name
        self.vectorstore = None
        
        # 한국어 특화 임베딩 모델
        self.embeddings = HuggingFaceEmbeddings(
            model_name="jhgan/ko-sroberta-multitask",
            encode_kwargs={'normalize_embeddings': True}
        )
        
    def build_company_db(self, document_dirs):
        """회사 문서 DB 구축"""
        print(f"🏢 {self.company_name} 문서 DB 구축 시작...")
        
        all_documents = []
        
        # 부서별 문서 로딩
        for dept, dir_path in document_dirs.items():
            print(f"📁 {dept} 문서 로딩 중...")
            
            loader = DirectoryLoader(
                dir_path,
                glob="**/*.txt",  # 중괄호 패턴은 지원되지 않음. 형식별로 따로 로드
                loader_cls=TextLoader,
                loader_kwargs={'encoding': 'utf-8'},
                show_progress=True
            )
            docs = loader.load()
            
            # 부서 정보를 메타데이터에 추가
            for doc in docs:
                doc.metadata.update({
                    "department": dept,
                    "company": self.company_name,
                    "doc_type": "internal",
                    "last_updated": "2025-01-01"
                })
            
            all_documents.extend(docs)
            print(f"   ✅ {len(docs)}개 문서 로딩 완료")
        
        # 텍스트 분할
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n## ", "\n### ", "\n\n", "\n", ".", "!", "?", " "]
        )
        
        chunks = text_splitter.split_documents(all_documents)
        
        # 벡터 스토어 구축
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=f"./company_db_{self.company_name.lower()}",
            collection_metadata={
                "hnsw:space": "cosine",
                "hnsw:M": 16
            }
        )
        
        print(f"🎉 총 {len(chunks)}개 청크로 DB 구축 완료!")
        
    def search_by_department(self, query, department=None, k=5):
        """부서별 문서 검색"""
        if not self.vectorstore:
            return "벡터 DB가 구축되지 않았습니다."
        
        # 부서 필터링
        filter_dict = {}
        if department:
            filter_dict["department"] = department
            
        results = self.vectorstore.similarity_search_with_score(
            query, 
            k=k,
            filter=filter_dict if filter_dict else None
        )
        
        search_results = []
        for doc, score in results:
            search_results.append({
                "content": doc.page_content,
                "department": doc.metadata.get("department", "알 수 없음"),
                "source": doc.metadata.get("source", "알 수 없음"),
                "relevance_score": round(1 - score, 3),  # 코사인 거리를 유사도로 환산
                "doc_type": doc.metadata.get("doc_type", "일반")
            })
        
        return search_results
    
    def generate_search_report(self, query, k=10):
        """검색 결과 보고서 생성"""
        results = self.search_by_department(query, k=k)
        
        print(f"🔍 '{query}' 검색 보고서")
        print("=" * 50)
        
        # 부서별 결과 통계
        dept_stats = {}
        for result in results:
            dept = result["department"]
            if dept not in dept_stats:
                dept_stats[dept] = 0
            dept_stats[dept] += 1
        
        print("📊 부서별 관련 문서 수:")
        for dept, count in sorted(dept_stats.items()):
            print(f"   {dept}: {count}개")
        
        print(f"\n📋 상위 {min(5, len(results))}개 검색 결과:")
        for i, result in enumerate(results[:5], 1):
            print(f"\n{i}. [{result['department']}] 관련도: {result['relevance_score']}")
            print(f"   출처: {result['source']}")
            print(f"   내용: {result['content'][:100]}...")
        
        return results
 
# 사용 예제
def main():
    # 회사 검색 시스템 초기화
    searcher = CompanyDocumentSearcher("테크스타트업")
    
    # 부서별 문서 경로 설정
    document_structure = {
        "개발팀": "./docs/dev/",
        "기획팀": "./docs/planning/", 
        "인사팀": "./docs/hr/",
        "마케팅팀": "./docs/marketing/"
    }
    
    # DB 구축
    searcher.build_company_db(document_structure)
    
    # 검색 테스트
    queries = [
        "신입사원 교육 프로그램",
        "API 개발 가이드라인",
        "마케팅 예산 계획",
        "코드리뷰 프로세스"
    ]
    
    for query in queries:
        print(f"\n{'='*60}")
        results = searcher.generate_search_report(query)
        
        # 부서별 상세 검색도 가능
        print(f"\n🏢 개발팀만 검색:")
        dev_results = searcher.search_by_department(query, "개발팀", k=3)
        for result in dev_results:
            print(f"   관련도: {result['relevance_score']} | {result['content'][:80]}...")
 
if __name__ == "__main__":
    main()

import gc
import psutil
 
def monitor_memory_usage():
    """메모리 사용량 모니터링"""
    process = psutil.Process()
    memory_mb = process.memory_info().rss / 1024 / 1024
    
    print(f"💾 현재 메모리 사용량: {memory_mb:.1f}MB")
    
    if memory_mb > 1000:  # 1GB 이상 사용시 경고
        print("⚠️ 메모리 사용량이 높습니다. 정리를 권장합니다.")
        
        # 가비지 컬렉션 실행
        gc.collect()
        
        new_memory_mb = process.memory_info().rss / 1024 / 1024
        print(f"🧹 정리 후: {new_memory_mb:.1f}MB (절약: {memory_mb - new_memory_mb:.1f}MB)")
 
# 대용량 처리시 배치 단위로 처리
def process_large_documents_in_batches(documents, batch_size=100):
    """대용량 문서를 배치로 처리"""
    
    total_batches = -(-len(documents) // batch_size)  # 올림
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        batch_num = i // batch_size + 1
        
        print(f"📦 배치 {batch_num}/{total_batches} 처리 중... ({len(batch)}개 문서)")
        
        # 벡터화 및 저장
        vectorstore.add_documents(batch)
        
        # 메모리 모니터링
        monitor_memory_usage()

def maintain_vectorstore_health():
    """벡터 스토어 건강 상태 체크"""
    
    print("🔧 벡터 DB 건강 상태 체크 중...")
    
    # 1. 컬렉션 정보 확인
    collection = vectorstore._collection
    print(f"📊 저장된 문서 수: {collection.count()}")
    
    # 2. 중복 문서 체크
    all_docs = vectorstore.get()
    content_hashes = set()
    duplicates = 0
    
    for doc_id, metadata in zip(all_docs['ids'], all_docs['metadatas']):
        content_hash = hash(all_docs['documents'][all_docs['ids'].index(doc_id)])
        if content_hash in content_hashes:
            duplicates += 1
        content_hashes.add(content_hash)
    
    print(f"🔍 중복 문서: {duplicates}개")
    
    # 3. 메타데이터 일관성 체크
    required_fields = ['source', 'chunk_id']
    missing_metadata = 0
    
    for metadata in all_docs['metadatas']:
        for field in required_fields:
            if field not in metadata:
                missing_metadata += 1
                break
    
    print(f"⚠️ 메타데이터 누락: {missing_metadata}개")
    
    # 4. 검색 성능 테스트
    test_query = "테스트 검색"
    start_time = time.time()
    vectorstore.similarity_search(test_query, k=5)
    search_time = time.time() - start_time
    
    print(f"⚡ 검색 응답 시간: {search_time:.3f}초")
    
    # 건강 점수 계산
    health_score = 100
    if duplicates > 10:
        health_score -= 20
    if missing_metadata > 5:
        health_score -= 15
    if search_time > 1.0:
        health_score -= 25
    
    print(f"💯 건강 점수: {health_score}/100")
    
    return {
        "total_docs": collection.count(),
        "duplicates": duplicates,
        "missing_metadata": missing_metadata,
        "search_time": search_time,
        "health_score": health_score
    }
 
# 정기적인 유지보수 실행
health_report = maintain_vectorstore_health()