사용자 질문 → 관련 문서 검색 → LLM에게 [문서 + 질문] 제공 → 답변 생성

from sentence_transformers import SentenceTransformer
import numpy as np

# 임베딩 모델 로드
model = SentenceTransformer('all-MiniLM-L6-v2')

# 문서 임베딩
documents = [
    "Python은 데이터 과학에 널리 사용되는 프로그래밍 언어입니다.",
    "머신러닝은 데이터에서 패턴을 학습하는 AI의 한 분야입니다.",
    "파리는 프랑스의 수도입니다.",
    "딥러닝은 신경망을 사용하는 머신러닝 방법입니다.",
]

doc_embeddings = model.encode(documents)
print(f"임베딩 shape: {doc_embeddings.shape}")  # (4, 384)

# 쿼리 임베딩
query = "인공지능과 데이터 분석"
query_embedding = model.encode([query])[0]

# 코사인 유사도 계산
def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

similarities = [cosine_similarity(query_embedding, doc_emb)
                for doc_emb in doc_embeddings]

# 결과 정렬
ranked = sorted(zip(similarities, documents), reverse=True)
print("\n유사도 순위:")
for score, doc in ranked:
    print(f"  {score:.4f}: {doc}")

from langchain.text_splitter import CharacterTextSplitter

text = """
머신러닝은 인공지능의 한 분야로, 컴퓨터가 명시적 프로그래밍 없이 데이터로부터
학습할 수 있게 하는 기술입니다. 지도학습, 비지도학습, 강화학습 등의 방법론이 있으며,
각각 다른 종류의 문제를 해결합니다.
"""

splitter = CharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    separator="\n"
)
chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"청크 {i}: {chunk[:50]}...")

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)

with open("document.txt", "r", encoding="utf-8") as f:
    text = f.read()

chunks = splitter.split_text(text)
print(f"총 청크 수: {len(chunks)}")
print(f"평균 청크 길이: {sum(len(c) for c in chunks) / len(chunks):.0f} 문자")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 의미 기반 청커 (임베딩 유사도 기반 분할)
semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # 또는 "standard_deviation"
    breakpoint_threshold_amount=95           # 상위 5% 유사도 변화 지점에서 분할
)

chunks = semantic_splitter.create_documents([text])
print(f"의미 기반 청킹 결과: {len(chunks)}개 청크")

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 자식 청커: 작은 청크 (검색용)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 부모 청커: 큰 청크 (컨텍스트용)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

# 벡터 스토어 (자식 청크 임베딩 저장)
vectorstore = Chroma(
    collection_name="full_documents",
    embedding_function=OpenAIEmbeddings()
)

# 부모 문서 저장소 (전체 부모 청크 저장)
store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

import numpy as np
import faiss
from langchain_community.vectorstores import FAISS, Chroma
from langchain_openai import OpenAIEmbeddings

# ===== FAISS 직접 사용 =====
# 랜덤 벡터 데이터 (실제로는 임베딩)
d = 384          # 벡터 차원
n = 10000        # 문서 수
vectors = np.random.randn(n, d).astype('float32')

# 인덱스 생성 (L2 거리)
index_flat = faiss.IndexFlatL2(d)
index_flat.add(vectors)
print(f"FAISS 인덱스 크기: {index_flat.ntotal}")

# 유사 벡터 검색
query = np.random.randn(1, d).astype('float32')
k = 5
distances, indices = index_flat.search(query, k)
print(f"상위 {k}개 결과: {indices[0]}")

# HNSW 인덱스 (빠른 근사 검색)
index_hnsw = faiss.IndexHNSWFlat(d, 32)  # M=32 연결
index_hnsw.add(vectors)
distances_hnsw, indices_hnsw = index_hnsw.search(query, k)
print(f"HNSW 결과: {indices_hnsw[0]}")

# ===== LangChain + Chroma =====
from langchain.schema import Document

documents = [
    Document(page_content="Python은 데이터 과학에 사용됩니다.", metadata={"source": "doc1"}),
    Document(page_content="머신러닝은 데이터에서 패턴을 학습합니다.", metadata={"source": "doc2"}),
    Document(page_content="딥러닝은 신경망 기반 머신러닝입니다.", metadata={"source": "doc3"}),
    Document(page_content="자연어 처리는 텍스트를 분석합니다.", metadata={"source": "doc4"}),
]

# Chroma 벡터스토어 생성
embeddings = OpenAIEmbeddings()
chroma_db = Chroma.from_documents(
    documents,
    embeddings,
    persist_directory="./chroma_db"
)

# 유사도 검색
results = chroma_db.similarity_search("AI와 머신러닝", k=2)
for doc in results:
    print(f"출처: {doc.metadata['source']}, 내용: {doc.page_content}")

# 점수와 함께 검색
results_with_score = chroma_db.similarity_search_with_score("딥러닝", k=2)
for doc, score in results_with_score:
    print(f"점수: {score:.4f}, 내용: {doc.page_content}")

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import os

# ===== 1. 문서 로딩 =====
# PDF 로더
loader = PyPDFLoader("company_handbook.pdf")
# 또는 디렉토리 전체 로드
# loader = DirectoryLoader("./docs/", glob="**/*.pdf", loader_cls=PyPDFLoader)
pages = loader.load()
print(f"로드된 페이지 수: {len(pages)}")

# ===== 2. 텍스트 분할 =====
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    length_function=len,
    separators=["\n\n", "\n", ".", "!", "?", ",", " "]
)
chunks = text_splitter.split_documents(pages)
print(f"생성된 청크 수: {len(chunks)}")

# ===== 3. 임베딩 생성 및 벡터 DB 저장 =====
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./rag_db"
)
print("벡터 DB 저장 완료")

# ===== 4. RAG 체인 구성 =====
# 커스텀 프롬프트 템플릿
prompt_template = """당신은 도움이 되는 AI 어시스턴트입니다.
주어진 컨텍스트만을 사용하여 질문에 답변하세요.
컨텍스트에 없는 정보는 "제공된 문서에서 찾을 수 없습니다"라고 말하세요.

컨텍스트:
{context}

질문: {question}

답변:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# RetrievalQA 체인
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# ===== 5. 질문 응답 =====
query = "회사의 휴가 정책은 어떻게 되나요?"
result = qa_chain.invoke({"query": query})

print(f"\n질문: {query}")
print(f"답변: {result['result']}")
print(f"\n참조 문서:")
for doc in result['source_documents']:
    print(f"  - 페이지 {doc.metadata.get('page', '?')}: {doc.page_content[:100]}...")

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 설정
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=20)

# 문서 로드
documents = SimpleDirectoryReader("./docs/").load_data()

# 인덱스 생성
index = VectorStoreIndex.from_documents(documents)

# 쿼리 엔진
query_engine = index.as_query_engine(
    similarity_top_k=4,
    response_mode="compact"  # "tree_summarize", "refine" 등
)

# 질의
response = query_engine.query("What is the main topic of the documents?")
print(f"답변: {response}")
print(f"\n소스 노드:")
for node in response.source_nodes:
    print(f"  - Score: {node.score:.4f}")
    print(f"    Text: {node.text[:100]}...")

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma

# BM25 검색 (키워드 기반)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 4

# 벡터 검색 (의미 기반)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 앙상블 (하이브리드)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]  # 각 방식의 가중치
)

results = ensemble_retriever.invoke("파이썬 프로그래밍 튜토리얼")
print(f"하이브리드 검색 결과 수: {len(results)}")

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

# 내부적으로 LLM이 질문을 여러 버전으로 재작성
# 예: "RAG의 장점은?"
# → "RAG 기술의 이점은 무엇인가?"
# → "검색 증강 생성이 일반 LLM보다 나은 점은?"
# → "RAG를 사용해야 하는 이유는?"

results = multi_query_retriever.invoke("RAG의 장점은?")
print(f"멀티 쿼리 검색 결과: {len(results)}개")

# MMR 검색
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,           # 최종 반환 수
        "fetch_k": 20,    # 초기 후보 수
        "lambda_mult": 0.5  # 유사도 vs 다양성 균형 (0=다양성, 1=유사도)
    }
)

results = mmr_retriever.invoke("머신러닝 알고리즘")
print(f"MMR 결과: {len(results)}개")

from langchain.schema import Document

# 메타데이터가 있는 문서
docs_with_metadata = [
    Document(
        page_content="2024년 1분기 매출은 100억원입니다.",
        metadata={"year": 2024, "quarter": "Q1", "category": "financial"}
    ),
    Document(
        page_content="2024년 2분기 매출은 120억원입니다.",
        metadata={"year": 2024, "quarter": "Q2", "category": "financial"}
    ),
    Document(
        page_content="기술 로드맵: AI 기능 강화 예정.",
        metadata={"year": 2024, "quarter": "Q1", "category": "strategy"}
    ),
]

# 메타데이터 필터로 검색 범위 좁히기
filtered_results = vectorstore.similarity_search(
    "매출 성과",
    k=2,
    filter={"category": "financial", "year": 2024}
)

from sentence_transformers import CrossEncoder
import numpy as np

# Cross-Encoder 모델 로드
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# 초기 검색 결과 (바이 인코더 기반)
query = "머신러닝 알고리즘의 종류"
initial_results = vectorstore.similarity_search(query, k=20)  # 많이 검색

# 크로스 인코더로 재정렬
pairs = [[query, doc.page_content] for doc in initial_results]
scores = cross_encoder.predict(pairs)

# 점수 기준 정렬
ranked = sorted(zip(scores, initial_results), reverse=True)
top_k = [doc for _, doc in ranked[:5]]  # 상위 5개만 사용

print("리랭킹 후 상위 결과:")
for score, doc in ranked[:3]:
    print(f"  점수 {score:.4f}: {doc.page_content[:80]}...")

import cohere
from langchain.retrievers.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever

co = cohere.Client("your-api-key")

# Cohere 리랭크 압축기
compressor = CohereRerank(
    cohere_api_key="your-api-key",
    top_n=3,
    model="rerank-multilingual-v3.0"  # 한국어 지원
)

# 압축 리트리버 (검색 + 리랭크)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

results = compression_retriever.invoke("회사 정책에 대해 알려주세요")
print(f"리랭크된 문서 수: {len(results)}")

from FlagEmbedding import FlagReranker

# BGE 리랭커 (오픈소스, 로컬 실행)
reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

query = "What is RAG?"
passages = [
    "RAG stands for Retrieval-Augmented Generation.",
    "A rag is a piece of cloth used for cleaning.",
    "RAG systems combine retrieval with generation for better LLM responses.",
]

# 쌍으로 점수 계산
scores = reranker.compute_score([[query, p] for p in passages])
ranked = sorted(zip(scores, passages), reverse=True)

for score, passage in ranked:
    print(f"  {score:.4f}: {passage}")

질문 → LLM이 가상 답변 생성 → 가상 답변 임베딩 → 실제 문서 검색

from langchain.chains import HypotheticalDocumentEmbedder
from langchain_openai import OpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# HyDE 프롬프트
hyde_prompt = """당신은 전문가입니다. 다음 질문에 대한 상세한 답변 문서를 작성하세요.
이 문서는 실제로 존재하는 것처럼 작성하세요.

질문: {question}

가상 답변 문서:"""

# HyDE 임베딩 생성기
llm = OpenAI()
embeddings = OpenAIEmbeddings()

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=llm,
    embeddings=embeddings,
    prompt_key="web_search"  # 또는 커스텀 프롬프트
)

# HyDE를 사용한 검색
query = "양자 컴퓨팅이 암호화에 미치는 영향"
# HyDE가 내부적으로 가상 문서 생성 후 임베딩
hyde_retriever = vectorstore.as_retriever(embedding_function=hyde_embeddings)
results = hyde_retriever.invoke(query)

# 수동 HyDE 구현
def manual_hyde(query, llm, embeddings, vectorstore, k=4):
    # 1. 가상 문서 생성
    hypothetical_doc = llm.invoke(
        f"다음 질문에 대한 상세한 답변을 작성하세요: {query}"
    )

    # 2. 가상 문서 임베딩
    hyp_embedding = embeddings.embed_query(hypothetical_doc.content)

    # 3. 가상 문서 임베딩으로 검색
    results = vectorstore.similarity_search_by_vector(hyp_embedding, k=k)

    return results, hypothetical_doc.content

from langchain_openai import ChatOpenAI
chat_llm = ChatOpenAI(temperature=0.7)
results, hyp_doc = manual_hyde(
    "딥러닝의 역사", chat_llm, embeddings, vectorstore
)
print(f"생성된 가상 문서: {hyp_doc[:200]}...")
print(f"검색된 실제 문서: {len(results)}개")

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Self-RAG 시뮬레이션 (실제 Self-RAG는 특별 훈련된 모델 필요)
class SelfRAGSimulator:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def should_retrieve(self, query: str) -> bool:
        """검색이 필요한지 판단"""
        prompt = ChatPromptTemplate.from_template("""
다음 질문에 답하기 위해 외부 문서 검색이 필요한지 판단하세요.
질문이 일반 상식이나 추론만으로 답할 수 있으면 'NO',
특정 사실이나 전문 지식이 필요하면 'YES'라고만 답하세요.

질문: {query}
판단 (YES/NO):""")
        response = self.llm.invoke(prompt.format_messages(query=query))
        return "YES" in response.content.upper()

    def is_relevant(self, query: str, doc_content: str) -> bool:
        """문서가 질문과 관련 있는지 판단"""
        prompt = ChatPromptTemplate.from_template("""
다음 문서가 질문에 관련이 있는지 판단하세요.
'RELEVANT' 또는 'IRRELEVANT'로만 답하세요.

질문: {query}
문서: {doc}
판단:""")
        response = self.llm.invoke(
            prompt.format_messages(query=query, doc=doc_content[:500])
        )
        return "RELEVANT" in response.content.upper()

    def generate_with_reflection(self, query: str) -> str:
        """Self-RAG 방식으로 응답 생성"""
        # 1. 검색 필요성 판단
        need_retrieve = self.should_retrieve(query)
        print(f"검색 필요: {need_retrieve}")

        if not need_retrieve:
            # 검색 없이 직접 답변
            response = self.llm.invoke(query)
            return response.content

        # 2. 문서 검색
        docs = self.retriever.invoke(query)

        # 3. 관련성 필터링
        relevant_docs = [d for d in docs if self.is_relevant(query, d.page_content)]
        print(f"관련 문서: {len(relevant_docs)}/{len(docs)}개")

        if not relevant_docs:
            return "관련 문서를 찾을 수 없습니다. 일반 지식으로 답변합니다: " + \
                   self.llm.invoke(query).content

        # 4. 컨텍스트로 답변 생성
        context = "\n\n".join([d.page_content for d in relevant_docs[:3]])
        prompt = f"""컨텍스트를 사용하여 질문에 답변하세요.
컨텍스트: {context}
질문: {query}
답변:"""
        return self.llm.invoke(prompt).content

from langchain_community.tools.tavily_search import TavilySearchResults
from typing import List, Tuple

class CorrectiveRAG:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
        self.web_search = TavilySearchResults(max_results=3)

    def evaluate_documents(self, query: str, docs: list) -> Tuple[str, List]:
        """
        문서 관련성 평가
        반환: ("CORRECT"|"INCORRECT"|"AMBIGUOUS", 필터링된 문서들)
        """
        evaluation_prompt = """질문에 대해 검색된 문서들의 관련성을 평가하세요.
- CORRECT: 문서가 질문에 직접적으로 답할 수 있음
- INCORRECT: 문서가 질문과 관련 없음
- AMBIGUOUS: 부분적으로 관련 있으나 불완전

질문: {query}
문서들:
{docs}

평가 (CORRECT/INCORRECT/AMBIGUOUS):"""

        docs_text = "\n---\n".join([d.page_content[:300] for d in docs[:4]])
        response = self.llm.invoke(
            evaluation_prompt.format(query=query, docs=docs_text)
        )

        evaluation = response.content.strip().upper()
        if "CORRECT" in evaluation:
            return "CORRECT", docs
        elif "INCORRECT" in evaluation:
            return "INCORRECT", []
        else:
            return "AMBIGUOUS", docs

    def run(self, query: str) -> str:
        # 1. 초기 검색
        docs = self.retriever.invoke(query)

        # 2. 문서 품질 평가
        status, filtered_docs = self.evaluate_documents(query, docs)
        print(f"문서 평가 결과: {status}")

        # 3. 상황에 따른 처리
        if status == "INCORRECT":
            # 웹 검색으로 보완
            print("웹 검색으로 보완 중...")
            web_results = self.web_search.invoke(query)
            context = "\n".join([r['content'] for r in web_results])
        elif status == "AMBIGUOUS":
            # 지식 정제 + 웹 검색 결합
            web_results = self.web_search.invoke(query)
            web_context = "\n".join([r['content'] for r in web_results])
            doc_context = "\n".join([d.page_content for d in filtered_docs[:2]])
            context = doc_context + "\n\n[웹 검색 보완]\n" + web_context
        else:
            # 검색 문서 직접 사용
            context = "\n\n".join([d.page_content for d in filtered_docs[:4]])

        # 4. 최종 응답 생성
        response = self.llm.invoke(
            f"컨텍스트:\n{context}\n\n질문: {query}\n답변:"
        )
        return response.content

class AdaptiveRAG:
    def __init__(self, simple_retriever, advanced_retriever, llm):
        self.simple_retriever = simple_retriever   # 단순 벡터 검색
        self.advanced_retriever = advanced_retriever  # 하이브리드 + 리랭킹
        self.llm = llm

    def classify_query(self, query: str) -> str:
        """쿼리 복잡도 분류"""
        prompt = f"""다음 질문의 복잡도를 분류하세요:
- simple: 단순 사실 확인 또는 직접 답변 가능
- complex: 여러 소스 조합, 다단계 추론 필요

질문: {query}
분류 (simple/complex):"""

        response = self.llm.invoke(prompt)
        return "complex" if "complex" in response.content.lower() else "simple"

    def run(self, query: str) -> str:
        query_type = self.classify_query(query)
        print(f"쿼리 유형: {query_type}")

        if query_type == "simple":
            docs = self.simple_retriever.invoke(query)
        else:
            docs = self.advanced_retriever.invoke(query)

        context = "\n\n".join([d.page_content for d in docs])
        return self.llm.invoke(
            f"컨텍스트:\n{context}\n\n질문: {query}\n답변:"
        ).content

# GraphRAG 설치 및 초기화
pip install graphrag

# 프로젝트 초기화
python -m graphrag.index --init --root ./ragtest

# 설정 파일 수정 후 인덱싱
python -m graphrag.index --root ./ragtest

# 글로벌 검색 (전체 문서 이해 필요)
python -m graphrag.query --root ./ragtest --method global "What are the main themes?"

# 로컬 검색 (특정 엔티티 중심)
python -m graphrag.query --root ./ragtest --method local "Tell me about company X"

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
)
from datasets import Dataset

# 평가 데이터 준비
evaluation_data = {
    "question": [
        "회사의 연차 제도는 어떻게 되나요?",
        "원격 근무 정책은 무엇인가요?",
    ],
    "answer": [
        "연차는 입사 1년 후 15일이 주어지며, 매년 1일씩 추가됩니다.",
        "주 3일 원격 근무가 가능합니다.",
    ],
    "contexts": [
        ["직원은 입사 1년 후 15일의 연차를 부여받습니다. 이후 매년 1일씩 추가됩니다."],
        ["직원은 주 2일 원격 근무를 할 수 있습니다. 특별 허가 시 추가 가능합니다."],
    ],
    "ground_truth": [
        "입사 1년 후 15일, 매년 1일 추가",
        "주 2일 원격 근무 기본, 허가 시 추가",
    ]
}

dataset = Dataset.from_dict(evaluation_data)

result = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision
    ]
)

print(result)
# faithfulness: 0.75 (답변이 컨텍스트와 불일치하는 부분 감지)
# answer_relevancy: 0.92
# context_recall: 0.85
# context_precision: 0.78

import json
from typing import List, Dict

class RAGEvaluator:
    def __init__(self, rag_chain, llm):
        self.rag_chain = rag_chain
        self.llm = llm

    def evaluate_faithfulness(self, answer: str, context: str) -> float:
        """답변이 컨텍스트에 충실한지 평가 (0-1)"""
        prompt = f"""다음 답변이 주어진 컨텍스트의 정보만을 기반으로 작성되었는지 평가하세요.
0.0 (전혀 아님) ~ 1.0 (완전히 충실)

컨텍스트: {context}
답변: {answer}

충실도 점수 (숫자만):"""
        response = self.llm.invoke(prompt)
        try:
            return float(response.content.strip())
        except:
            return 0.5

    def evaluate_answer_relevancy(self, question: str, answer: str) -> float:
        """답변이 질문과 관련 있는지 평가"""
        prompt = f"""다음 답변이 질문에 얼마나 관련 있는지 0.0~1.0으로 평가하세요.

질문: {question}
답변: {answer}

관련도 점수 (숫자만):"""
        response = self.llm.invoke(prompt)
        try:
            return float(response.content.strip())
        except:
            return 0.5

    def run_evaluation(self, test_cases: List[Dict]) -> Dict:
        results = []
        for case in test_cases:
            question = case["question"]
            expected = case.get("expected_answer", "")

            # RAG 응답 생성
            result = self.rag_chain.invoke({"query": question})
            answer = result["result"]
            context = "\n".join([d.page_content for d in result["source_documents"]])

            # 평가
            faithfulness_score = self.evaluate_faithfulness(answer, context)
            relevancy_score = self.evaluate_answer_relevancy(question, answer)

            results.append({
                "question": question,
                "answer": answer,
                "faithfulness": faithfulness_score,
                "relevancy": relevancy_score,
            })

        # 집계
        avg_faithfulness = sum(r["faithfulness"] for r in results) / len(results)
        avg_relevancy = sum(r["relevancy"] for r in results) / len(results)

        return {
            "results": results,
            "avg_faithfulness": avg_faithfulness,
            "avg_relevancy": avg_relevancy,
            "overall_score": (avg_faithfulness + avg_relevancy) / 2
        }

import hashlib
import json
from functools import lru_cache
import redis

class CachedRAGSystem:
    def __init__(self, rag_chain, redis_client=None, ttl=3600):
        self.rag_chain = rag_chain
        self.redis = redis_client
        self.ttl = ttl

    def _get_cache_key(self, query: str) -> str:
        return f"rag:{hashlib.md5(query.encode()).hexdigest()}"

    def query(self, query: str) -> dict:
        cache_key = self._get_cache_key(query)

        # 캐시 확인
        if self.redis:
            cached = self.redis.get(cache_key)
            if cached:
                print("캐시 히트!")
                return json.loads(cached)

        # RAG 실행
        result = self.rag_chain.invoke({"query": query})
        response = {
            "answer": result["result"],
            "sources": [d.metadata for d in result["source_documents"]]
        }

        # 캐시 저장
        if self.redis:
            self.redis.setex(cache_key, self.ttl, json.dumps(response))

        return response

# 임베딩 캐싱
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache

# 인메모리 캐시 (개발용)
set_llm_cache(InMemoryCache())

# SQLite 캐시 (프로덕션용)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI

# 스트리밍 LLM
streaming_llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# FastAPI를 통한 스트리밍 엔드포인트
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def generate_rag_stream(query: str):
    docs = retriever.invoke(query)
    context = "\n\n".join([d.page_content for d in docs])

    async for chunk in streaming_llm.astream(
        f"컨텍스트:\n{context}\n\n질문: {query}\n답변:"
    ):
        if chunk.content:
            yield f"data: {chunk.content}\n\n"

@app.get("/rag/stream")
async def rag_stream_endpoint(query: str):
    return StreamingResponse(
        generate_rag_stream(query),
        media_type="text/event-stream"
    )

# 토큰 사용량 추적
from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = qa_chain.invoke({"query": "질문 텍스트"})
    print(f"총 토큰: {cb.total_tokens}")
    print(f"프롬프트 토큰: {cb.prompt_tokens}")
    print(f"완성 토큰: {cb.completion_tokens}")
    print(f"비용: ${cb.total_cost:.6f}")

# 청크 압축으로 컨텍스트 토큰 절약
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever

# 관련 부분만 추출하여 컨텍스트 압축
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 8})
)

# 압축된 문서만 LLM에 전달 (토큰 절약)
compressed_docs = compression_retriever.invoke("질문")
total_tokens = sum(len(d.page_content.split()) for d in compressed_docs)
print(f"압축된 컨텍스트 토큰 수 (추정): {total_tokens}")

import time
import logging
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class RAGMetrics:
    query: str
    retrieval_time: float = 0.0
    generation_time: float = 0.0
    num_docs_retrieved: int = 0
    answer_length: int = 0
    error: Optional[str] = None

class MonitoredRAGSystem:
    def __init__(self, rag_chain, logger=None):
        self.rag_chain = rag_chain
        self.logger = logger or logging.getLogger(__name__)
        self.metrics_history = []

    def query(self, query: str) -> dict:
        metrics = RAGMetrics(query=query)
        start_total = time.time()

        try:
            # 검색 타이밍
            retrieval_start = time.time()
            docs = retriever.invoke(query)
            metrics.retrieval_time = time.time() - retrieval_start
            metrics.num_docs_retrieved = len(docs)

            # 생성 타이밍
            gen_start = time.time()
            result = self.rag_chain.invoke({"query": query})
            metrics.generation_time = time.time() - gen_start
            metrics.answer_length = len(result["result"])

        except Exception as e:
            metrics.error = str(e)
            self.logger.error(f"RAG 오류: {e}")
            raise

        finally:
            total_time = time.time() - start_total
            self.metrics_history.append(metrics)
            self.logger.info(
                f"쿼리 처리 완료 | "
                f"검색: {metrics.retrieval_time:.2f}s | "
                f"생성: {metrics.generation_time:.2f}s | "
                f"총: {total_time:.2f}s | "
                f"문서: {metrics.num_docs_retrieved}개"
            )

        return result

    def get_stats(self) -> dict:
        if not self.metrics_history:
            return {}
        retrieval_times = [m.retrieval_time for m in self.metrics_history if not m.error]
        gen_times = [m.generation_time for m in self.metrics_history if not m.error]
        return {
            "total_queries": len(self.metrics_history),
            "error_rate": sum(1 for m in self.metrics_history if m.error) / len(self.metrics_history),
            "avg_retrieval_time": sum(retrieval_times) / len(retrieval_times) if retrieval_times else 0,
            "avg_generation_time": sum(gen_times) / len(gen_times) if gen_times else 0,
        }