192 lines
7.6 KiB
Python
192 lines
7.6 KiB
Python
"""backend.engines.dev_chatbot
|
||
|
||
개발챗봇 엔진 모듈.
|
||
|
||
구성
|
||
1. TOOL_ID / TOOL_INFO – 엔진 메타 데이터 (레지스트리에 자동 수집)
|
||
2. prepare_context() – Question ↔ VectorStore 유사도 검색 후 컨텍스트 생성
|
||
3. 벡터/파일 유틸 – index_pdf, delete_pdf, get_vector_store
|
||
4. FastAPI router – PDF 업로드·조회·삭제 등 API
|
||
|
||
다른 엔진을 만들 때 이 파일을 템플릿으로 삼으면 일관된 구조를 유지할 수 있다.
|
||
"""
|
||
|
||
# General Chatbot engine configuration
|
||
|
||
TOOL_ID = "dev_chatbot"
|
||
|
||
TOOL_INFO = {
|
||
"name": "개발챗봇",
|
||
"description": "일반적인 챗봇입니다.",
|
||
"system_prompt": (
|
||
"당신은 다목적 AI 어시스턴트입니다. 사용자의 질문에 대해 PDF 문서 분석 및 실시간 웹 검색을 통해 가장 신뢰할 수 있는 정보를 찾아 제공해야 합니다. "
|
||
"1) 질문과 관련된 내부·외부 PDF 자료가 있으면 우선적으로 내용을 요약·분석하여 근거와 함께 답변하세요. "
|
||
"2) 추론·계산·코드 예시·표·수식이 도움이 되면 적극 활용하세요. "
|
||
"3) 답변의 신뢰도를 높이기 위해 항상 출처(페이지 번호, 링크 등)를 명시하고, 확인되지 않은 정보는 가정임을 분명히 밝히세요. "
|
||
"4) 모호한 요청이나 추가 정보가 필요한 경우에는 명확한 follow-up 질문을 통해 요구 사항을 파악한 뒤 답변하세요. "
|
||
"5) 사실과 다른 정보를 임의로 생성하지 마세요. 필요한 정보가 없을 경우 솔직히 설명하고 가능한 대안을 제시합니다."
|
||
"6) 한국어로 대답해주세요."
|
||
)
|
||
}
|
||
|
||
from typing import Tuple, List
|
||
import os
|
||
from langchain.schema import Document
|
||
|
||
# get_vector_store 함수는 app.py 에 정의되어 주입받는다.
|
||
|
||
import urllib.parse
|
||
from fastapi import APIRouter, UploadFile, File
|
||
from fastapi.responses import StreamingResponse
|
||
from datetime import datetime
|
||
|
||
# -------------------------------------------------
|
||
# 벡터스토어 및 PDF 색인/삭제 기능
|
||
# -------------------------------------------------
|
||
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir))
|
||
UPLOAD_DIR = os.path.join(ROOT_DIR, "uploads")
|
||
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
||
|
||
# 벡터스토어 기능 비활성화(GPT 전용 모드)
|
||
def get_vector_store():
|
||
return None
|
||
|
||
# PDF 처리 기능 비활성화
|
||
try:
|
||
from langchain.text_splitter import RecursiveCharacterTextSplitter
|
||
except ModuleNotFoundError:
|
||
from langchain_text_splitters import RecursiveCharacterTextSplitter
|
||
|
||
try:
|
||
import pdfplumber # type: ignore
|
||
except ImportError:
|
||
pdfplumber = None
|
||
|
||
|
||
def delete_pdf(filename: str):
|
||
file_path = os.path.join(UPLOAD_DIR, filename)
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
vectordb = get_vector_store()
|
||
try:
|
||
vectordb.delete(where={"source": filename, "tool": TOOL_ID})
|
||
except Exception:
|
||
pass
|
||
|
||
# -------------------------------------------------
|
||
# FastAPI Router (PDF 업로드/조회/삭제)
|
||
# -------------------------------------------------
|
||
router = APIRouter()
|
||
|
||
@router.post("/upload_pdf")
|
||
async def upload_pdf(files: List[UploadFile] = File(...)):
|
||
saved_files = []
|
||
for up in files:
|
||
if not up.filename.lower().endswith(".pdf"):
|
||
continue
|
||
filename = f"{datetime.now().strftime('%Y%m%d%H%M%S%f')}_{up.filename}"
|
||
file_path = os.path.join(UPLOAD_DIR, filename)
|
||
with open(file_path, "wb") as f:
|
||
f.write(await up.read())
|
||
try:
|
||
if pdfplumber is None:
|
||
# pdfplumber 미설치 시 인덱싱 건너뜀
|
||
return
|
||
|
||
filename = os.path.basename(file_path)
|
||
docs = []
|
||
saved_files.append(filename)
|
||
except Exception as e:
|
||
return {"status": "error", "message": str(e)}
|
||
return {"status": "success", "files": saved_files}
|
||
|
||
@router.get("/files")
|
||
async def list_files():
|
||
pdfs = [f for f in os.listdir(UPLOAD_DIR) if f.lower().endswith('.pdf')]
|
||
return {"files": pdfs}
|
||
|
||
@router.get("/file_content")
|
||
async def get_file_content(filename: str):
|
||
file_path = os.path.join(UPLOAD_DIR, filename)
|
||
if not os.path.isfile(file_path):
|
||
return {"status": "error", "message": "파일을 찾을 수 없습니다."}
|
||
try:
|
||
# Assuming PDFPlumberLoader is available or needs to be imported
|
||
# from langchain.document_loaders import PDFPlumberLoader
|
||
# loader = PDFPlumberLoader(file_path)
|
||
# docs = loader.load()
|
||
# text = "\n\n".join(d.page_content for d in docs)
|
||
# return {"status": "success", "content": text}
|
||
# Placeholder for actual PDF content extraction if PDFPlumberLoader is not available
|
||
return {"status": "success", "content": "PDF content extraction is not implemented in this engine."}
|
||
except Exception as e:
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
@router.get("/pdf")
|
||
async def serve_pdf(filename: str):
|
||
# 1차: uploads 디렉터리
|
||
file_path = os.path.join(UPLOAD_DIR, filename)
|
||
# 2차: scripts/gxp 원본 디렉터리 (GxP 문서용)
|
||
if not os.path.isfile(file_path):
|
||
alt_dir = os.path.join(ROOT_DIR, "scripts", "gxp")
|
||
file_path = os.path.join(alt_dir, filename)
|
||
if not os.path.isfile(file_path):
|
||
return {"status":"error","message":"파일을 찾을 수 없습니다."}
|
||
file_stream = open(file_path, "rb")
|
||
safe_name = urllib.parse.quote(filename)
|
||
headers = {"Content-Disposition": f'inline; filename="{safe_name}"'}
|
||
return StreamingResponse(file_stream, media_type="application/pdf", headers=headers)
|
||
|
||
@router.delete("/file")
|
||
async def remove_file(filename: str):
|
||
try:
|
||
delete_pdf(filename)
|
||
return {"status": "success"}
|
||
except Exception as e:
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
def prepare_context(message: str, knowledge_mode: str, vectordb) -> Tuple[str, List[Document]]:
|
||
"""사용자 메시지에 대한 벡터 검색 컨텍스트와 문서 리스트를 반환합니다.
|
||
vectordb: langchain_chroma.Chroma 인스턴스 (지연 로드)
|
||
"""
|
||
context_text = ""
|
||
retrieved_docs: List[Document] = []
|
||
if not vectordb or vectordb._collection.count() == 0:
|
||
return context_text, retrieved_docs
|
||
|
||
# dev_chatbot 문서만 대상으로 검색
|
||
docs_with_scores = vectordb.similarity_search_with_score(
|
||
message,
|
||
k=4,
|
||
filter={"tool": TOOL_ID}
|
||
)
|
||
score_threshold = 0.9
|
||
filtered = [d for d, s in docs_with_scores if s is not None and s < score_threshold]
|
||
retrieved_docs = filtered if filtered else [d for d, _ in docs_with_scores[:2]]
|
||
|
||
if not retrieved_docs:
|
||
return context_text, []
|
||
|
||
# 페이지별 그룹화 후 상위 2페이지만 사용
|
||
page_groups = {}
|
||
for d in retrieved_docs:
|
||
src = d.metadata.get("source", "")
|
||
pg = d.metadata.get("page_number", "")
|
||
key = (src, pg)
|
||
page_groups.setdefault(key, []).append(d.page_content)
|
||
|
||
selected_pages = list(page_groups.keys())[:2]
|
||
snippets = []
|
||
for (src, pg) in selected_pages:
|
||
texts = page_groups[(src, pg)]
|
||
joined = "\n".join(texts)
|
||
snippet_txt = joined[:1500]
|
||
header = f"[출처:{src} p{pg}]"
|
||
snippets.append(header + "\n" + snippet_txt)
|
||
|
||
context_text = "\n\n[관련 문서 발췌]\n" + "\n---\n".join(snippets)
|
||
|
||
# retrieved_docs 축소
|
||
retrieved_docs = [d for d in retrieved_docs if (d.metadata.get("source",""), d.metadata.get("page_number","")) in selected_pages]
|
||
return context_text, retrieved_docs |