LLM Day13 - RAG 기반 챗봇+평가모델

🎯 목표:

이 코드는 RAG (Retrieval-Augmented Generation) 기반 챗봇 시스템을 구현하는 것으로,
사용자의 질문을 받아 관련 문서를 검색하고, 신뢰도 점수를 매긴 후 답변을 생성하는 것이 목표다.
즉, 단순한 AI 챗봇이 아니라 "근거가 있는 답변"을 제공하는 챗봇이다.

🔄 코드의 전체적인 흐름

1️⃣ 사용자 질문 입력 → 2️⃣ 문서 검색 → 3️⃣ 관련성 평가 → 4️⃣ 답변 생성 및 신뢰도 점수 제공

1️⃣ 사용자 질문 입력 📝

사용자가 질문을 입력하면, 챗봇은 먼저 그 질문을 받아 저장한다.
✅ generate_answer(message, history): 사용자가 입력한 질문을 처리하는 핵심 함

2️⃣ 문서 검색 (Retrieval) 🔍

LangChain의 VectorStoreRetriever를 활용해,
질문과 관련된 문서를 벡터(숫자 배열) 기반으로 검색한다.
✅ search_documents(question): 질문과 관련된 문서를 찾아서 반환하는 함수

3️⃣ 문서와 질문의 관련성 평가 (Evaluation) 🤔

검색된 문서와 질문이 얼마나 관련이 있는지 평가한다.
두 가지 방법을 사용하여 점수를 계산함.

OpenAI GPT-4o 평가 모델을 활용한 점수 (직접 연관성, 추론 가능성, 신뢰도 등)
BERT 기반 유사도 분석 모델을 활용한 점수
✅ evaluation_chain: GPT 기반 평가 체인
✅ evaluate_with_bert(question, context): 문장 유사도를 계산하는 BERT 평가 모델

📌 점수는 0100점 척도로 변환되며,
🔴 낮음 (040), 🟡 보통 (4170), 🟢 높음 (71100) 의 신뢰도 등급도 함께 제공됨.

4️⃣ 최종 답변 생성 💡

점수를 바탕으로 최종 신뢰도 점수를 계산
관련성이 높은 경우 답변을 생성
관련성이 낮거나 문서를 찾을 수 없는 경우 사용자에게 적절한 안내 제공
✅ generate_answer(message, history): 점수를 종합하여 답변을 반환하는 함수
✅ normalize_score(score): 평가 점수를 100점 기준으로 변환
✅ get_relevance_label(score): 점수에 따라 신뢰도 등급 설정

이 코드의 특징

✅ 단순 챗봇이 아니라 신뢰할 수 있는 답변을 제공하는 시스템
✅ GPT + BERT 결합 → 보다 정확한 문서 검색 및 평가 가능
✅ Gradio를 이용해 웹에서 손쉽게 실행 가능

👉 질문을 하면 AI가 문서를 찾아서 답을 주고, 신뢰도를 평가해주는 스마트한 챗봇 시스템! 😊

# 필요한 라이브러리 임포트
import gradio as gr  # Gradio: 웹에서 AI 모델을 실행할 수 있도록 인터페이스를 제공하는 라이브러리
from langchain_core.language_models import BaseChatModel  # LangChain의 기본 챗봇 모델
from langchain_core.vectorstores import VectorStoreRetriever  # 검색 엔진 역할을 하는 벡터 저장소에서 데이터를 가져오는 기능
from langchain_core.output_parsers import StrOutputParser  # AI의 출력값을 문자열로 변환하는 기능
from langchain_core.prompts import ChatPromptTemplate  # AI에게 주어지는 질문 형식을 정의하는 템플릿
from langchain_core.runnables import RunnableLambda  # 특정 함수를 LangChain의 실행 가능한 체인에 포함시키는 기능
from langchain_core.pydantic_v1 import BaseModel, Field  # 데이터 구조를 정리하고 검증하는 데 사용되는 Pydantic 라이브러리
from langchain_openai import ChatOpenAI  # OpenAI의 GPT 모델을 LangChain에서 사용하도록 연결하는 모듈
from sentence_transformers import SentenceTransformer, util  # 문장 간의 의미 유사도를 평가하는 BERT 기반 모델
from typing import List, Optional  # Python의 타입 힌팅을 위한 모듈
from dataclasses import dataclass  # 데이터 클래스를 생성하는 모듈 (클래스를 더 간결하게 정의할 수 있음)

@dataclass
class SearchResult:
    context: str
    source_documents: Optional[List]

class RelevanceEvaluation(BaseModel):
    """문서와 질문의 관련성을 평가하는 JSON 출력 구조"""
    question: str = Field(..., description="사용자의 원본 질문")
    direct_relevance: float = Field(..., description="문서의 직접적 연관성 점수 (0~1)")
    inference_potential: float = Field(..., description="문서로부터 답변을 추론할 수 있는 가능성 (0~1)")
    overall_trustworthiness: float = Field(..., description="문서의 신뢰도 점수 (0~1)")
    explanation: str = Field(..., description="문서가 관련이 있는 이유 또는 없는 이유 설명")

def normalize_score(score):
    """ 점수를 정규화하여 신뢰도를 조정하고 100점 척도로 변환 """
    normalized = min(max((score - 0.2) / 0.8, 0), 1)
    return int(normalized * 100)  # 100점 기준 변환

def get_relevance_label(score):
    """ 관련성 점수에 따른 등급 반환 """
    if score <= 40:
        return "🔴 낮음"
    elif score <= 70:
        return "🟡 보통"
    else:
        return "🟢 높음"

# ✅ BERT 기반 평가 모델 추가
bert_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def evaluate_with_bert(question, context):
    """ BERT 기반 문서와 질문의 유사도 평가 """
    question_embedding = bert_model.encode(question, convert_to_tensor=True)
    context_embedding = bert_model.encode(context, convert_to_tensor=True)
    similarity_score = util.pytorch_cos_sim(question_embedding, context_embedding).item()
    return min(max(similarity_score, 0), 1)  # 정규화 (0~1 범위 유지)

evaluation_prompt = ChatPromptTemplate.from_template("""
사용자의 질문과 문서의 관련성을 다각도로 평가하세요.

[사용자 질문]
{question}

[문서 내용]
{context}

각 항목에 대한 점수를 0~1 범위에서 부여하세요.
- **직접적 연관성**: 문서가 질문과 직접 관련 있는가? (0: 관련 없음, 1: 매우 관련)
- **추론 가능성**: 문서 내용을 기반으로 논리적으로 답변할 수 있는가? (0: 불가능, 1: 가능)
- **전반적 신뢰도**: 문서가 올바른 정보로 보이는가? (0: 신뢰 낮음, 1: 신뢰 높음)

### 출력 형식 (JSON)
{{
    "question": "{question}",
    "direct_relevance": 직접적 연관성 점수 (0~1),
    "inference_potential": 추론 가능성 점수 (0~1),
    "overall_trustworthiness": 전반적 신뢰도 점수 (0~1),
    "explanation": "문서가 관련이 있는 이유 또는 없는 이유 설명"
}}
""")

evaluation_chain = (
    {
        "question": RunnableLambda(lambda x: x["question"]),
        "context": RunnableLambda(lambda x: x["context"])
    }
    | evaluation_prompt
    | ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(RelevanceEvaluation)
)

summary_prompt = ChatPromptTemplate.from_template("""
다음 대화 이력을 요약하여 핵심 내용을 유지하세요.

[대화 이력]
{history}

### 출력 형식
요약된 대화 내용:
""")

class RAGSystem:
    def __init__(
            self, 
            llm: BaseChatModel, 
            eval_llm: BaseChatModel,
            retriever: VectorStoreRetriever
        ):
        self.llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.eval_llm = eval_llm or ChatOpenAI(model="gpt-4o", temperature=0)
        if not retriever:
            raise ValueError("검색기(retriever)가 필요합니다.")
        self.retriever = retriever
        self.chat_history = []  # 대화 이력 관리
    
    def summarize_history(self):
        if len(self.chat_history) > 5:
            chain = summary_prompt | self.llm | StrOutputParser()
            self.chat_history = [{"summary": chain.invoke({"history": self.chat_history})}]
    
    def search_documents(self, question: str) -> SearchResult:
        docs = self.retriever.invoke(question)
        return SearchResult(
            context="\n\n".join(doc.page_content for doc in docs) if docs else "관련 문서를 찾을 수 없습니다.",
            source_documents=docs,
        )
    
    def generate_answer(self, message: str, history: List):
        self.chat_history.append({"question": message})
        if len(self.chat_history) > 5:
            self.summarize_history()
        
        search_result = self.search_documents(message)
        if not search_result.source_documents:
            return "죄송합니다. 관련 문서를 찾을 수 없어 답변하기 어렵습니다."
        
        llm_evaluation = evaluation_chain.invoke({"context": search_result.context, "question": message})
        bert_evaluation = evaluate_with_bert(message, search_result.context)
        
        final_score = (llm_evaluation.direct_relevance + llm_evaluation.inference_potential + llm_evaluation.overall_trustworthiness) / 3
        final_score = (final_score + bert_evaluation) / 2  # BERT 점수 반영
        
        normalized_score = normalize_score(final_score)
        relevance_label = get_relevance_label(normalized_score)
        
        return f"{message}\n\n{relevance_label} 관련성 점수: {normalized_score}/100\n📝 설명: {llm_evaluation.explanation}"

# Gradio 인터페이스 설정
rag_system = RAGSystem(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),   
    eval_llm=ChatOpenAI(model="gpt-4o", temperature=0),
    retriever=vector_store.as_retriever(search_kwargs={"k": 2})
)

demo = gr.ChatInterface(
    fn=rag_system.generate_answer,
    title="RAG QA 시스템",
    description="질문을 입력하면 관련 문서를 검색하여 답변을 생성합니다.",
    theme=gr.themes.Soft(primary_hue="blue", secondary_hue="gray"),
    examples=[
        ["수원시의 주택건설지역은 어디에 해당하나요?"],
        ["무주택 세대에 대해서 설명해주세요."],
        ["2순위로 당첨된 사람이 청약통장을 다시 사용할 수 있나요?"],
    ]
)

demo.launch()