프로젝트 2 — 전공 지식 챗봇: 정보통신 학습 도우미

import os
import re
from pathlib import Path

import pandas as pd
from dotenv import find_dotenv, load_dotenv
from langchain_community.document_loaders import PyPDFLoader
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)

HEADER = "정보통신 기초 수업 노트 (연습용 가상 문서)"
docs = []
for p in PyPDFLoader("network_notes.pdf").load():
    text = re.sub(r"\s+", " ", p.page_content.replace(HEADER, "")).strip()
    # "노트 N. " 바로 앞에서 자른다. 첫 토막은 "1주차 네트워크의 짜임" 같은 주차 제목
    parts = re.split(r"(?=노트 \d+\. )", text)
    week = parts[0].strip()
    for part in parts[1:]:
        note = re.match(r"노트 \d+", part).group()
        docs.append(Document(page_content=part.strip(),
                             metadata={"note": note, "week": week, "page": p.metadata["page"] + 1}))
print("노트 조각:", len(docs), "개 /", [d.metadata["note"] for d in docs[:3]], "…")

INDEX = "out_ch08_notes_faiss"
if Path(INDEX).exists():
    # 내가 만든 색인 폴더라 pickle 을 열어도 된다(6차시)
    db = FAISS.load_local(INDEX, emb, allow_dangerous_deserialization=True)
    how = "불러옴"
else:
    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, MessagesPlaceholder

NOT_IN_NOTES = "수업 노트에 없는 내용입니다. 담당 강사에게 물어 주세요."
retriever = db.as_retriever(search_kwargs={"k": 2})

answer_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "너는 정보통신 기초 수업의 학습 도우미다. 처음 배우는 학생에게 아래 [노트]의 내용으로만 두세 문장으로 쉽게 설명한다. "
     "용어와 숫자는 노트의 표기 그대로 쓴다. 노트에 없는 내용이면 '" + NOT_IN_NOTES + "'라고만 답한다.\n\n[노트]\n{context}"),
    MessagesPlaceholder("history"),          # 앞의 대화(4차시)
    ("human", "{question}"),
])
answer_chain = answer_prompt | model | StrOutputParser()


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


def ask(question, history=()):
    """검색할 질문으로 노트를 찾아 답하고, 근거 노트 번호를 함께 돌려준다. 출처는 코드가 metadata 에서 붙인다"""
    found = retriever.invoke(question)
    text = answer_chain.invoke({"context": format_notes(found), "history": list(history), "question": question})
    return {"answer": text, "notes": [d.metadata["note"] for d in found]}


for q in ["TCP는 연결할 때 무엇을 주고받나요?", "/24 네트워크에서 기기에 줄 수 있는 주소는 몇 개인가요?"]:
    r = ask(q)
    print(f"Q. {q}\nA. {r['answer']}\n   근거: {', '.join(r['notes'])}")

rewrite_prompt = ChatPromptTemplate.from_messages([
    ("system", "대화 기록을 보고 마지막 질문을 앞의 대화 없이도 뜻이 통하는 질문 한 문장으로 바꿔 쓴다. "
               "답하지 말고 바꾼 질문만 쓴다."),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])
rewrite_chain = rewrite_prompt | model | StrOutputParser()

history = []
for q in ["DNS는 무슨 일을 하나요?", "그건 몇 번 포트를 써요?", "그럼 DHCP는요?"]:
    # 첫 질문은 기록이 비어 있어 바꾸지 않는다
    standalone = rewrite_chain.invoke({"history": history, "question": q}) if history else q
    plain = [d.metadata["note"] for d in retriever.invoke(q)]          # 바꿔 쓰지 않고 찾았다면
    r = ask(standalone, history)
    # 기록에는 학생이 쓴 원래 질문을 넣는다
    history += [{"role": "user", "content": q}, {"role": "assistant", "content": r["answer"]}]
    print(f"Q. {q}\n   바꿔 쓴 질문: {standalone}\n   바꾸지 않고 찾은 노트: {plain} → 바꿔 쓰고 찾은 노트: {r['notes']}")
    print(f"A. {r['answer']}")

questions = pd.read_csv("notes_questions.csv")
OUTSIDE = ["6G 이동통신은 언제부터 쓸 수 있나요?", "이번 학기 시험 범위가 어디까지예요?"]

rows = []
for row in questions.itertuples():
    r = ask(row.질문)
    rows.append({"번호": row.번호, "질문": row.질문, "답": r["answer"], "근거": ", ".join(r["notes"]),
                 "근거에 정답": row.정답노트 in r["notes"], "핵심어 있음": row.핵심어 in r["answer"],
                 "노트에 없음": NOT_IN_NOTES in r["answer"]})
for k, q in enumerate(OUTSIDE, start=len(questions) + 1):
    r = ask(q)
    rows.append({"번호": k, "질문": q, "답": r["answer"], "근거": ", ".join(r["notes"]),
                 "근거에 정답": None, "핵심어 있음": None, "노트에 없음": NOT_IN_NOTES in r["answer"]})

report = pd.DataFrame(rows)
print(report[["번호", "근거", "근거에 정답", "핵심어 있음", "노트에 없음"]].to_string(index=False))
inside = report.iloc[:len(questions)]
print("근거에 정답 노트:", int(inside["근거에 정답"].sum()), "/", len(inside),
      "/ 답에 핵심어:", int(inside["핵심어 있음"].sum()), "/", len(inside),
      "/ 노트 밖에 노트에 없음:", int(report.iloc[len(questions):]["노트에 없음"].sum()), "/", len(OUTSIDE))
# 핵심어가 빠진 답은 눈으로 읽어 본다
for row in inside[~inside["핵심어 있음"].astype(bool)].itertuples():
    print(f"[{row.번호}] {row.질문}\n    → {row.답}")

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