







pip install aiohttp
import asyncioimport aiohttpfrom typing import List, Dict, Anyimport timeclass NanoBananaBatchGenerator: """Nano Banana Pro 高并发批量生成客户端""" def __init__(self, api_key: str, max_concurrent: int = 10, retries: int = 3): """ 初始化客户端 :param api_key: GrSAI API密钥 :param max_concurrent: 最大并发数,控制请求“洪峰” :param retries: 失败重试次数 """ self.api_key = api_key self.max_concurrent = max_concurrent self.retries = retries # 注意:此处使用Nano Banana Pro的专用端点 self.endpoint = "https://api.grsai.com/v1/draw/nano-banana" self.session = None # aiohttp会话对象,稍后创建
async def __aenter__(self):
"""异步上下文管理器入口,创建可复用的会话和连接池"""
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=60) # 设置总超时
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口,确保会话被关闭"""
if self.session:
await self.session.close()
async def _generate_single_with_retry(self, prompt: str, params: Dict) -> Dict[str, Any]:
"""执行单次生成,内置指数退避重试机制"""
payload = {
"model": "nano-banana-Pro", # 指定模型
"prompt": prompt,
**params # 合并其他参数(如尺寸,参考图等)
}
last_error = None
for attempt in range(self.retries):
try:
async with self.session.post(self.endpoint, json=payload) as response:
if response.status == 200:
data = await response.json()
return {"status": "success", "data": data, "prompt": prompt}
else:
error_text = await response.text()
last_error = f"HTTP {response.status}: {error_text[:200]}"
# 429错误代表请求过快,采用指数退避等待
if response.status == 429:
wait_time = 2 ** (attempt + 1)
print(f"提示词 '{prompt[:30]}...' 触发频率限制,{wait_time}秒后重试...")
await asyncio.sleep(wait_time)
continue
elif response.status >= 500:
# 服务器错误,稍后重试
await asyncio.sleep(1)
continue
else:
# 客户端错误(如参数错误)通常无需重试
break
except (asyncio.TimeoutError, aiohttp.ClientError) as e:
last_error = str(e)
await asyncio.sleep(1) # 网络异常,简单等待后重试
# 所有重试均失败
return {"status": "failed", "error": last_error, "prompt": prompt}
async def generate_batch(self, prompts: List[str], **params) -> List[Dict[str, Any]]:
"""
批量生成主方法
:param prompts: 提示词列表
:param params: 额外API参数,如 aspectRatio="16:9"
:return: 生成结果列表
"""
if not self.session:
raise RuntimeError("请使用异步上下文管理器(async with)来调用此客户端")
# 创建信号量以严格控制最大并发数
semaphore = asyncio.Semaphore(self.max_concurrent)
async def bounded_task(prompt: str) -> Dict[str, Any]:
async with semaphore: # 只有获得“通行证”的任务才能执行
# 在请求间添加微小延迟,平滑流量,避免瞬时峰值
await asyncio.sleep(0.05)
return await self._generate_single_with_retry(prompt, params)
print(f"开始批量生成,总计 {len(prompts)} 个任务,最大并发数 {self.max_concurrent}...")
start_time = time.time()
# 创建所有异步任务对象
tasks = [bounded_task(prompt) for prompt in prompts]
# 并发执行所有任务,并等待它们全部完成
results = await asyncio.gather(*tasks, return_exceptions=True)
elapsed_time = time.time() - start_time print(f"批量生成完成!总耗时 {elapsed_time:.2f} 秒")
# 统一处理结果,将异常转换为格式化的字典
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append({
"status": "error",
"error": str(result),
"prompt": prompts[i]
})
else:
processed_results.append(result)
return processed_results
# ==================== 实战用例:亚马逊电商套图生成 ====================async def main():
"""实战案例:为智能咖啡杯生成亚马逊产品套图"""
API_KEY = "your_grsi_api_key_here" # 请替换为Grsai.com提供的APiKey
# 产品核心描述
product_desc = "a modern smart coffee mug with temperature display"
# 构建亚马逊产品页面所需的多样化提示词列表
prompts = [
f"{product_desc}, on a pure white background, professional product photography, Amazon main image",
f"{product_desc}, on a wooden desk in a cozy home office, lifestyle shot",
f"Close-up of {product_desc}'s LED screen showing 65°C, macro photography",
f"A person's hands holding {product_desc} in a modern kitchen, action shot",
f"{product_desc} shown next to a regular coffee mug for size comparison",
]
# API调用参数
generate_params = {
"aspectRatio": "1:1", # 正方形构图,适合电商平台
"imageSize": "2K", # 可选:输出尺寸(1K, 2K, 4K)
"urls": ["https://example.com/example.png"], # 参考图
}
# 执行批量生成:使用async with自动管理客户端生命周期
async with NanoBananaBatchGenerator(API_KEY, max_concurrent=5) as generator:
results = await generator.generate_batch(prompts, **generate_params)
# 分析并打印结果
success_count = 0
for i, result in enumerate(results):
print(f"\n任务 {i+1}: '{result['prompt'][:50]}...'")
if result["status"] == "success":
success_count += 1
image_data = result.get("data", {})
# 图像URL通常位于 data.results[0].url
if "results" in image_data:
print(f"✅ 生成成功!")
else:
print(f"❌ 失败: {result.get('error')}")
# 运行主函数if __name__ == "__main__":
asyncio.run(main())

payload = {
“model”: “nano-banana-pro”,
“prompt”: “一只猫”,
“aspectRatio”: “1:1”,
“webHook”: “https://your-server.com/api/callback”, # 你的回调地址
“shutProgress”: true # 回调只接收最终结果,可选}
# 使用Flask的简单示例from flask import Flask, request
app = Flask(__name__)@app.route(‘/api/callback’, methods=[‘POST’])def handle_callback():
data = request.json # 根据ID匹配你的任务
task_id = data[‘id’]
if data[‘status’] == ‘succeeded’:
image_url = data[‘results’][0][‘url’]
# 保存URL或触发后续操作
print(f“任务 {task_id} 完成,图片: {image_url}”)
return ‘OK’ # 务必返回成功响应
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70048307/viewspace-3107323/ ,如需转载,请注明出处,否则将追究法律责任。