Model Gallery quick start

This sample code calls the large model service in Alibaba Cloud Model Studio. You need a Model Studio API key. The code uses qwen-plus-latest to generate business data and qwen3-235b-a22b to create structured labels for it.

# -*- 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 the environment variable is not configured, 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 = [
    "北京市", "天津市", "河北省", "山西省", "内蒙古自治区", "辽宁省", "吉林省", "黑龙江省",
    "上海市", "江苏省", "浙江省", "安徽省", "福建省", "江西省", "山东省", "河南省",
    "湖北省", "湖南省", "广东省", "广西壮族自治区", "海南省", "重庆市", "四川省", "贵州省",
    "云南省", "西藏自治区", "陕西省", "甘肃省", "青海省", "宁夏回族自治区", "新疆维吾尔自治区"
]

# Recipient templates.
recipient_templates = [
    "收件人{name}", "收件人：{name}", "收件人是{name}", "收件:{name}",
    "收件人为{name}", "{name}", "姓名：{name}", "姓名{name}",
    "联系人{name}", "联系人：{name}", "接收人{name}", "接收人：{name}",
    "收货人{name}", "收货人：{name}", "寄给{name}", "给{name}",
    "收件者{name}", "收件者：{name}", "接收者{name}", "接收者：{name}"
]

# Phone number templates.
phone_templates = [
    "tel：{phone}", "tel:{phone}", "mobile：{phone}", "mobile:{phone}",
    "手机号码{phone}", "手机号码：{phone}", "手机：{phone}", "手机{phone}",
    "电话：{phone}", "电话{phone}", "联系电话{phone}", "联系电话：{phone}",
    "号码：{phone}", "号码{phone}", "TEL：{phone}", "MOBILE：{phone}",
    "contact：{phone}", "phone：{phone}", "{phone}", "call：{phone}",
    "联系方式{phone}", "联系方式：{phone}", "电话号码{phone}", "电话号码：{phone}",
    "手机号{phone}", "手机号：{phone}", "电话号码是{phone}", "联系电话是{phone}"
]

# Generate a synthetic mobile number (starting with 2 to avoid conflicts 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):
    """Use a large model to generate a recipient's name and address for a specified province."""
    prompt = f"""Please generate recipient information for {province}, including:
1. A realistic Chinese name (can be common or less common, for diversity).
2. A city name within that province.
3. An administrative district name within that city (such as a district or county).
4. A specific street address (such as road name + number, community name + building number, or commercial building + floor, must be realistic).

Return the output directly in JSON format:
{{"name": "收件人姓名", "city": "城市名", "district": "行政区名", "specific_location": "具体地址"}}

Do not include any other content. Return only the JSON. Names should be diverse and not always 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 names more diverse.
        )
        
        result = response.choices[0].message.content.strip()
        # Clean up potential 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 fallback.")
        # Fallback.
        backup_names = ["王建军", "李春燕", "张志华", "陈美玲", "刘德强", "赵敏慧", "孙文博", "周晓丽"]
        return {
            "name": random.choice(backup_names),
            "city": f"{province.replace('省', '').replace('市', '').replace('自治区', '')}市",
            "district": "市辖区", 
            "specific_location": f"人民路{random.randint(1, 999)}号"
        }

# Generate one 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 (70% chance of a mobile number, 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 a batch of data.
async def generate_batch_data(count: int) -> List[str]:
    """Generates a specified number of data records."""
    print(f"Starting to generate {count} data records...")
    data = []
    
    # Use a semaphore to control concurrency. With a QPM of 1500, set concurrency to 20.
    semaphore = asyncio.Semaphore(20)
    
    async def generate_single_record(index):
        async with semaphore:
            record = await generate_record()
            print(f"Generated record #{index+1}: {record}")
            return record
    
    # Concurrently generate data.
    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 records.
    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 records")
    print("\nSample data:")
    for i, record in enumerate(data[:3]):  # Show the first 3 records 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
From the given input text, extract the following six fields and generate a JSON output:
- province: The province, municipality, or autonomous region. Use the full official name, 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 adhere to the JSON format below. Do not add any explanatory text.
{
  "province": "省份名称",
  "city": "城市名称", 
  "district": "区县名称",
  "specific_location": "详细地址",
  "name": "收件人姓名",
  "phone": "联系电话"
}"""

# 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 potential 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 fallback.
        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...")
        # A simple and clear system message improves training and inference speed.
        system_prompt = "你是一个专业的信息抽取助手，专门负责从中文文本中提取收件人的JSON信息，包含的Key有province（省份）、city（城市名称）、district（区县名称）、specific_location（街道、门牌号、小区、楼栋等详细信息）、name（收件人姓名）、phone（联系电话） 现在输入如下:" 
        output_file = 'recipient_sft_data.json'
        
        # Use a semaphore to control concurrency.
        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 record #{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! {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 combined data processing pipeline...")
    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 completed successfully!")
        print("Generated files:")
        print("- recipient_data.json: Raw data list")
        print("- recipient_sft_data.json: SFT-formatted training data")
        print("=" * 50)
    else:
        print("Data generation phase failed. Terminating.")

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)