from fastapi.responses import JSONResponse # <--- [추가] from fastapi import FastAPI from pydantic import BaseModel from mistral_inference.transformer import Transformer from mistral_inference.generate import generate from mistral_common.tokens.tokenizers.mistral import MistralTokenizer from pathlib import Path import uvicorn import pymysql import re from fastapi.encoders import jsonable_encoder # --------------------------------------------------------- # [설정 1] 데이터베이스 연결 정보 # --------------------------------------------------------- DB_HOST = "qst-s.iptime.org" DB_PORT = 33063 DB_USER = "ai_read_only" DB_PASSWORD = "qsentech!1233" DB_NAME = "paradise" # --------------------------------------------------------- # [설정 2] 모델 로드 및 클래스 찾기 # --------------------------------------------------------- try: from mistral_common.protocol.instruct.messages import ChatCompletionRequest, UserMessage, SystemMessage except ImportError: try: from mistral_common.protocol.instruct.request import ChatCompletionRequest from mistral_common.protocol.instruct.messages import UserMessage, SystemMessage except ImportError: import mistral_common.protocol.instruct.messages as msg_module UserMessage = msg_module.UserMessage SystemMessage = msg_module.SystemMessage pass mistral_models_path = Path.home().joinpath('mistral_models', '7B-Instruct-v0.3') tokenizer_path = mistral_models_path / "tokenizer.model.v3" print("=== 모델 로딩 중 ===") tokenizer = MistralTokenizer.from_file(str(tokenizer_path)) model = Transformer.from_folder(mistral_models_path) print("=== 준비 완료 ===") app = FastAPI() class ChatRequest(BaseModel): prompt: str max_tokens: int = 1024 # --------------------------------------------------------- # [Helper] 모델 호출 함수 # --------------------------------------------------------- def ask_mistral(messages, max_tokens=1024, temperature=0.1): chat_request = ChatCompletionRequest(messages=messages) tokens = tokenizer.encode_chat_completion(chat_request).tokens out_tokens, _ = generate( [tokens], model, max_tokens=max_tokens, temperature=temperature, eos_id=tokenizer.instruct_tokenizer.tokenizer.eos_id ) return tokenizer.instruct_tokenizer.tokenizer.decode(out_tokens[0]) # --------------------------------------------------------- # [프롬프트 정의] # --------------------------------------------------------- # ▼▼▼ [수정된 부분] AI에게 정확한 카테고리 족보를 줍니다 ▼▼▼ def get_sql_prompt(): return """ You are a SQL expert. Convert the user's question into a MariaDB SQL query. [Schema] Table: tbl_casino_info Table COMMENT: Casino 정보 - casinoId (varchar(10)): Casino id - casinoName (carchar(50)): Casino name Table: tbl_emap_info Table COMMENT: DID eMAP 대시보드 이미지 정보 - eMapId (varchar(10)): eMap id - eMapName (varchar(50)): eMap name - eMapFileName (varchar(100)): eMap file name - eMapVer (varchar(20)): eMap Version - useYn (char(1)): use check (default Y) Table: tbl_game_info Table COMMENT: 카지노 게임 정보 - gameId (varchar(10)): Casino game slot id - gameName (varchar(50)): Casino game slot name Table: tbl_group_info Table COMMENT: Jackpot group information - groupId (varchar(10)): Jackpot group id - groupName (varchar(50)): Jackpot group name - groupType (char(1)): Jackpot group type Table: tbl_icon_info Table COMMENT: DID eMAP 대시보드 아이콘 정보 - iconId (varchar(10)): Digital information display dashboard icon id - iconName (varchar(50)): Digital information display dashboard icon name - iconFileName (varchar(100)): Digital information display dashboard icon file name - didType (char(1)): Digital information display type - didStatus (char(10)): Digital information display status Table: tbl_jackpot_hitlog Table COMMENT: 잭팟 당첨 이력(당첨 정보 로그/ 당첨정보 내역) - jackpotId (varchar(10)): Hit jackpot id - jackpotName (varchar(50)): Hit jackpot name - groupId (varchar(10)): Jackpot group id - casinoId (varchar(10)): Casino id - machineId (varchar(10)): Hit slot machineid or tableid - spotNo (varchar(10)): Table game's spot number - gameId (varchar(10)): Table or game slot id - typeId (char(1)): Jackpot typeId (tbl_jptype_info) - sizeId (char(1)): Jackpot sizeId (tbl_jpsize_info) - mwId (varchar(10)): Middleware agent id - hitPrize (double): Hit jackpot money - upTime (datetime): Hit jackpot time Table: tbl_jackpot_info Table COMMENT: 실시간 잭팟 금액 적립 정보 - jackpotId (varchar(10)): Accumulated jackpot id - jackpotName (varchar(50)): Accumulated jackpot name - groupId (varchar(10)): Jackpot group id - casinoId (varchar(10)): Casino id - mwId (varchar(10)): Middleware agent id - prizeCurrency (double): Accumulated jackpot money - minPrize (double): Minimum jackpot money - maxPrize (double): Maximum jackpot money - typeId (varchar(10)): Jackpot typeId - sizeId (varchar(10)): Jackpot sizeId - delaySec (int(11)): The time stored in the tbl_jackpot_hit table - upTime (datetime): Lastest accumulated jackpot money update iime Table: tbl_jpc_info Table COMMENT: Jackpot controller information - jpcId (varchar(10)): Jackpot controller id - jpcName (varchar(50)): Jackpot controller name - product (varchar(100)): Jackpot controller product company Table: tbl_jpsize_info Table COMMENT: Jackpot size information - sizeId (varchar(10)): Jackpot size id - sizeName (varchar(50)): Jackpot size name Table: tbl_jptype_info Table COMMENT: Jackpot type information - typeId (varchar(10)): Jackpot type id - typeName (varchar(50)): Jackpot type name Table: tbl_machine_info Table COMMENT: Casino slot machine information - machineId (varchar(10)): Casino slot machine id - machineName (varchar(50)): Casino slot machine name - gameId (varchar(10)): Casino game id - jpcId (varchar(10)): Jacppot controller id - eMapId (varchar(10)): eMap id - eMapAxis (varchar(50)): eMap Axis - useYn (char(1)): Machine use yn Table: tbl_middleware_info Table COMMENT: Middleware agent information - mwId (varchar(10)): Middleware agent id - mwName (varchar(50)): Middleware agent name - mwConfigInfo (longtext): Middleware config information - mwVer (varchar(20)): Middleware agent file version - useYn (char(1)): use check (default Y) Table: tbl_resource_info Table COMMENT: 웹 서버 리소스 정보 - resName (varchar(50)): Resource name - resValue (varchar(50)): Resource value Table: tbl_table_info Table COMMENT: 카지노 테이블 정보 - tableId (varchar(10)): Casino table id - tableName (varchar(50)): Casino table name - tableStatus (char(1)): Casino table status - gameId (varchar(10)): Casino game id - eMapAxis (varchar(20)): eMapAxis - useYn (char(1)): use check (default Y) Table: tbl_user_info Table COMMENT: 사용자 정보 - userId (varchar(20)): User login id - userPw (varchar(100)): User login password - userLevel (char(1)): User authority level - userName (varchar(50)): User name - casinoId (varchar(10)): User casino id - employeeId (varchar(30)): User employee id - officeCode_01 (varchar(10)): User office affiiation 01 - officeCode_02 (varchar(10)): User office affiiation 02 - officeCode_03 (varchar(10)): User office affiiation 03 - officeCode_04 (varchar(10)): User office affiiation 04 - tel_01 (varchar(20)): User telephone number 01 - tel_02 (varchar(20)): User telephone number 02 - connStatus (char(1)): User connection status - useYn (char(1)): use check (default Y) [Table Relationships & JOIN Keys] 1. tbl_jackpot_hitlog (Main Transaction) - JOIN tbl_casino_info ON tbl_jackpot_hitlog.casinoId = tbl_casino_info.casinoId - JOIN tbl_game_info ON tbl_jackpot_hitlog.gameId = tbl_game_info.gameId - JOIN tbl_group_info ON tbl_jackpot_hitlog.groupId = tbl_group_info.groupId - JOIN tbl_jptype_info ON tbl_jackpot_hitlog.typeId = tbl_jptype_info.typeId - JOIN tbl_jpsize_info ON tbl_jackpot_hitlog.sizeId = tbl_jpsize_info.sizeId - JOIN tbl_middleware_info ON tbl_jackpot_hitlog.mwId = tbl_middleware_info.mwId 2. tbl_jackpot_info - JOIN tbl_group_info ON tbl_jackpot_info.groupId = tbl_group_info.groupId - JOIN tbl_casino_info ON tbl_jackpot_info.casinoId = tbl_casino_info.casinoId - JOIN tbl_middleware_info ON tbl_jackpot_info.mwId = tbl_middleware_info.mwId - JOIN tbl_game_info ON tbl_jackpot_info.gameId = tbl_game_info.gameId - JOIN tbl_jptype_info ON tbl_jackpot_info.typeId = tbl_jptype_info.typeId - JOIN tbl_jpsize_info ON tbl_jackpot_info.sizeId = tbl_jpsize_info.sizeId 3. tbl_machine_info - JOIN tbl_game_info ON tbl_machine_info.gameId = tbl_game_info.gameId - JOIN tbl_jpc_info ON tbl_machine_info.jpcId = tbl_jpc_info.jpcId - JOIN tbl_emap_info ON tbl_machine_info.eMapId = tbl_emap_info.eMapId 4. tbl_table_info - JOIN tbl_game_info ON tbl_table_info.gameId = tbl_game_info.gameId - JOIN tbl_emap_info ON tbl_table_info.eMapId = tbl_emap_info.eMapId 5. tbl_user_info - JOIN tbl_casino_info ON ui.casinoId = ci.casinoId [Rules] 1. Output ONLY the SQL query inside a code block (```sql ... ```). 2. Use SELECT statement only. 3. [IMPORTANT] You MUST use Korean aliases for all columns in the SELECT clause. - Format: column_name AS 'Korean_Name' 4. [CRITICAL] When searching for user information, ALWAYS check both 'userId' and 'userName' columns using 'OR'. - Syntax: WHERE (userId = 'INPUT_VALUE' OR userName = 'INPUT_VALUE') 5. [JACKPOT JOIN] When querying 'tbl_jackpot_hitlog' (tbl_jackpot_hitlog), you must perform a LEFT JOIN: - on tbl_jackpot_hitlog.machineId = tbl_machine_info.machineId - on tbl_jackpot_hitlog.machineId = tbl_table_info.tableId 6. [NO WILDCARDS] Do not use '*'. Explicitly list columns from all joined tables with Korean aliases. [Examples] User: "홍길동 전화번호 알려줘" SQL: ```sql SELECT userName AS '이름', tel_01 AS '전화번호1', tel_02 AS '전화번호2' FROM tbl_user_info WHERE userId = '홍길동' OR userName = '홍길동'; ``` """ # ▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲▲ def get_general_prompt(): return "You are a helpful AI assistant. Answer kindly in Korean." # --------------------------------------------------------- # [기능] SQL 실행 함수 # --------------------------------------------------------- def execute_sql_query(sql: str): try: conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) with conn: with conn.cursor() as cursor: cursor.execute(sql) result = cursor.fetchall() print(f"DEBUG: DB에서 가져온 레코드 수: {len(result)}") # 이 로그를 확인하세요 return list(result) except Exception as e: return f"SQL_ERROR: {str(e)}" # --------------------------------------------------------- # [필수 추가] 에러 발생 시 자연어 변환 프롬프트 # --------------------------------------------------------- def get_error_response_prompt(user_input, error_type, detail_msg=""): return f""" The user asked: "{user_input}" Situation: - We tried to search the database but found nothing or an error occurred. - Status: {error_type} - Detail: {detail_msg} Task: - Write a SHORT, kind, and helpful response in Korean explaining the situation. - If Status is 'EMPTY', say "searched for it but couldn't find any matching data." - If Status is 'ERROR', say "an internal error occurred while searching." - Do NOT mention technical details (SQL, column names) to the user. """ # --------------------------------------------------------- # [API 엔드포인트] - 무조건 SQL 모드로 동작하게 수정됨 # --------------------------------------------------------- @app.post("/chat") async def chat_endpoint(request: ChatRequest): try: user_input = request.prompt print(f">> 질문: {user_input}") print(">> 모드: 강제 SQL 실행") # ------------------------------------------------------- # 1. SQL 생성 # ------------------------------------------------------- sql_messages = [ SystemMessage(content=get_sql_prompt()), UserMessage(content=f"Question: {user_input}\nSQL Query:") ] # SQL 생성은 길게 허용 (2048) generated_text = ask_mistral(sql_messages, max_tokens=2048, temperature=0.1) # 정규식으로 SQL 추출 match = re.search(r"```(sql)?(.*?)```", generated_text, re.DOTALL | re.IGNORECASE) if match: clean_sql = match.group(2).strip() else: clean_sql = generated_text.strip() if "select" in clean_sql.lower(): clean_sql = clean_sql[clean_sql.lower().find("select"):] if ";" in clean_sql: clean_sql = clean_sql.split(";")[0] + ";" print(f">> 추출된 SQL: {clean_sql}") # ------------------------------------------------------- # 2. SQL 유효성 검사 실패 시 # ------------------------------------------------------- if not clean_sql.lower().startswith("select"): print(">> 에러: SQL이 아님 -> AI 에러 설명 생성 중...") err_msg = ask_mistral( [UserMessage(content=get_error_response_prompt(user_input, "ERROR", "Invalid SQL Generated"))], max_tokens=512, # <--- [중요] 답변 길이 제한 temperature=0.7 ) print(f">> 에러 답변 완료: {err_msg}") return {"response": err_msg} # ------------------------------------------------------- # 3. DB 실행 # ------------------------------------------------------- db_result = execute_sql_query(clean_sql) # (A) DB 에러 발생 시 if isinstance(db_result, str) and "SQL_ERROR" in db_result: error_explanation = ask_mistral( [UserMessage(content=get_error_response_prompt(user_input, "ERROR", db_result))], max_tokens=512, temperature=0.7 ) return JSONResponse(content={"status": "error", "message": error_explanation}, headers={"Connection": "close"}) # (B) 검색 결과가 0건일 때 if isinstance(db_result, list) and not db_result: empty_explanation = ask_mistral( [UserMessage(content=get_error_response_prompt(user_input, "EMPTY", "No records found"))], max_tokens=512, temperature=0.7 ) return JSONResponse(content={"status": "empty", "message": empty_explanation, "data": []}, headers={"Connection": "close"}) # (C) 정상 결과 반환 (JSON 리스트 변환 로직) # pymysql.cursors.DictCursor를 사용하므로 db_result는 이미 리스트[딕셔너리] 형태입니다. # jsonable_encoder를 사용하면 datetime 객체 등이 자동으로 ISO 포맷 문자열로 변환됩니다. json_compatible_data = jsonable_encoder(db_result) return JSONResponse( content={ "status": "success", "message": f"{len(db_result)}건의 데이터를 찾았습니다.", "data": json_compatible_data, # 델파이에서 파싱할 데이터 배열 "generated_sql": clean_sql # 디버깅용 }, headers={"Connection": "close"} ) except Exception as e: import traceback traceback.print_exc() return JSONResponse(content={"status": "critical_error", "message": "서버 시스템 오류가 발생했습니다."}, status_code=500) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)