from flask import Blueprint, request, jsonify import os import sys import json import re import unicodedata # 이모티콘 처리를 위해 추가 from dotenv import load_dotenv import chromadb from chromadb.utils import embedding_functions from openai import OpenAI from flask_cors import cross_origin import logging from src.models.db import db from src.models.gif_feedback import GifFeedback, GifUsage from datetime import datetime import uuid from flask import Blueprint from src.routes.smalltalk_manager import SmalltalkPatternManager from src.routes.inappropriate_filter import is_inappropriate_question, get_inappropriate_response from src.routes.gif_manager import analyze_gif_content, generate_contextual_response, GifResponseManager from src.routes.emoji_manager import is_emoji_only_message, get_emoji_response, handle_emoji_message chatbot_bp = Blueprint('chatbot', __name__) chatbot_bp2 = Blueprint('chatbot2', __name__) # 전역 매니저 인스턴스 smalltalk_manager = SmalltalkPatternManager() # .env 로드 load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env')) OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") FINE_TUNE_MODEL = os.getenv("FINE_TUNE_MODEL", "gpt-3.5-turbo") CHROMA_DB_PATH = "./output_final0718/chroma_db" RAG_COLLECTION_NAME = "mlink_knowledge" PREFIX = "공지 내용을 요약해 주세요. " KEYWORDS = ["엠링크", "엘림", "상품등록", "주문연동", "자동등록", "MLINK", "Yellim"] logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger(__name__) # 전역 변수로 선언 (모듈 로딩 시 한 번만 초기화) original_contents = {} collection = None client = None openai_ef = None def initialize_chatbot(): """챗봇 초기화 함수 - 모듈 로딩 시 한 번만 실행""" global original_contents, collection, client, openai_ef # 환경 변수 검증 if not OPENAI_API_KEY: logger.error("OPENAI_API_KEY가 설정되지 않았습니다.") raise ValueError("OPENAI_API_KEY 환경 변수가 필요합니다.") # OpenAI 클라이언트 초기화 client = OpenAI(api_key=OPENAI_API_KEY) openai_ef = embedding_functions.OpenAIEmbeddingFunction(api_key=OPENAI_API_KEY, model_name="text-embedding-ada-002") # ChromaDB 연결 및 컬렉션 설정 try: chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH) try: collection = chroma_client.get_collection(RAG_COLLECTION_NAME) logger.info(f"ChromaDB 컬렉션 '{RAG_COLLECTION_NAME}' 로드 성공") except Exception as e: logger.info(f"새 컬렉션 '{RAG_COLLECTION_NAME}' 생성 중...") collection = chroma_client.create_collection( name=RAG_COLLECTION_NAME, embedding_function=openai_ef ) logger.info("ChromaDB 컬렉션 생성 완료") except Exception as e: logger.error(f"ChromaDB 연결 실패: {e}") raise # 원문 데이터 로드 metadata_file_path = './output_final0718/org_metadata_data.jsonl' if not os.path.exists(metadata_file_path): logger.warning(f"메타데이터 파일을 찾을 수 없습니다: {metadata_file_path}") return try: loaded_count = 0 with open(metadata_file_path, encoding='utf-8') as f: for line_num, line in enumerate(f, start=1): try: entry = json.loads(line) except json.JSONDecodeError: logger.warning(f"JSON 파싱 오류 (라인 {line_num}): {line[:50]}...") continue article_id = entry.get('article_id') content = entry.get('original_content') if article_id is None or content is None: continue try: article_id = int(article_id) except (ValueError, TypeError): pass original_contents[article_id] = content loaded_count += 1 # 진행 상황 로깅 (1000개마다) if loaded_count % 1000 == 0: logger.info(f"원문 데이터 로딩 진행: {loaded_count}개 완료") logger.info(f"원문 데이터 로드 완료: 총 {loaded_count}개 항목") except Exception as e: logger.error(f"원문 데이터 로드 실패: {e}") raise # 모듈 로딩 시 초기화 실행 try: initialize_chatbot() logger.info("챗봇 모듈 초기화 완료") except Exception as e: logger.error(f"챗봇 모듈 초기화 실패: {e}") logger.warning("챗봇 기능이 비활성화됩니다. OPENAI_API_KEY를 설정해주세요.") # 초기화 실패 시에도 모듈은 로드되도록 함 pass # =============================== # 이모티콘 메시지 처리 함수들 # =============================== def needs_prefix(text): return not any(kw.lower() in text.lower() for kw in KEYWORDS) def apply_prefix(text): return PREFIX + text if needs_prefix(text) else text def rag_search(query, top_k=3): try: emb = client.embeddings.create(model="text-embedding-ada-002", input=query).data[0].embedding results = collection.query( query_embeddings=[emb], n_results=top_k, where={"chunk_type": "faq"}, include=["documents", "metadatas"]) docs = results.get('documents') metas = results.get('metadatas') if not docs or not isinstance(docs, list) or not docs[0]: logger.warning("RAG 검색 결과 문서가 없습니다.") return [], [] if not metas or not isinstance(metas, list) or not metas[0]: logger.warning("RAG 검색 결과 메타데이터가 없습니다.") return [], [] return docs[0], metas[0] except Exception as e: logger.error(f"RAG 검색 실패: {e}") return [], [] # =============================== # rag_search (paragraph + page-level 보강) # =============================== def rag_search2(user_query, top_k=3): """ FAQ(Excel 기반) + PDF 매뉴얼(paragraph-level + page-level) 동시 검색 반환: faq_docs, faq_metas, para_docs, para_metas, page_results """ try: emb = client.embeddings.create( model="text-embedding-ada-002", input=user_query ).data[0].embedding # 1. FAQ 검색 faq_results = collection.query( query_embeddings=[emb], n_results=top_k, where={"chunk_type": "faq"}, include=["documents", "metadatas"] ) faq_docs, faq_metas = [], [] if faq_results["documents"]: faq_docs = faq_results["documents"][0] faq_metas = faq_results["metadatas"][0] # 2. 매뉴얼 paragraph 검색 para_results = collection.query( query_embeddings=[emb], n_results=top_k, where={"chunk_type": "paragraph"}, include=["documents", "metadatas"] ) para_docs, para_metas = [], [] if para_results["documents"]: para_docs = para_results["documents"][0] para_metas = para_results["metadatas"][0] # 3. paragraph가 속한 page-level 가져오기 page_results = None if para_metas: page_ids = list(set([m["page_id"] for m in para_metas if "page_id" in m])) if page_ids: page_results = collection.get(ids=page_ids) return faq_docs, faq_metas, para_docs, para_metas, page_results except Exception as e: logger.error(f"RAG 검색 실패: {e}") return [], [], [], [], None def fine_tune_answer(user_query): """Fine-tuned 모델 답변만 생성 (원문은 hybrid_answer에서 제어)""" try: messages = [ {"role": "system", "content": "너는 MLink 공지/FAQ 전문가다. 친근하고 이해하기 쉽게 답변해라."}, {"role": "user", "content": user_query} ] resp = client.chat.completions.create( model=FINE_TUNE_MODEL, messages=messages, temperature=0.0 ) answer = resp.choices[0].message.content.strip() # fallback 필터링 (혹시라도 모델이 무시할 경우 대비) unrelated_keywords = ["대통령", "대한민국", "노벨", "정치", "역사", "인물", "세계사"] if any(k in user_query for k in unrelated_keywords): return "제가 학습하지 않은 질문입니다." return answer except Exception as e: logger.error(f"Fine-tune 답변 생성 실패: {e}") return "⚠️ 답변을 생성하지 못했습니다. 다른 질문을 시도해 주세요." def hybrid_answer(user_query, top_k=3, default_open=3, user_name=''): """FAQ + 원문(문단 단위, 토글) + 매뉴얼을 합쳐 하나의 응답 생성 출처(page/paragraph)는 '참고자료 보기' 버튼을 눌러야 표시됨 """ try: # (0) 부적절한 질문 체크 (가장 먼저) - 사용자 이름도 함께 체크 is_inappropriate, category, severity = is_inappropriate_question(user_query, user_name) if is_inappropriate: return { "final_answer": get_inappropriate_response(category, severity), "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" } # (1) 단순 인사/감사 멘트는 바로 처리 if is_smalltalk_or_thanks(user_query): return { "final_answer": "네, 감사합니다. 도움이 필요하시면 언제든 말씀해주세요 😊", "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" } # (2) FAQ + 매뉴얼 동시 검색 faq_docs, faq_metas, para_docs, para_metas, page_results = rag_search2(user_query, top_k=top_k) # 검색 결과 전혀 없음 → 학습 외 질문으로 처리 if not faq_docs and not para_docs: return { "final_answer": "제가 학습하지 않은 질문입니다.", "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" } # (3) Fine-tune 응답 (FAQ 스타일) faq_answer = fine_tune_answer(user_query) # Fine-tune에서도 범위 밖으로 판정된 경우 if faq_answer == "제가 학습하지 않은 질문입니다.": return { "final_answer": faq_answer, "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" } # 3. FAQ 원문 문단 단위 분리 original_paragraphs = [] # if original_text: # original_paragraphs = [p.strip() for p in re.split(r"\n{2,}", original_text) if p.strip()] for i, meta in enumerate(faq_metas): if not meta: print(f"⚠️ metas[{i}] 항목이 None입니다.") original_paragraphs.append("메타데이터 없음") continue article_id = meta.get('article_id') if not article_id: print(f"❓ metas[{i}]에 'article_id' 없음: {meta}") original_paragraphs.append("article_id 누락") continue original = original_contents.get(article_id) if original: original_paragraphs.append(original) else: print(f"❓ article_id {article_id}에 해당하는 원문 없음") original_paragraphs.append("원문을 찾을 수 없습니다.") if len(original_paragraphs) <= default_open: open_paragraphs = original_paragraphs collapsed_paragraphs = [] else: open_paragraphs = original_paragraphs[:default_open] collapsed_paragraphs = original_paragraphs[default_open:] # 4. 매뉴얼 context (출처 포함) → 따로 저장, 기본 응답에서는 제외 rag_context = "" for i, (doc, meta) in enumerate(zip(para_docs, para_metas)): rag_context += f"[출처 {i+1} - {meta['para_id']} (page={meta['page_id']})]\n{doc}\n\n" if page_results: for pid, ptext in zip(page_results["ids"], page_results["documents"]): rag_context += f"[페이지 전체: {pid}]\n{ptext[:400]}...\n\n" # 5. 최종 응답 (FAQ + 매뉴얼 통합) combined_context = f"📌 FAQ 응답:\n{faq_answer}\n\n" if open_paragraphs: combined_context += " FAQ 원문 일부:\n" + "\n".join(open_paragraphs) + "\n\n" if para_docs: combined_context += " 매뉴얼 내용:\n" + "\n".join(para_docs[:2]) + "\n\n" # 대표 문단 2개만 표시 # GPT 최종 응답 (출처는 숨기고 답변만 생성) messages = [ {"role": "system", "content": "너는 MLink 고객지원 전문가다. FAQ와 매뉴얼 내용을 참고해 사용자의 질문에 자연스럽고 정확하게 답변해라. 출처는 답변에 직접 포함하지 마라."}, {"role": "user", "content": f"[참고자료]\n{combined_context}"}, {"role": "user", "content": f"위 내용을 종합해서 '{user_query}'에 답변해줘."} ] resp = client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.3 ) final_answer = resp.choices[0].message.content.strip() return { "final_answer": final_answer, "original_paragraphs_open": open_paragraphs, "original_paragraphs_collapsed": collapsed_paragraphs, "collapsed_label": "… 여기를 눌러 나머지 원문 보기 ▾", "expanded_label": "원문 접기 ▴", "references": rag_context, # 참고자료 (버튼 눌렀을 때만 보여줌) "reference_label": "📖 참고자료 보기 ▾", "reference_hide_label": " 참고자료 접기 ▴" } except Exception as e: logger.error(f"하이브리드 답변 생성 실패: {e}") return { "final_answer": "⚠️ AI 서비스에 일시적인 오류가 발생했습니다. 잠시 후 다시 시도해 주세요.", "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" } def generate_answer(context, user_query): try: messages = [ {"role": "user", "content": f"[참고지식]\n{context}"}, {"role": "user", "content": apply_prefix(user_query)} ] response = client.chat.completions.create( model=FINE_TUNE_MODEL, messages=messages, temperature=0.3 ) if not response.choices[0].message.content.strip(): return "⚠️ 답변을 생성하지 못했습니다. 다른 질문을 시도해 주세요." return response.choices[0].message.content except Exception as e: logger.error(f"OpenAI API 호출 실패: {e}") return "⚠️ AI 서비스에 일시적인 오류가 발생했습니다. 잠시 후 다시 시도해 주세요." @chatbot_bp2.route('/api/chatbot/ask2', methods=['POST']) def ask_chatbot2(): try: # 챗봇 초기화 상태 확인 if client is None or collection is None: return jsonify({ "error": "챗봇 서비스가 초기화되지 않았습니다.", "answer": "죄송합니다. 챗봇 서비스가 현재 사용할 수 없습니다. 관리자에게 문의해주세요.", "original_content": [] }), 503 data = request.get_json() if not data: return jsonify({"error": "요청 데이터가 없습니다."}), 400 user_input = data.get("question", "") if not user_input.strip(): return jsonify({"error": "질문 내용이 없습니다."}), 400 logger.info(f"챗봇 질문 수신: {user_input[:50]}...") docs, metas = rag_search(user_input) answer = generate_answer("\n".join(docs), user_input) original_texts = [] for i, meta in enumerate(metas): if not meta: logger.warning(f"metas[{i}] 항목이 None입니다.") original_texts.append("메타데이터 없음") continue article_id = meta.get('article_id') if not article_id: logger.warning(f"metas[{i}]에 'article_id' 없음: {meta}") original_texts.append("article_id 누락") continue original = original_contents.get(article_id) if original: original_texts.append(original) else: logger.warning(f"article_id {article_id}에 해당하는 원문 없음") original_texts.append("원문을 찾을 수 없습니다.") logger.info(f"챗봇 응답 완료: {len(original_texts)}개 원문") return jsonify({ "answer": answer, "original_content": original_texts }) except Exception as e: logger.error(f"챗봇 처리 중 오류 발생: {e}") return jsonify({ "error": "서버 내부 오류가 발생했습니다.", "answer": "죄송합니다. 일시적인 오류가 발생했습니다. 잠시 후 다시 시도해 주세요.", "original_content": [] }), 500 @chatbot_bp.route('/api/chatbot/ask', methods=['POST']) def ask_chatbot(): try: # 챗봇 초기화 상태 확인 if client is None or collection is None: return jsonify({ "error": "챗봇 서비스가 초기화되지 않았습니다.", "answer": "죄송합니다. 챗봇 서비스가 현재 사용할 수 없습니다. 관리자에게 문의해주세요.", "original_content": [] }), 503 data = request.get_json() if not data: return jsonify({"error": "요청 데이터가 없습니다."}), 400 user_input = data.get("question", "") file_content = data.get("file_content", "") # 첨부파일 내용 추가 user_name = data.get("user_name", "") # 사용자 이름 추가 if not user_input.strip(): return jsonify({"error": "질문 내용이 없습니다."}), 400 logger.info(f"챗봇 질문 수신: {user_input[:50]}... (사용자: {user_name})") # 이모티콘만으로 구성된 메시지인지 확인 if is_emoji_only_message(user_input): logger.info(f"이모티콘 메시지 감지: {user_input}") answer = handle_emoji_message(user_input) logger.info("이모티콘 응답 완료") return jsonify(answer) # 하이브리드 답변 생성 (사용자 이름 포함) combined_input = user_input.strip() if file_content: combined_input += "\n[첨부파일 내용]\n" + file_content[:1000] # 너무 긴 경우 자르기 answer = hybrid_answer(combined_input, top_k=3, user_name=user_name) logger.info(f"챗봇 응답 완료") return jsonify(answer) except Exception as e: logger.error(f"챗봇 처리 중 오류 발생: {e}") return jsonify({ "error": "서버 내부 오류가 발생했습니다.", "final_answer": "죄송합니다. 일시적인 오류가 발생했습니다. 잠시 후 다시 시도해 주세요.", "original_paragraphs_open": [], "original_paragraphs_collapsed": [], "collapsed_label": "", "expanded_label": "", "references": "", "reference_label": "", "reference_hide_label": "" }), 500 @chatbot_bp.route('/api/chatbot/gif-response', methods=['POST']) @cross_origin() def generate_gif_response(): """GIF에 대한 자동 응답 생성""" try: data = request.get_json() # 필수 필드 검증 if 'gif_id' not in data or 'gif_title' not in data: return jsonify({"error": "필수 필드가 누락되었습니다."}), 400 gif_id = data['gif_id'] gif_title = data['gif_title'] gif_url = data.get('gif_url', '') # GIF 제목을 기반으로 감정/상황 분석 gif_analysis = analyze_gif_content(gif_title) # 분석 결과에 따른 응답 생성 response = generate_contextual_response(gif_analysis, gif_title) # GIF 사용 추적 try: usage = GifUsage( gif_id=gif_id, gif_title=gif_title, gif_url=gif_url, user_id=data.get('user_id'), session_id=data.get('session_id'), usage_type='sent', context='chat' ) db.session.add(usage) db.session.commit() except Exception as e: logging.warning(f"GIF 사용 추적 실패: {e}") return jsonify({ "response": response, "analysis": gif_analysis, "gif_id": gif_id }), 200 except Exception as e: logging.error(f"GIF 응답 생성 실패: {str(e)}") return jsonify({"error": "GIF 응답 생성에 실패했습니다."}), 500 # =============================== # 스몰토크 패턴 관리 API 엔드포인트 # =============================== @chatbot_bp.route('/api/chatbot/smalltalk/patterns', methods=['GET']) def get_smalltalk_patterns(): """스몰토크 패턴 목록 조회""" try: category = request.args.get('category') if category: patterns = smalltalk_manager.get_patterns_by_category(category) else: patterns = [p for p in smalltalk_manager.patterns if p.enabled] return jsonify({ "patterns": [ { "pattern": p.pattern, "category": p.category, "description": p.description, "enabled": p.enabled } for p in patterns ], "categories": list(smalltalk_manager.get_all_categories()), "statistics": smalltalk_manager.get_statistics() }) except Exception as e: logger.error(f"스몰토크 패턴 조회 실패: {e}") return jsonify({"error": "패턴 조회에 실패했습니다."}), 500 @chatbot_bp.route('/api/chatbot/smalltalk/patterns', methods=['POST']) def add_smalltalk_pattern(): """새 스몰토크 패턴 추가""" try: data = request.get_json() if not data or 'pattern' not in data: return jsonify({"error": "필수 필드가 누락되었습니다."}), 400 pattern = data['pattern'] category = data.get('category', 'custom') description = data.get('description', '사용자 정의 패턴') enabled = data.get('enabled', True) # 패턴 유효성 검사 try: re.compile(pattern) except re.error as e: return jsonify({"error": f"잘못된 정규식 패턴: {e}"}), 400 smalltalk_manager.add_pattern(pattern, category, description, enabled) return jsonify({ "message": "패턴이 성공적으로 추가되었습니다.", "pattern": { "pattern": pattern, "category": category, "description": description, "enabled": enabled } }) except Exception as e: logger.error(f"스몰토크 패턴 추가 실패: {e}") return jsonify({"error": "패턴 추가에 실패했습니다."}), 500 @chatbot_bp.route('/api/chatbot/smalltalk/patterns/', methods=['DELETE']) def remove_smalltalk_pattern(pattern): """스몰토크 패턴 제거""" try: if smalltalk_manager.remove_pattern(pattern): return jsonify({"message": "패턴이 성공적으로 제거되었습니다."}) else: return jsonify({"error": "패턴을 찾을 수 없습니다."}), 404 except Exception as e: logger.error(f"스몰토크 패턴 제거 실패: {e}") return jsonify({"error": "패턴 제거에 실패했습니다."}), 500 @chatbot_bp.route('/api/chatbot/smalltalk/patterns//toggle', methods=['PUT']) def toggle_smalltalk_pattern(pattern): """스몰토크 패턴 활성화/비활성화""" try: data = request.get_json() enabled = data.get('enabled', True) if smalltalk_manager.toggle_pattern(pattern, enabled): return jsonify({ "message": f"패턴이 {'활성화' if enabled else '비활성화'}되었습니다.", "enabled": enabled }) else: return jsonify({"error": "패턴을 찾을 수 없습니다."}), 404 except Exception as e: logger.error(f"스몰토크 패턴 토글 실패: {e}") return jsonify({"error": "패턴 토글에 실패했습니다."}), 500 @chatbot_bp.route('/api/chatbot/smalltalk/test', methods=['POST']) def test_smalltalk_pattern(): """스몰토크 패턴 테스트""" try: data = request.get_json() if not data or 'query' not in data: return jsonify({"error": "테스트할 쿼리가 필요합니다."}), 400 query = data['query'] is_smalltalk, category = smalltalk_manager.is_smalltalk(query) return jsonify({ "query": query, "is_smalltalk": is_smalltalk, "category": category, "matched_patterns": [ { "pattern": p.pattern, "category": p.category, "description": p.description } for p in smalltalk_manager.patterns if p.enabled and re.match(p.pattern, query.strip().lower()) ] }) except Exception as e: logger.error(f"스몰토크 패턴 테스트 실패: {e}") return jsonify({"error": "패턴 테스트에 실패했습니다."}), 500