This commit is contained in:
dsyoon
2025-12-27 14:06:26 +09:00
parent 23f5388c56
commit 46460b77f8
33 changed files with 4600 additions and 1 deletions

View File

@@ -0,0 +1,28 @@
"""backend.engines package
엔진 레지스트리.
서브패키지를 순회해 각 엔진 모듈이 노출한
`TOOL_ID` 와 `TOOL_INFO` 를 자동 수집하여 `TOOLS` dict 로 제공한다.
외부(backend.app 등)에서는 from engines import TOOLS 로 공통 접근.
"""
from importlib import import_module
import pkgutil
# Dictionary mapping tool_id to its info
TOOLS = {}
# Discover and import all subpackages to collect TOOL_INFO
package_name = __name__
for _, module_name, is_pkg in pkgutil.iter_modules(__path__):
if not is_pkg:
# Skip if it's not a package (we expect dirs)
continue
try:
module = import_module(f"{package_name}.{module_name}")
# Each engine package must expose TOOL_ID and TOOL_INFO
if hasattr(module, "TOOL_ID") and hasattr(module, "TOOL_INFO"):
TOOLS[module.TOOL_ID] = module.TOOL_INFO
except ModuleNotFoundError:
# If submodule import fails, skip silently
continue

View File

@@ -0,0 +1,11 @@
TOOL_ID = "chatgpt"
TOOL_INFO = {
"name": "ChatGPT",
"description": "OpenAI ChatGPT 모델과 대화할 수 있는 도구입니다.",
"system_prompt": (
"You are ChatGPT, a large language model trained by OpenAI. "
"Provide thorough, well-structured answers in Korean using 전문지식과 신뢰할 수 있는 공개 자료를 바탕으로 설명하세요. 필요 시 표/리스트/소제목을 활용해 가독성을 높이십시오. "
"만약 기업의 위치나 연락처 등 요청이 오면 본사·연구소·공장 등 주요 거점을 빠짐없이 요약해 주세요."
)
}

View File

@@ -0,0 +1,53 @@
from fastapi import APIRouter, Form
from typing import Optional
import os
import openai
from dotenv import load_dotenv
load_dotenv()
OPEN_API_KEY = os.getenv("OPENAI_API_KEY", "")
openai_client = openai.OpenAI(api_key=OPEN_API_KEY)
router = APIRouter(prefix="/chatgpt", tags=["ChatGPT"])
# 모델 매핑 테이블 (프론트 선택값 ➜ OpenAI 모델명)
MODEL_MAP = {
"auto": "gpt-5",
"gpt-5": "gpt-5",
"gpt-5-mini": "gpt-5-mini",
"gpt-5-nano": "gpt-5-nano",
}
SYSTEM_PROMPT = (
"You are ChatGPT, a large language model trained by OpenAI. "
"Provide thorough, well-structured answers in Korean with rich formatting. "
"Always include 적절한 소제목과 불릿, 줄바꿈을 활용하고, 각 소제목 앞에 관련 이모지(예: 🏢 본사, 🧪 연구소, 🏭 공장 등)를 붙여 가독성을 높이십시오. "
"필요하면 표 또는 번호 리스트를 사용하세요. "
"회사의 위치·연락처·교통편을 묻는 질문에는 반드시 본사, 연구소, 공장 등 주요 거점 정보를 빠짐없이 상세히 제공합니다. "
"답변 길이 제한 없이 충분히 상세히 작성하세요."
)
@router.post("/chat")
async def chat_gpt_endpoint(
message: str = Form(...),
model: str = Form("auto"),
session_id: Optional[str] = Form(None)
):
# 모델 매핑
model_name = MODEL_MAP.get(model, "gpt-5")
# 직접 OpenAI ChatCompletion 호출
completion = openai_client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": message},
]
)
return {
"status": "success",
"response": completion.choices[0].message.content.strip(),
"session_id": session_id or "",
"tool_name": "ChatGPT",
}

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,192 @@
"""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

View File

@@ -0,0 +1,346 @@
"""backend.engines.doc_translation
문서번역 엔진 모듈.
MS Word 문서를 업로드하여 한글을 영어로 번역하는 기능을 제공합니다.
"""
import os
import shutil
import re
from datetime import datetime
from typing import List
from fastapi import APIRouter, UploadFile, File, HTTPException, Form
from fastapi.responses import FileResponse
from docx import Document as DocxDocument
import openai
from dotenv import load_dotenv
# .env 파일 로드
load_dotenv()
TOOL_ID = "doc_translation"
TOOL_INFO = {
"name": "문서번역",
"description": "MS Word 문서를 업로드하면 한글을 영어로 번역합니다.",
"system_prompt": (
"당신은 전문 번역가입니다. 한국어 문서를 영어로 정확하고 자연스럽게 번역해주세요. "
"번역할 때는 원문의 의미와 뉘앙스를 최대한 보존하면서도 영어로 자연스럽게 표현해주세요. "
"대상 문장 다음 줄에 번역 문장을 제공해주세요."
)
}
# 업로드 디렉토리 설정 - 프로젝트 루트/uploads (개발챗봇과 동일)
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)
# OpenAI API 키 설정
openai.api_key = os.getenv("OPENAI_API_KEY")
def translate_text_with_gpt(text: str) -> str:
"""ChatGPT를 사용하여 텍스트를 한글에서 영어로 번역 (채팅용)"""
try:
client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.chat.completions.create(
model="gpt-5",
messages=[
{
"role": "system",
"content": TOOL_INFO["system_prompt"]
},
{
"role": "user",
"content": f"다음 한국어 텍스트를 영어로 번역해주세요:\n\n{text}"
}
]
)
return response.choices[0].message.content
except Exception as e:
print(f"번역 오류: {e}")
return f"번역 오류가 발생했습니다: {str(e)}"
# (Ollama 내부 모델 번역 기능 제거)
def translate_paragraph(paragraph: str) -> str:
"""단일 단락을 번역"""
try:
client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.chat.completions.create(
model="gpt-5",
messages=[
{
"role": "system",
"content": "당신은 전문 번역가입니다. 한국어 문단을 영어로 정확하고 자연스럽게 번역해주세요. 번역문만 제공하세요."
},
{
"role": "user",
"content": f"다음 한국어 문단을 영어로 번역해주세요:\n\n{paragraph}"
}
]
)
return response.choices[0].message.content.strip()
except Exception as e:
print(f"단락 번역 오류: {e}")
return f"[번역 오류: {str(e)}]"
def extract_paragraphs_from_docx(file_path: str) -> List[str]:
"""Word 문서에서 단락별 텍스트 추출"""
try:
doc = DocxDocument(file_path)
paragraphs = []
for paragraph in doc.paragraphs:
if paragraph.text.strip():
paragraphs.append(paragraph.text.strip())
return paragraphs
except Exception as e:
raise HTTPException(status_code=400, detail=f"Word 파일 읽기 오류: {str(e)}")
def create_bilingual_docx(original_paragraphs: List[str], translated_paragraphs: List[str], output_path: str):
"""원본과 번역이 번갈아 나타나는 이중언어 Word 문서 생성"""
try:
doc = DocxDocument()
# 각 단락별로 원본과 번역을 번갈아 추가
for i, (original, translated) in enumerate(zip(original_paragraphs, translated_paragraphs)):
# 원본 단락 추가 (굵은 글씨로)
original_para = doc.add_paragraph()
original_run = original_para.add_run(original)
original_run.bold = True
# 번역 단락 추가 (일반 글씨로)
translated_para = doc.add_paragraph(translated)
# 단락 사이에 여백 추가 (마지막 단락이 아닌 경우)
if i < len(original_paragraphs) - 1:
doc.add_paragraph("") # 빈 줄 추가
doc.save(output_path)
except Exception as e:
raise HTTPException(status_code=500, detail=f"이중언어 파일 생성 오류: {str(e)}")
def prepare_context(question: str, vector_store=None, model: str = "external") -> str:
"""채팅용 컨텍스트 준비 (실시간 번역)"""
if not question.strip():
return ""
# 모델 선택에 따른 번역 처리
translated = translate_text_with_gpt(question)
return translated
# FastAPI Router
router = APIRouter()
@router.post("/upload_doc")
async def upload_document(files: List[UploadFile] = File(...)):
"""MS Word 문서 업로드 및 번역 처리"""
if not files:
raise HTTPException(status_code=400, detail="파일이 업로드되지 않았습니다.")
processed_files = []
for file in files:
# MS Word 파일 확인
if not (file.filename.lower().endswith('.doc') or file.filename.lower().endswith('.docx')):
raise HTTPException(status_code=400, detail="MS Word 파일(.doc, .docx)만 업로드 가능합니다.")
# 파일명 생성: [doc_translation]_admin_YYYYMMDDHHMMSSMMM_원본파일명
timestamp = datetime.now().strftime('%Y%m%d%H%M%S%f')[:-3] # 밀리초 3자리까지
original_name = os.path.splitext(file.filename)[0]
extension = os.path.splitext(file.filename)[1]
uploaded_filename = f"[{TOOL_ID}]_admin_{timestamp}_{original_name}{extension}"
uploaded_path = os.path.join(UPLOAD_DIR, uploaded_filename)
# 파일 저장
try:
with open(uploaded_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# 단락별 텍스트 추출
paragraphs = extract_paragraphs_from_docx(uploaded_path)
if not paragraphs:
raise HTTPException(status_code=400, detail="문서에 번역할 텍스트가 없습니다.")
# 각 단락을 개별적으로 번역
translated_paragraphs = []
for paragraph in paragraphs:
translated = translate_paragraph(paragraph)
translated_paragraphs.append(translated)
# 이중언어 문서 저장 (원본 + 번역)
result_filename = f"[{TOOL_ID}]_admin_{timestamp}_{original_name}_결과{extension}"
result_path = os.path.join(UPLOAD_DIR, result_filename)
create_bilingual_docx(paragraphs, translated_paragraphs, result_path)
processed_files.append({
"original_filename": uploaded_filename,
"result_filename": result_filename,
"status": "success"
})
except Exception as e:
# 업로드된 파일이 있다면 삭제
if os.path.exists(uploaded_path):
os.remove(uploaded_path)
raise HTTPException(status_code=500, detail=f"파일 처리 오류: {str(e)}")
return {"status": "success", "files": processed_files}
@router.get("/files")
async def list_files():
"""업로드된 파일 목록 조회"""
try:
files = []
for filename in os.listdir(UPLOAD_DIR):
if filename.startswith(f"[{TOOL_ID}]") and not filename.endswith("_결과.doc") and not filename.endswith("_결과.docx"):
file_path = os.path.join(UPLOAD_DIR, filename)
stat = os.stat(file_path)
# 결과 파일 존재 여부 확인
# 파일명에서 확장자 분리
base_name, ext = os.path.splitext(filename)
result_filename = f"{base_name}_결과{ext}"
result_exists = os.path.exists(os.path.join(UPLOAD_DIR, result_filename))
files.append({
"filename": filename,
"upload_time": datetime.fromtimestamp(stat.st_ctime).isoformat(),
"size": stat.st_size,
"has_result": result_exists,
"result_filename": result_filename if result_exists else None
})
# 업로드 시간 순으로 정렬 (최신순)
files.sort(key=lambda x: x["upload_time"], reverse=True)
return files
except Exception as e:
raise HTTPException(status_code=500, detail=f"파일 목록 조회 오류: {str(e)}")
@router.get("/download/{filename}")
async def download_file(filename: str):
"""번역된 파일 다운로드"""
file_path = os.path.join(UPLOAD_DIR, filename)
if not os.path.exists(file_path):
raise HTTPException(status_code=404, detail="파일을 찾을 수 없습니다.")
return FileResponse(
path=file_path,
filename=filename,
media_type='application/vnd.openxmlformats-officedocument.wordprocessingml.document'
)
@router.post("/translate_chat")
async def translate_chat(message: str = Form(...)):
"""실시간 채팅 번역"""
if not message.strip():
raise HTTPException(status_code=400, detail="번역할 메시지가 없습니다.")
try:
translated = translate_text_with_gpt(message)
return {"translated_text": translated}
except Exception as e:
raise HTTPException(status_code=500, detail=f"번역 오류: {str(e)}")
@router.delete("/files/{filename}")
async def delete_file(filename: str):
"""파일 삭제 (서버 저장 파일명으로)"""
file_path = os.path.join(UPLOAD_DIR, filename)
if os.path.exists(file_path):
os.remove(file_path)
# 해당하는 결과 파일도 삭제
if not filename.endswith("_결과.doc") and not filename.endswith("_결과.docx"):
base_name, ext = os.path.splitext(filename)
result_filename = f"{base_name}_결과{ext}"
result_path = os.path.join(UPLOAD_DIR, result_filename)
if os.path.exists(result_path):
os.remove(result_path)
return {"status": "success", "message": "파일이 삭제되었습니다."}
@router.delete("/delete_by_original_name/{original_filename}")
async def delete_file_by_original_name(original_filename: str):
"""파일 삭제 (원본 파일명으로) - 외부 접근용"""
try:
# 업로드된 파일 중에서 원본 파일명과 일치하는 파일 찾기
deleted_files = []
for filename in os.listdir(UPLOAD_DIR):
if filename.startswith(f"[{TOOL_ID}]"):
# 실제 파일명 추출
pattern = r'^\[doc_translation\]_admin_\d{14,17}_(.+)$'
match = re.match(pattern, filename)
if match and match.group(1) == original_filename:
# 원본 파일 삭제
file_path = os.path.join(UPLOAD_DIR, filename)
if os.path.exists(file_path):
os.remove(file_path)
deleted_files.append(filename)
# 결과 파일도 삭제
if not filename.endswith("_결과.doc") and not filename.endswith("_결과.docx"):
base_name, ext = os.path.splitext(filename)
result_filename = f"{base_name}_결과{ext}"
result_path = os.path.join(UPLOAD_DIR, result_filename)
if os.path.exists(result_path):
os.remove(result_path)
deleted_files.append(result_filename)
if deleted_files:
return {"status": "success", "message": f"파일이 삭제되었습니다.", "deleted_files": deleted_files}
else:
return {"status": "error", "message": "해당하는 파일을 찾을 수 없습니다."}
except Exception as e:
raise HTTPException(status_code=500, detail=f"파일 삭제 오류: {str(e)}")
@router.get("/list_files")
async def list_all_files():
"""모든 파일 목록 조회 (원본 파일명으로) - 외부 접근용"""
try:
files = []
for filename in os.listdir(UPLOAD_DIR):
if filename.startswith(f"[{TOOL_ID}]") and not filename.endswith("_결과.doc") and not filename.endswith("_결과.docx"):
# 실제 파일명 추출
pattern = r'^\[doc_translation\]_admin_\d{14,17}_(.+)$'
match = re.match(pattern, filename)
if match:
original_name = match.group(1)
file_path = os.path.join(UPLOAD_DIR, filename)
stat = os.stat(file_path)
# 결과 파일 존재 여부 확인
base_name, ext = os.path.splitext(filename)
result_filename = f"{base_name}_결과{ext}"
result_exists = os.path.exists(os.path.join(UPLOAD_DIR, result_filename))
files.append({
"original_filename": original_name,
"server_filename": filename,
"upload_time": datetime.fromtimestamp(stat.st_ctime).isoformat(),
"size": stat.st_size,
"has_result": result_exists,
"result_filename": result_filename if result_exists else None
})
# 업로드 시간 순으로 정렬 (최신순)
files.sort(key=lambda x: x["upload_time"], reverse=True)
return {"status": "success", "files": files}
except Exception as e:
raise HTTPException(status_code=500, detail=f"파일 목록 조회 오류: {str(e)}")