lexistudio / app.py
scipious's picture
Update app.py
2000df9 verified
import os
from flask import Flask, render_template, jsonify, request, Response
from flask_socketio import SocketIO, emit
import uuid
import threading
import sqlite3
import gc
import time
import re
import traceback
import requests # API 호출을 위해 필요
from typing import Optional, Tuple, Any, Dict, List
# --- Together AI SDK ---
from together import Together
# --- eventlet monkey patch (Gunicorn + SocketIO 필수!) ---
import eventlet
eventlet.monkey_patch()
# --- Flask & SocketIO 설정 ---
app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*", async_mode='eventlet')
import logging
# 로거 설정: 레벨을 INFO로 설정하고, 포맷을 지정합니다.
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# --- 외부 모듈 임포트 ---
import reg_embedding_system
import leximind_prompts
# --- 전역 변수 ---
connected_clients = 0
search_document_number = 30
# --- 경로 설정 ---
current_dir = os.path.dirname(os.path.abspath(__file__))
ResultFile_FolderAddress = os.path.join(current_dir, 'result.txt')
# --- RAG 데이터 경로 ---
# NOTE: Hugging Face Spaces에서 데이터가 /app/data에 있는지 확인해야 합니다.
region_paths = {
"국내": "/app/data/KMVSS_RAG",
"북미": "/app/data/FMVSS_RAG",
"유럽": "/app/data/EUR_RAG"
}
# --- 프롬프트 ---
lexi_prompts = leximind_prompts.PromptLibrary()
# 세션별 요청 추적을 위한 딕셔너리
active_sessions = {}
# --- RAG 객체 ---
region_rag_objects = {}
# --- Together AI 설정 (SDK 대신 API 호출에 사용) ---
TOGETHER_API_KEY = os.getenv("TOGETHER_API_KEY")
if not TOGETHER_API_KEY:
raise EnvironmentError("TOGETHER_API_KEY가 설정되지 않았습니다. Hugging Face Secrets에 추가하세요.")
# client = Together(api_key=TOGETHER_API_KEY) # <--- Together SDK 클라이언트 제거
try:
# TOGETHER_API_KEY를 사용해 클라이언트 초기화 (TOGETHER_API_KEY가 코드 내에 정의되어 있다고 가정)
client = Together(api_key=TOGETHER_API_KEY)
except NameError:
# TOGETHER_API_KEY가 정의되지 않은 경우 환경 변수 사용을 시도
client = Together()
rag_connection_status_info = ""
# --- RAG 로딩 ---
def load_rag_objects():
global region_rag_objects
global rag_connection_status_info
# 로딩 스레드 시작 로그를 추가하여 Gunicorn 로그에서 확인 가능하게 함
logger.info(">>> [RAG_LOADER] RAG 로딩 스레드 시작 <<<")
for region, path in region_paths.items():
if not os.path.exists(path):
msg = f"[{region}] 경로 없음: {path}"
socketio.emit('message', {'message': msg})
logger.info(msg)
continue
try:
socketio.emit('message', {'message': f"[{region}] RAG 로딩 중..."})
rag_connection_status_info = f"[{region}] RAG 로딩 중..."
# NOTE: reg_embedding_system 모듈이 현재 환경에 설치/존재하는지 확인해야 합니다.
ensemble_retriever, vectorstore, sqlite_conn = reg_embedding_system.load_embedding_from_faiss(path)
sqlite_conn.close()
db_path = os.path.join(path, "metadata_mapping.db")
new_conn = sqlite3.connect(db_path, check_same_thread=False)
region_rag_objects[region] = {
"ensemble_retriever": ensemble_retriever,
"vectorstore": vectorstore,
"sqlite_conn": new_conn
}
socketio.emit('message', {'message': f"[{region}] 로딩 완료"})
logger.info(f"[{region}] RAG 로딩 완료")
rag_connection_status_info = f"[{region}] RAG 로딩 완료"
except Exception as e:
error_msg = f"[{region}] 로딩 실패: {str(e)}"
logger.info(error_msg)
# [수정]: 상세한 에러 추적을 위해 traceback 추가
traceback.logger.info_exc()
socketio.emit('message', {'message': error_msg})
socketio.emit('message', {'message': "Ready to Search"})
logger.info("Ready to Search")
rag_connection_status_info = "Ready to Search"
# --- 웹 ---
@app.route('/')
def index():
return render_template('chat_v03.html')
# 전역 변수에 기본값 추가
Search_each_all_mode = True # 기본값을 클라이언트에서 제어 가능
@socketio.on('search_query')
def handle_search_query(data):
global Search_each_all_mode
global current_dir
# 세션 ID 생성
session_id = str(uuid.uuid4())
active_sessions[session_id] = True
# 클라이언트에 session_id 전달
emit('search_started', {'session_id': session_id})
try:
# 클라이언트에서 전송된 검색 모드 사용
Search_each_all_mode = data.get('searchEachMode', True)
query = data.get('query', '')
regions = data.get('regions', [])
selected_regulations = data.get('selectedRegulations', [])
emit('search_status', {'status': 'processing', 'message': '검색 요청을 처리하는 중입니다...'})
logger.info("선택된 지역: %s", regions)
logger.info("선택된 법규: %s", selected_regulations)
logger.info("검색 모드: %s", "각각 검색" if Search_each_all_mode else "통합 검색")
# 번역 진행 상황 알림
emit('search_status', {'status': 'translating', 'message': '질문에 대해 생각 중입니다...'})
if session_id not in active_sessions:
emit('search_cancelled', {'message': '검색이 취소되었습니다.'})
emit('search_status', {'status': 'processing', 'message': 'Ready to search'})
return
Translated_query = Gemma3_AI_Translate(query)
emit('search_status', {'status': 'translated', 'message': f'번역 완료: {Translated_query}'})
logger.info(f"Query: Original query : {query}")
logger.info(f"Query: Translated_query : {Translated_query}")
if selected_regulations:
cont_selected_num = 0
# 파일로 저장
output_path = os.path.join(current_dir, "merged_ai_messages.txt")
if os.path.exists(output_path):
os.remove(output_path)
logger.info(f"기존 파일 삭제 완료: {output_path}")
# 통합 검색 모드 - 타입별로 그룹화
grouped_regulations = group_regulations_by_type(selected_regulations)
emit('search_status', {'status': 'searching', 'message': f'선택된 {len(selected_regulations)}개 법규를 타입별로 통합하여 검색 중...'})
# 타입별로 필터 생성
combined_filters = create_combined_filters(grouped_regulations)
logger.info(f"통합 필터: {combined_filters}")
combined_cleaned_filter = {k: v for k, v in combined_filters.items() if v}
if Search_each_all_mode:
# 각각 검색 모드
emit('search_status', {'status': 'searching', 'message': f'선택된 {len(combined_cleaned_filter)}개 법규를 각각 검색 중...'})
total_search_num = sum(len(v) for v in combined_cleaned_filter.values())
i = 0
for RegType, RegNames in combined_cleaned_filter.items():
if RegNames: # 값이 비어있지 않은 경우만 처리
for RegName in RegNames:
i = i + 1
#RegType는 법규 유형, RegName은 법규 명칭
if session_id not in active_sessions:
emit('search_cancelled', {'message': '검색이 취소되었습니다.'})
emit('search_status', {'status': 'processing', 'message': 'Ready to search'})
return
emit('search_status', {
'status': 'searching_regulation',
'message': f'법규 {i}/{len(combined_cleaned_filter)}: {RegName} 검색 중...',
'progress': (i / len(combined_cleaned_filter)) * 100
})
# 법규 타입별 필터 생성
current_filters = create_filter_by_type(RegType, RegName)
logger.info(f"생성된 필터: {current_filters}")
Rag_Results = search_DB_from_multiple_regions(Translated_query, regions, region_rag_objects, current_filters, False) #마지막 False값은 유사한 값에 대한 검색을 하지 않겠다는 의미
if Rag_Results:
if session_id not in active_sessions:
emit('search_cancelled', {'message': '검색이 취소되었습니다.'})
emit('search_status', {'status': 'processing', 'message': 'Ready to search'})
return
emit('search_status', {
'status': 'ai_processing',
'message': f'AI가 {RegName}에 대한 답변을 생성 중...'
})
AImessage = RegAI(query, Rag_Results, ResultFile_FolderAddress)
logger.info(f"Answer: {AImessage}")
if session_id not in active_sessions:
emit('search_cancelled', {'message': '검색이 취소되었습니다.'})
return
# 각 법규별 결과를 실시간으로 전송 (타입 정보 포함)
emit('regulation_result', {
'regulation_title': f"[{RegName}]",
'regulation_index': i,
'total_regulations': total_search_num,
'result': AImessage
})
# 파일에 저장
if isinstance(AImessage, str) and AImessage.strip():
with open(output_path, "a", encoding="utf-8") as f:
cont_selected_num += 1
from datetime import datetime
stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
f.write(f"\n--- [{stamp}] message #{cont_selected_num} --- Regulation Type: {RegType} --- Regulation Name : {RegName} ---\n {AImessage}")
emit('search_complete', {'status': 'completed', 'message': '모든 법규 검색이 완료되었습니다.'})
else:
Rag_Results = search_DB_from_multiple_regions(Translated_query, regions, region_rag_objects, combined_filters, True)
if session_id in active_sessions:
emit('search_status', {'status': 'ai_processing', 'message': 'AI가 통합 답변을 생성 중...'})
AImessage = RegAI(query, Rag_Results, ResultFile_FolderAddress)
logger.info(f"Answer: {AImessage}")
if session_id in active_sessions:
emit('search_result', {'result': AImessage})
emit('search_complete', {'status': 'completed', 'message': '통합 검색이 완료되었습니다.'})
else:
emit('search_status', {'status': 'searching_all', 'message': '전체 법규에서 검색 중...'})
# 필터 없이 검색
Rag_Results = search_DB_from_multiple_regions(Translated_query, regions, region_rag_objects, None, True)
if session_id in active_sessions:
emit('search_status', {'status': 'ai_processing', 'message': 'AI가 답변을 생성 중...'})
AImessage = RegAI(query, Rag_Results, ResultFile_FolderAddress)
logger.info(f"Answer: {AImessage}")
if session_id in active_sessions:
emit('search_result', {'result': AImessage})
emit('search_complete', {'status': 'completed', 'message': '검색이 완료되었습니다.'})
except Exception as e:
print(f"검색 오류: {e}")
emit('search_error', {'error': str(e), 'message': '검색 중 오류가 발생했습니다.'})
finally:
# 세션 정리
if session_id in active_sessions:
del active_sessions[session_id]
@socketio.on('cancel_search')
def handle_cancel_search(data):
session_id = data.get('session_id')
if session_id and session_id in active_sessions:
del active_sessions[session_id]
emit('search_cancelled', {'message': '검색이 취소되었습니다.'})
# --- 법규 리스트 ---
@app.route('/get_reg_list', methods=['POST'])
def get_reg_list():
data = request.get_json()
selected_regions = data.get('regions', [])
# 지역이 선택되지 않았으면 전체 지역으로 설정
if not selected_regions:
selected_regions = ["국내", "북미", "유럽"]
all_reg_list_part = []
all_reg_list_section = []
all_reg_list_chapter = []
all_reg_list_jo = []
for region in selected_regions:
rag = region_rag_objects.get(region)
if not rag:
continue # 해당 지역 RAG가 없으면 건너뜀
try:
# 이미 로드된 SQLite 연결 재사용
sqlite_conn = rag["sqlite_conn"]
reg_list_part = get_unique_metadata_values(sqlite_conn, "regulation_part")
reg_list_section = get_unique_metadata_values(sqlite_conn, "regulation_section")
reg_list_chapter = get_unique_metadata_values(sqlite_conn, "chapter_section")
reg_list_jo = get_unique_metadata_values(sqlite_conn, "jo")
# 문자열이면 리스트로 변환
if isinstance(reg_list_part, str):
reg_list_part = [reg_list_part]
if isinstance(reg_list_section, str):
reg_list_section = [reg_list_section]
if isinstance(reg_list_chapter, str):
reg_list_chapter = [reg_list_chapter]
if isinstance(reg_list_jo, str):
reg_list_jo = [reg_list_jo]
all_reg_list_part.extend(reg_list_part)
all_reg_list_section.extend(reg_list_section)
all_reg_list_chapter.extend(reg_list_chapter)
all_reg_list_jo.extend(reg_list_jo)
except Exception as e:
print(f"[{region}] DB 연결 오류: {e}")
# 중복 제거
#unique_reg_list_part = list(set(all_reg_list_part))
unique_reg_list_part = sorted(set(all_reg_list_part), key=reg_embedding_system.natural_sort_key)
#unique_reg_list_section = list(set(all_reg_list_section))
unique_reg_list_section = sorted(set(all_reg_list_section), key=reg_embedding_system.natural_sort_key)
#unique_reg_list_chapter = list(set(all_reg_list_chapter))
unique_reg_list_chapter = sorted(set(all_reg_list_chapter), key=reg_embedding_system.natural_sort_key)
#unique_reg_list_jo = list(set(all_reg_list_jo))
unique_reg_list_jo = sorted(set(all_reg_list_jo), key=reg_embedding_system.natural_sort_key)
text_result_part = "\n".join(str(item) for item in unique_reg_list_part)
text_result_section = "\n".join(str(item) for item in unique_reg_list_section)
text_result_chapter = "\n".join(str(item) for item in unique_reg_list_chapter)
text_result_jo = "\n".join(str(item) for item in unique_reg_list_jo)
return jsonify(reg_list_part=text_result_part,
reg_list_section=text_result_section,
reg_list_chapter=text_result_chapter,
reg_list_jo=text_result_jo)
# --- SocketIO ---
@socketio.on('connect')
def handle_connect():
global connected_clients
connected_clients += 1
# 클라이언트 IP 가져오기
client_ip = request.remote_addr
# 프록시(Nginx, Cloudflare 등)를 거치는 경우 실제 IP는 헤더에 들어있을 수 있음
if request.headers.get('X-Forwarded-For'):
# X-Forwarded-For 는 "client, proxy1, proxy2" 형태로 여러 IP가 있을 수 있음
client_ip = request.headers.get('X-Forwarded-For').split(',')[0].strip()
elif request.headers.get('X-Real-IP'):
client_ip = request.headers.get('X-Real-IP')
# Cloudflare의 경우
elif request.headers.get('CF-Connecting-IP'):
client_ip = request.headers.get('CF-Connecting-IP')
logger.info(f"클라이언트 연결 | IP: {client_ip} | 현재 접속자: {connected_clients}명")
global rag_connection_status_info
socketio.emit('message', {'message': rag_connection_status_info})
@socketio.on('disconnect')
def handle_disconnect():
global connected_clients
connected_clients -= 1
logger.info(f"클라이언트 연결: {connected_clients}명")
#if connected_clients <= 0:
# cleanup_connections()
# logger.info("서버 종료")
# os._exit(0)
def cleanup_connections():
for region, rag in region_rag_objects.items():
try:
rag["sqlite_conn"].close()
logger.info(f"[{region}] DB 연결 종료")
except:
pass
# --- Together AI 분석 (SDK -> requests 직접 호출로 변경) ---
def Gemma3_AI_analysis(query_txt, content_txt):
content_txt = "\n".join(doc.page_content for doc in content_txt) if isinstance(content_txt, list) else str(content_txt)
query_txt = str(query_txt)
prompt = lexi_prompts.use_prompt(lexi_prompts.AI_system_prompt, query_txt=query_txt, content_txt=content_txt)
try:
response = client.chat.completions.create(
#model="meta-llama/Llama-4-Scout-17B-16E-Instruct", #비용 효율 측면 최고
model="moonshotai/Kimi-K2-Instruct-0905", #오픈소스 최고 성능
messages=[
{
"role": "user",
"content": prompt,
}
],
)
# 응답에서 결과 텍스트를 추출
AI_Result = response.choices[0].message.content
return AI_Result
except Exception as e:
# Together SDK의 오류는 requests.exceptions.RequestException이 아닌 다른 종류의 예외로 발생합니다.
# 따라서 일반적인 Exception으로 처리하는 것이 안전합니다.
logger.info(f"Together AI 분석 API 호출 실패: {e}")
traceback.print_exc() # traceback.logger.info_exc() 대신 일반 print_exc()를 사용하거나, logging 모듈 설정을 확인하세요.
return f"AI 분석 중 오류가 발생했습니다: {e}"
# --- Together AI 번역 (SDK -> requests 직접 호출로 변경) ---
def Gemma3_AI_Translate(query_txt):
query_txt = str(query_txt)
prompt = lexi_prompts.use_prompt(lexi_prompts.query_translator, query_txt=query_txt)
try:
response = client.chat.completions.create(
#model="meta-llama/Llama-4-Scout-17B-16E-Instruct", #비용 효율 측면 최고
model="moonshotai/Kimi-K2-Instruct-0905", #오픈소스 최고 성능
messages=[
{
"role": "user",
"content": prompt,
}
],
)
# 응답에서 결과 텍스트를 추출
AI_Result = response.choices[0].message.content
return AI_Result
except Exception as e:
# API 호출 실패 시 처리 (SDK 사용 시 일반 Exception으로 처리)
logger.info(f"Together AI 번역 API 호출 실패: {e}")
# traceback.logger.info_exc() 대신 traceback.print_exc() 사용 (권장)
# 만약 기존 로깅 시스템에서 해당 함수를 정의해 사용하고 있다면 그대로 두셔도 됩니다.
# 여기서는 표준 traceback 모듈을 사용합니다.
traceback.print_exc()
# 번역 실패 시 query_txt 변수를 반환 (기존 코드 로직 반영)
return query_txt
# --- 검색 ---
# 검색 함수 수정
def search_DB_from_multiple_regions(query, selected_regions, region_rag_objects, custom_filters=None, failsafe_mode=True):
if not selected_regions:
selected_regions = list(region_rag_objects.keys())
print(f"Translated Query : {query}")
# None일 때 안전하게 처리
if custom_filters is None:
custom_filters = {} # 빈 딕셔너리로 대체 (필터 없음 = 전체 검색)
# 필터가 설정되어 있는지 확인
has_filters = any(custom_filters.get(key, []) for key in custom_filters.keys())
print(f"사용된 검색 필터: {custom_filters}")
print(f"필터 사용 여부: {has_filters}")
combined_results = []
for region in selected_regions:
rag = region_rag_objects.get(region)
if not rag:
continue
ensemble_retriever = rag["ensemble_retriever"]
vectorstore = rag["vectorstore"]
sqlite_conn = rag["sqlite_conn"]
if ensemble_retriever:
if has_filters:
results = reg_embedding_system.search_with_metadata_filter(
ensemble_retriever=ensemble_retriever,
vectorstore=vectorstore,
query=query,
k=search_document_number,
metadata_filter=custom_filters,
sqlite_conn=sqlite_conn,
failsafe_search=failsafe_mode
)
else:
results = reg_embedding_system.smart_search_vectorstore(
retriever=ensemble_retriever,
query=query,
k=search_document_number,
vectorstore=vectorstore,
sqlite_conn=sqlite_conn,
enable_detailed_search=True
)
print(f"[{region}] 검색 완료: {len(results)}건")
combined_results.extend(results)
return combined_results
# --- 최종 AI ---
def RegAI(query, Rag_Results, ResultFile_FolderAddress):
gc.collect()
AI_Result = "검색 결과가 없습니다." if not Rag_Results else Gemma3_AI_analysis(query, Rag_Results)
#with open(ResultFile_FolderAddress, 'w', encoding='utf-8') as f:
# print("검색된 문서:", file=f)
# logger.info("검색된 문서:")
# for i, doc in enumerate(Rag_Results):
# print(f"문서 {i+1}: {doc.page_content[:200]}... (메타: {doc.metadata})", file=f)
# logger.info(f"문서 {i+1}: {doc.page_content[:200]}... (메타: {doc.metadata})")
# print("\n답변:", file=f)
# logger.info("\n답변:")
# print(AI_Result, file=f)
# logger.info(AI_Result)
return AI_Result
# 법규 타입별 필터 생성 함수
def create_filter_by_type(regulation_type, regulation_title):
"""법규 타입에 따라 적절한 필터 딕셔너리 생성"""
filter_dict = {
"regulation_part": [],
"regulation_section": [],
"chapter_section": [],
"jo": []
}
# 타입별 매핑
# 전체 키를 지원하는 매핑 (입력으로 'regulation_section' 등을 받는 경우)
type_mapping = {
"regulation_part": "regulation_part",
"regulation_section": "regulation_section",
"chapter_section": "chapter_section",
"jo": "jo",
# 혹시 짧은 형태로 들어오는 경우도 함께 지원
"part": "regulation_part",
"section": "regulation_section",
"chapter": "chapter_section",
}
filter_key = type_mapping.get(regulation_type, "regulation_part")
filter_dict[filter_key].append(regulation_title)
return filter_dict
# 법규들을 타입별로 그룹화하는 함수
def group_regulations_by_type(selected_regulations):
"""선택된 법규들을 타입별로 그룹화"""
grouped = {
"part": [],
"section": [],
"chapter": [],
"jo": []
}
for regulation in selected_regulations:
regulation_type = regulation.get('type', 'part')
regulation_title = regulation.get('title', '')
if regulation_title and regulation_type in grouped:
grouped[regulation_type].append(regulation_title)
return grouped
# 통합 필터 생성 함수
def create_combined_filters(grouped_regulations):
"""그룹화된 법규들로부터 통합 필터 생성"""
filters = {
"regulation_part": grouped_regulations["part"],
"regulation_section": grouped_regulations["section"],
"chapter_section": grouped_regulations["chapter"],
"jo": grouped_regulations["jo"]
}
return filters
def natural_sort_key(text):
"""숫자가 포함된 문자열을 자연스럽게 정렬 (예: item1, item2, item10)"""
return [int(c) if c.isdigit() else c.lower() for c in re.split('([0-9]+)', str(text))]
def get_unique_metadata_values(
sqlite_conn: sqlite3.Connection,
key_name: str,
partial_match: Optional[str] = None
) -> List[str]:
"""
SQLite 'documents' 테이블에서 특정 컬럼(key_name)의 중복되지 않은
모든 고유 값 리스트를 반환합니다.
Args:
sqlite_conn: SQLite 데이터베이스 연결 객체.
key_name: 고유한 값을 가져올 컬럼 이름 (예: 'regulation_name', 'part_name').
partial_match: (선택 사항) 해당 문자열을 포함하는 값만 검색할 때 사용.
Returns:
중복이 제거된 고유한 값들의 리스트.
"""
text_result = ""
if not sqlite_conn:
print("[경고] SQLite 연결이 없어 고유 값 검색을 수행할 수 없습니다.")
return text_result
cursor = sqlite_conn.cursor()
# SQL 쿼리 구성
# 1. 컬럼 이름에 백틱(`)을 사용하여 안전성 확보
# 2. DISTINCT를 사용하여 중복 제거
sql_query = f"SELECT DISTINCT `{key_name}` FROM documents"
params = []
# 부분 문자열 검색 (LIKE) 조건 추가
if partial_match:
sql_query += f" WHERE `{key_name}` LIKE ?"
params.append(f"%{partial_match}%")
try:
cursor.execute(sql_query, params)
# 쿼리 결과에서 첫 번째 항목 (값)만 추출
unique_values = [row[0] for row in cursor.fetchall() if row[0] is not None]
unique_values.sort(key=natural_sort_key)
text_result = "\n".join(str(value) for value in unique_values)
return text_result
except sqlite3.OperationalError as e:
# 컬럼 이름이 DB에 없을 때 발생하는 에러 처리
print(f"[에러] SQLite 쿼리 실행 실패 (컬럼 '{key_name}' 이름 오류 가능): {e}")
return text_result
except Exception as e:
print(f"[에러] 고유 값 검색 중 알 수 없는 오류 발생: {e}")
return text_result
# --- 실행 ---
if __name__ == '__main__':
# 로컬 개발용
threading.Thread(target=load_rag_objects, daemon=True).start()
time.sleep(2)
socketio.emit('message', {'message': '데이터 로딩 시작...'})
socketio.run(app, host='0.0.0.0', port=7860, debug=False)
else:
# Gunicorn용: 워커 시작 후 로딩
import atexit
loading_thread = threading.Thread(target=load_rag_objects, daemon=True)
loading_thread.start()
atexit.register(cleanup_connections)