Model Gallery quick start

To run the following code, you need to create an Alibaba Cloud Model Studio API key. The code uses qwen-plus-latest to generate business data and qwen3-235b-a22b for labeling.

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

# Create an asynchronous client instance.
client = AsyncOpenAI(
    # If you have not configured environment variables, replace the following line with your Model Studio API key: api_key="sk-xxx",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# List of Chinese provinces.
provinces = [
    "Beijing", "Tianjin", "Hebei", "Shanxi", "Inner Mongolia", "Liaoning", "Jilin", "Heilongjiang",
    "Shanghai", "Jiangsu", "Zhejiang", "Anhui", "Fujian", "Jiangxi", "Shandong", "Henan",
    "Hubei", "Hunan", "Guangdong", "Guangxi", "Hainan", "Chongqing", "Sichuan", "Guizhou",
    "Yunnan", "Tibet", "Shaanxi", "Gansu", "Qinghai", "Ningxia", "Xinjiang"
]

# Recipient writing templates.
recipient_templates = [
    "Recipient {name}", "Recipient: {name}", "Recipient is {name}", "To: {name}",
    "Recipient is {name}", "{name}", "Name: {name}", "Name {name}",
    "Contact {name}", "Contact: {name}", "Receiver {name}", "Receiver: {name}",
    "Consignee {name}", "Consignee: {name}", "Send to {name}", "To {name}",
    "Recipient {name}", "Recipient: {name}", "Receiver {name}", "Receiver: {name}"
]

# Phone number writing templates.
phone_templates = [
    "tel: {phone}", "tel:{phone}", "mobile: {phone}", "mobile:{phone}",
    "Mobile number {phone}", "Mobile number: {phone}", "Mobile: {phone}", "Mobile {phone}",
    "Phone: {phone}", "Phone {phone}", "Contact number {phone}", "Contact number: {phone}",
    "Number: {phone}", "Number {phone}", "TEL: {phone}", "MOBILE: {phone}",
    "contact: {phone}", "phone: {phone}", "{phone}", "call: {phone}",
    "Contact method {phone}", "Contact method: {phone}", "Phone number {phone}", "Phone number: {phone}",
    "Mobile No. {phone}", "Mobile No.: {phone}", "Phone number is {phone}", "Contact number is {phone}"
]

# Generate a virtual mobile number that starts with 2 to avoid overlapping with real numbers.
def generate_mobile():
    prefixes = ['200', '201', '202', '203', '204', '205', '206', '207', '208', '209',
               '210', '211', '212', '213', '214', '215', '216', '217', '218', '219',
               '220', '221', '222', '223', '224', '225', '226', '227', '228', '229',
               '230', '231', '232', '233', '234', '235', '236', '237', '238', '239']
    return random.choice(prefixes) + ''.join([str(random.randint(0, 9)) for _ in range(8)])

# Generate a landline number.
def generate_landline():
    area_codes = ['010', '021', '022', '023', '024', '025', '027', '028', '029', '0311', '0351', '0431', '0451']
    area_code = random.choice(area_codes)
    number = ''.join([str(random.randint(0, 9)) for _ in range(random.choice([7, 8]))])
    return f"{area_code}-{number}"

# Use a large model to generate recipient and address information.
async def generate_recipient_and_address_by_llm(province: str):
    """Uses a large model to generate the recipient's name and address information for a specified province."""
    prompt = f"""Please generate recipient information for {province}, including the following:
1. A real Chinese name. It can be a common name or a less common one for diversity.
2. A city name within that province.
3. An administrative region name within that city, such as a district or county.
4. A specific street address, such as a road name and house number, residential area name and building number, or commercial building and floor, which should be realistic.

Please return the information in JSON format:
{{"name": "Recipient Name", "city": "City Name", "district": "Administrative Region Name", "specific_location": "Specific Address"}}

Do not include any other content. Return only the JSON. The names should be diverse, not just common ones like Zhang San or Li Si."""

    try:
        response = await client.chat.completions.create(
            messages=[{"role": "user", "content": prompt}],
            model="qwen-plus-latest",
            temperature=1.7,  # Increase the temperature to make the names more diverse.
        )
        
        result = response.choices[0].message.content.strip()
        # Clean up possible markdown code block markers.
        if result.startswith('```'):
            result = result.split('\n', 1)[1]
        if result.endswith('```'):
            result = result.rsplit('\n', 1)[0]
        
        # Try to parse the JSON.
        info = json.loads(result)
        return info
    except Exception as e:
        print(f"Failed to generate recipient and address: {e}. Using a backup plan.")
        # Backup plan.
        backup_names = ["Wang Jianjun", "Li Chunyan", "Zhang Zhihua", "Chen Meiling", "Liu Deqiang", "Zhao Minhui", "Sun Wenbo", "Zhou Xiaoli"]
        return {
            "name": random.choice(backup_names),
            "city": f"{province.replace('Province', '').replace('City', '').replace('Autonomous Region', '')} City",
            "district": "Municipal District", 
            "specific_location": f"Renmin Road No. {random.randint(1, 999)}"
        }

# Generate a record.
async def generate_record():
    # Randomly select a province.
    province = random.choice(provinces)
    
    # Use a large model to generate recipient and address information.
    info = await generate_recipient_and_address_by_llm(province)
    
    # Generate the recipient information format.
    recipient = random.choice(recipient_templates).format(name=info['name'])
    
    # Generate a phone number. There is a 70% chance of a mobile number and a 30% chance of a landline number.
    if random.random() < 0.7:
        phone = generate_mobile()
    else:
        phone = generate_landline()
    
    phone_info = random.choice(phone_templates).format(phone=phone)
    
    # Assemble the address.
    full_address = f"{info['city']}{info['district']}{info['specific_location']}"
    
    # Assemble the data.
    components = [recipient, phone_info, full_address]
    
    # Randomly shuffle the order.
    random.shuffle(components)
    
    # Randomly select a separator.
    separators = [' ', ',', ',', ';', ';', ':', ':', ',', '|', '\t', '', '  ', ' | ', ' , ', ' ; ', '/']
    separator = random.choice(separators)
    
    # Merge the data.
    if separator == '':
        # No separator.
        combined_data = ''.join(components)
    else:
        combined_data = separator.join(components)    
    return combined_data

# Generate data in batches.
async def generate_batch_data(count: int) -> List[str]:
    """Generates a specified amount of data."""
    print(f"Starting to generate {count} data entries...")
    data = []
    
    # Use a semaphore to control the number of concurrent requests. QPM=1500, set to 20 concurrent requests.
    semaphore = asyncio.Semaphore(20)
    
    async def generate_single_record(index):
        async with semaphore:
            record = await generate_record()
            print(f"Generating data entry {index+1}: {record}")
            return record
    
    # Generate data concurrently.
    tasks = [generate_single_record(i) for i in range(count)]
    data = await asyncio.gather(*tasks)
    
    return data

# Save data to a file.
def save_data(data: List[str], filename: str = "recipient_data.json"):
    """Saves data to a JSON file."""
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"Data saved to {filename}")

# Data generation phase.
async def produce_data_phase():
    print("=== Phase 1: Start generating recipient data ===")
    
    # Generate 2,000 data entries.
    batch_size = 2000
    data = await generate_batch_data(batch_size)
    
    # Save the data.
    save_data(data)
    
    print(f"\nGenerated a total of {len(data)} data entries")
    print("\nSample data:")
    for i, record in enumerate(data[:3]):  # Display the first 3 entries as examples.
        print(f"{i+1}. Raw data: {record}")
        print()
    
    print("=== Phase 1 complete ===\n")
    return True

def get_system_prompt():
    """Returns the system prompt."""
    return """You are a professional information extraction assistant that specializes in extracting structured recipient information from Chinese text.

## Task description
Based on the input text, accurately extract and generate a JSON-formatted output that contains the following six fields:
- province: The province, municipality, or autonomous region. The full official name is required, such as "河南省", "上海市", or "新疆维吾尔自治区".
- city: The city name, which must include "市", such as "郑州市" or "西安市".
- district: The district or county name, which must include "区" or "县", such as "金水区" or "雁塔区".
- specific_location: The detailed address, including street, house number, residential area, and building.
- name: The full name of the recipient in Chinese.
- phone: The complete phone number, including the area code.

## Extraction rules
1. **Address information processing**:
   - Accurately identify the hierarchical relationship between the province, city, and district.
   - Use the full official name for the province, such as "河南省" instead of "河南".
   - For municipalities, the values of the province and city fields must be the same, such as "上海市".
   - The specific_location field must contain the detailed street address, residential area name, and building number.

2. **Name identification**:
   - Accurately extract the full Chinese name, including compound surnames.
   - Include names of ethnic minorities.

3. **Phone number processing**:
   - Extract the complete phone number and keep its original format.

## Output format
Strictly follow the JSON format below. Do not add any explanatory text.
{
  "province": "Province Name",
  "city": "City Name", 
  "district": "District Name",
  "specific_location": "Detailed Address",
  "name": "Recipient Name",
  "phone": "Phone Number"
}"""

# Use a large model to predict structured data.
async def predict_structured_data(raw_data: str):
    """Uses the qwen3-235b-a22b model to predict structured data."""
    system_prompt = get_system_prompt()
    
    try:
        response = await client.chat.completions.create(
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": raw_data}
            ],
            model="qwen3-235b-a22b",
            temperature=0.1,  # Lower the temperature to improve prediction accuracy.
            response_format={"type": "json_object"},
            extra_body={"enable_thinking":False}
        )
        
        result = response.choices[0].message.content.strip()
        
        # Clean up possible markdown code block markers.
        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]
        
        # Try to parse the JSON.
        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 empty structured data as a backup.
        return {
            "province": "",
            "city": "",
            "district": "",
            "specific_location": "",
            "name": "",
            "phone": ""
        }

# Data conversion phase.
async def convert_data_phase():
    """Converts the data format and uses a large model to predict structured data."""
    print("=== Phase 2: Start converting data format ===")
    
    try:
        print("Start reading the recipient_data.json file...")
        
        # Read the raw data.
        with open('recipient_data.json', 'r', encoding='utf-8') as f:
            raw_data_list = json.load(f)
        
        print(f"Successfully read the data. A total of {len(raw_data_list)} records")
        print("Start using the qwen3-235b-a22b model to predict structured data...")
        # Using a simple and clear system message helps improve training and inference speed.
        system_prompt = "You are a professional information extraction assistant that specializes in extracting JSON information of recipients from Chinese text. The keys to include are province, city, district, specific_location (detailed information such as street, house number, residential area, and building), name (recipient's name), and phone (contact number). The input is as follows:" 
        output_file = 'recipient_sft_data.json'
        
        # Use a semaphore to control the number of concurrent requests.
        semaphore = asyncio.Semaphore(10)
        
        async def process_single_item(index, raw_data):
            async with semaphore:
                # Use a large model to predict structured data.
                structured_data = await predict_structured_data(raw_data)
                print(f"Processing data entry {index+1}: {raw_data}")

                conversation = {
                    "instruction": system_prompt + raw_data,
                    "output": json.dumps(structured_data, ensure_ascii=False)
                }
            
                return conversation
        
        print(f"Start converting data to {output_file}...")
        
        # Process all data concurrently.
        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. A total of {len(raw_data_list)} records processed")
        print(f"Output file: {output_file}")
        print("=== Phase 2 complete ===")
        
    except FileNotFoundError:
        print("Error: The recipient_data.json file was not found.")
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"JSON parsing error: {e}")
        sys.exit(1)
    except Exception as e:
        print(f"An error occurred during conversion: {e}")
        sys.exit(1)

# Main function.
async def main():
    print("Starting the merged data processing flow...")
    print("This program will execute two phases in sequence:")
    print("1. Generate raw recipient data.")
    print("2. Use the qwen3-235b-a22b model to predict structured data and convert it to SFT training format.")
    print("-" * 50)
    
    # Phase 1: Generate data.
    success = await produce_data_phase()
    
    if success:
        # Phase 2: Convert data.
        await convert_data_phase()
        
        print("\n" + "=" * 50)
        print("All processes are complete.")
        print("Generated files:")
        print("- recipient_data.json: Raw data list")
        print("- recipient_sft_data.json: SFT training format data")
        print("=" * 50)
    else:
        print("Data generation phase failed. Execution terminated.")

if __name__ == '__main__':
    # Set the event loop policy.
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # Run the main coroutine.
    asyncio.run(main(), debug=False)