이미지 생성 및 분석을 위한 GPT-4 사용 방법

### code block translation starts ###
%pip install openai --quiet
%pip install tenacity --quiet
%pip install tqdm --quiet
%pip install numpy --quiet
%pip install typing --quiet
%pip install tiktoken --quiet
%pip install concurrent --quiet
### code block translation ends ###
 
### code block translation starts ###
import pandas as pd
import numpy as np
import json
import ast
import tiktoken
import concurrent
from openai import OpenAI
from tqdm import tqdm
from tenacity import retry, wait_random_exponential, stop_after_attempt
from IPython.display import Image, display, HTML
from typing import List
 
client = OpenAI()
 
GPT_MODEL = "gpt-4-vision-preview"
EMBEDDING_MODEL = "text-embedding-3-large"
EMBEDDING_COST_PER_1K_TOKENS = 0.00013
### code block translation ends ###
 
### markdown translation starts ###
### Creating the Embeddings
# 지식 베이스를 설정하고 그에 대한 임베딩을 생성함으로써 지식 베이스를 설정합니다. 데이터 폴더 내의 'sample_styles.csv' 파일을 사용합니다. 이것은 '약 44K'개의 항목을 포함한 더 큰 데이터셋의 샘플입니다. 또한, 이 단계는 기본 제공 벡터 베이스를 사용하여 대체될 수도 있습니다. 예를 들어, [여기 cookbooks 중 하나](https://github.com/openai/openai-cookbook/tree/main/examples/vector_databases)를 따라하여 벡터 베이스를 설정할 수 있습니다.
### markdown translation ends ###
 
### code block translation starts ###
styles_filepath = "data/sample_clothes/sample_styles.csv"
styles_df = pd.read_csv(styles_filepath, on_bad_lines='skip')
print(styles_df.head())
print("Opened dataset successfully. Dataset has {} items of clothing.".format(len(styles_df)))
### code block translation ends ###
 
### markdown translation starts ###
Now we will generate embeddings for the entire dataset. We can parallelize the execution of these embeddings to ensure that the script scales up for larger datasets. With this logic, the time to create embeddings for the full `44K` entry dataset decreases from ~4h to ~2-3min.
### markdown translation ends ###
 
```python
## Batch Embedding Logic
# Simple function to take in a list of text objects and return them as a list of embeddings
@retry(wait=wait_random_exponential(min=1, max=40), stop=stop_after_attempt(10))
def get_embeddings(input: List):
    response = client.embeddings.create(
        input=input,
        model=EMBEDDING_MODEL
    ).data
    return [data.embedding for data in response]
 
 
# Splits an iterable into batches of size n.
def batchify(iterable, n=1):
    l = len(iterable)
    for ndx in range(0, l, n):
        yield iterable[ndx : min(ndx + n, l)]
     
 
# Function for batching and parallel processing the embeddings
def embed_corpus(
    corpus: List[str],
    batch_size=64,
    num_workers=8,
    max_context_len=8191,
):
    # Encode the corpus, truncating to max_context_len
    encoding = tiktoken.get_encoding("cl100k_base")
    encoded_corpus = [
        encoded_article[:max_context_len] for encoded_article in encoding.encode_batch(corpus)
    ]
 
    # Calculate corpus statistics: the number of inputs, the total number of tokens, and the estimated cost to embed
    num_tokens = sum(len(article) for article in encoded_corpus)
    cost_to_embed_tokens = num_tokens / 1000 * EMBEDDING_COST_PER_1K_TOKENS
    print(
        f"num_articles={len(encoded_corpus)}, num_tokens={num_tokens}, est_embedding_cost={cost_to_embed_tokens:.2f} USD"
    )
 
    # Embed the corpus
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
        
        futures = [
            executor.submit(get_embeddings, text_batch)
            for text_batch in batchify(encoded_corpus, batch_size)
        ]
 
        with tqdm(total=len(encoded_corpus)) as pbar:
            for _ in concurrent.futures.as_completed(futures):
                pbar.update(batch_size)
 
        embeddings = []
        for future in futures:
            data = future.result()
            embeddings.extend(data)
 
        return embeddings
    
# Function to generate embeddings for a given column in a DataFrame
def generate_embeddings(df, column_name):
    # Initialize an empty list to store embeddings
    descriptions = df[column_name].astype(str).tolist()
    embeddings = embed_corpus(descriptions)
 
    # Add the embeddings as a new column to the DataFrame
    df['embeddings'] = embeddings
    print("Embeddings created successfully.")

generate_embeddings(styles_df, 'productDisplayName')
print("Writing embeddings to file ...")
styles_df.to_csv('data/sample_clothes/sample_styles_with_embeddings.csv', index=False)
print("Embeddings successfully stored in sample_styles_with_embeddings.csv")

# styles_df = pd.read_csv('data/sample_clothes/sample_styles_with_embeddings.csv', on_bad_lines='skip')
 
# # Convert the 'embeddings' column from string representations of lists to actual lists of floats
# styles_df['embeddings'] = styles_df['embeddings'].apply(lambda x: ast.literal_eval(x))
 
print(styles_df.head())
print("Opened dataset successfully. Dataset has {} items of clothing along with their embeddings.".format(len(styles_df)))

def cosine_similarity_manual(vec1, vec2):
    """두 벡터 사이의 코사인 유사도를 계산합니다."""
    vec1 = np.array(vec1, dtype=float)
    vec2 = np.array(vec2, dtype=float)
 
 
    dot_product = np.dot(vec1, vec2)
    norm_vec1 = np.linalg.norm(vec1)
    norm_vec2 = np.linalg.norm(vec2)
    return dot_product / (norm_vec1 * norm_vec2)
 
 
def find_similar_items(input_embedding, embeddings, threshold=0.5, top_k=2):
    """코사인 유사도를 기반으로 가장 유사한 항목을 찾습니다."""
    
    # 입력 임베딩과 다른 모든 임베딩 사이의 코사인 유사도 계산
    similarities = [(index, cosine_similarity_manual(input_embedding, vec)) for index, vec in enumerate(embeddings)]
    
    # 임계값 이하의 유사도 제외
    filtered_similarities = [(index, sim) for index, sim in similarities if sim >= threshold]
    
    # 유사도 점수에 따라 필터링된 유사도 정렬
    sorted_indices = sorted(filtered_similarities, key=lambda x: x[1], reverse=True)[:top_k]
 
    # 상위 k개의 가장 유사한 항목 반환
    return sorted_indices

def find_matching_items_with_rag(df_items, item_descs):
   """입력 항목 설명을 기반으로 코사인 유사도에 따라 가장 유사한 항목을 찾습니다."""
   
   # 데이터프레임에서 임베딩 선택
   embeddings = df_items['embeddings'].tolist()
 
   
   similar_items = []
   for desc in item_descs:
      
      # 입력 항목에 대한 임베딩 생성
      input_embedding = get_embeddings([desc])
    
      # 코사인 유사도에 따라 가장 유사한 항목 찾기
      similar_indices = find_similar_items(input_embedding, embeddings, threshold=0.6)
      similar_items += [df_items.iloc[i] for i in similar_indices]
    
   return similar_items

def analyze_image(image_base64, subcategories):
    response = client.chat.completions.create(
        model=GPT_MODEL,
        messages=[
            {
            "role": "user",
            "content": [
                {
                "type": "text",
                "text": """의류 항목의 이미지를 제공하면 해당 항목을 분석하고 "items", "category" 및 "gender"와 같은 필드로 구성된 JSON 출력을 생성합니다.
                           패션 트렌드, 스타일 및 성별 기호 등을 이용하여 아웃핏을 완성하는 정확하고 관련성 높은 제안을 제공하세요.
                           items 필드는 이미지 내 항목과 잘 어울리는 항목들의 리스트여야 합니다. 각 항목은 스타일, 색상 및 성별 정보가 포함된 의류 항목의 제목입니다."""

                           이 목록의 유형 중에서 카테고리를 선택해야 합니다: {subcategories}.
                           이 목록의 성별 사이에서 선택해야 합니다: [Men, Women, Boys, Girls, Unisex]
                           사진의 아이템 설명을 포함하지 마세요. 출력에서 `json` 태그를 포함하지 마세요.
 
                           예시 입력: 검은 가죽 재킷을 나타내는 이미지.
 
                           예시 출력: {"items": ["여성용 흰색 반팔 티셔츠", "화이트 캔버스 스니커즈", "여성용 검정 스키니 진"], "category": "자켓", "gender": "여성"}
                           """,
                },
                {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{image_base64}",
                },
                }
            ],
            }
        ],
        max_tokens=300,
    )
    # 응답에서 관련 특징 추출
    features = response.choices[0].message.content
    return features

import base64
 
def encode_image_to_base64(image_path):
    with open(image_path, 'rb') as image_file:
        encoded_image = base64.b64encode(image_file.read())
        return encoded_image.decode('utf-8')

# 이미지 경로 설정 및 테스트 이미지 선택
image_path = "data/sample_clothes/sample_images/"
test_images = ["2133.jpg", "7143.jpg", "4226.jpg"]
 
# 테스트 이미지를 base64로 인코딩
reference_image = image_path + test_images[0]
encoded_image = encode_image_to_base64(reference_image)

# DataFrame에서 고유한 서브카테고리 선택
unique_subcategories = styles_df['articleType'].unique()
 
# 이미지 분석 수행하고 결과 반환
analysis = analyze_image(encoded_image, unique_subcategories)
image_analysis = json.loads(analysis)
 
# 이미지와 분석 결과 출력
display(Image(filename=reference_image))
print(image_analysis)

# 분석 결과에서 관련 기능 추출
item_descs = image_analysis['items']
item_category = image_analysis['category']
item_gender = image_analysis['gender']
 
 
# 동일한 성별 (또는 유니섹스)이고 다른 카테고리의 아이템만 사용하여 데이터 필터링
filtered_items = styles_df.loc[styles_df['gender'].isin([item_gender, 'Unisex'])]
filtered_items = filtered_items[filtered_items['articleType'] != item_category]
print(str(len(filtered_items)) + " Remaining Items")
 
# 입력 아이템 설명과 가장 유사한 아이템 찾기
matching_items = find_matching_items_with_rag(filtered_items, item_descs)
 
# 일치하는 항목 표시 (이미지 분석의 각 설명에 대해 2개의 항목이 표시됩니다)
html = ""
paths = []
for i, item in enumerate(matching_items):
    item_id = item['id']
        
    # 이미지 파일 경로
    image_path = f'data/sample_clothes/sample_images/{item_id}.jpg'
    paths.append(image_path)
    html += f'<img src="{image_path}" style="display:inline;margin:1px"/>'
 
# 일치하는 항목 설명 출력
print(item_descs)
# 이미지 표시
display(HTML(html))

def check_match(reference_image_base64, suggested_image_base64):
    response = client.chat.completions.create(
        model=GPT_MODEL,
        messages=[
            {
            "role": "user",
            "content": [
                {
                "type": "text",
                "text": """ 두 가지 다른 옷의 이미지가 제공됩니다.
                            목표는 두 이미지의 옷이 함께 잘어울리는지를 결정하는 것입니다.
                            첫 번째 이미지는 기준 아이템(사용자가 다른 아이템과 맞추려고 하는 아이템)입니다.
                            두 번째 아이템이 기준 아이템과 잘 어울리는지를 결정해야 합니다.
                            답변은 "answer", "reason" 두 가지 필드를 포함하는 JSON 출력이어야 합니다.
                            "answer" 필드는 두 아이템이 서로 잘 어울릴 때 "yes" 또는 "no"가됩니다.
                            "reason" 필드는 답변에 대한 추론의 간단한 설명이어야 합니다. 2개 이미지의 설명은 포함하지 마십시오.
                            출력에 ```json``` 태그를 포함하지 마십시오.
                           """,
                },
                {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{reference_image_base64}",
                },
                },
                {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{suggested_image_base64}",
                },
                }
            ],
            }
        ],
        max_tokens=300,
    )
    # 응답에서 관련 기능을 추출합니다.
    features = response.choices[0].message.content
    return features

# 생성된 이미지의 고유 경로를 선택합니다.
paths = list(set(paths))
 
for path in paths:
    # 테스트 이미지를 base64로 인코딩합니다.
    suggested_image = encode_image_to_base64(path)
    
    # 아이템이 일치하는지 확인합니다.
    match = json.loads(check_match(encoded_image, suggested_image))
    
    # 이미지와 분석 결과를 표시합니다.
    if match["answer"] == 'yes':
        display(Image(filename=path))
        print("아이템이 일치합니다!")
        print(match["reason"])