프로젝트 3 — 구름다리여행 종합 상담 챗봇

from pathlib import Path

BOT_CODE = r'''
"""구름다리여행 종합 상담 챗봇의 부품: 약관·FAQ 검색 도구, 상품 도구 둘, 에이전트"""
import os
import re
from pathlib import Path

import pandas as pd
from langchain.agents import create_agent
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

HERE = Path(__file__).parent
HEADER = "구름다리여행 여행 약관 (연습용 가상 문서)"
SYSTEM = (
    "너는 구름다리여행의 상담 챗봇이다. 취소, 환불, 짐, 아동 요금, 반려동물, 보험, 여권 같은 규정은 "
    "search_terms 로 찾아 답하고 끝에 출처를 괄호로 적는다. 상품은 find_products 로 찾고, "
    "요금은 calc_price 로 계산한다. 숫자를 직접 셈하지 않는다. "
    "도구로 찾은 자료에 없으면 '그 내용은 제가 가진 자료에 없습니다. 상담 시간(평일 오전 9시~오후 6시)에 "
    "상담원에게 물어 주세요.'라고 답한다. 답은 두세 문장으로 짧게 쓴다."
)


def load_docs():
    """약관을 조항 단위로 나누고 FAQ 를 더해 Document 목록을 만든다"""
    docs = []
    for p in PyPDFLoader(str(HERE / "travel_terms.pdf")).load():
        text = re.sub(r"\s+", " ", p.page_content.replace(HEADER, "")).strip()
        for part in re.split(r"(?=제\d+조 )", text)[1:]:
            number = re.match(r"제\d+조", part).group()
            docs.append(Document(page_content=part.strip(), metadata={"source": f"약관 {number}({p.metadata['page'] + 1}쪽)"}))
    for r in pd.read_csv(HERE / "faq.csv").itertuples():
        docs.append(Document(page_content=f"{r.질문} {r.답}", metadata={"source": f"FAQ {r.의도}"}))
    return docs


def build_agent(api_key):
    """키를 받아 색인과 도구 셋, 에이전트를 만든다. (에이전트, 색인 조각 수)를 돌려준다"""
    emb = OpenAIEmbeddings(model=os.getenv("OPENAI_EMBED_MODEL", "text-embedding-3-small"), api_key=api_key)
    db = FAISS.from_documents(load_docs(), emb)
    products = pd.read_csv(HERE / "trip_products.csv")

    @tool
    def search_terms(query: str) -> str:
        """구름다리여행의 약관과 자주 묻는 질문(FAQ)에서 query 와 가까운 글 3개를 출처와 함께 찾는다."""
        return "\n".join(f"[{d.metadata['source']}] {d.page_content}" for d in db.similarity_search(query, k=3))

    @tool(parse_docstring=True)
    def find_products(region: str = "", max_price: int = 0) -> str:
        """구름다리여행 상품을 지역과 가격으로 찾는다.

        Args:
            region: 지역 이름. 예: 제주, 부산. 비우면 모든 지역
            max_price: 어른 1명 가격의 상한(만 원). 0 이면 가격을 따지지 않는다
        """
        df = products
        if region:
            df = df[df["지역"] == region]
        if max_price > 0:
            df = df[df["가격"] <= max_price]
        if df.empty:
            return "조건에 맞는 상품이 없습니다."
        return "\n".join(
            f"{r.상품코드} {r.상품명} | {r.가격}만 원 | 남은 자리 {r.남은자리} | 출발 {r.출발일}" for r in df.itertuples()
        )

    @tool(parse_docstring=True)
    def calc_price(code: str, adults: int, children: int = 0) -> str:
        """상품코드와 인원으로 총 여행 요금(만 원)을 계산한다. 아동 요금은 어른의 80% 이다.

        Args:
            code: 상품코드. 예: P01
            adults: 어른 수
            children: 아동(만 2~12세 미만) 수
        """
        rows = products[products["상품코드"] == code]
        if rows.empty:
            return f"없는 상품코드입니다: {code}"
        row = rows.iloc[0]
        total = round(row["가격"] * adults + row["가격"] * 0.8 * children, 1)
        return f"{code} {row['상품명']} · 어른 {adults}명, 아동 {children}명 · {float(total):g}만 원"

    model = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"), api_key=api_key, temperature=0)
    agent = create_agent(model, tools=[search_terms, find_products, calc_price], system_prompt=SYSTEM)
    return agent, db.index.ntotal


def reply(agent, history, question):
    """화면의 기록(history)에 새 질문을 더해 에이전트를 부르고, 답과 이번에 부른 도구 이름을 돌려준다"""
    out = agent.invoke({"messages": history + [{"role": "user", "content": question}]})
    new = out["messages"][len(history) + 1:]          # 이번 질문 뒤에 생긴 메시지
    return out["messages"][-1].content, [m.name for m in new if m.type == "tool"]
'''

Path("out_ch08_gureum_bot.py").write_text(BOT_CODE.lstrip(), encoding="utf-8")
print("저장: out_ch08_gureum_bot.py,", len(BOT_CODE.strip().splitlines()), "줄")

import os
import sys

from dotenv import find_dotenv, load_dotenv

# 작업 폴더(data)의 부품 파일을 불러온다
sys.path.insert(0, ".")
import out_ch08_gureum_bot as bot

load_dotenv(find_dotenv(usecwd=True))
agent, n_docs = bot.build_agent(os.getenv("OPENAI_API_KEY"))
print("검색 조각:", n_docs, "개(약관 16 + FAQ 10)")

answer, used = bot.reply(agent, [], "강아지랑 같이 갈 수 있어요?")
print("부른 도구:", used)
print("답:", answer)

history = []
for q in ["12월에 제주 가는 40만 원 이하 상품 있어요?", "그거 어른 둘에 아이 하나면 얼마예요?",
          "출발 5일 전에 취소하면 수수료는요?"]:
    answer, used = bot.reply(agent, history, q)
    # 화면에 그릴 기록과 같은 꼴로 쌓는다. 다음 차례에 함께 넘어가 "그거"를 알아듣게 한다
    history += [{"role": "user", "content": q}, {"role": "assistant", "content": answer}]
    print(f"Q. {q}\n   부른 도구: {used}\nA. {answer}")
print("기록:", len(history), "개")

import re

import pandas as pd
import streamlit as st
from streamlit.testing.v1 import AppTest

APP_CODE = r'''
import os
import sys
from pathlib import Path

import streamlit as st
from dotenv import find_dotenv, load_dotenv

HERE = Path(__file__).parent
sys.path.insert(0, str(HERE))
import out_ch08_gureum_bot as bot  # noqa: E402

PLACEHOLDER = "발급받은_인증키"


def get_api_key():
    """클라우드에서는 Secrets 칸, 내 PC 에서는 .env 에서 키를 읽는다(7차시 실습 3)"""
    try:
        return st.secrets["OPENAI_API_KEY"]
    except (FileNotFoundError, KeyError):
        load_dotenv(find_dotenv(usecwd=True))
        return os.getenv("OPENAI_API_KEY")


@st.cache_resource(show_spinner="약관과 FAQ 로 색인을 만드는 중입니다...")
def load_agent(api_key):
    return bot.build_agent(api_key)


st.set_page_config(page_title="구름다리여행 상담 챗봇")
st.title("구름다리여행 상담 챗봇")
api_key = get_api_key()
if not api_key or api_key == PLACEHOLDER:
    # 이 앱은 진짜 모델이 도구를 고르므로 키가 없으면 멈춘다
    st.warning("API 키가 없어 상담 챗봇을 열 수 없습니다. .env 나 Secrets 에 키를 넣어 주세요.")
    st.stop()
agent, n_docs = load_agent(api_key)
with st.sidebar:
    st.header("자료")
    st.caption(f"약관·FAQ 조각: {n_docs}개")
    st.caption("도구: search_terms, find_products, calc_price")
    st.caption("연습용으로 지어낸 가상 여행사의 자료입니다.")

if "gureum_messages" not in st.session_state:
    st.session_state.gureum_messages = []
for m in st.session_state.gureum_messages:
    with st.chat_message(m["role"]):
        st.markdown(m["content"])

if question := st.chat_input("상품, 요금, 약관을 물어보세요"):
    with st.chat_message("user"):
        st.markdown(question)
    with st.spinner("자료를 찾는 중입니다..."):
        # 그리기 전의 기록을 넘긴다(7차시: 질문은 부른 뒤에 기록에 넣는다)
        answer, used = bot.reply(agent, st.session_state.gureum_messages, question)
    with st.chat_message("assistant"):
        st.markdown(answer)
        st.caption("부른 도구: " + (", ".join(used) if used else "없음"))
    st.session_state.gureum_messages += [{"role": "user", "content": question},
                                         {"role": "assistant", "content": answer}]
'''
Path("out_ch08_gureum_app.py").write_text(APP_CODE.lstrip(), encoding="utf-8")
print("저장: out_ch08_gureum_app.py,", len(APP_CODE.strip().splitlines()), "줄")

checks = []
st.cache_resource.clear()
# 진짜 모델이 도구를 여러 번 부르므로 기다리는 시간을 넉넉히 준다
at = AppTest.from_file("out_ch08_gureum_app.py", default_timeout=90).run()
checks.append({"점검": "첫 화면", "결과": "통과" if not at.exception and not at.warning else "고칠 것",
               "까닭": " / ".join(c.value for c in at.sidebar.caption[:2])})

# (질문, 불러야 할 도구, 답에 있어야 할 글)
DEMO = [("12월에 제주 가는 40만 원 이하 상품 있어요?", "find_products", "P01"),
        ("그거 어른 둘에 아이 하나면 얼마예요?", "calc_price", "109.2"),
        ("출발 5일 전에 취소하면 수수료는요?", "search_terms", "30%"),
        ("엔화 환율이 얼마예요?", None, "자료에 없습니다")]
for q, tool_name, must in DEMO:
    at.chat_input[0].set_value(q).run()
    reply = at.chat_message[-1]
    said, used = reply.markdown[0].value, reply.caption[0].value
    ok = must in said and (tool_name is None or tool_name in used)
    checks.append({"점검": f"질문: {q}", "결과": "통과" if ok else "고칠 것", "까닭": used})

checks.append({"점검": "기록", "결과": "통과" if len(at.session_state["gureum_messages"]) == 8 else "고칠 것",
               "까닭": f"말풍선 {len(at.chat_message)}개"})
leaks = [f for f in ["out_ch08_gureum_bot.py", "out_ch08_gureum_app.py"]
         if re.search(r"sk-[A-Za-z0-9_-]{20,}", Path(f).read_text(encoding="utf-8"))]
checks.append({"점검": "코드에 적힌 키", "결과": "고칠 것" if leaks else "통과",
               "까닭": ", ".join(leaks) if leaks else "키 꼴의 글 없음"})

demo = pd.DataFrame(checks)
print(demo.to_string(index=False))
demo.to_csv("out_ch08_gureum_demo.csv", index=False, encoding="utf-8-sig")
print("저장: out_ch08_gureum_demo.csv,", len(demo), "행")