Platform For AI:モデルギャラリーのクイックスタート

次のコードを実行するには、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,  # より多様な名前と住所のために温度を上げます。
        )

        result = response.choices[0].message.content.strip()
        # 潜在的なマークダウンコードブロックマーカーをクリーンアップします。
        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: 完全な州名 (例: "California" であり "CA" ではない)。
- zip_code: 5 桁または 9 桁の郵便番号。
- phone: 完全な連絡先電話番号。

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

## 出力形式
以下の JSON 形式に厳密に従ってください。説明文やマークダウンは追加しないでください。
{
  "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,  # 抽出の精度を高めるために温度を下げます。
            response_format={"type": "json_object"},
            extra_body={"enable_thinking": False}
        )

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

        # 潜在的なマークダウンコードブロックマーカーをクリーンアップします。
        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)