knn/nmf_FAST_API -> knn/nmf/svd_test

[qkrtnqls1216]

NMF를 사용하여 유사도 확인한 모델을 웹 서버에서 과거 id값에 따른 현재 데이터 추천이 잘되는지 확인하기 위해 rest api 사용

FastAPI 인스턴스 생성

MySQL 데이터베이스에서 데이터를 로드하는 함수 정의

뮤지컬 추천을 위한 musical_id에 기반한 엔드포인트 정의

@app_nmf.get("/recommend/{musical_id}")
def recommend(musical_id: int):
    try:
        # 선택한 작품의 인덱스 찾기
        selected_work_index_past = past_data[past_data['musical_id'] == musical_id].index[0]

        # 과거 데이터 파싱 및 스케일링 
        past_data['synopsis_numpy_scale'] = past_data['synopsis_numpy_scale'].apply(lambda x: np.array(json.loads(x.decode('utf-8'))))
        scaler_past = StandardScaler()
        past_data_scaled = scaler_past.fit_transform(np.vstack(past_data['synopsis_numpy_scale']))
        past_data_scaled = past_data_scaled - np.min(past_data_scaled) + 1e-10

        # 현재 작품 선택
        present_data['synopsis_numpy_scale'] = present_data['synopsis_numpy_scale'].apply(lambda x: np.array(json.loads(x.decode('utf-8'))))
        scaler_present = StandardScaler()
        present_data_scaled = scaler_present.fit_transform(np.vstack(present_data['synopsis_numpy_scale']))
        present_data_scaled = present_data_scaled - np.min(present_data_scaled) + 1e-10

        # NMF 모델 초기화
        # n_components: 추출할 특성의 수로, 작품을 나타내는 잠재적인 특징의 개수를 설정
        #  init: 행렬을 초기화하는 방법을 설정 -> 'random'은 무작위 초기화
        # random_state: 난수 발생을 제어하여 모델이 항상 일관된 결과를 생성하도록
        nmf = NMF(n_components=10, init='random', random_state=42)

        # 특성 행렬 생성
        # 행렬을 수직으로 쌓아서 새로운 행렬을 생성
        # 'past_data_scaled'는 각 작품의 특성을 행으로 가지고 있는 2D 배열
        # np.vstack 함수를 사용하여 이를 수직으로 쌓아서 특성 행렬 'V'를 생성
        V = np.vstack(past_data_scaled)

        # NMF 모델 훈련
        W = nmf.fit_transform(V) # W는 특성 행렬 -> W는 데이터를 특성으로 표현
        # H = nmf.components_ # H는 NMF 모델에서 생성된 행렬 중 하나로 주로 특성을 나타냄

        # 현재 상영중인 데이터에 대한 특성 행렬 생성
        V_present = np.vstack(present_data_scaled)

        # NMF 모델을 사용하여 현재 상영중인 데이터의 특성 행렬 분해
        W_present = nmf.transform(V_present)

        # 선택한 작품과 다른 작품 간의 코사인 유사도 계산
        selected_work = W[selected_work_index_past].reshape(1, -1)
        similarities = cosine_similarity(W_present, selected_work)

        # 유사도가 높은 순서대로 정렬하여 유사한 작품 인덱스를 찾기
        # argsort 함수를 사용하여 정렬된 인덱스를 얻고, [::-1]을 사용하여 역순으로 정렬
        # 이후 flatten 함수를 사용하여 1차원 배열로 변환
        similar_work_indices = similarities.argsort(axis=0)[::-1].flatten()
        # 상위 N개의 유사한 작품을 선택하되, 실제 유사한 작품의 수를 벗어나지 않도록
        top_n = min(5, len(similar_work_indices))

        # 상위 N개의 유사한 작품에 대한 정보를 추출하고 결과 리스트에 추가
        # 결과 리스트에는 각 작품의 제목, musical_id, 그리고 코사인 유사도(similarity)가 포함
        result = []  # 추천 사항을 저장할 빈 리스트 생성
        for i in range(top_n):
            index = similar_work_indices[i]
            similarity = float(similarities[index])  # NumPy float로 변환
            # 작품 정보 추출
            title = present_data.loc[index, 'title']
            musical_id = int(present_data.loc[index, 'musical_id'])  # NumPy int64를 Python int로 변환
             # 결과 리스트에 작품 정보를 추가
            result.append({"title": title, "musical_id": musical_id, "similarity": similarity})

        return result  # 추천 목록을 반환
    except Exception as e:
        # 예외가 발생한 경우, 에러 응답을 반환
        return JSONResponse(content={"error": f"An error occurred: {str(e)}"}, status_code=500)

결과확인

실행방법 : uvicorn app_nmf:app_nmf --reload --host 0.0.0.0 --port 8080 주소검색 : http://localhost:8080/recommend/{musical_id}

---

knn 사용하여 유사도 확인한 모델을 웹 서버에서 과거 id값에 따른 현재 데이터 추천이 잘되는지 확인하기 위해 rest api 사용

FastAPI 인스턴스 생성

MySQL 데이터베이스에서 데이터를 로드하는 함수 정의

뮤지컬 추천을 위한 musical_id에 기반한 엔드포인트 정의

@app_knn.get("/recommend/{musical_id}")
def recommend(musical_id: int):
    try:
        # 선택한 작품의 인덱스 찾기
        selected_work_index_past = past_data[past_data['musical_id'] == musical_id].index[0]

        # 데이터프레임에서 synopsis_numpy_scale 열의 값을 파싱하여(JSON으로 로드하여) 리스트로 변환
        past_data['synopsis_numpy_scale'] = past_data['synopsis_numpy_scale'].apply(json.loads)
        # StandardScaler를 사용하여 특성들을 표준 스케일링
        scaler_past = StandardScaler()
        past_data_scaled = scaler_past.fit_transform(np.vstack(past_data['synopsis_numpy_scale']))

        # KNN 모델 초기화
        knn_model_past = NearestNeighbors(n_neighbors=7, metric='euclidean')
        knn_model_past.fit(past_data_scaled)

        # 현재 작품 선택
        # 데이터프레임에서 synopsis_numpy_scale 열의 값을 파싱하여 리스트로 변환
        present_data['synopsis_numpy_scale'] = present_data['synopsis_numpy_scale'].apply(json.loads)
        # StandardScaler를 사용하여 특성들을 표준 스케일링
        scaler_present = StandardScaler()
        present_data_scaled = scaler_present.fit_transform(np.vstack(present_data['synopsis_numpy_scale']))

        # KNN 모델 초기화
        knn_model_present = NearestNeighbors(n_neighbors=6, metric='euclidean')
        knn_model_present.fit(present_data_scaled)

        # 선택한 작품과 유사한 작품 찾기
        distances, indices = knn_model_present.kneighbors([present_data_scaled[selected_work_index_past]]) # 내적으로 유클리디안 거리가 계산됨

        # 최대 거리와 최소 거리 계산
        max_distance = distances.max()
        min_distance = distances.min()

        # 정규화된 유사도 계산
        # 최대 거리와 최소 거리를 이용하여 거리 값을 [0, 1] 범위로 정규화
        # 정규화된 유사도를 계산하여 유사도 값이 1에 가까울수록 더 유사한 항목
        # distances 배열에서 각 거리 값을 정규화
        normalized_distances = (1 - (distances - min_distance) / (max_distance - min_distance))

        if len(distances[0]) <= 1:  # 유사한 항목이 부족할 경우, 즉, 적어도 2개 이상의 데이터가 필요한 경우 에러 응답을 반환
            return JSONResponse(content={"error": "Not enough similar items found."}, status_code=500)
            # 클라이언트에게 충분한 유사한 항목을 찾을 수 없다는 내용의 에러 메시지를 전송하고 상태 코드 500을 반환

        # 유사도가 높은 순서대로 정렬하여 유사한 작품 인덱스를 찾기
        # argsort 함수를 사용하여 정렬된 인덱스를 얻고, [::-1]을 사용하여 역순으로 정렬
        # 이후 flatten 함수를 사용하여 1차원 배열로 변환
        similar_work_indices = normalized_distances.argsort(axis=0)[::-1].flatten()
        # 상위 N개의 유사한 작품을 선택하되, 실제 유사한 작품의 수를 벗어나지 않도록
        top_n = min(5, len(similar_work_indices))

        # 상위 N개의 유사한 작품에 대한 정보를 추출하고 결과 리스트에 추가
        # 결과 리스트에는 각 작품의 제목, musical_id, 그리고 정규화된 유사도(similarity)가 포함
        result = []  # 추천 사항을 저장할 빈 리스트 생성
        # 유사한 작품 출력
        for i in range(top_n):
            index = similar_work_indices[i]
            similarity = float(distances[0][index])  # NumPy float로 변환
            # 작품 정보 추출
            title = present_data.loc[index, 'title']
            musical_id = int(present_data.loc[index, 'musical_id'])  # NumPy int64를 Python int로 변환
            # 결과 리스트에 작품 정보를 추가
            result.append({"title": title, "musical_id": musical_id, "similarity": similarity})
        # 추천 목록을 반환
        return result 
    except Exception as e:
         # 예외가 발생한 경우, 에러 응답을 반환
        return JSONResponse(content={"error": f"An error occurred: {str(e)}"}, status_code=500)

결과확인

실행방법 : uvicorn app_knn:app_knn --reload --host 0.0.0.0 --port 8080 주소검색 : http://localhost:8080/recommend/{musical_id}

---

해당코드는 0.1.2/FAST -> Fast_API폴더에서 확인가능합니다.