AI Challenge Day 参加記！！画像データのRAG実装に挑戦！

# Parameters
target_image_extensions = ["jpg", "jpeg", "png", "webp", "gif"]
current_dir = os.getcwd()
target_dir = os.path.join(current_dir, "dataset/images/")


# Get File Names
for extension in target_image_extensions:
   for full_path in glob.glob(os.path.join(target_dir, f"*/*.{extension}")):
       file_name = os.path.basename(full_path)


	 # ターゲットファイルのパスとファイル名は後で使うので保存
       image_info_list.append({"file_name": file_name, "full_path": full_path})

# Transform target_files to indeices
for target in image_info_list:
   # base64に変換した画像をgpt-4-turbo-visionで分析
   base64str = local_image_to_base64(target["full_path"])
   contexts = get_contexts_of_image(base64str)


   # computer visionで画像をembedding
   image_vector = get_vectorized_image(target["full_path"])

def get_contexts_of_image(base64str: str):


   api_base = os.getenv("AZURE_OPENAI_ENDPOINT")
   aoai_key = os.getenv("AZURE_OPENAI_KEY")
   deployment_name = os.getenv("AZURE_OPENAI_VISION_DEPLOYMENT_NAME")
   api_version = "2024-02-01"
   base_url = f"{api_base}openai/deployments/{deployment_name}"


   aoai_client = AzureOpenAI(
       api_key=aoai_key,
       api_version=api_version,
       base_url=base_url,
   )


   try:
       response = aoai_client.chat.completions.create(
           model=deployment_name,
           messages=[
               {
                   "role": "system",
                   "content": [
                       {
                           "type": "text",
                           "text": 'You, as the AI assistant, will output a sentence containing a detailed and detailed description of the image you have been given, as well as keywords. It must be strictly in Japanese and follow the following format.\n\n {"keywords":String[],"description":String}',
                       },
                       {"type": "image_url", "image_url": {"url": base64str}},
                   ],
               },
               {
                   "role": "user",
                   "content": [
                       {"type": "text", "text": "この画像を説明して下さい"},
                       {"type": "image_url", "image_url": {"url": base64str}},
                   ],
               },
           ],
           temperature=0.7,
           top_p=0.95,
           max_tokens=1000,
       )
   except Exception as e:
       raise SystemExit(f"Failed to make the request. Error: {e}")


   str_result = response.json()
   json_result = json.loads(str_result)
   metadata_str = json_result["choices"][0]["message"]["content"]
   metadata = json.loads(metadata_str)


   return {
       "keywords": metadata["keywords"],
       "content": metadata["description"],
   }

def get_vectorized_image(image_path: str):

    url = os.getenv("VISION_ENDPOINT")
    vision_key = os.getenv("VISION_KEY")

    with open(image_path, "rb") as img:
        data = img.read()

    headers = {
        "Ocp-Apim-Subscription-Key": vision_key,
        "Content-Type": "application/octet-stream",
    }
    try:
        response = requests.post(url, data=data, headers=headers)
        if response.status_code == 200:
            image_vector = response.json()["vector"]
            return image_vector
        else:
            print(
                f"An error occurred while processing {image_path}. Error code: {response.status_code}."
            )
    except Exception as e:
        print(f"An error occurred while processing {image_path}: {e}")

    return None

def search_images(text_queries: object, vector_query: object):
    ai_search_service_name = os.getenv("AI_SEARCH_SERVICE_NAME")
    ai_search_endpoint = "https://{}.search.windows.net/".format(ai_search_service_name)
    ai_search_key = os.getenv("AI_SEARCH_SERVIVE_ADMIN_KEY")
    index_name = "image-processed"

    credential = AzureKeyCredential(ai_search_key)

    filter_values = ""
    high_score_result = []

    for query in text_queries:
        print(f"Search for '${query}'")
        try:
            search_client = SearchClient(ai_search_endpoint, index_name, credential)
            results = search_client.search(
                search_text=query,
                filter=filter_values,
                query_type=QueryType.SEMANTIC,
                query_caption=QueryCaptionType.EXTRACTIVE,
                query_answer=QueryAnswerType.EXTRACTIVE,
                query_answer_threshold=0.9,
                query_language="ja-JP",
                query_speller="none",
                vector_queries=[vector_query],
                semantic_configuration_name="semantic-config",
                search_fields=["file_name", "content", "keyword"],
                select=["file_name", "content", "keyword"],
                top=5,
            )

            if results:
                # 最も高い@search.scoreと@search.reranker_scoreを持つresultを特定する
                max_search_score_result = None
                max_reranker_score_result = None

                max_search_score = float("-inf")
                max_reranker_score = float("-inf")

                for result in results:
                    if result is not None:
                        if result["@search.score"] > max_search_score:
                            max_search_score = result["@search.score"]
                            max_search_score_result = result
                        if result["@search.reranker_score"] > max_reranker_score:
                            max_reranker_score = result["@search.reranker_score"]
                            max_reranker_score_result = result
                if max_search_score_result:
                    high_score_result.append(max_search_score_result)
                if (
                    max_reranker_score_result
                    and max_reranker_score_result != max_search_score_result
                ):
                    high_score_result.append(max_reranker_score_result)

        except Exception as e:
            print(f"An exception was occurred!!!: {e}")

    return high_score_result

i = 0
for result in search_results:
    i = i + 1
    content = f"{result['file_name']}: {result['content']}"
    content_query += f"{i}. {content}\\n"
content_query = content_query[:-2]

input_prompt = (
    f"# Question\n{target['user_prompt']}\n# Source Informations\n{content_query}"
)
print("input_prompt is\n", input_prompt)

generate_answer(base64str, input_prompt)

def generate_answer(base64str: str, input_prompt):
    api_base = os.getenv("AZURE_OPENAI_ENDPOINT")
    aoai_key = os.getenv("AZURE_OPENAI_KEY")
    deployment_name = os.getenv("AZURE_OPENAI_VISION_DEPLOYMENT_NAME")
    api_version = "2024-02-01"
    base_url = f"{api_base}openai/deployments/{deployment_name}"

    aoai_client = AzureOpenAI(
        api_key=aoai_key,
        api_version=api_version,
        base_url=base_url,
    )

    try:
        response = aoai_client.chat.completions.create(
            model=deployment_name,
            messages=[
                {
                    "role": "system",
                    "content": [
                        {
                            "type": "text",
                            "text": "As a Professional Trip Advisor, you will produce a sentence that accurately responds to questions about World Heritage sites based on the provided source information and images. When you can identify place names or building names, include that information in your description. Additionally, provide enticing descriptions of the location's characteristics, highlights and historical backgrounds to spark the user's interest. Files that appear multiple times in Source Information are considered reliable, so prioritize them in your descriptions. The response must be strictly in Japanese.",
                        },
                    ],
                },
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": input_prompt},
                    ],
                },
            ],
            temperature=0.3,
            top_p=0.95,
            max_tokens=1000,
        )
    except Exception as e:
        raise SystemExit(f"Failed to make the request. Error: {e}")

    str_response = response.json()
    json_response = json.loads(str_response)
    answer = json_response["choices"][0]["message"]["content"]
    print(f"Generated answer is\n{answer}")

# Question
写真を見てどこの寺社か名前を教えてください。
# Source Informations
1. 花ケ咲神社_001.png: この画像は、美しい日本の庭園を映しています。前景には整然と剪定された緑豊かな丸い形の樹木があり、背景には桜の木がピンクの花を咲かせています。画像の中央には、赤と白の色が特徴的な伝統的な日本の寺院の屋根が見えます。この風景は日本の伝統的な美しさと自然の調和を表しています。\n2. 仁和寺_011.jpg: 画像には、満開の桜の木々が並ぶ小道があり、背景には日本の伝統的な五重塔が見えます。春の季節を象徴する美しい桜が青空の下で華やかに咲いており、地面には落ちた桜の花びらが散らばっています。五重塔は黒と灰色の屋根が特徴的で、日本の伝統的な建築美を示しています。\n3. 花ケ咲神社_001.png: この画像は、静寂と自然の美しさが感じられる日本の伝統的な庭園を捉えています。画像の前景には、整形された緑豊かな木々があり、その背後には赤と白の色が映える伝統的な日本建築の一部が見えます。さらに背景には、満開の桜の枝が庭園の上に広がり、鮮やかなピンク色が周囲の緑と調和しています。この場所は、日本の美を象徴する風景として、静けさと和の精神を感じさせる場です。\n4. 仁和寺_011.jpg: 画像には、満開の桜の木々が並ぶ小道があり、背景には日本の伝統的な五重塔が見えます。春の季節を象徴する美しい桜が青空の下で華やかに咲いており、地面には落ちた桜の花びらが散らばっています。五重塔は黒と灰色の屋根が特徴的で、日本の伝統的な建築美を示しています。

この写真は「花ケ咲神社」のものです。美しい日本の庭園が映されており、整然と剪定された緑豊かな丸い形の樹木と、満開の桜の木が背景にあります。中央に見えるのは、赤と白の色が特徴的な伝統的な日本の寺院の屋根です。この場所は、日本の伝統的な美しさと自然の調和を感じさせる静寂な空間です。