"RAG" 태그 — 2개 게시물

# 메모랜덤: 폰트 크기 기반 목차 추출
class MemorandumNaiveParser:
    def extract_toc(self, page):
        if font_size >= 13.5:
            return "대제목"  # "1. 목적"
        elif font_size >= 11.5:
            return "소제목"  # "2.1. 균주"
        else:
            return "본문"

# 연구노트: 키워드 기반 섹션 분류
SECTION_KEYWORDS = {
    "실험개요": ["실험개요", "실험 개요"],
    "실험방법": ["실험방법", "실험 방법"],
    "실험결과": ["실험결과", "Task Results"],
    "결론": ["결론", "고찰", "결과 및 토의"],
}

# Amazon Titan Embed Text V2 설정
embedding_config = {
    "model_id": "amazon.titan-embed-text-v2:0",
    "dimensions": 1024,
    "normalize": True,  # L2 정규화로 코사인 유사도 계산
    "max_tokens": 8192,
    "safety_margin": 0.85  # 실제 최대: 6,553 토큰
}

# ChromaDB 디스크 기반 저장
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.create_collection(
    name="memorandum_eln",
    metadata={"hnsw:space": "cosine"}
)

# Top-K=5 벡터 검색
results = collection.query(
    query_embeddings=[eln_embedding],
    n_results=5,
    where={"session_id": session_id}
)

# Claude Sonnet 4.5 매칭 판정
prompt = f"""
다음 연구노트와 메모랜덤 청크가 같은 실험을 다루는지 판단하세요.

[판정 기준]
- True: 표(Table)에 구체적 실험 데이터(OD, 수율, 농도 등) 있음
- False: 결론/목적/향후 계획 섹션
- False: 판단 어려움 (보수적 판정)

[연구노트]
{eln_content}

[메모랜덤 청크]
{memorandum_chunk}

JSON 형식으로 응답: {{"match": true/false, "confidence": "high/medium/low"}}
"""

response = bedrock.invoke_model(
    modelId="global.anthropic.claude-sonnet-4-5-20250929-v1:0",
    body=json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
        "temperature": 0.0,  # 일관성 최대화
        "messages": [{"role": "user", "content": prompt}]
    })
)

def bcubed_precision(item_i, cluster_assignments, ground_truth):
    """아이템 i에 대한 BCubed Precision"""
    cluster_i = cluster_assignments[item_i]
    category_i = ground_truth[item_i]
    
    # 같은 클러스터에 속한 아이템들
    same_cluster = [j for j, c in enumerate(cluster_assignments) if c == cluster_i]
    
    # 그 중 실제로 같은 카테고리인 아이템들
    correct = [j for j in same_cluster if ground_truth[j] == category_i]
    
    return len(correct) / len(same_cluster)

# 전체 BCubed F1 계산
precision = np.mean([bcubed_precision(i, clusters, labels) for i in range(n)])
recall = np.mean([bcubed_recall(i, clusters, labels) for i in range(n)])
f1 = 2 * precision * recall / (precision + recall)

# Label Propagation 작동 원리
def label_propagation(graph, max_iter=100):
    # 초기: 모든 노드에 고유 라벨
    labels = {node: i for i, node in enumerate(graph.nodes())}
    
    for _ in range(max_iter):
        for node in graph.nodes():
            # 이웃 라벨 중 가장 많은 것 선택
            neighbor_labels = [labels[n] for n in graph.neighbors(node)]
            labels[node] = max(set(neighbor_labels), key=neighbor_labels.count)
    
    return labels

# 최적 조합: (min_cluster_size=2~3, min_samples=3)
best_config = {
    "min_cluster_size": 3,  # 최소 클러스터 크기
    "min_samples": 3,       # 핵심 포인트 판정 기준
    "metric": "euclidean",
    "cluster_selection_method": "eom"  # Excess of Mass
}

# 성능 개선
# Before (5, 1): F1 0.540, 평균 16.5개 클러스터
# After  (3, 3): F1 0.736, 평균 5.3개 클러스터

# Euclidean vs Cosine
# Euclidean: 평균 F1 0.731
# Cosine:    평균 F1 0.718

# 의사결정: Euclidean 선택
# 이유: Titan V2는 이미 L2 정규화 적용하여 방향성보다 거리가 유의미

# 개선 방향 1: 메타데이터 통합
chunk_metadata = {
    "content": text,
    "date": extract_date(text),
    "experiment_phase": classify_phase(text),  # "준비", "진행", "분석"
    "section_type": section_type
}

# 개선 방향 2: 하이브리드 임베딩
hybrid_embedding = concatenate([
    semantic_embedding,  # Titan V2
    temporal_embedding,  # 날짜 정보
    structural_embedding  # 섹션 타입
])

def select_algorithm(data_characteristics):
    if data_characteristics["topic_diversity"] == "single":
        return "connected_components"  # F1 1.0 기대
    
    elif data_characteristics["cluster_boundaries"] == "clear":
        return "louvain"  # F1 0.95+ 기대
    
    elif data_characteristics["noise_level"] == "high":
        return {
            "algorithm": "hdbscan",
            "params": {"min_cluster_size": 3, "min_samples": 3}
        }
    
    else:
        return "label_propagation"  # 범용적으로 안정적

# 과분할 탐지
if avg_clusters > expected_clusters * 2 and f1 < 0.7:
    print("과분할 의심: min_cluster_size 증가 필요")

# 과소분할 탐지
if avg_clusters < expected_clusters * 0.5 and f1 < 0.7:
    print("과소분할 의심: k_neighbors 감소 또는 algorithm 변경")

# 핵심: 명시적 기준 + 보수적 판정
prompt_template = """
[판정 기준] (우선순위 순)
1. 표(Table) 데이터 존재 여부
2. 정량적 수치 (농도, 수율, 온도 등) 존재 여부
3. 섹션 타입 (결론/계획은 제외)

[보수적 판정 원칙]
- 애매한 경우 False 반환
- 추론 과정을 reasoning 필드에 기록
"""

# 머릿속 기대: 전수 탐색 후 정렬
def ideal_search(query_vector, all_vectors):
    distances = [cosine_distance(query_vector, v) for v in all_vectors]
    sorted_results = sorted(distances)
    return sorted_results[:n_results]  # 상위 n개 자르기

# 실제 동작: 그래프 기반 근사 탐색 (HNSW 예시)
def ann_search(query_vector, hnsw_graph, n_results):
    entry_point = hnsw_graph.top_layer_node
    visited = set()
    candidates = []
    
    # 상위 레이어부터 탐색
    for layer in range(top_layer, 0, -1):
        entry_point = search_layer(query_vector, entry_point, layer, ef=1)
    
    # 최하위 레이어에서 ef 크기만큼 탐색
    candidates = search_layer(query_vector, entry_point, layer=0, ef=n_results)
    
    # 탐색한 candidates 중 상위 n_results 반환
    return heapq.nsmallest(n_results, candidates, key=lambda x: x.distance)

# 추상화된 시각화
벡터 공간 상의 분포:

  "러닝화 벗기" 클러스터 (초고밀도)
  ┌─────────────────┐
  │ ●●●●●●●●●●●●●● │  ← 날짜/시간만 다른 수백 개 문서
  │ ●●●●●●●●●●●●●● │     (의미적으로 거의 동일)
  │ ●●●●●●●●●●●●●● │
  └─────────────────┘

              vs

  "별자리 관측" 영역 (저밀도)
         ●                  ← "스카이뷰 앱" 문서
                            (적은 빈도, 멀리 떨어짐)

# 쿼리: "7키로 러닝을 하다"
query_embedding = model.encode("7키로 러닝을 하다")

# 문제 1: "러닝" 키워드가 포함된 고밀도 클러스터가 탐색 시작점
# 문제 2: "7키로"라는 구체적 정보는 임베딩에서 약하게 표현됨
# 문제 3: n_results=100이면 고밀도 클러스터 내부만 탐색하고 종료

# n_results=500 탐색 과정
1. "러닝" 고밀도 클러스터 탐색 (0~200번째)
2. 유사도가 낮아지면서 인접 클러스터로 확장
3. "운동 기록" 관련 클러스터 진입 (300~400번째)
4. "7km 기록" 문서 발견! (425번째)
5. 재정렬 후 상위권으로 부상

from sklearn.neighbors import NearestNeighbors
import numpy as np

def analyze_density(embeddings, sample_size=1000):
    """벡터 공간의 밀도 분석"""
    sample_indices = np.random.choice(len(embeddings), sample_size)
    sample_vectors = embeddings[sample_indices]
    
    nbrs = NearestNeighbors(n_neighbors=100, metric='cosine').fit(embeddings)
    distances, indices = nbrs.kneighbors(sample_vectors)
    
    # 10번째, 50번째, 100번째 이웃까지의 평균 거리
    print(f"10th neighbor avg distance: {distances[:, 9].mean():.4f}")
    print(f"50th neighbor avg distance: {distances[:, 49].mean():.4f}")
    print(f"100th neighbor avg distance: {distances[:, 99].mean():.4f}")
    
    # 거리 증가율
    growth_rate = distances[:, 99].mean() / distances[:, 9].mean()
    print(f"Distance growth rate (10th->100th): {growth_rate:.2f}x")
    
    if growth_rate < 1.5:
        print("⚠️ HIGH DENSITY - 큰 n_results 필요")
    elif growth_rate < 2.5:
        print("⚡ MEDIUM DENSITY - 적절한 n_results 필요")
    else:
        print("✅ LOW DENSITY - 작은 n_results도 안전")

# 실행 결과 예시 (LGE 프로젝트)
# 10th neighbor: 0.1234
# 50th neighbor: 0.1456  
# 100th neighbor: 0.1589
# Growth rate: 1.29x ← 매우 고밀도!

def estimate_query_abstraction(query):
    """쿼리의 추상화 정도 추정"""
    concrete_markers = ["스카이뷰", "7km", "09:25", "평창 공원"]
    abstract_markers = ["러닝", "별 보기", "운동", "휴식"]
    
    concrete_score = sum(1 for m in concrete_markers if m in query)
    abstract_score = sum(1 for m in abstract_markers if m in query)
    
    if abstract_score > concrete_score:
        return "ABSTRACT", 500  # 추상적이면 큰 n_results
    else:
        return "CONCRETE", 100  # 구체적이면 작은 n_results

# 예시
estimate_query_abstraction("스마트폰 앱으로 별 보기")  
# → ("ABSTRACT", 500)

estimate_query_abstraction("평창 공원에서 스카이뷰 앱 실행")
# → ("CONCRETE", 100)

from sentence_transformers import CrossEncoder

class AdaptiveRetriever:
    def __init__(self, vectordb, reranker_model="cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.vectordb = vectordb
        self.reranker = CrossEncoder(reranker_model)
        
    def search(self, query, final_k=10, safety_factor=10):
        """
        안전한 검색 전략:
        1. 넉넉한 후보 추출 (final_k * safety_factor)
        2. Reranking으로 정확도 보정
        3. 최종 k개 반환
        """
        # Step 1: 넉넉한 초기 검색
        initial_n = final_k * safety_factor  # 예: 10 * 10 = 100
        candidates = self.vectordb.query(
            query_texts=[query],
            n_results=initial_n
        )
        
        # Step 2: Cross-encoder로 재정렬
        pairs = [[query, doc] for doc in candidates['documents'][0]]
        rerank_scores = self.reranker.predict(pairs)
        
        # Step 3: 상위 k개 추출
        sorted_indices = np.argsort(rerank_scores)[::-1][:final_k]
        final_results = [candidates['documents'][0][i] for i in sorted_indices]
        
        return final_results

# 사용 예시
retriever = AdaptiveRetriever(chroma_collection)

# 안전한 검색 (내부적으로 100개 검색 후 10개로 rerank)
results = retriever.search("스마트폰 앱으로 별 보기", final_k=10)

def find_optimal_n_results(queries, ground_truth, vectordb):
    """다양한 n_results 값 비교"""
    test_n_values = [50, 100, 200, 500, 1000]
    results = {}
    
    for n in test_n_values:
        recall_scores = []
        for query, true_docs in zip(queries, ground_truth):
            retrieved = vectordb.query(query_texts=[query], n_results=n)
            retrieved_ids = set(retrieved['ids'][0][:10])  # 상위 10개만 평가
            true_ids = set(true_docs)
            
            recall = len(retrieved_ids & true_ids) / len(true_ids)
            recall_scores.append(recall)
        
        results[n] = {
            'recall@10': np.mean(recall_scores),
            'std': np.std(recall_scores)
        }
    
    return results

# 실행 결과 예시
# n=50:  recall@10=0.62
# n=100: recall@10=0.71  
# n=200: recall@10=0.85
# n=500: recall@10=0.94 ← 최적점
# n=1000: recall@10=0.95 (미미한 개선, 비용 증가)

# 1. HNSW ef_search 파라미터 조정 (Chroma 예시)
collection = client.create_collection(
    name="optimized_collection",
    metadata={
        "hnsw:space": "cosine",
        "hnsw:search_ef": 500,  # n_results와 연동
        "hnsw:M": 32  # 그래프 연결성 증가
    }
)

# 2. Hybrid Search로 보완
from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, vectordb, documents):
        self.vectordb = vectordb
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        
    def search(self, query, n_results=100, alpha=0.7):
        # Vector search
        vec_results = self.vectordb.query(query, n_results=n_results)
        vec_scores = {id: score for id, score in 
                     zip(vec_results['ids'][0], vec_results['distances'][0])}
        
        # BM25 search  
        bm25_scores = self.bm25.get_scores(query.split())
        
        # Hybrid ranking
        final_scores = {}
        for doc_id in set(vec_scores.keys()):
            vec_score = vec_scores.get(doc_id, 0)
            bm25_score = bm25_scores[doc_id] if doc_id < len(bm25_scores) else 0
            final_scores[doc_id] = alpha * vec_score + (1-alpha) * bm25_score
        
        return sorted(final_scores.items(), key=lambda x: x[1], reverse=True)