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():
    """ランダムな 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"""Please generate recipient information for a location in {state}, USA, including the following:
1. A realistic full English name. Aim for diversity.
2. A real city name within that state.
3. A specific street address, such as street number and name, and apartment number. It should be realistic.
4. A corresponding 5-digit ZIP code for that city or area.

Please return only the JSON object in the following format:
{{"name": "Recipient Name", "city": "City Name", "street_address": "Specific Street Address", "zip_code": "ZIP Code"}}

Do not include any other text, just the JSON. Ensure names are diverse, not just 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"Failed to generate recipient and address: {e}, using fallback.")
        # フォールバックメカニズム。
        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"Starting to generate {count} records...")

    # セマフォを使用して同時実行数を制御します (例: 最大 20 件の同時リクエスト)。
    semaphore = asyncio.Semaphore(20)

    async def generate_single_record(index):
        async with semaphore:
            try:
                record = await generate_record()
                print(f"Generated record #{index + 1}: {record}")
                return record
            except Exception as e:
                print(f"Failed to generate record #{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"Data has been saved to {filename}")


# フェーズ 1: データ生成。
async def produce_data_phase():
    """生の受信者データの生成を処理します。"""
    print("=== Phase 1: Starting Raw Recipient Data Generation ===")

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

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

    print(f"\nTotal records generated: {len(data)}")
    print("\nSample Data:")
    for i, record in enumerate(data[:3]):  # 最初の 3 件を例として表示します。
        print(f"{i + 1}. Raw Data: {record}\n")

    print("=== Phase 1 Complete ===\n")
    return True


# 抽出モデル用のシステムプロンプトを定義します。
def get_system_prompt_for_extraction():
    """情報抽出タスク用のシステムプロンプトを返します。"""
    return """You are a professional information extraction assistant specializing in parsing US shipping addresses from unstructured text.

## Task Description
Based on the given input text, accurately extract and generate a JSON object containing the following six fields:
- name: The full name of the recipient.
- street_address: The complete street address, including number, street name, and any apartment or suite number.
- city: The city name.
- state: The full state name (e.g., "California", not "CA").
- zip_code: The 5 or 9-digit ZIP code.
- phone: The complete contact phone number.

## Extraction Rules
1.  **Address Handling**:
    -   Accurately identify the components: street, city, state, and ZIP code.
    -   The `state` field must be the full official name (e.g., "New York", not "NY").
    -   The `street_address` should contain all details before the city, such as "123 Apple Lane, Apt 4B".
2.  **Name Identification**:
    -   Extract the full recipient name.
3.  **Phone Number Handling**:
    -   Extract the complete phone number, preserving its original format.
4.  **ZIP Code**:
    -   Extract the 5-digit or 9-digit (ZIP+4) code.

## Output Format
Strictly adhere to the following JSON format. Do not add any explanatory text or markdown.
{
  "name": "Recipient's Full Name",
  "street_address": "Complete Street Address",
  "city": "City Name",
  "state": "Full State Name",
  "zip_code": "ZIP Code",
  "phone": "Contact Phone Number"
}
"""


# 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()

        # 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"Failed to predict structured data: {e}, Raw data: {raw_data}")
        # 失敗時に空の構造を返します。
        return {
            "name": "",
            "street_address": "",
            "city": "",
            "state": "",
            "zip_code": "",
            "phone": ""
        }


# フェーズ 2: データ変換。
async def convert_data_phase():
    """生データを読み込み、構造化形式を予測し、SFT データとして保存します。"""
    print("=== Phase 2: Starting Data Conversion to SFT Format ===")

    try:
        print("Reading us_recipient_data.json file...")
        with open('us_recipient_data.json', 'r', encoding='utf-8') as f:
            raw_data_list = json.load(f)

        print(f"Successfully read {len(raw_data_list)} records.")
        print("Starting to predict structured data using the extraction model...")

        # 単純で明確なシステムメッセージは、トレーニングおよび推論速度を向上させます。
        system_prompt = "You are an expert assistant for extracting structured JSON from US shipping information. The JSON keys are name, street_address, city, state, zip_code, and 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"Processing record #{index + 1}: {raw_data}")

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

                return conversation

        print(f"Starting conversion to {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"Conversion complete! Processed {len(raw_data_list)} records.")
        print(f"Output file: {output_file}")
        print("=== Phase 2 Complete ===")

    except FileNotFoundError:
        print("Error: us_recipient_data.json not found.")
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"JSON decoding error: {e}")
        sys.exit(1)
    except Exception as e:
        print(f"An error occurred during conversion: {e}")
        sys.exit(1)


# メイン関数。
async def main():
    print("Starting the data processing pipeline...")
    print("This program will execute two phases in sequence:")
    print("1. Generate raw US recipient data.")
    print("2. Predict structured data and convert it to SFT format.")
    print("-" * 50)

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

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

        print("\n" + "=" * 50)
        print("All processes completed successfully!")
        print("Generated files:")
        print("- us_recipient_data.json: Raw, unstructured data list.")
        print("- us_recipient_sft_data.json: SFT-formatted training data.")
        print("=" * 50)
    else:
        print("Data generation phase failed. Terminating.")


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

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