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 = [
    "To: {name}", "Recipient: {name}", "Deliver to {name}", "For: {name}",
    "ATTN: {name}", "{name}", "Name: {name}", "Contact: {name}", "Receiver: {name}"
]

# 電話番号テンプレート。
phone_templates = [
    "Tel: {phone}", "Tel. {phone}", "Mobile: {phone}", "Phone: {phone}",
    "Contact number: {phone}", "Phone number {phone}", "TEL: {phone}", "MOBILE: {phone}",
    "Contact: {phone}", "P: {phone}", "{phone}", "Call: {phone}",
]


# もっともらしい米国スタイルの電話番号を生成します。
def generate_us_phone():
    """(XXX) XXX-XXXX 形式でランダムな 10 桁の米国電話番号を生成します。"""
    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 桁の郵便番号。

以下の形式で JSON オブジェクトのみを返してください。
{{"name": "受信者名", "city": "市区町村名", "street_address": "特定の住所", "zip_code": "郵便番号"}}

他のテキストは含めず、JSON のみを含めてください。名前が John Doe だけでなく、多様であることを確認してください。
"""

    try:
        response = await client.chat.completions.create(
            messages=[{"role": "user", "content": prompt}],
            model="qwen-plus-latest",
            temperature=1.5,  # temperature を高くして、より多様な名前と住所を生成します。
        )

        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)}"
        }


# 単一の生データレコードを生成します。
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: 州の正式名称 (例: "CA" ではなく "California")。
- zip_code: 5 桁または 9 桁の郵便番号。
- phone: 完全な連絡先電話番号。

## 抽出ルール
1.  **住所の処理**:
    -   住所、市区町村、州、郵便番号の各コンポーネントを正確に特定します。
    -   `state` フィールドは、州の正式名称でなければなりません (例: "NY" ではなく "New York")。
    -   `street_address` には、市区町村名の前にあるすべての詳細 (例: "123 Apple Lane, Apt 4B") を含める必要があります。
2.  **名前の特定**:
    -   受信者のフルネームを抽出します。
3.  **電話番号の処理**:
    -   元のフォーマットを維持したまま、完全な電話番号を抽出します。
4.  **郵便番号**:
    -   5 桁または 9 桁 (ZIP+4) の郵便番号を抽出します。

## 出力フォーマット
以下の JSON フォーマットに厳密に従ってください。説明文や Markdown を追加しないでください。
{
  "name": "受信者のフルネーム",
  "street_address": "完全な住所",
  "city": "市区町村名",
  "state": "州の正式名称",
  "zip_code": "郵便番号",
  "phone": "連絡先電話番号"
}
"""


# LLM を使用して生テキストから構造化データを予測します。
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,  # 抽出の精度を高めるために temperature を低くします。
            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)