GPT-image-1 API 并发限制：每分钟能生成几张图，停用后怎么迁移

import base64
from openai import OpenAI

# 官方方式
# client = OpenAI(api_key="sk-xxxxxxxxxxxxxx")

# 中转方式 - 仅需修改base_url
client = OpenAI(
    api_key="sk-xxxxxxxxxxxxxx",  # 使用你的中转服务API密钥
    base_url="https://api.laozhang.ai/v1"  # 中转服务endpoint
)

response = client.images.generate(
    model="gpt-image-2.5-flare",  # gpt-image-1 将于 2026-10-23 停用；以服务商实际支持的模型 ID 为准
    prompt="一只可爱的白色暹罗猫坐在窗台上看夕阳",
    size="1024x1024",
    quality="high",
    n=1
)

image_bytes = base64.b64decode(response.data[0].b64_json)
with open("cat.png", "wb") as f:
    f.write(image_bytes)
print("已保存 cat.png")

const OpenAI = require('openai');
const fs = require('fs');

// 创建中转服务客户端
const openai = new OpenAI({
  apiKey: 'sk-xxxxxxxxxxxxxx',
  baseURL: 'https://api.laozhang.ai/v1'
});

async function generateImage() {
  const response = await openai.images.generate({
    model: "gpt-image-2.5-flare",
    prompt: "夜晚的赛博朋克城市街道，霓虹灯闪烁",
    size: "1024x1024",
    quality: "high"
  });

  fs.writeFileSync('city.png', Buffer.from(response.data[0].b64_json, 'base64'));
  console.log('已保存 city.png');
}

import asyncio
import base64
import aiohttp
from typing import List
import time

class BatchImageGenerator:
    def __init__(self, api_key: str, base_url: str, max_concurrent: int = 20):
        self.api_key = api_key
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        
    async def generate_single(self, session: aiohttp.ClientSession, prompt: str, idx: int) -> dict:
        """生成单张图片并保存"""
        async with self.semaphore:  # 控制并发数
            headers = {"Authorization": f"Bearer {self.api_key}"}
            data = {
                "model": "gpt-image-2.5-flare",
                "prompt": prompt,
                "size": "1024x1024",
                "quality": "medium"
            }
            
            try:
                async with session.post(
                    f"{self.base_url}/images/generations",
                    headers=headers,
                    json=data
                ) as response:
                    if response.status == 429:
                        return {"prompt": prompt, "status": "rate_limited"}
                    result = await response.json()
                    path = f"image_{idx}.png"
                    with open(path, "wb") as f:
                        f.write(base64.b64decode(result["data"][0]["b64_json"]))
                    return {"prompt": prompt, "path": path, "status": "success"}
            except Exception as e:
                return {"prompt": prompt, "error": str(e), "status": "failed"}
    
    async def generate_batch(self, prompts: List[str]) -> List[dict]:
        """批量生成图片"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.generate_single(session, p, i) for i, p in enumerate(prompts)]
            return await asyncio.gather(*tasks)

# 使用示例
async def main():
    generator = BatchImageGenerator(
        api_key="sk-xxxxxx",
        base_url="https://api.laozhang.ai/v1",
        max_concurrent=10  # 按所在层级的 IPM 调整
    )
    
    prompts = [f"未来科技风格的产品展示图 #{i}" for i in range(20)]
    
    start_time = time.time()
    results = await generator.generate_batch(prompts)
    elapsed = time.time() - start_time
    
    success_count = sum(1 for r in results if r["status"] == "success")
    print(f"生成完成！成功: {success_count}/{len(prompts)}，总耗时: {elapsed:.2f}秒")

asyncio.run(main())

import heapq
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PriorityItem:
    priority: int
    item: Any = field(compare=False)
    
class PriorityQueue:
    def __init__(self, maxsize: int = 1000):
        self.items = []
        self.maxsize = maxsize
        
    def put(self, item: Any, priority: int = 5):
        if len(self.items) >= self.maxsize:
            # 队列满时，移除优先级最低的项目
            if priority < self.items[0].priority:
                return False
            heapq.heappop(self.items)
        
        heapq.heappush(self.items, PriorityItem(priority, item))
        return True
    
    def get(self):
        if self.items:
            return heapq.heappop(self.items).item
        return None

import hashlib
from typing import Optional

class ImageCache:
    def __init__(self):
        self.cache = {}
        
    def get_cache_key(self, prompt: str) -> str:
        # 标准化提示词
        normalized = prompt.lower().strip()
        # 移除数字等变化部分
        normalized = ''.join(c for c in normalized if c.isalpha() or c.isspace())
        return hashlib.md5(normalized.encode()).hexdigest()
    
    def get(self, prompt: str) -> Optional[str]:
        key = self.get_cache_key(prompt)
        return self.cache.get(key)
    
    def set(self, prompt: str, path: str):
        key = self.get_cache_key(prompt)
        self.cache[key] = path

class MultiLevelCache:
    def __init__(self):
        # L1: 内存缓存（快速，容量小）
        self.memory_cache = LRUCache(maxsize=1000)
        # L2: Redis缓存（较快，容量中等）
        self.redis_client = redis.StrictRedis()
        # L3: OSS存储（慢，容量大）
        self.oss_client = OSSClient()
        
    async def get_image(self, prompt: str) -> Optional[str]:
        # 先检查L1
        url = self.memory_cache.get(prompt)
        if url:
            return url
            
        # 再检查L2
        url = await self.redis_client.get(f"img:{prompt}")
        if url:
            self.memory_cache.put(prompt, url)
            return url
            
        # 最后检查L3
        url = await self.oss_client.get(prompt)
        if url:
            await self.redis_client.set(f"img:{prompt}", url, ex=86400)
            self.memory_cache.put(prompt, url)
            return url
            
        return None

class BudgetController:
    def __init__(self, daily_budget: float):
        self.daily_budget = daily_budget
        # gpt-image-1 1024x1024 单张参考价；迁移到 2.5 后按 usage 实际计算
        self.costs = {"low": 0.011, "medium": 0.042, "high": 0.167}
        
    async def can_generate(self, quality: str):
        today_usage = await self.get_today_usage()
        cost = self.costs[quality]
        
        if today_usage + cost > self.daily_budget:
            # 预算即将超支，尝试降级
            if quality == "high" and today_usage + self.costs["medium"] <= self.daily_budget:
                return "medium"  # 建议使用中等质量
            return False
        
        return True