프로젝트 1 — 여행 약관 챗봇 만들고 점검하기

import os
import warnings
from pathlib import Path

warnings.filterwarnings("ignore", category=DeprecationWarning)

import pandas as pd
from dotenv import find_dotenv, load_dotenv
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# .env 에서 키와 모델 이름을 읽는다
load_dotenv(find_dotenv(usecwd=True))
MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")
EMBED_MODEL = os.getenv("OPENAI_EMBED_MODEL", "text-embedding-3-small")
model = ChatOpenAI(model=MODEL, temperature=0)
emb = OpenAIEmbeddings(model=EMBED_MODEL)

INDEX = "out_ch06_faiss_openai"
if Path(INDEX).exists():
    # 실습 5에서 내가 만든 색인이라 pickle 을 열어도 된다
    db = FAISS.load_local(INDEX, emb, allow_dangerous_deserialization=True)
    how = "불러옴"
else:
    saved = pd.read_csv("out_ch06_chunks.csv")
    docs = [Document(page_content=row.내용, metadata={"article": row.article, "page": int(row.page)})
            for row in saved.itertuples()]
    db = FAISS.from_documents(docs, emb)
    db.save_local(INDEX)
    how = "새로 만들어 저장함"
print(f"색인 {how}: 조각 {db.index.ntotal}개 / 모델 {MODEL} / 임베딩 {EMBED_MODEL}")

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

NOT_FOUND = "약관에서 찾지 못했습니다."
retriever = db.as_retriever(search_kwargs={"k": 3})


def format_docs(docs):
    return "\n".join(f"[{d.metadata['article']} · {d.metadata['page']}쪽] {d.page_content}" for d in docs)


prompt = ChatPromptTemplate.from_messages([
    ("system",
     "너는 구름다리여행의 약관 안내 직원이다. 아래 [근거]에 적힌 내용으로만 두세 문장으로 답한다. "
     "숫자와 조건은 근거에 적힌 표기 그대로 쓴다.\n"
     f"근거에 없는 내용이면 '{NOT_FOUND}'라고만 답한다.\n\n[근거]\n{{context}}"),
    ("human", "{question}"),
])
answer_chain = prompt | model | StrOutputParser()
rag = RunnableParallel(docs=retriever, question=RunnablePassthrough()).assign(
    answer=lambda x: answer_chain.invoke({"context": format_docs(x["docs"]), "question": x["question"]})
)


def ask(question):
    """답과 근거(조항 목록, 출처 글)를 사전으로 돌려준다. 출처는 metadata 에서 우리 코드가 붙인다"""
    out = rag.invoke(question)
    articles = [d.metadata["article"] for d in out["docs"]]
    sources = ", ".join(f"{d.metadata['article']}({d.metadata['page']}쪽)" for d in out["docs"])
    return {"answer": out["answer"], "articles": articles, "sources": sources}


for q in ["강아지 데려가도 돼요?", "출발 5일 전에 취소하면 수수료가 얼마예요?"]:
    r = ask(q)
    print(f"Q. {q}\nA. {r['answer']}\n   근거: {r['sources']}")

questions = pd.read_csv("rag_questions.csv")
OUTSIDE = ["제주도 맛집 추천해 주세요.", "지금 엔화 환율이 얼마예요?", "항공 마일리지는 어떻게 쌓나요?"]

rows = []
for row in questions.itertuples():
    r = ask(row.질문)
    rows.append({"번호": row.번호, "질문": row.질문, "답": r["answer"], "근거": r["sources"],
                 "근거에 정답": row.정답조항 in r["articles"],
                 "핵심어 있음": row.핵심어 in r["answer"],
                 "찾지 못함이라 답함": NOT_FOUND in r["answer"]})
for k, q in enumerate(OUTSIDE, start=len(questions) + 1):
    r = ask(q)
    rows.append({"번호": k, "질문": q, "답": r["answer"], "근거": r["sources"],
                 "근거에 정답": None, "핵심어 있음": None,
                 "찾지 못함이라 답함": NOT_FOUND in r["answer"]})

report = pd.DataFrame(rows)
print(report[["번호", "근거에 정답", "핵심어 있음", "찾지 못함이라 답함"]].to_string(index=False))

inside, outside = report.iloc[:len(questions)], report.iloc[len(questions):]
print("근거에 정답 조항:", int(inside["근거에 정답"].sum()), "/", len(inside))
print("답에 핵심어:", int(inside["핵심어 있음"].sum()), "/", len(inside))
print("약관에 없는 질문에 찾지 못함:", int(outside["찾지 못함이라 답함"].sum()), "/", len(outside))

# 핵심어가 빠진 답은 눈으로 읽어 본다
for row in inside[~inside["핵심어 있음"].astype(bool)].itertuples():
    print(f"[{row.번호}] {row.질문}\n    → {row.답}")
print("[13]", outside.iloc[0]["답"])

report.to_csv("out_ch06_bot_answers.csv", index=False, encoding="utf-8-sig")
print("저장: out_ch06_bot_answers.csv")