商品名の名寄せに生成AIを使ってみた

import time
import textwrap
import pandas as pd
import boto3
import json
import io

class ProductNameDeduper:
    """
    商品名の名寄せを行うクラス。
    名寄せルールに基づき表記を統一する。
    """

    def __init__(self, aws_region: str = "us-east-1"):
        session = boto3.Session(region_name=aws_region)
        self._model_identifier = "anthropic.claude-3-5-sonnet-20240620-v1:0"
        self._bedrock_runtime_client = session.client(
            service_name="bedrock-runtime",
            region_name=aws_region
        )

    def __invoke_bedrock(self, context_message: str, customer_input: str) -> str:
        """
        Bedrockにリクエストを行う
        ThrottlingExceptionが出た際は3回までリトライする。
        """

        # リクエストのペイロードを作成
        payload = json.dumps(
            {
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 4096,
                "temperature": 0,
                "system": context_message,
                "messages": [
                    {
                        "role": "user",
                        "content": [{"type": "text", "text": customer_input}],
                    }
                ],
            }
        )

        for _ in range(3):
            try:
                response = self._bedrock_runtime_client.invoke_model(
                    modelId=self._model_identifier,
                    body=payload
                )
                api_result = json.loads(response.get("body").read())
                # 使用トークン数の出力
                print(api_result["usage"])
                return api_result["content"][0]["text"]
            except self._bedrock_runtime_client.exceptions.ThrottlingException as te:
                print("Exceeded rate: wait 10 seconds...")
                time.sleep(10)
            except Exception:
                raise

        raise RuntimeError("Failed to get response from Bedrock service.")

    def identify_and_normalize_products(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        商品名の名寄せを行って結果をDataFrameで返す
        """

        system_prompt_msg = textwrap.dedent("""
                あなたはECサイトに掲載された商品名の名寄せを行うアシスタントです。
                各商品にはIDと商品名が入力として与えられます。商品名にばらつきがある場合でも、できる限り本来の正しい商品名を推測し、一定のルールに基づいて表記を統一してください。

                ▼ 名寄せルール
                1. 不要な接尾辞や修飾語（例: "～セット", "数量限定"など）は基本的に削除し、商品名の本質的な部分のみを残してください。
                2. メーカーやブランド名などが商品名の一部として確実に必要な場合のみ残し、それらにゆらぎがあれば統一した表記に整えてください（例： "Sony"/"SONY"/"ソニー"→"Sony"）。
                3. バージョンや型番など、商品本体に必須の情報が含まれている場合は残してください（例： "iPhone 13 Pro"など）。
                4. カラー、サイズ、容量といった情報は今回の集計には不要なので削除してください。
                5. 上記ルールに当てはまらない場合は、可能な範囲で統一し、未知の場合は元の表記を尊重しつつ簡潔にしてください。

                ▼ 出力フォーマットのルール
                1. 出力はCSV形式でヘッダーは不要です。
                2. 返事や解説などは一切不要で、CSVの中身のみを返してください。
                3. 列は必ず"id","product"の2列だけにしてください。
                4. "id"は入力で与えられた値をそのまま出力すること。
                5. "product"は上記の名寄せルールに従って統一した商品名を出力してください。

                以下がIDと商品名の一覧です。
                """)[1:-1]

        target_columns = ["id", "product"]
        # 結果をまとめるデータフレーム
        consolidated_data = pd.DataFrame(columns=target_columns)

        batch_size = 20  # 1回あたりの処理件数

        for i in range(0, len(input_data), batch_size):
            batch_df = input_data[i : i + batch_size].copy()
            csv_data = batch_df.to_csv(index=False)

            completion = self.__invoke_bedrock(
                context_message=system_prompt_msg,
                customer_input=csv_data
            )

            # 返ってきたCSVからDataFrameを生成、列名をそろえる
            parsed_result = pd.read_csv(
                io.StringIO(completion),
                header=None,
                names=target_columns
            )

            consolidated_data = pd.concat(
                [consolidated_data, parsed_result],
                ignore_index=True
            )

            print(f"len(consolidated_data) = {len(consolidated_data)}")

        return consolidated_data

if __name__ == "__main__":
    source_csv = './input/product_list.csv'
    print(f"starting: {source_csv}")

    # 入力データの読み込み
    df_input = pd.read_csv(source_csv)
    df_input = df_input[["id","text"]]
    print(df_input.head())

    # クラスインスタンス化
    normalizer = ProductNameDeduper()
    # 名寄せ処理を実行
    consolidated_result = normalizer.identify_and_normalize_products(df_input)

    print(consolidated_result.head())

    # 元のDataFrameとマージして出力
    final_output = df_input.merge(consolidated_result, how="left", on="id")
    final_output.to_csv('./output/result.csv', index=False)

プロダクト名,件数
AirPods Pro 第2世代,8
Amazon Kindle Oasis Pro,3
Kindle Oasis Pro,4
Apple iPad Pro 13,4
iPad Pro 13,1
iPad Pro 13 M3,2
iPad Air 第6世代,6
Apple iPhone 15 Pro Max,1
iPhone 15 Pro Max,5
Apple Watch Series 9,2
Apple Watch Series 9 45mm,3
Apple Watch Series 9 45mm GPS,2
ASUS ZenBook 14 OLED,7
Bose QuietComfort 45,7
Canon EOS R8,7
DJI Air 3,7
DualSense Edge コントローラー,7
Dyson V15 Detect,1
Dyson V15 Detect Complete,1
Dyson V15 Detect Submarine,6
FUJIFILM X-T5,7
Google Pixel 8 Pro,7
GoPro HERO12 Black,7
HP Spectre x360 14,6
HP x360 14,1
Lenovo ThinkPad X1 Carbon,3
Lenovo ThinkPad X1 Carbon Gen11,4
MacBook Air 15 M2,3
MacBook Air 15インチ M2,3
Nintendo Switch Lite,7
Nintendo Switch Pro コントローラー Zelda Edition,7
Nintendo Switch 有機ELモデル,8
Panasonic LUMIX S5II,7
PlayStation 5 Digital Edition,4
PlayStation 5 デジタルエディション,4
Razer Blade 16,7
Samsung Galaxy S24 Ultra,6
Sony WF-1000XM5,7
Sony WH-1000XM5,7
Surface Pro 9,4
Xiaomi Redmi Note 13 Pro,1
Xiaomi Redmi Note 13 Pro 5G,6