import os
import re
import csv
import uuid
import html
import json
import fitz
from pathlib import Path
from datetime import datetime
from typing import TypedDict, Literal
import streamlit as st
import streamlit.components.v1 as components
from openai import OpenAI
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document
from langgraph.graph import StateGraph, END
# =========================================================
# 0. 기본 설정
# =========================================================
st.set_page_config(
page_title="AIVLE 학습도우미",
page_icon="🤖",
layout="wide"
)
BASE_DIR = Path(__file__).resolve().parent
# 통합 백서 PDF 하나만 사용합니다.
# 파일명이 다를 수 있어, 아래 후보 순서대로 먼저 존재하는 PDF를 사용합니다.
PDF_CANDIDATES = [
BASE_DIR / "AIVLE_School_백서_통합본.pdf",
]
PDF_PATH = next((path for path in PDF_CANDIDATES if path.exists()), PDF_CANDIDATES[0])
UPLOAD_DIR = BASE_DIR / "uploads"
TEMP_DIR = BASE_DIR / "temp"
IMAGE_BANNER_PATH = BASE_DIR / "images" / "이미지.png"
LOGO_PATH = BASE_DIR / "images" / "에이블학습도우미 로고.png"
UPLOAD_DIR.mkdir(exist_ok=True)
TEMP_DIR.mkdir(exist_ok=True)
client = OpenAI()
# =========================================================
# 1. CSS: 1번 사이드바/로그인 UI + 2번 채팅 UI 결합
# =========================================================
st.markdown("""
""", unsafe_allow_html=True)
# =========================================================
# 2. 세션 상태 초기화
# =========================================================
def init_session_state():
if "logged_in" not in st.session_state:
st.session_state.logged_in = False
if "user_id" not in st.session_state:
st.session_state.user_id = ""
if "user_name" not in st.session_state:
st.session_state.user_name = ""
if "page" not in st.session_state:
st.session_state.page = "home"
if "chats" not in st.session_state:
st.session_state.chats = {}
if "current_chat_id" not in st.session_state:
st.session_state.current_chat_id = None
if "recommended_questions" not in st.session_state:
st.session_state.recommended_questions = [
"출석 인정 요청은 어떻게 하나요?",
"강의 다시보기를 할 수 있나요?",
"훈련장려금은 언제 지급되나요?",
"개인 포트폴리오로 활용할 수 있는 범위가 어떻게 되나요?"
]
if "rag_upload_signature" not in st.session_state:
st.session_state.rag_upload_signature = None
if "retriever" not in st.session_state:
st.session_state.retriever = None
if "rag_chain" not in st.session_state:
st.session_state.rag_chain = None
if "format_docs" not in st.session_state:
st.session_state.format_docs = None
init_session_state()
# =========================================================
# 3. 대화 세션 관리 함수: 1번 코드 기반
# =========================================================
def create_new_chat():
chat_id = str(uuid.uuid4())
st.session_state.chats[chat_id] = {
"title": "새 대화",
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
"messages": []
}
st.session_state.current_chat_id = chat_id
st.session_state.page = "home"
def get_current_messages():
if st.session_state.current_chat_id is None:
create_new_chat()
return st.session_state.chats[st.session_state.current_chat_id]["messages"]
def update_chat_title(question):
chat_id = st.session_state.current_chat_id
if chat_id is None:
return
current_title = st.session_state.chats[chat_id]["title"]
if current_title == "새 대화":
title = question.strip()
if len(title) > 24:
title = title[:24] + "..."
st.session_state.chats[chat_id]["title"] = title
def delete_chat(chat_id):
if chat_id in st.session_state.chats:
del st.session_state.chats[chat_id]
if st.session_state.current_chat_id == chat_id:
if st.session_state.chats:
latest_chat_id = list(st.session_state.chats.keys())[-1]
st.session_state.current_chat_id = latest_chat_id
else:
create_new_chat()
st.session_state.page = "home"
# =========================================================
# 4. 문서 로딩 함수: 1번 코드 기반 + UPLOAD_DIR 보완
# =========================================================
def safe_filename(filename):
filename = re.sub(r"[^가-힣a-zA-Z0-9_.-]", "_", filename)
return filename
def load_txt_file(file_path):
try:
text = file_path.read_text(encoding="utf-8")
except UnicodeDecodeError:
text = file_path.read_text(encoding="cp949")
return [
Document(
page_content=text,
metadata={"source": str(file_path)}
)
]
def load_csv_file(file_path):
rows = []
try:
f = open(file_path, "r", encoding="utf-8")
except UnicodeDecodeError:
f = open(file_path, "r", encoding="cp949")
with f:
reader = csv.reader(f)
for row in reader:
rows.append(" | ".join(row))
text = "\n".join(rows)
return [
Document(
page_content=text,
metadata={"source": str(file_path)}
)
]
def load_pdf_file(file_path):
loader = PyMuPDFLoader(str(file_path))
return loader.load()
def load_faq_from_pdf(pdf_path):
"""
업로드된 통합 백서 PDF 안의 제4장 FAQ 표를 읽어서
category/question/answer 구조로 변환한다.
이 함수는 별도 FAQ PDF를 사용하지 않는다.
현재 파일처럼 FAQ가 18~31쪽에 여러 표로 나뉘어 있고,
일부 페이지에는 표 헤더가 반복되지 않거나 답변이 다음 페이지로 이어지는 경우까지 처리한다.
"""
if not pdf_path.exists():
st.warning(f"통합 백서 PDF 파일을 찾을 수 없습니다: {pdf_path}")
return []
faq_data = []
# 이 PDF에서 실제 FAQ 표에 쓰이는 카테고리만 허용한다.
# 이렇게 해야 1~8기 우수 프로젝트 목록 같은 다른 표가 FAQ로 잘못 들어가는 것을 막을 수 있다.
valid_categories = {
"모집/선발",
"출석",
"강의",
"KDT",
"AIVLE-EDU",
"교육장",
"기타",
"코딩 학습",
"노트북",
"국민취업제도",
"취업지원/채용연계",
"학습",
"프로젝트",
}
def clean_text(text):
if text is None:
return ""
text = str(text).replace("\n", " ")
text = re.sub(r"\s+", " ", text)
return text.strip()
def normalize_category(text):
category = clean_text(text)
# PDF 표 추출 시 줄바꿈 때문에 카테고리가 쪼개지는 경우 보정
category = category.replace("국민취업제 도", "국민취업제도")
category = category.replace("취업지원/ 채용연계", "취업지원/채용연계")
return category
try:
doc = fitz.open(str(pdf_path))
in_faq_section = False
last_item = None
for page in doc:
page_text = page.get_text("text")
# 목차에도 "제4장" 문구가 있기 때문에, FAQ 본문 안내 문장 또는 실제 본문 페이지 조건으로 시작점을 잡는다.
if (
"FAQ 는 지원 전 궁금증" in page_text
or (
re.search(r"제\s*4\s*장\.?\s*자주 묻는 질문", page_text)
and "목 차" not in page_text
and page.number > 5
)
):
in_faq_section = True
if not in_faq_section:
continue
tables = page.find_tables()
for table in tables:
rows = table.extract()
for row in rows:
# PyMuPDF가 9열짜리 표처럼 읽더라도 빈 칸을 제거하면
# 실제 값은 ["구분", "질문", "답변"] 또는 [카테고리, 질문, 답변] 형태가 된다.
cells = [clean_text(cell) for cell in row if clean_text(cell)]
if not cells:
continue
# 표 헤더 제거
if cells == ["구분", "질문", "답변"]:
continue
# 이전 답변이 다음 페이지/다음 행으로 이어진 경우
# 예: 앞 페이지 마지막 답변의 나머지 문장이 다음 페이지 첫 행에 단독으로 잡히는 경우
if len(cells) == 1:
if last_item and not cells[0].startswith("KT AIVLE"):
last_item["answer"] = f"{last_item['answer']}\n{cells[0]}".strip()
continue
if len(cells) < 3:
continue
category = normalize_category(cells[0])
question = clean_text(cells[1])
answer = clean_text(cells[2])
if category not in valid_categories:
continue
if not question or not answer:
continue
if question == "질문" or answer == "답변":
continue
item = {
"category": category,
"question": question,
"answer": answer
}
faq_data.append(item)
last_item = item
# 제5장이 나오면 FAQ는 끝난다.
# 같은 페이지에 FAQ 표가 먼저 있고 제5장이 아래에 붙어 있을 수 있으므로
# 현재 페이지 처리는 마친 뒤 종료한다.
if re.search(r"제\s*5\s*장\.?\s*선배", page_text):
break
doc.close()
except Exception as e:
st.warning(f"통합 백서에서 FAQ 표 추출 중 오류가 발생했습니다: {e}")
return []
# 중복 제거: 같은 질문이 요약 FAQ 표에 한 번 더 나올 수 있으므로 질문 기준으로 한 번만 표시
unique_faq_data = []
seen_questions = set()
for item in faq_data:
normalized_question = re.sub(r"\s+", " ", item["question"]).strip()
if normalized_question in seen_questions:
continue
seen_questions.add(normalized_question)
unique_faq_data.append(item)
return unique_faq_data
def save_uploaded_files(uploaded_files):
saved_paths = []
if not uploaded_files:
return saved_paths
for uploaded_file in uploaded_files:
filename = safe_filename(uploaded_file.name)
save_path = UPLOAD_DIR / filename
with open(save_path, "wb") as f:
f.write(uploaded_file.getbuffer())
saved_paths.append(save_path)
return saved_paths
def load_all_documents(uploaded_files):
docs = []
if PDF_PATH.exists():
docs.extend(load_pdf_file(PDF_PATH))
else:
st.warning(f"기본 백서 PDF를 찾을 수 없습니다: {PDF_PATH}")
# FAQ는 별도 PDF가 아니라 통합 백서 PDF 안에 포함되어 있으므로
# 여기에서 따로 추가하지 않습니다. PDF_PATH 하나만 문서로 사용합니다.
saved_paths = save_uploaded_files(uploaded_files)
for path in saved_paths:
suffix = path.suffix.lower()
if suffix == ".pdf":
docs.extend(load_pdf_file(path))
elif suffix == ".txt":
docs.extend(load_txt_file(path))
elif suffix == ".csv":
docs.extend(load_csv_file(path))
return docs
def get_upload_signature(uploaded_files):
if not uploaded_files:
return "no_upload"
return "|".join([f"{f.name}:{f.size}" for f in uploaded_files])
FAQ_DATA = load_faq_from_pdf(PDF_PATH)
if not FAQ_DATA:
st.warning("통합 백서에서 FAQ 데이터를 불러오지 못했습니다. FAQ 표 구조를 확인하세요.")
# =========================================================
# 5. RAG 구성 함수: 1번 코드 기반
# =========================================================
def build_rag(uploaded_files):
docs = load_all_documents(uploaded_files)
if not docs:
st.error("RAG에 사용할 문서가 없습니다. 기본 백서 PDF 또는 업로드 파일을 확인하세요.")
st.stop()
splitter = RecursiveCharacterTextSplitter(
chunk_size=1200,
chunk_overlap=200,
separators=["\n\n", "\n", ".", " ", ""]
)
chunks = splitter.split_documents(docs)
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embedding)
retriever = vectorstore.as_retriever(
search_kwargs={"k": 8}
)
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0
)
prompt = ChatPromptTemplate.from_template("""
당신은 AIVLE School 백서, FAQ, 사용자가 업로드한 자료를 기반으로 답변하는 학습도우미 챗봇입니다.
규칙:
1. 반드시 [문서 내용]에 근거해서 답변하세요.
2. 문서에서 확인되지 않는 내용은 "문서에서 확인되지 않습니다."라고 답하세요.
3. 한국어로 친절하고 간결하게 답변하세요.
4. 질문과 직접 관련된 내용만 답변하세요.
[문서 내용]
{context}
[질문]
{question}
""")
chain = prompt | llm | StrOutputParser()
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
return retriever, chain, format_docs
def ensure_rag_ready(uploaded_files):
current_signature = get_upload_signature(uploaded_files)
if (
st.session_state.retriever is None
or st.session_state.rag_chain is None
or st.session_state.rag_upload_signature != current_signature
):
with st.spinner("문서를 읽고 AI 검색 인덱스를 준비하는 중입니다..."):
retriever, rag_chain, format_docs = build_rag(uploaded_files)
st.session_state.retriever = retriever
st.session_state.rag_chain = rag_chain
st.session_state.format_docs = format_docs
st.session_state.rag_upload_signature = current_signature
# =========================================================
# 6. FAQ 기반 추천 질문: 1번 코드 기반
# =========================================================
@st.cache_resource
def build_faq_retriever():
faq_docs = []
for item in FAQ_DATA:
faq_docs.append(
Document(
page_content=item["question"],
metadata={
"category": item["category"],
"answer": item["answer"]
}
)
)
if not faq_docs:
return None
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
faq_vectorstore = FAISS.from_documents(faq_docs, embedding)
return faq_vectorstore.as_retriever(search_kwargs={"k": 4})
def recommend_questions(user_question):
try:
faq_retriever = build_faq_retriever()
if faq_retriever is None:
return []
docs = faq_retriever.invoke(user_question)
recommended = []
for doc in docs:
q = doc.page_content.strip()
if q and q not in recommended:
recommended.append(q)
return recommended[:4]
except Exception as e:
st.warning(f"추천질문 생성 중 오류가 발생했습니다: {e}")
return []
# =========================================================
# 7. LangGraph 질문 분기 처리: 1번 코드 기반
# =========================================================
class ChatState(TypedDict):
question: str
route: str
context: str
answer: str
def classify_question_node(state: ChatState) -> ChatState:
question = state["question"]
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0
)
prompt = ChatPromptTemplate.from_template("""
당신은 사용자 질문을 분류하는 라우터입니다.
아래 기준에 따라 질문을 반드시 둘 중 하나로만 분류하세요.
[AIVLE]
- KT AIVLE School, 에이블스쿨, 에이블러 관련 질문
- 출석, 결석, 지각, 조퇴, 외출, 출석 인정 관련 질문
- 강의, 다시보기, 교육 일정, 체크인/체크아웃 관련 질문
- KDT, 훈련장려금, 내일배움카드, 국민취업지원제도 관련 질문
- 교육장, 노트북, 코딩마스터스, 코딩 학습 플랫폼 관련 질문
- 채용연계, 포트폴리오, AX 챌린지 등 에이블스쿨 제도 관련 질문
- 사용자가 업로드한 백서/FAQ/문서에서 답해야 할 질문
[GENERAL]
- 위와 무관한 일반 지식, 코딩, 생활, 음식, 건강, 문서 작성, 번역, 상담 질문
- AIVLE 문서 근거가 필요 없는 일반 질문
출력 규칙:
- AIVLE 관련이면 AIVLE
- 일반 질문이면 GENERAL
- 다른 설명 없이 단어 하나만 출력
[사용자 질문]
{question}
""")
chain = prompt | llm | StrOutputParser()
route = chain.invoke({
"question": question
}).strip().upper()
if "AIVLE" in route:
state["route"] = "aivle"
else:
state["route"] = "general"
return state
def route_condition(state: ChatState) -> Literal["aivle", "general"]:
return state["route"]
def aivle_rag_node(state: ChatState) -> ChatState:
question = state["question"]
docs = st.session_state.retriever.invoke(question)
context = st.session_state.format_docs(docs)
answer = st.session_state.rag_chain.invoke({
"context": context,
"question": question
})
state["context"] = context
state["answer"] = answer
return state
def general_llm_node(state: ChatState) -> ChatState:
question = state["question"]
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.3
)
prompt = ChatPromptTemplate.from_template("""
당신은 친절한 AI 학습 도우미입니다.
사용자의 질문에 대해 한국어로 자연스럽고 이해하기 쉽게 답변하세요.
[사용자 질문]
{question}
""")
chain = prompt | llm | StrOutputParser()
answer = chain.invoke({
"question": question
})
state["context"] = ""
state["answer"] = answer
return state
def build_question_graph():
graph = StateGraph(ChatState)
graph.add_node("classify", classify_question_node)
graph.add_node("aivle_rag", aivle_rag_node)
graph.add_node("general_llm", general_llm_node)
graph.set_entry_point("classify")
graph.add_conditional_edges(
"classify",
route_condition,
{
"aivle": "aivle_rag",
"general": "general_llm"
}
)
graph.add_edge("aivle_rag", END)
graph.add_edge("general_llm", END)
return graph.compile()
def run_langgraph_answer(question):
app = build_question_graph()
result = app.invoke({
"question": question,
"route": "",
"context": "",
"answer": ""
})
return result
# =========================================================
# 8. 채팅 부가 기능: 2번 코드 기반
# =========================================================
def copy_button(text, key):
text_json = json.dumps(text, ensure_ascii=False)
components.html(
f"""
""",
height=40
)
def generate_tts(text, filename):
speech_file = TEMP_DIR / f"{filename}.mp3"
response = client.audio.speech.create(
model="gpt-4o-mini-tts",
voice="nova",
input=text
)
response.stream_to_file(str(speech_file))
return str(speech_file)
def transcribe_audio(audio_file):
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
)
return transcript.text
def render_text_for_html(text):
return html.escape(text).replace("\n", "
")
def answer_question(question):
messages = get_current_messages()
messages.append({
"role": "user",
"content": question
})
update_chat_title(question)
result = run_langgraph_answer(question)
answer = result["answer"]
route = result["route"]
if route == "aivle":
route_label = "📚 FAQ/백서 기반 답변"
else:
route_label = "💬 일반 AI 답변"
final_answer = f"{route_label}\n\n{answer}"
messages.append({
"role": "assistant",
"content": final_answer
})
if route == "aivle":
st.session_state.recommended_questions = recommend_questions(question)
else:
st.session_state.recommended_questions = []
# =========================================================
# 9. 사이드바: 1번 코드 기반
# =========================================================
uploaded_files = None
with st.sidebar:
if LOGO_PATH.exists():
st.image(str(LOGO_PATH), width=240)
else:
st.markdown('