프로젝트 1 — 실무 문서 챗봇: 사내 규정 도우미

import re

import pandas as pd
from langchain_community.document_loaders import PyPDFLoader
from langchain_core.documents import Document

PDF = "solmaru_rules.pdf"
# 쪽마다 맨 위에 붙은 머리글. 조각에 섞이면 검색을 흐리므로 지운다(6차시 실습 2)
HEADER = "솔마루상사 복무·출장 규정 (연습용 가상 문서)"

pages = PyPDFLoader(PDF).load()
print("쪽 수:", len(pages), "/ 쪽별 글자 수:", [len(p.page_content) for p in pages])


def split_articles(pages):
    """머리글을 지우고 '제N조 ' 앞에서 잘라 조항 하나를 조각 하나로 만든다(6차시 실습 2와 같은 방법)"""
    docs = []
    for p in pages:
        text = re.sub(r"\s+", " ", p.page_content.replace(HEADER, "")).strip()
        parts = re.split(r"(?=제\d+조 )", text)
        chapter = parts[0].strip()                      # 첫 토막은 "제1장 근무" 같은 장 제목
        for part in parts[1:]:
            number = re.match(r"제\d+조", part).group()
            docs.append(Document(page_content=part.strip(),
                                 metadata={"article": number, "chapter": chapter, "page": p.metadata["page"] + 1}))
    return docs


docs = split_articles(pages)
print("조항 조각:", len(docs), "개")
for d in docs[2:4]:
    print(d.metadata, "|", d.page_content[:45], "…")

import zlib

from langchain_community.vectorstores import FAISS
from langchain_core.embeddings import Embeddings


class BigramEmbeddings(Embeddings):
    """글자 두 개 묶음(바이그램)을 세어 숫자 목록을 만드는 연습용 임베딩"""

    def __init__(self, size=512):
        self.size = size

    def _vector(self, text):
        text = text.replace(" ", "").replace("\n", "")
        vec = [0.0] * self.size
        for i in range(len(text) - 1):
            # 글자 두 개를 crc32 로 0~511 사이의 칸 번호로 바꿔 그 칸을 1 올린다
            vec[zlib.crc32(text[i:i + 2].encode("utf-8")) % self.size] += 1.0
        norm = sum(v * v for v in vec) ** 0.5 or 1.0
        return [v / norm for v in vec]

    def embed_documents(self, texts):
        return [self._vector(t) for t in texts]

    def embed_query(self, text):
        return self._vector(text)


db = FAISS.from_documents(docs, BigramEmbeddings())
questions = pd.read_csv("rules_questions.csv")


def measure(db, questions, rewrite=lambda q: q):
    """질문마다 가까운 조각 3개를 찾아 1위 조항, 1위 거리, 적중(1위, 3위 안)을 표로 돌려준다"""
    rows = []
    for row in questions.itertuples():
        hits = db.similarity_search_with_score(rewrite(row.질문), k=3)
        found = [d.metadata["article"] for d, _ in hits]
        rows.append({"번호": row.번호, "정답": row.정답조항, "찾은 조항": " ".join(found),
                     "1위 거리": round(float(hits[0][1]), 3),
                     "1위 적중": found[0] == row.정답조항, "3위 안 적중": row.정답조항 in found})
    return pd.DataFrame(rows)


before = measure(db, questions)
print(before.to_string(index=False))
print("1위 적중:", int(before["1위 적중"].sum()), "/", len(before),
      "/ 3위 안 적중:", int(before["3위 안 적중"].sum()), "/", len(before))

# 시연에서 받은 피드백(지어낸 것): "직원들은 규정의 낱말 대신 평소 말로 묻는다"
# 그래서 평소 말 → 규정의 낱말로 바꾸는 별칭 사전을 만들어 질문을 고쳐 쓴다
ALIASES = {"집에서 일": "재택근무", "아내": "배우자", "낳으면": "출산하면",
           "택시비": "택시 교통비", "다녀와서": "돌아온 날부터 경비 정산"}


def rewrite(question):
    for word, rule_word in ALIASES.items():
        question = question.replace(word, rule_word)
    return question


after = measure(db, questions, rewrite)
print(after[["번호", "정답", "찾은 조항", "1위 거리", "1위 적중"]].to_string(index=False))
print("1위 적중:", int(after["1위 적중"].sum()), "/", len(after),
      "/ 3위 안 적중:", int(after["3위 안 적중"].sum()), "/", len(after))

# 사전에 없는 새 말투로도 찾는지 본다(평가 질문에 맞춰 사전을 만들었으므로 새 질문으로 따로 잰다)
NEW = [("재택은 한 주에 며칠까지예요?", "제3조"), ("와이프 출산 휴가 며칠이에요?", "제7조"),
       ("택시 타도 돼요?", "제11조"), ("출장 경비 정산은 언제 들어와요?", "제12조")]
for q, answer in NEW:
    d, s = db.similarity_search_with_score(rewrite(q), k=1)[0]
    print(f"새 질문 {q} → {d.metadata['article']} (정답 {answer}, 거리 {round(float(s), 3)})")

# 규정에 없는 질문. 1위 거리가 문턱보다 멀면 "찾지 못했습니다"라고 답하게 한다
OUTSIDE = ["회사 주차장은 몇 대까지 대요?", "올해 성과급은 언제 나와요?", "사내 식당 메뉴 알려 주세요."]
outside_dist = [round(float(db.similarity_search_with_score(rewrite(q), k=1)[0][1]), 3) for q in OUTSIDE]
hit_far = float(after.loc[after["1위 적중"], "1위 거리"].max())     # 맞게 찾은 질문 가운데 가장 먼 거리
out_near = min(outside_dist)                                       # 규정 밖 질문 가운데 가장 가까운 거리
THRESHOLD = round((hit_far + out_near) / 2, 2)
print("규정 밖 질문의 1위 거리:", outside_dist)
print("맞게 찾은 가장 먼 거리:", hit_far, "/ 밖 질문의 가장 가까운 거리:", out_near, "→ 문턱:", THRESHOLD)

after["판정"] = ["모름" if dist > THRESHOLD else ("맞음" if ok else "틀림")
                 for dist, ok in zip(after["1위 거리"], after["1위 적중"])]
print("평가 질문 판정:", after["판정"].value_counts().to_dict())
print("규정 밖 질문에 모름:", sum(d > THRESHOLD for d in outside_dist), "/", len(OUTSIDE))

report = before[["번호", "정답", "찾은 조항", "1위 적중"]].merge(
    after[["번호", "찾은 조항", "1위 거리", "1위 적중", "판정"]], on="번호", suffixes=("_전", "_후"))
report.to_csv("out_ch08_rules_eval.csv", index=False, encoding="utf-8-sig")
print("저장: out_ch08_rules_eval.csv,", len(report), "행")

from pathlib import Path

# 챗봇 로직(부품)을 한 파일에, 화면을 다른 파일에 둔다(7차시 "챗봇 로직과 웹 화면 잇기")
BOT_CODE = r'''
"""솔마루상사 규정 도우미의 부품: 조각 나누기, 연습용 임베딩, 검색, 문턱, 답하기"""
import re
import zlib

from langchain_community.document_loaders import PyPDFLoader
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

HEADER = "솔마루상사 복무·출장 규정 (연습용 가상 문서)"
ALIASES = {"집에서 일": "재택근무", "아내": "배우자", "낳으면": "출산하면",
           "택시비": "택시 교통비", "다녀와서": "돌아온 날부터 경비 정산"}
THRESHOLD = 1.67
NOT_FOUND = "규정에서 찾지 못했습니다. 인사팀에 물어 주세요."
PLACEHOLDER = "발급받은_인증키"


class BigramEmbeddings(Embeddings):
    """글자 두 개 묶음(바이그램)을 세어 숫자 목록을 만드는 연습용 임베딩"""

    def __init__(self, size=512):
        self.size = size

    def _vector(self, text):
        text = text.replace(" ", "").replace("\n", "")
        vec = [0.0] * self.size
        for i in range(len(text) - 1):
            # 글자 두 개를 crc32 로 0~511 사이의 칸 번호로 바꿔 그 칸을 1 올린다
            vec[zlib.crc32(text[i:i + 2].encode("utf-8")) % self.size] += 1.0
        norm = sum(v * v for v in vec) ** 0.5 or 1.0
        return [v / norm for v in vec]

    def embed_documents(self, texts):
        return [self._vector(t) for t in texts]

    def embed_query(self, text):
        return self._vector(text)


def load_chunks(pdf):
    """PDF 를 읽어 머리글을 지우고 조항 하나를 조각 하나로 만든다"""
    docs = []
    for p in PyPDFLoader(str(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={"article": number, "page": p.metadata["page"] + 1}))
    return docs


def build_index(docs):
    return FAISS.from_documents(docs, BigramEmbeddings())


def rewrite(question):
    """직원의 평소 말을 규정의 낱말로 바꾼다"""
    for word, rule_word in ALIASES.items():
        question = question.replace(word, rule_word)
    return question


def make_llm(api_key):
    """키가 있으면 진짜 모델, 없거나 자리표시면 연습 모드의 가짜 모델"""
    if api_key and api_key != PLACEHOLDER:
        import os

        from langchain_openai import ChatOpenAI
        return ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"), api_key=api_key, temperature=0), "OpenAI"
    return FakeListChatModel(responses=["(연습 모드) 진짜 모델 없이 규정만 찾았습니다. 아래 근거 조항을 확인해 주세요."]), "연습 모드"


PROMPT = ChatPromptTemplate.from_messages([
    ("system", "너는 솔마루상사의 규정 안내 도우미다. 아래 [근거]의 조항에 적힌 내용으로만 두세 문장으로 답하고, "
               "숫자는 근거의 표기 그대로 쓴다. 근거에 없으면 '" + NOT_FOUND + "'라고만 답한다.\n\n[근거]\n{context}"),
    ("human", "{question}"),
])


def answer(db, llm, question, k=3):
    """답, 근거 목록, 1위 거리를 돌려준다. 1위가 문턱보다 멀면 모델을 부르지 않고 모른다고 답한다"""
    hits = [(d, float(s)) for d, s in db.similarity_search_with_score(rewrite(question), k=k)]
    best = hits[0][1]
    if best > THRESHOLD:
        return {"answer": NOT_FOUND, "sources": [], "best": best}
    near = [(d, s) for d, s in hits if s <= THRESHOLD]          # 문턱 안의 조각만 근거로 쓴다
    context = "\n".join(f"[{d.metadata['article']} · {d.metadata['page']}쪽] {d.page_content}" for d, _ in near)
    text = (PROMPT | llm | StrOutputParser()).invoke({"context": context, "question": question})
    sources = [f"{d.metadata['article']}({d.metadata['page']}쪽) · 거리 {s:.3f} · {d.page_content[:30]}…" for d, s in near]
    return {"answer": text, "sources": sources, "best": best}
'''

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_rules_bot as bot  # noqa: E402


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="규정으로 색인을 만드는 중입니다...")
def load_index():
    docs = bot.load_chunks(HERE / "solmaru_rules.pdf")
    return bot.build_index(docs), len(docs)


@st.cache_resource
def load_llm(api_key):
    return bot.make_llm(api_key)


def show(message):
    with st.chat_message(message["role"]):
        st.markdown(message["content"])
        if message.get("sources"):
            with st.expander("근거 조항 보기"):
                for s in message["sources"]:
                    st.caption(s)


st.set_page_config(page_title="솔마루상사 규정 도우미")
st.title("솔마루상사 규정 도우미")
db, n_chunks = load_index()
llm, mode = load_llm(get_api_key())
with st.sidebar:
    st.header("문서")
    st.caption(f"규정 조각: {n_chunks}개")
    st.caption(f"답하는 모델: {mode}")
    st.caption("연습용으로 지어낸 가상 회사의 규정입니다.")

if "rule_messages" not in st.session_state:
    st.session_state.rule_messages = []
for m in st.session_state.rule_messages:
    show(m)

if question := st.chat_input("근무, 휴가, 출장 규정을 물어보세요"):
    user = {"role": "user", "content": question}
    show(user)
    with st.spinner("규정에서 찾는 중입니다..."):
        result = bot.answer(db, llm, question)
    reply = {"role": "assistant", "content": result["answer"], "sources": result["sources"]}
    show(reply)
    st.session_state.rule_messages.extend([user, reply])
'''

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

# 부품 파일을 불러와 3단계의 값과 같은지 본다
import sys

sys.path.insert(0, ".")
import out_ch08_rules_bot as bot

print("문턱이 3단계와 같은가:", bot.THRESHOLD == THRESHOLD, "/ 별칭 사전이 같은가:", bot.ALIASES == ALIASES)
llm, mode = bot.make_llm(None)
r = bot.answer(bot.build_index(bot.load_chunks(PDF)), llm, "해외 출장은 언제까지 신청하나요?")
print("모드:", mode, "/ 1위 거리:", round(r["best"], 3))
print("근거:", r["sources"])

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

KEY_PATTERN = re.compile(r"sk-[A-Za-z0-9_-]{20,}")
checks = []


def check(name, ok, why):
    checks.append({"점검": name, "결과": "통과" if ok else "고칠 것", "까닭": why})


st.cache_resource.clear()
at = AppTest.from_file("out_ch08_rules_app.py", default_timeout=30)
# 자리표시를 Secrets 에 넣어 연습 모드로 시험한다(진짜 모델을 부르지 않는다)
at.secrets["OPENAI_API_KEY"] = "발급받은_인증키"
at.run()
side = [c.value for c in at.sidebar.caption]
check("첫 화면", len(at.exception) == 0 and at.title[0].value == "솔마루상사 규정 도우미", " / ".join(side))

# 시연 질문 셋: 규정에 있는 질문, 사전에 없는 말투의 질문, 규정에 없는 질문
DEMO = [("해외 출장은 언제까지 신청하나요?", "제9조"), ("재택은 한 주에 며칠까지예요?", "제3조"),
        ("회사 주차장은 몇 대까지 대요?", None)]
for q, want in DEMO:
    at.chat_input[0].set_value(q).run()
    reply = at.chat_message[-1]
    sources = [c.value for c in reply.expander[0].caption] if reply.expander else []
    if want:
        first = sources[0].split("(")[0] if sources else "없음"
        check(f"질문: {q}", first == want, f"첫 근거 {first}, 근거 {len(sources)}개")
    else:
        said = reply.markdown[0].value
        check(f"질문: {q}", said == bot.NOT_FOUND and not sources, said)

check("기록", len(at.session_state["rule_messages"]) == 2 * len(DEMO), f"말풍선 {len(at.chat_message)}개")
leaks = [f for f in ["out_ch08_rules_bot.py", "out_ch08_rules_app.py"]
         if KEY_PATTERN.search(Path(f).read_text(encoding="utf-8"))]
check("코드에 적힌 키", not leaks, "키 꼴의 글이 든 파일: " + ", ".join(leaks) if leaks else "키 꼴의 글 없음")

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