Platform For AI:Model Gallery クイックスタート

以下のコードを実行するには、Alibaba Cloud Model Studio API キーを作成する 必要があります。コードでは qwen-plus-latest を用いて業務データを生成し、qwen3-235b-a22b を用いてラベル付けを行います。

# -*- coding: utf-8 -*-
import os
import asyncio
import random
import json
import sys
from typing import List
import platform
from openai import AsyncOpenAI

# 非同期クライアントインスタンスを作成します。
# 注：このスクリプトでは DashScope 互換 API エンドポイントを使用します。
# 異なる OpenAI 互換サービスを使用する場合は、base_url を変更してください。
client = AsyncOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 米国の州および準州のリスト。
us_states = [
    "Alabama", "Alaska", "Arizona", "Arkansas", "California", "Colorado", "Connecticut", "Delaware",
    "Florida", "Georgia", "Hawaii", "Idaho", "Illinois", "Indiana", "Iowa", "Kansas", "Kentucky",
    "Louisiana", "Maine", "Maryland", "Massachusetts", "Michigan", "Minnesota", "Mississippi",
    "Missouri", "Montana", "Nebraska", "Nevada", "New Hampshire", "New Jersey", "New Mexico",
    "New York", "North Carolina", "North Dakota", "Ohio", "Oklahoma", "Oregon", "Pennsylvania",
    "Rhode Island", "South Carolina", "South Dakota", "Tennessee", "Texas", "Utah", "Vermont",
    "Virginia", "Washington", "West Virginia", "Wisconsin", "Wyoming", "District of Columbia",
    "Puerto Rico", "Guam", "American Samoa", "U.S. Virgin Islands", "Northern Mariana Islands"
]

# 受信者テンプレート。
recipient_templates = [
    "宛先：{name}", "受信者：{name}", "配達先：{name}", "対象：{name}",
    "ATTN：{name}", "{name}", "氏名：{name}", "連絡先：{name}", "受取人：{name}"
]

# 電話番号テンプレート。
phone_templates = [
    "Tel：{phone}", "Tel. {phone}", "携帯：{phone}", "電話：{phone}",
    "連絡先電話番号：{phone}", "電話番号 {phone}", "TEL：{phone}", "MOBILE：{phone}",
    "連絡先：{phone}", "P：{phone}", "{phone}", "通話：{phone}",
]


# 米国スタイルの電話番号を生成します。
def generate_us_phone():
    """10 桁の米国電話番号を (XXX) XXX-XXXX 形式でランダムに生成します。"""
    area_code = random.randint(201, 999)  # 0xx、1xx の市外局番を避ける。
    exchange = random.randint(200, 999)
    line = random.randint(1000, 9999)
    return f"({area_code}) {exchange}-{line}"


# LLM を用いて受信者および住所情報を生成します。
async def generate_recipient_and_address_by_llm(state: str):
    """指定された州に対して、LLM を用いて受信者氏名および住所情報を生成します。"""
    prompt = f"""米国 {state} の所在地に関する受信者情報を生成してください。以下の内容を含めてください：
1. 現実的な英語のフルネーム。多様性を確保してください。
2. その州内の実在する都市名。
3. 具体的な住所（例：建物番号および通り名、アパート番号など）。現実的である必要があります。
4. その都市または地域に対応する 5 桁の ZIP コード。

以下の JSON オブジェクトのみを返してください：
{{"name": "受信者氏名", "city": "都市名", "street_address": "具体的な住所", "zip_code": "ZIP コード"}}

その他のテキストは一切含めず、JSON のみを返してください。氏名は「John Doe」のような一般的な名前だけではなく、多様なものにしてください。
"""

    try:
        response = await client.chat.completions.create(
            messages=[{"role": "user", "content": prompt}],
            model="qwen-plus-latest",
            temperature=1.5,  # 氏名および住所の多様性を高めるために温度を上げます。
        )

        result = response.choices[0].message.content.strip()
        # Markdown コードブロックマーカーをクリーンアップします。
        if result.startswith('```'):
            result = result.split('\n', 1)[1]
        if result.endswith('```'):
            result = result.rsplit('\n', 1)[0]

        # JSON の解析を試みます。
        info = json.loads(result)
        print(info)
        return info
    except Exception as e:
        print(f"受信者および住所情報の生成に失敗しました：{e}。代替プランを使用します。")
        # 代替プラン。
        backup_names = ["Michael Johnson", "Emily Williams", "David Brown", "Jessica Jones", "Christopher Davis",
                        "Sarah Miller"]
        return {
            "name": random.choice(backup_names),
            "city": "Anytown",
            "street_address": f"{random.randint(100, 9999)} Main St",
            "zip_code": f"{random.randint(10000, 99999)}"
        }


# 1 件の生データレコードを生成します。
async def generate_record():
    """米国の住所情報を含む 1 件の雑然とした文字列を生成します。"""
    # 州をランダムに選択します。
    state = random.choice(us_states)

    # LLM を用いて受信者および住所情報を生成します。
    info = await generate_recipient_and_address_by_llm(state)

    # 受信者氏名のフォーマットを設定します。
    recipient = random.choice(recipient_templates).format(name=info['name'])

    # 電話番号を生成します。
    phone = generate_us_phone()
    phone_info = random.choice(phone_templates).format(phone=phone)

    # 完全な住所行を組み立てます。
    full_address = f"{info['street_address']}, {info['city']}, {state} {info['zip_code']}"

    # すべてのコンポーネントを結合します。
    components = [recipient, phone_info, full_address]

    # コンポーネントの順序をランダムにシャッフルします。
    random.shuffle(components)

    # ランダムなセパレーターを選択します。
    separators = [' ', ', ', '; ', ' | ', '\t', ' - ', ' // ', '', '  ']
    separator = random.choice(separators)

    # コンポーネントを結合します。
    combined_data = separator.join(components)
    return combined_data.strip()


# データをバッチ単位で生成します。
async def generate_batch_data(count: int) -> List[str]:
    """指定した件数のデータレコードを生成します。"""
    print(f"{count} 件のレコードの生成を開始しています...")

    # 同時リクエスト数を制御するためのセマフォを使用します（例：最大 20 件の同時リクエスト）。
    semaphore = asyncio.Semaphore(20)

    async def generate_single_record(index):
        async with semaphore:
            try:
                record = await generate_record()
                print(f"{index + 1} 件目のレコードを生成しました：{record}")
                return record
            except Exception as e:
                print(f"{index + 1} 件目のレコードの生成に失敗しました：{e}")
                return None

    # 同時にデータを生成します。
    tasks = [generate_single_record(i) for i in range(count)]

    data = await asyncio.gather(*tasks)

    successful_data = [record for record in data if record is not None]

    return successful_data


# データをファイルに保存します。
def save_data(data: List[str], filename: str = "us_recipient_data.json"):
    """生成されたデータを JSON ファイルに保存します。"""
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"データを {filename} に保存しました")


# フェーズ 1：データ生成。
async def produce_data_phase():
    """生の受信者データの生成を処理します。"""
    print("=== フェーズ 1：生の受信者データ生成の開始 ===")

    # 2,000 件のレコードを生成します。
    batch_size = 2000
    data = await generate_batch_data(batch_size)

    # データを保存します。
    save_data(data, "us_recipient_data.json")

    print(f"\n合計 {len(data)} 件のレコードを生成しました")
    print("\nサンプルデータ：")
    for i, record in enumerate(data[:3]):  # 最初の 3 件を例として表示します。
        print(f"{i + 1}. 生データ：{record}\n")

    print("=== フェーズ 1 完了 ===\n")
    return True


# 情報抽出タスクのシステムプロンプトを定義します。
def get_system_prompt_for_extraction():
    """情報抽出タスクのシステムプロンプトを返します。"""
    return """あなたは、非構造化テキストから米国の配送住所を解析することに特化した専門的な情報抽出アシスタントです。

## タスクの説明
与えられた入力テキストに基づき、以下の 6 つのフィールドを含む JSON オブジェクトを正確に抽出および生成します：
- name：受信者のフルネーム。
- street_address：完全な住所（建物番号、通り名、アパートまたはスイート番号を含む）。
- city：都市名。
- state：完全な州名（例：「California」、略称の「CA」ではない）。
- zip_code：5 桁または 9 桁の ZIP コード。
- phone：完全な連絡先電話番号。

## 抽出ルール
1.  **住所の取り扱い**：
    -   通り、都市、州、ZIP コードの構成要素を正確に識別します。
    -   `state` フィールドには完全な正式名称（例：「New York」、略称の「NY」ではない）を使用します。
    -   `street_address` には、都市の前にあるすべての詳細（例：「123 Apple Lane, Apt 4B」）を含める必要があります。
2.  **氏名の識別**：
    -   受信者のフルネームを抽出します。
3.  **電話番号の取り扱い**：
    -   完全な電話番号を抽出し、元の形式を保持します。
4.  **ZIP コード**：
    -   5 桁または 9 桁（ZIP+4）のコードを抽出します。

## 出力形式
以下の JSON 形式を厳密に遵守してください。説明文や Markdown は一切追加しないでください。
{
  "name": "受信者のフルネーム",
  "street_address": "完全な住所",
  "city": "都市名",
  "state": "完全な州名",
  "zip_code": "ZIP コード",
  "phone": "連絡先電話番号"
}
"""


# 生の文字列から構造化データを予測します。
async def predict_structured_data(raw_data: str):
    """LLM を用いて生の文字列から構造化データを予測します。"""
    system_prompt = get_system_prompt_for_extraction()

    try:
        response = await client.chat.completions.create(
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": raw_data}
            ],
            model="qwen3-235b-a22b",  # このタスクには強力なモデルが推奨されます。
            temperature=0.0,  # 抽出の精度を高めるために温度を下げます。
            response_format={"type": "json_object"},
            extra_body={"enable_thinking": False}
        )

        result = response.choices[0].message.content.strip()

        # Markdown コードブロックマーカーをクリーンアップします。
        if result.startswith('```'):
            lines = result.split('\n')
            for i, line in enumerate(lines):
                if line.strip().startswith('{'):
                    result = '\n'.join(lines[i:])
                    break
        if result.endswith('```'):
            result = result.rsplit('\n```', 1)[0]

        structured_data = json.loads(result)
        return structured_data

    except Exception as e:
        print(f"構造化データの予測に失敗しました：{e}、生データ：{raw_data}")
        # 失敗時には空の構造を返します。
        return {
            "name": "",
            "street_address": "",
            "city": "",
            "state": "",
            "zip_code": "",
            "phone": ""
        }


# フェーズ 2：データ変換。
async def convert_data_phase():
    """生データを読み込み、構造化形式を予測し、SFT 形式に変換します。"""
    print("=== フェーズ 2：SFT 形式へのデータ変換の開始 ===")

    try:
        print("us_recipient_data.json ファイルの読み込みを開始しています...")
        with open('us_recipient_data.json', 'r', encoding='utf-8') as f:
            raw_data_list = json.load(f)

        print(f"{len(raw_data_list)} 件のレコードを正常に読み込みました。")
        print("抽出モデルを用いて構造化データの予測を開始しています...")

        # トレーニングおよび推論の速度を向上させるため、シンプルで明確なシステムメッセージを使用します。
        system_prompt = "あなたは、米国の配送情報から構造化された JSON を抽出する専門アシスタントです。JSON のキーは name、street_address、city、state、zip_code、phone です。"
        output_file = 'us_recipient_sft_data.json'

        # 同時リクエスト数を制御するためのセマフォを使用します。
        semaphore = asyncio.Semaphore(10)

        async def process_single_item(index, raw_data):
            async with (semaphore):
                structured_data = await predict_structured_data(raw_data)
                print(f"{index + 1} 件目のレコードを処理中：{raw_data}")

                conversation = {
                        "instruction": system_prompt + '  ' + raw_data,
                        "output": json.dumps(structured_data, ensure_ascii=False)
                }

                return conversation

        print(f"{output_file} への変換を開始しています...")

        tasks = [process_single_item(i, raw_data) for i, raw_data in enumerate(raw_data_list)]
        conversations = await asyncio.gather(*tasks)

        with open(output_file, 'w', encoding='utf-8') as outfile:
            json.dump(conversations, outfile, ensure_ascii=False, indent=4)

        print(f"変換が完了しました！{len(raw_data_list)} 件のレコードを処理しました。")
        print(f"出力ファイル：{output_file}")
        print("=== フェーズ 2 完了 ===")

    except FileNotFoundError:
        print("エラー：us_recipient_data.json が見つかりません。")
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"JSON 解析エラー：{e}")
        sys.exit(1)
    except Exception as e:
        print(f"変換中にエラーが発生しました：{e}")
        sys.exit(1)


# メイン関数。
async def main():
    print("データ処理パイプラインの開始...")
    print("このプログラムは以下の 2 つのフェーズを順次実行します：")
    print("1. 生の米国受信者データを生成します。")
    print("2. 構造化データを予測し、SFT 形式に変換します。")
    print("-" * 50)

    # フェーズ 1：データの生成。
    success = await produce_data_phase()

    if success:
        # フェーズ 2：データの変換。
        await convert_data_phase()

        print("\n" + "=" * 50)
        print("すべての処理が正常に完了しました！")
        print("生成されたファイル：")
        print("- us_recipient_data.json：非構造化データの一覧。")
        print("- us_recipient_sft_data.json：SFT 形式のトレーニングデータ。")
        print("=" * 50)
    else:
        print("データ生成フェーズに失敗しました。終了します。")


if __name__ == '__main__':
    # Windows の場合、イベントループポリシーを設定します。
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

    # メインコルーチンを実行します。
    asyncio.run(main(), debug=False)