3주차 실습 - 주택청약 챗봇

gistfile1.txt

# 여기에 코드를 작성하세요.

import gradio as gr
from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.language_models import BaseChatModel
from langchain_core.vectorstores import VectorStoreRetriever
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List, Optional, Generator
from dataclasses import dataclass


# ── 1. 답변 품질 평가 스키마 ──────────────────────────────────────────
class AnswerQuality(BaseModel):
    is_sufficient: bool = Field(description="답변이 질문에 충분히 답하고 있는지 여부")
    reason: str = Field(description="판단 이유 (1문장)")


# ── 2. 대화 이력 관리 (트리밍) ────────────────────────────────────────
class ConversationHistory:
    """최근 N턴만 유지하는 대화 이력 관리자"""

    def __init__(self, max_turns: int = 3):
        self.max_turns = max_turns
        self.messages: List[dict] = []

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_turns * 2:
            self.messages = self.messages[-(self.max_turns * 2):]

    def format_history(self) -> str:
        if not self.messages:
            return ""
        lines = ["[이전 대화]"]
        for msg in self.messages:
            prefix = "사용자" if msg["role"] == "user" else "어시스턴트"
            lines.append(f"{prefix}: {msg['content']}")
        return "\n".join(lines)

    def reset(self):
        self.messages = []


# ── 3. 검색 결과 dataclass ────────────────────────────────────────────
@dataclass
class SearchResult:
    context: str
    source_documents: Optional[List] = None


# ── 4. 개선된 RAG 시스템 ─────────────────────────────────────────────
class ImprovedRAGSystem:
    def __init__(
        self,
        llm: BaseChatModel,
        eval_llm: BaseChatModel,
        retriever: VectorStoreRetriever,
        max_history_turns: int = 3,
    ):
        self.llm = llm
        self.eval_llm = eval_llm
        self.retriever = retriever   # self.retriever로 저장
        self.history = ConversationHistory(max_turns=max_history_turns)

    def _format_docs(self, docs: List) -> str:
        # page_content 대신 metadata의 answer를 컨텍스트로 활용
        parts = []
        for doc in docs:
            meta = doc.metadata
            q = meta.get("question", "")
            a = meta.get("answer", doc.page_content)
            parts.append(f"Q: {q}\nA: {a}")
        return "\n\n".join(parts)

    def _format_source_documents(self, docs: Optional[List]) -> str:
        if not docs:
            return "\n\nℹ️ 관련 문서를 찾을 수 없습니다."
        lines = []
        for i, doc in enumerate(docs, 1):
            meta = doc.metadata
            info = []
            if "question_id" in meta:
                info.append(f"ID: {meta['question_id']}")
            if "keyword" in meta:
                info.append(f"키워드: {meta['keyword']}")
            if "summary" in meta:
                info.append(f"요약: {meta['summary']}")
            lines.append(
                f"📚 참조 문서 {i}\n"
                f"• {' | '.join(info) if info else '출처 정보 없음'}\n"
                f"• 내용: {doc.page_content}"
            )
        return "\n\n" + "\n\n".join(lines)

    def _check_relevance(self, docs: List, question: str) -> List:
        """관련성 평가 - metadata의 answer를 기준으로 평가"""
        if not docs:
            return []
        prompt = ChatPromptTemplate.from_messages([
            ("system", """컨텍스트가 질문에 답변하는데 필요한 정보를 포함하는지 평가하세요.
기준 중 하나라도 충족하면 'Yes', 모두 불충족이면 'No'로만 답하세요.
1. 직접적으로 필요한 정보를 포함하는가?
2. 논리적으로 추론 가능한가?"""),
            ("human", "[컨텍스트]\n{context}\n\n[질문]\n{question}")
        ])
        chain = prompt | self.eval_llm | StrOutputParser()

        relevant = []
        for doc in docs:
            # page_content(짧은 요약 or QA포맷) 대신 실제 answer로 평가
            context = doc.metadata.get("answer", doc.page_content)
            result = chain.invoke({"context": context, "question": question}).lower()
            if "yes" in result:
                relevant.append(doc)
        return relevant

    def _evaluate_answer_quality(self, question: str, answer: str) -> AnswerQuality:
        prompt = ChatPromptTemplate.from_messages([
            ("system", """생성된 답변이 질문에 충분히 답하고 있는지 평가하세요.
- 질문의 핵심 내용에 직접 답하고 있는가?
- "모르겠습니다", "근거 없음" 등 불충분한 내용만 있지 않은가?"""),
            ("human", "[질문]\n{question}\n\n[답변]\n{answer}")
        ])
        chain = prompt | self.eval_llm.with_structured_output(AnswerQuality)
        return chain.invoke({"question": question, "answer": answer})

    def _search_documents(self, question: str) -> SearchResult:
        try:
            # 관련성 평가 없이 검색된 문서를 그대로 사용
            docs = self.retriever.invoke(question)
            return SearchResult(
                context=self._format_docs(docs) if docs else "",
                source_documents=docs if docs else None,
            )
        except Exception as e:
            print(f"문서 검색 오류: {e}")
            return SearchResult(context="", source_documents=None)

    def generate_answer(self, message: str, history: List) -> Generator[str, None, None]:
        # 1. 문서 검색
        search_result = self._search_documents(message)

        if not search_result.source_documents:
            yield "죄송합니다. 관련 문서를 찾을 수 없어 답변하기 어렵습니다. 다른 질문을 해주시겠습니까?"
            return

        # 2. 대화 이력 포함 프롬프트 구성
        history_text = self.history.format_history()

        prompt = ChatPromptTemplate.from_messages([
            ("system", """다음 지침을 따라 질문에 답변해주세요:
1. 주어진 문서의 내용만을 기반으로 답변하세요.
2. 문서에 명확한 근거가 없는 내용은 "근거 없음"이라고 답변하세요.
3. 이전 대화 맥락을 참고하여 자연스럽게 답변하세요.

{history}"""),
            ("human", "문서들:\n{context}\n\n질문: {question}")
        ])

        chain = prompt | self.llm | StrOutputParser()

        full_answer = ""
        try:
            for chunk in chain.stream({
                "history": history_text,
                "context": search_result.context,
                "question": message,
            }):
                full_answer += chunk
                yield full_answer

            # 3. 답변 품질 평가
            quality = self._evaluate_answer_quality(message, full_answer)
            if not quality.is_sufficient:
                insufficient_msg = (
                    f"{full_answer}\n\n"
                    f"⚠️ **답변 품질 경고**: {quality.reason}\n"
                    f"더 구체적인 질문을 해주시면 더 정확한 답변을 드릴 수 있습니다."
                )
                yield insufficient_msg
                self.history.add("user", message)
                self.history.add("assistant", insufficient_msg)
                return

            # 4. 참조 문서 추가 및 이력 저장
            sources = self._format_source_documents(search_result.source_documents)
            final_response = f"{full_answer}\n\n---\n{sources}"
            yield final_response

            self.history.add("user", message)
            self.history.add("assistant", full_answer)

        except Exception as e:
            yield f"답변 생성 중 오류가 발생했습니다: {str(e)}"


# ── 5. 벡터 스토어 및 시스템 초기화 ─────────────────────────────────
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store_loaded = Chroma(
    collection_name="housing_faq_db",
    persist_directory="./chroma_db",
    embedding_function=embeddings,
)

rag_system = ImprovedRAGSystem(
    llm=ChatOpenAI(model="gpt-4.1-mini", temperature=0),
    eval_llm=ChatOpenAI(model="gpt-4.1-mini", temperature=0),
    retriever=vector_store_loaded.as_retriever(
        search_type="mmr",
        search_kwargs={"fetch_k": 10, "k": 3, "lambda_mult": 0.5}
    ),
    max_history_turns=3,
)


# ── 6. Gradio ChatInterface ───────────────────────────────────────────
demo = gr.ChatInterface(
    fn=rag_system.generate_answer,
    title="주택청약 FAQ 챗봇",
    description="""
    주택청약 관련 질문을 입력하면 FAQ 문서를 기반으로 답변합니다.
    - MMR 검색으로 다양한 관련 문서를 검색합니다.
    - 답변 품질을 자동으로 평가하여 불충분한 경우 안내합니다.
    - 최근 3턴의 대화 이력을 기반으로 맥락을 유지합니다.
    """,
    examples=[
        ["수원시의 주택건설지역은 어디에 해당하나요?"],
        ["무주택 세대에 대해서 설명해주세요."],
        ["2순위로 당첨된 사람이 청약통장을 다시 사용할 수 있나요?"],
        ["청약통장 가입 조건은 무엇인가요?"],
    ],
)

demo.launch()