1. 项目缘起当Python遇上AI绘画最近在折腾一个内容创作的小项目需要批量生成一些特定风格的配图。手动去设计网站找图效率低不说版权和风格统一性都是问题。就在琢磨有没有更“懒”的办法时阿里云新上线的“万象”绘图API进入了我的视线。这玩意儿号称能通过文本描述生成图片听起来不就是为我这种“懒人程序员”量身定做的吗用Python写个脚本把文案丢进去图片就自动生成了这才是真正的“自动化图像创作”。这个想法其实挺直接的利用Python作为胶水语言调用阿里云万象绘图API实现从文本到图像的自动化生成流水线。无论是做自媒体配图、电商商品图、营销海报的初稿还是为游戏、小说生成概念图都能大幅提升效率。整个过程的核心就是理解API怎么用然后用Python把它封装成稳定、好用的工具。听起来简单但里面有不少细节比如怎么处理API的异步响应、怎么设计合理的重试和错误处理机制、怎么根据返回结果调整提示词Prompt等等都是实战中才会遇到的“坑”。接下来我就把自己从零开始搭建这个自动化工具的过程以及踩过的坑、总结的经验完整地分享出来。如果你也对用代码“创造”图像感兴趣或者正想找一个稳定的AI绘图接口来做些自动化的事情那这篇内容应该能给你提供一条清晰的路径。2. 前期准备环境、账号与权限动手写代码之前有几件“脏活累活”必须得先搞定。这就像做饭前得先备好菜和锅一样基础打好了后面才能顺风顺水。2.1 Python环境与必要库首先确保你的电脑上安装了Python。我个人推荐使用Python 3.8或以上的版本兼容性和库支持都更好。检查版本很简单打开终端Windows是CMD或PowerShellMac/Linux是Terminal输入python --version # 或 python3 --version如果显示版本号大于等于3.8那就没问题。如果没有安装或者版本太低去Python官网下载安装包记得安装时勾选“Add Python to PATH”选项。接下来是安装必要的Python库。我们这个项目主要依赖两个库requests用于发送HTTP请求调用APIPillow(PIL) 用于可能的图片处理比如查看、缩放、格式转换。在终端里用pip一键安装pip install requests pillow如果网络环境导致pip安装慢或失败可以尝试使用国内的镜像源比如清华源pip install requests pillow -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后可以在Python交互环境里快速测试一下是否成功import requests, PIL print(requests.__version__)没有报错就说明环境准备妥当了。2.2 阿里云账号与万象绘图API开通这是最关键的一步因为所有的能力都来源于阿里云的万象绘图服务。注册阿里云账号如果你还没有阿里云账号需要先去官网注册一个。这个过程需要手机号和实名认证属于标准流程。开通“通义万相”服务阿里云的AI绘图能力集成在“通义万相”这个产品里。登录阿里云控制台在顶部搜索栏搜索“通义万相”进入产品页面。通常新用户会有一定的免费额度用于体验和测试务必仔细阅读当时的计费说明。获取AccessKey调用API需要身份凭证。在阿里云控制台鼠标移到右上角头像点击“AccessKey管理”。在这里你可以创建一对AccessKey ID和AccessKey Secret。这组密钥非常重要相当于你账号的密码绝对不能泄露或提交到公开的代码仓库如GitHub。我建议立即在代码中通过环境变量来使用它们。找到API文档与Endpoint在通义万相的控制台或文档中心找到“图像生成”或“文生图”相关的API文档。你需要重点关注两个信息API的Endpoint请求地址这是你发送请求的URL。不同区域Region的Endpoint可能不同例如https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis。API的请求参数与格式文档会详细说明需要以什么格式通常是JSON发送请求包含哪些必选和可选参数比如模型名称、提示词、图片尺寸、生成数量等。重要安全提示永远不要将AccessKey硬编码在脚本里。我吃过亏有一次不小心把带密钥的脚本传到了GitHub公共仓库虽然很快删除了但依然心惊胆战。正确的做法是使用环境变量。在终端中临时设置重启后失效export ALIBABA_CLOUD_ACCESS_KEY_IDyour-id export ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-secret或者在项目根目录创建.env文件记得加入.gitignoreALIBABA_CLOUD_ACCESS_KEY_IDyour-id ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-secret然后在Python代码中使用os.getenv()来读取。3. 核心实现从第一行代码到第一张图环境备齐密钥在手现在可以开始编写最核心的API调用部分了。我们的目标是写一个函数输入一段文字描述输出一张生成的图片。3.1 构建基础的API请求函数我们先从最简单的功能开始发送一次请求并保存返回的图片。根据阿里云文档万象绘图的API通常是一个HTTP POST请求请求体是JSON格式并且需要在请求头Header中携带认证信息。下面是一个最基础的实现示例import os import requests import json from pathlib import Path def generate_image_with_dashscope(prompt, modelwanx-v1, size1024*1024, n1, save_dir./generated_images): 调用阿里云万象绘图API生成图片。 参数: prompt (str): 图片描述文本。 model (str): 使用的模型名称默认为wanx-v1。 size (str): 生成图片的尺寸如1024*1024, 720*1280。 n (int): 生成图片的数量默认为1。 save_dir (str): 图片保存的目录。 返回: list: 成功保存的图片文件路径列表。 # 1. 从环境变量读取密钥安全 access_key_id os.getenv(ALIBABA_CLOUD_ACCESS_KEY_ID) access_key_secret os.getenv(ALIBABA_CLOUD_ACCESS_KEY_SECRET) if not access_key_id or not access_key_secret: raise ValueError(请在环境变量中设置 ALIBABA_CLOUD_ACCESS_KEY_ID 和 ALIBABA_CLOUD_ACCESS_KEY_SECRET) # 2. 设置API端点请替换为实际Endpoint api_url https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis # 3. 准备请求头 # 阿里云API通常使用Bearer Token认证Token由AK/SK计算得来但部分简化版API可能直接使用AK。 # 这里假设使用DashScope的认证方式具体请以官方文档为准。 # 一种常见方式是在Header中添加 Authorization: Bearer ${api_key}而api_key可能需要在控制台单独生成。 # 本例假设我们使用AK/SK在Header中进行简单认证仅作示例非真实 headers { Content-Type: application/json, Authorization: fBearer {access_key_id}:{access_key_secret}, # 注意这仅为示例格式真实格式请查文档 X-DashScope-Async: enable # 如果支持异步可添加此头 } # 4. 准备请求体JSON数据 payload { model: model, input: { prompt: prompt }, parameters: { size: size, n: n } } # 5. 发送POST请求 print(f正在生成: {prompt}) response requests.post(api_url, headersheaders, jsonpayload) # 6. 检查响应 if response.status_code ! 200: print(f请求失败状态码: {response.status_code}) print(f响应内容: {response.text}) return [] result response.json() # 7. 处理响应并保存图片 # 注意阿里云API的响应结构需要根据实际文档解析。 # 常见结构同步API直接返回图片URL或Base64数据异步API返回任务ID需再轮询获取结果。 # 此处以假设返回了图片的Base64编码数据为例 saved_paths [] Path(save_dir).mkdir(parentsTrue, exist_okTrue) # 创建保存目录 # 假设响应格式为{output:{task_id:xxx, image_urls:[url1, url2]}} 或直接包含image_data # 这里我们需要根据真实的API响应来写解析逻辑。 # 示例1如果返回的是图片URL列表 if output in result and image_urls in result[output]: image_urls result[output][image_urls] for i, img_url in enumerate(image_urls): # 下载图片 img_response requests.get(img_url) if img_response.status_code 200: file_path Path(save_dir) / fgenerated_{i}_{hash(prompt) % 10000}.png with open(file_path, wb) as f: f.write(img_response.content) saved_paths.append(str(file_path)) print(f图片已保存至: {file_path}) # 示例2如果返回的是Base64编码的图片数据更常见于同步API elif output in result and image_data in result[output]: import base64 image_data_b64 result[output][image_data] image_data base64.b64decode(image_data_b64) file_path Path(save_dir) / fgenerated_{hash(prompt) % 10000}.png with open(file_path, wb) as f: f.write(image_data) saved_paths.append(str(file_path)) print(f图片已保存至: {file_path}) else: print(无法从响应中解析图片数据。响应体, json.dumps(result, indent2)) return saved_paths # 使用示例 if __name__ __main__: # 请确保已设置环境变量 my_prompt 一只戴着眼镜、在电脑前打代码的卡通柴犬赛博朋克风格 generated_images generate_image_with_dashscope(my_prompt, size1024*1024) print(f生成了 {len(generated_images)} 张图片。)代码要点解析安全第一密钥通过os.getenv从环境变量读取这是必须遵守的规范。请求头HeadersContent-Type: application/json是必须的告诉服务器我们发送的是JSON数据。Authorization头的格式是关键且易错点阿里云不同服务、不同版本的认证方式可能有差异。上述代码中的Bearer {access_key_id}:{access_key_secret}只是一个占位示例你必须查阅当前“通义万相”API的最新文档找到正确的认证方式。常见的可能是使用X-DashScope-API-Key头直接传递一个在控制台生成的API Key。请求体Payload结构通常包含model指定模型、input输入如prompt、parameters参数如尺寸、数量、风格等。这些字段名和层级需要严格按文档填写。响应处理这是另一个核心易错点。AI绘图API可能是同步或异步的。同步请求后直接返回图片数据Base64格式或图片URL。处理简单但耗时请求可能会超时。异步请求后立即返回一个task_id你需要用这个ID再去轮询另一个“任务查询”接口直到任务完成并获取结果。这能避免HTTP请求超时是更稳健的方式。上述代码注释中提到了X-DashScope-Async: enable头就是用来启用异步模式的如果API支持。错误处理我们检查了HTTP状态码是否为200。但即使状态码是200业务逻辑也可能出错比如额度不足、参数错误。完整的响应体里通常会有code和message字段来描述业务状态需要进一步解析。3.2 处理异步任务与完善错误处理鉴于图像生成比较耗时阿里云的API很可能默认或推荐使用异步模式。我们需要完善代码来处理异步任务的生命周期。import time def generate_image_async(prompt, modelwanx-v1, size1024*1024, n1, save_dir./generated_images, max_wait_seconds120): 异步方式生成图片包含轮询逻辑。 # ... (前面的密钥读取、headers设置与上面相同) ... # 1. 发起异步生成任务 task_payload { model: model, input: {prompt: prompt}, parameters: {size: size, n: n} } task_response requests.post(api_url, headersheaders, jsontask_payload) if task_response.status_code ! 200: print(f创建任务失败: {task_response.text}) return [] task_result task_response.json() task_id task_result.get(output, {}).get(task_id) if not task_id: print(响应中未找到task_id) return [] print(f任务已创建ID: {task_id}开始轮询结果...) # 2. 轮询任务结果 # 假设查询任务结果的API端点是 {api_url}/tasks/{task_id} task_query_url f{api_url}/tasks/{task_id} start_time time.time() while time.time() - start_time max_wait_seconds: time.sleep(3) # 每3秒查询一次避免请求过于频繁 query_response requests.get(task_query_url, headersheaders) if query_response.status_code ! 200: print(f查询任务状态失败: {query_response.text}) continue query_result query_response.json() task_status query_result.get(output, {}).get(task_status) if task_status SUCCEEDED: print(任务成功) # 解析成功的图片数据并保存逻辑同前 return parse_and_save_images(query_result, prompt, save_dir) elif task_status in [FAILED, CANCELED]: print(f任务失败或取消: {query_result}) return [] elif task_status PENDING or task_status RUNNING: print(f任务状态: {task_status} 继续等待...) else: print(f未知任务状态: {task_status}) break print(f轮询超时{max_wait_seconds}秒任务可能仍在处理中。) return [] def parse_and_save_images(api_result, prompt, save_dir): 解析API结果并保存图片的通用函数 saved_paths [] Path(save_dir).mkdir(parentsTrue, exist_okTrue) # ... (具体的解析逻辑根据实际API响应格式编写参考上一节的示例) ... return saved_paths这个版本增加了健壮性。它先提交任务然后进入一个循环定期查询任务状态直到成功、失败或超时。这对于生成高质量或复杂图片非常必要。4. 工程化与优化打造可靠的生产力工具一个能跑通的脚本和一個可靠的生产工具之间隔着许多工程细节。要让这个自动化流程真正可用我们需要考虑更多。4.1 参数化与配置管理硬编码模型、尺寸等参数不利于复用。我们可以使用Python的配置文件如config.yaml或.json或命令行参数来管理。使用argparse处理命令行参数import argparse def main(): parser argparse.ArgumentParser(description阿里云万象绘图自动化脚本) parser.add_argument(--prompt, typestr, requiredTrue, help图片描述文本) parser.add_argument(--config, typestr, defaultconfig.json, help配置文件路径) parser.add_argument(--output-dir, typestr, default./output, help图片输出目录) args parser.parse_args() # 从配置文件读取其他参数 import json with open(args.config, r) as f: config json.load(f) model config.get(model, wanx-v1) size config.get(size, 1024*1024) generate_image_async(args.prompt, modelmodel, sizesize, save_dirargs.output_dir) if __name__ __main__: main()这样我们就可以通过命令python image_generator.py --prompt 一片星空下的孤山 --output-dir ./my_art来运行脚本了。4.2 实现批量处理与队列自动化创作的核心价值之一是批量处理。我们可以从一个文本文件或CSV文件中读取多条提示词依次或并发地生成图片。import csv from concurrent.futures import ThreadPoolExecutor, as_completed def batch_generate_from_csv(csv_file, output_base_dir./batch_output): 从CSV文件批量生成图片。CSV格式第一列为提示词第二列可选为输出子目录名。 with open(csv_file, r, encodingutf-8) as f: reader csv.reader(f) tasks [] for row in reader: if not row: # 跳过空行 continue prompt row[0].strip() if not prompt: continue # 如果有第二列用作子目录名 sub_dir row[1].strip() if len(row) 1 else default output_dir Path(output_base_dir) / sub_dir tasks.append((prompt, output_dir)) print(f从 {csv_file} 中读取了 {len(tasks)} 个任务。) # 使用线程池并发执行注意API可能有速率限制 max_workers 3 # 并发数不宜过高避免触发API限流 with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_task {executor.submit(generate_image_async, prompt, save_dirstr(output_dir)): (prompt, output_dir) for prompt, output_dir in tasks} for future in as_completed(future_to_task): prompt, output_dir future_to_task[future] try: saved_paths future.result() print(f提示词 {prompt[:30]}... 处理完成生成 {len(saved_paths)} 张图。) except Exception as e: print(f处理提示词 {prompt[:30]}... 时发生错误: {e})这里使用了ThreadPoolExecutor进行简单的并发控制。但务必注意免费或低阶的API套餐通常有严格的QPS每秒查询率限制盲目开大量线程会导致请求被拒绝。设置一个较小的max_workers如2-3是稳妥的做法。更好的做法是实现一个带延迟的任务队列。4.3 增强鲁棒性重试、降级与日志网络请求和远程API调用充满不确定性必须增加重试机制。import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) # 使用tenacity库实现智能重试 retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)), before_sleeplambda retry_state: logger.warning(f第{retry_state.attempt_number}次重试异常: {retry_state.outcome.exception()}) ) def call_api_with_retry(api_url, headers, payload): 带重试机制的API调用 response requests.post(api_url, headersheaders, jsonpayload, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 return response.json()然后在主要的生成函数里用call_api_with_retry替换直接的requests.post调用。这样遇到网络抖动或临时性服务错误时脚本会自动重试大大提高了成功率。此外还应该记录详细的日志包括每次请求的参数、响应时间、是否成功等便于后期排查问题和分析使用情况。4.4 提示词Prompt工程浅谈AI绘图的质量极大程度上取决于提示词。这里分享几个从实战中总结的小技巧具体化“一只猫”和“一只银渐层英国短毛猫碧绿色眼睛在阳光下慵懒地躺在波斯地毯上”的效果天差地别。使用权重部分API支持用(word:weight)或[word]的语法强调或弱化某些元素。例如(masterpiece:1.2), best quality强调高质量。负面提示词Negative Prompt告诉AI你不想要什么。这在万象绘图的参数里可能是negative_prompt。例如加上low quality, blurry, ugly可以过滤掉低质量输出。风格化在提示词中加入风格关键词如digital art, concept art, trending on artstation, 4k或Chinese ink painting style。迭代优化很少有提示词能一次就生成完美图片。通常需要根据第一次的结果调整描述增减细节多次尝试。可以将这个迭代过程也稍作自动化比如生成不同参数组合的图片进行对比。5. 实战踩坑与效能提升指南理论说得再多不如踩一次坑。下面是我在开发和实际使用这个自动化工具过程中遇到的一些典型问题及解决方案。5.1 高频错误码与排查思路调用API时返回的错误信息是你最好的朋友。以下是一些可能遇到的错误及应对400 Bad Request请求参数错误。这是最常见的问题。检查点JSON格式是否正确字段名是否拼写错误比如prompt写成promtsize参数的格式是否是API要求的精确字符串如1024*1024而非1024x1024model名称是否准确例如是wanx-v1还是wanx-1.0行动仔细对照官方API文档逐个参数检查。可以用json.dumps(payload, indent2)打印出完整的请求体进行核对。401 Unauthorized或403 Forbidden认证失败。检查点AccessKey ID和Secret是否正确是否已经过期或被禁用Authorization请求头的格式是否完全符合文档要求如果是用API Key是否放在了正确的Header字段里如X-DashScope-API-Key行动去阿里云控制台重新生成一对Key试试。确保代码中读取环境变量的名字没错。429 Too Many Requests请求频率超限。检查点你是否在短时间内发送了太多请求免费套餐的QPS通常很低如1-2次/秒。行动立即降低并发数。在批量脚本中增加time.sleep(1)之类的延迟。考虑使用更高级的流控策略如令牌桶算法。500 Internal Server Error或502/503/504服务器端错误。检查点这可能是阿里云服务临时问题或者你的请求触发了某些内部错误。行动首先实现重试机制如上节所述。如果重试多次仍失败等待一段时间再试。检查阿里云的服务状态公告。任务一直处于PENDING或RUNNING状态异步任务卡住。检查点任务是否过于复杂服务端队列是否繁忙行动增加轮询的超时时间max_wait_seconds。对于非常重要的任务可以实现一个持久化的任务状态跟踪器即使脚本重启也能继续轮询。5.2 成本控制与额度监控AI绘图是计费服务自动化脚本一旦跑起来如果提示词列表很长可能不知不觉就消耗了大量额度。估算成本在阿里云控制台查看计费规则通常是按生成图片的尺寸和数量计费。在批量运行前先手动生成几张测试了解单张图片的成本再乘以任务数量进行估算。设置预算告警在阿里云费用中心设置预算告警当消费达到一定阈值时通过短信、邮件等方式通知你。代码层面限流在批量脚本中不仅限制并发数还可以增加每日/每月总额度检查。例如维护一个计数器每成功生成一张图片就加1当接近额度上限时自动停止脚本。使用免费额度充分利用新用户的免费额度进行开发和测试。5.3 生成结果的质检与后处理全自动化并不意味着完全放任不管。生成图片的质量需要抽查。自动抽样预览可以在批量脚本中每隔N张图片使用PIL库自动打开一张进行快速预览或者将缩略图保存到一个专门的“预览”目录。from PIL import Image import io def generate_and_preview(prompt, ...): saved_paths generate_image_async(prompt, ...) if saved_paths: # 打开第一张生成的图片 img Image.open(saved_paths[0]) img.thumbnail((400, 400)) # 缩略图 preview_path Path(./previews) / Path(saved_paths[0]).name img.save(preview_path) print(f预览图已保存: {preview_path}) return saved_paths后处理流水线生成的图片可能需要进行统一的后处理比如调整到固定尺寸、添加统一的水印、转换为WebP格式以节省空间等。可以将这些操作封装成函数集成到生成流程的末尾。元数据记录将每次生成的提示词、使用的参数、模型版本、生成时间、图片路径记录到一个日志文件或数据库中。这对于后续分析哪种提示词效果好、优化生成策略至关重要。5.4 应对API变更与版本升级云服务的API和模型更新是常态。你的脚本需要有应对变化的能力。隔离变化点将API的Endpoint、认证方式、请求/响应格式的解析逻辑封装在独立的函数或类中。当API升级时你只需要修改这一个地方。版本化配置在配置文件中注明当前脚本适配的API版本号。例如api_version: 2024-01-01。关注官方动态订阅阿里云相关的技术博客、公告或GitHub仓库及时了解服务更新、废弃通知和新功能发布。走到这一步你已经拥有了一个相当健壮的Python自动化图像创作工具。它不仅能处理单次请求还能进行安全的批量作业具备基本的错误恢复能力并且便于维护和扩展。从“能用”到“好用”这些工程化的思考和实践往往才是提升效率的关键。 
长河君 · 2026/8/2 9:47:23 · 阅读 1.2k Python调用阿里云万象API实现AI绘画自动化：从环境配置到工程化实践
1. 项目缘起当Python遇上AI绘画最近在折腾一个内容创作的小项目需要批量生成一些特定风格的配图。手动去设计网站找图效率低不说版权和风格统一性都是问题。就在琢磨有没有更“懒”的办法时阿里云新上线的“万象”绘图API进入了我的视线。这玩意儿号称能通过文本描述生成图片听起来不就是为我这种“懒人程序员”量身定做的吗用Python写个脚本把文案丢进去图片就自动生成了这才是真正的“自动化图像创作”。这个想法其实挺直接的利用Python作为胶水语言调用阿里云万象绘图API实现从文本到图像的自动化生成流水线。无论是做自媒体配图、电商商品图、营销海报的初稿还是为游戏、小说生成概念图都能大幅提升效率。整个过程的核心就是理解API怎么用然后用Python把它封装成稳定、好用的工具。听起来简单但里面有不少细节比如怎么处理API的异步响应、怎么设计合理的重试和错误处理机制、怎么根据返回结果调整提示词Prompt等等都是实战中才会遇到的“坑”。接下来我就把自己从零开始搭建这个自动化工具的过程以及踩过的坑、总结的经验完整地分享出来。如果你也对用代码“创造”图像感兴趣或者正想找一个稳定的AI绘图接口来做些自动化的事情那这篇内容应该能给你提供一条清晰的路径。2. 前期准备环境、账号与权限动手写代码之前有几件“脏活累活”必须得先搞定。这就像做饭前得先备好菜和锅一样基础打好了后面才能顺风顺水。2.1 Python环境与必要库首先确保你的电脑上安装了Python。我个人推荐使用Python 3.8或以上的版本兼容性和库支持都更好。检查版本很简单打开终端Windows是CMD或PowerShellMac/Linux是Terminal输入python --version # 或 python3 --version如果显示版本号大于等于3.8那就没问题。如果没有安装或者版本太低去Python官网下载安装包记得安装时勾选“Add Python to PATH”选项。接下来是安装必要的Python库。我们这个项目主要依赖两个库requests用于发送HTTP请求调用APIPillow(PIL) 用于可能的图片处理比如查看、缩放、格式转换。在终端里用pip一键安装pip install requests pillow如果网络环境导致pip安装慢或失败可以尝试使用国内的镜像源比如清华源pip install requests pillow -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后可以在Python交互环境里快速测试一下是否成功import requests, PIL print(requests.__version__)没有报错就说明环境准备妥当了。2.2 阿里云账号与万象绘图API开通这是最关键的一步因为所有的能力都来源于阿里云的万象绘图服务。注册阿里云账号如果你还没有阿里云账号需要先去官网注册一个。这个过程需要手机号和实名认证属于标准流程。开通“通义万相”服务阿里云的AI绘图能力集成在“通义万相”这个产品里。登录阿里云控制台在顶部搜索栏搜索“通义万相”进入产品页面。通常新用户会有一定的免费额度用于体验和测试务必仔细阅读当时的计费说明。获取AccessKey调用API需要身份凭证。在阿里云控制台鼠标移到右上角头像点击“AccessKey管理”。在这里你可以创建一对AccessKey ID和AccessKey Secret。这组密钥非常重要相当于你账号的密码绝对不能泄露或提交到公开的代码仓库如GitHub。我建议立即在代码中通过环境变量来使用它们。找到API文档与Endpoint在通义万相的控制台或文档中心找到“图像生成”或“文生图”相关的API文档。你需要重点关注两个信息API的Endpoint请求地址这是你发送请求的URL。不同区域Region的Endpoint可能不同例如https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis。API的请求参数与格式文档会详细说明需要以什么格式通常是JSON发送请求包含哪些必选和可选参数比如模型名称、提示词、图片尺寸、生成数量等。重要安全提示永远不要将AccessKey硬编码在脚本里。我吃过亏有一次不小心把带密钥的脚本传到了GitHub公共仓库虽然很快删除了但依然心惊胆战。正确的做法是使用环境变量。在终端中临时设置重启后失效export ALIBABA_CLOUD_ACCESS_KEY_IDyour-id export ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-secret或者在项目根目录创建.env文件记得加入.gitignoreALIBABA_CLOUD_ACCESS_KEY_IDyour-id ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-secret然后在Python代码中使用os.getenv()来读取。3. 核心实现从第一行代码到第一张图环境备齐密钥在手现在可以开始编写最核心的API调用部分了。我们的目标是写一个函数输入一段文字描述输出一张生成的图片。3.1 构建基础的API请求函数我们先从最简单的功能开始发送一次请求并保存返回的图片。根据阿里云文档万象绘图的API通常是一个HTTP POST请求请求体是JSON格式并且需要在请求头Header中携带认证信息。下面是一个最基础的实现示例import os import requests import json from pathlib import Path def generate_image_with_dashscope(prompt, modelwanx-v1, size1024*1024, n1, save_dir./generated_images): 调用阿里云万象绘图API生成图片。 参数: prompt (str): 图片描述文本。 model (str): 使用的模型名称默认为wanx-v1。 size (str): 生成图片的尺寸如1024*1024, 720*1280。 n (int): 生成图片的数量默认为1。 save_dir (str): 图片保存的目录。 返回: list: 成功保存的图片文件路径列表。 # 1. 从环境变量读取密钥安全 access_key_id os.getenv(ALIBABA_CLOUD_ACCESS_KEY_ID) access_key_secret os.getenv(ALIBABA_CLOUD_ACCESS_KEY_SECRET) if not access_key_id or not access_key_secret: raise ValueError(请在环境变量中设置 ALIBABA_CLOUD_ACCESS_KEY_ID 和 ALIBABA_CLOUD_ACCESS_KEY_SECRET) # 2. 设置API端点请替换为实际Endpoint api_url https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis # 3. 准备请求头 # 阿里云API通常使用Bearer Token认证Token由AK/SK计算得来但部分简化版API可能直接使用AK。 # 这里假设使用DashScope的认证方式具体请以官方文档为准。 # 一种常见方式是在Header中添加 Authorization: Bearer ${api_key}而api_key可能需要在控制台单独生成。 # 本例假设我们使用AK/SK在Header中进行简单认证仅作示例非真实 headers { Content-Type: application/json, Authorization: fBearer {access_key_id}:{access_key_secret}, # 注意这仅为示例格式真实格式请查文档 X-DashScope-Async: enable # 如果支持异步可添加此头 } # 4. 准备请求体JSON数据 payload { model: model, input: { prompt: prompt }, parameters: { size: size, n: n } } # 5. 发送POST请求 print(f正在生成: {prompt}) response requests.post(api_url, headersheaders, jsonpayload) # 6. 检查响应 if response.status_code ! 200: print(f请求失败状态码: {response.status_code}) print(f响应内容: {response.text}) return [] result response.json() # 7. 处理响应并保存图片 # 注意阿里云API的响应结构需要根据实际文档解析。 # 常见结构同步API直接返回图片URL或Base64数据异步API返回任务ID需再轮询获取结果。 # 此处以假设返回了图片的Base64编码数据为例 saved_paths [] Path(save_dir).mkdir(parentsTrue, exist_okTrue) # 创建保存目录 # 假设响应格式为{output:{task_id:xxx, image_urls:[url1, url2]}} 或直接包含image_data # 这里我们需要根据真实的API响应来写解析逻辑。 # 示例1如果返回的是图片URL列表 if output in result and image_urls in result[output]: image_urls result[output][image_urls] for i, img_url in enumerate(image_urls): # 下载图片 img_response requests.get(img_url) if img_response.status_code 200: file_path Path(save_dir) / fgenerated_{i}_{hash(prompt) % 10000}.png with open(file_path, wb) as f: f.write(img_response.content) saved_paths.append(str(file_path)) print(f图片已保存至: {file_path}) # 示例2如果返回的是Base64编码的图片数据更常见于同步API elif output in result and image_data in result[output]: import base64 image_data_b64 result[output][image_data] image_data base64.b64decode(image_data_b64) file_path Path(save_dir) / fgenerated_{hash(prompt) % 10000}.png with open(file_path, wb) as f: f.write(image_data) saved_paths.append(str(file_path)) print(f图片已保存至: {file_path}) else: print(无法从响应中解析图片数据。响应体, json.dumps(result, indent2)) return saved_paths # 使用示例 if __name__ __main__: # 请确保已设置环境变量 my_prompt 一只戴着眼镜、在电脑前打代码的卡通柴犬赛博朋克风格 generated_images generate_image_with_dashscope(my_prompt, size1024*1024) print(f生成了 {len(generated_images)} 张图片。)代码要点解析安全第一密钥通过os.getenv从环境变量读取这是必须遵守的规范。请求头HeadersContent-Type: application/json是必须的告诉服务器我们发送的是JSON数据。Authorization头的格式是关键且易错点阿里云不同服务、不同版本的认证方式可能有差异。上述代码中的Bearer {access_key_id}:{access_key_secret}只是一个占位示例你必须查阅当前“通义万相”API的最新文档找到正确的认证方式。常见的可能是使用X-DashScope-API-Key头直接传递一个在控制台生成的API Key。请求体Payload结构通常包含model指定模型、input输入如prompt、parameters参数如尺寸、数量、风格等。这些字段名和层级需要严格按文档填写。响应处理这是另一个核心易错点。AI绘图API可能是同步或异步的。同步请求后直接返回图片数据Base64格式或图片URL。处理简单但耗时请求可能会超时。异步请求后立即返回一个task_id你需要用这个ID再去轮询另一个“任务查询”接口直到任务完成并获取结果。这能避免HTTP请求超时是更稳健的方式。上述代码注释中提到了X-DashScope-Async: enable头就是用来启用异步模式的如果API支持。错误处理我们检查了HTTP状态码是否为200。但即使状态码是200业务逻辑也可能出错比如额度不足、参数错误。完整的响应体里通常会有code和message字段来描述业务状态需要进一步解析。3.2 处理异步任务与完善错误处理鉴于图像生成比较耗时阿里云的API很可能默认或推荐使用异步模式。我们需要完善代码来处理异步任务的生命周期。import time def generate_image_async(prompt, modelwanx-v1, size1024*1024, n1, save_dir./generated_images, max_wait_seconds120): 异步方式生成图片包含轮询逻辑。 # ... (前面的密钥读取、headers设置与上面相同) ... # 1. 发起异步生成任务 task_payload { model: model, input: {prompt: prompt}, parameters: {size: size, n: n} } task_response requests.post(api_url, headersheaders, jsontask_payload) if task_response.status_code ! 200: print(f创建任务失败: {task_response.text}) return [] task_result task_response.json() task_id task_result.get(output, {}).get(task_id) if not task_id: print(响应中未找到task_id) return [] print(f任务已创建ID: {task_id}开始轮询结果...) # 2. 轮询任务结果 # 假设查询任务结果的API端点是 {api_url}/tasks/{task_id} task_query_url f{api_url}/tasks/{task_id} start_time time.time() while time.time() - start_time max_wait_seconds: time.sleep(3) # 每3秒查询一次避免请求过于频繁 query_response requests.get(task_query_url, headersheaders) if query_response.status_code ! 200: print(f查询任务状态失败: {query_response.text}) continue query_result query_response.json() task_status query_result.get(output, {}).get(task_status) if task_status SUCCEEDED: print(任务成功) # 解析成功的图片数据并保存逻辑同前 return parse_and_save_images(query_result, prompt, save_dir) elif task_status in [FAILED, CANCELED]: print(f任务失败或取消: {query_result}) return [] elif task_status PENDING or task_status RUNNING: print(f任务状态: {task_status} 继续等待...) else: print(f未知任务状态: {task_status}) break print(f轮询超时{max_wait_seconds}秒任务可能仍在处理中。) return [] def parse_and_save_images(api_result, prompt, save_dir): 解析API结果并保存图片的通用函数 saved_paths [] Path(save_dir).mkdir(parentsTrue, exist_okTrue) # ... (具体的解析逻辑根据实际API响应格式编写参考上一节的示例) ... return saved_paths这个版本增加了健壮性。它先提交任务然后进入一个循环定期查询任务状态直到成功、失败或超时。这对于生成高质量或复杂图片非常必要。4. 工程化与优化打造可靠的生产力工具一个能跑通的脚本和一個可靠的生产工具之间隔着许多工程细节。要让这个自动化流程真正可用我们需要考虑更多。4.1 参数化与配置管理硬编码模型、尺寸等参数不利于复用。我们可以使用Python的配置文件如config.yaml或.json或命令行参数来管理。使用argparse处理命令行参数import argparse def main(): parser argparse.ArgumentParser(description阿里云万象绘图自动化脚本) parser.add_argument(--prompt, typestr, requiredTrue, help图片描述文本) parser.add_argument(--config, typestr, defaultconfig.json, help配置文件路径) parser.add_argument(--output-dir, typestr, default./output, help图片输出目录) args parser.parse_args() # 从配置文件读取其他参数 import json with open(args.config, r) as f: config json.load(f) model config.get(model, wanx-v1) size config.get(size, 1024*1024) generate_image_async(args.prompt, modelmodel, sizesize, save_dirargs.output_dir) if __name__ __main__: main()这样我们就可以通过命令python image_generator.py --prompt 一片星空下的孤山 --output-dir ./my_art来运行脚本了。4.2 实现批量处理与队列自动化创作的核心价值之一是批量处理。我们可以从一个文本文件或CSV文件中读取多条提示词依次或并发地生成图片。import csv from concurrent.futures import ThreadPoolExecutor, as_completed def batch_generate_from_csv(csv_file, output_base_dir./batch_output): 从CSV文件批量生成图片。CSV格式第一列为提示词第二列可选为输出子目录名。 with open(csv_file, r, encodingutf-8) as f: reader csv.reader(f) tasks [] for row in reader: if not row: # 跳过空行 continue prompt row[0].strip() if not prompt: continue # 如果有第二列用作子目录名 sub_dir row[1].strip() if len(row) 1 else default output_dir Path(output_base_dir) / sub_dir tasks.append((prompt, output_dir)) print(f从 {csv_file} 中读取了 {len(tasks)} 个任务。) # 使用线程池并发执行注意API可能有速率限制 max_workers 3 # 并发数不宜过高避免触发API限流 with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_task {executor.submit(generate_image_async, prompt, save_dirstr(output_dir)): (prompt, output_dir) for prompt, output_dir in tasks} for future in as_completed(future_to_task): prompt, output_dir future_to_task[future] try: saved_paths future.result() print(f提示词 {prompt[:30]}... 处理完成生成 {len(saved_paths)} 张图。) except Exception as e: print(f处理提示词 {prompt[:30]}... 时发生错误: {e})这里使用了ThreadPoolExecutor进行简单的并发控制。但务必注意免费或低阶的API套餐通常有严格的QPS每秒查询率限制盲目开大量线程会导致请求被拒绝。设置一个较小的max_workers如2-3是稳妥的做法。更好的做法是实现一个带延迟的任务队列。4.3 增强鲁棒性重试、降级与日志网络请求和远程API调用充满不确定性必须增加重试机制。import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) # 使用tenacity库实现智能重试 retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)), before_sleeplambda retry_state: logger.warning(f第{retry_state.attempt_number}次重试异常: {retry_state.outcome.exception()}) ) def call_api_with_retry(api_url, headers, payload): 带重试机制的API调用 response requests.post(api_url, headersheaders, jsonpayload, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 return response.json()然后在主要的生成函数里用call_api_with_retry替换直接的requests.post调用。这样遇到网络抖动或临时性服务错误时脚本会自动重试大大提高了成功率。此外还应该记录详细的日志包括每次请求的参数、响应时间、是否成功等便于后期排查问题和分析使用情况。4.4 提示词Prompt工程浅谈AI绘图的质量极大程度上取决于提示词。这里分享几个从实战中总结的小技巧具体化“一只猫”和“一只银渐层英国短毛猫碧绿色眼睛在阳光下慵懒地躺在波斯地毯上”的效果天差地别。使用权重部分API支持用(word:weight)或[word]的语法强调或弱化某些元素。例如(masterpiece:1.2), best quality强调高质量。负面提示词Negative Prompt告诉AI你不想要什么。这在万象绘图的参数里可能是negative_prompt。例如加上low quality, blurry, ugly可以过滤掉低质量输出。风格化在提示词中加入风格关键词如digital art, concept art, trending on artstation, 4k或Chinese ink painting style。迭代优化很少有提示词能一次就生成完美图片。通常需要根据第一次的结果调整描述增减细节多次尝试。可以将这个迭代过程也稍作自动化比如生成不同参数组合的图片进行对比。5. 实战踩坑与效能提升指南理论说得再多不如踩一次坑。下面是我在开发和实际使用这个自动化工具过程中遇到的一些典型问题及解决方案。5.1 高频错误码与排查思路调用API时返回的错误信息是你最好的朋友。以下是一些可能遇到的错误及应对400 Bad Request请求参数错误。这是最常见的问题。检查点JSON格式是否正确字段名是否拼写错误比如prompt写成promtsize参数的格式是否是API要求的精确字符串如1024*1024而非1024x1024model名称是否准确例如是wanx-v1还是wanx-1.0行动仔细对照官方API文档逐个参数检查。可以用json.dumps(payload, indent2)打印出完整的请求体进行核对。401 Unauthorized或403 Forbidden认证失败。检查点AccessKey ID和Secret是否正确是否已经过期或被禁用Authorization请求头的格式是否完全符合文档要求如果是用API Key是否放在了正确的Header字段里如X-DashScope-API-Key行动去阿里云控制台重新生成一对Key试试。确保代码中读取环境变量的名字没错。429 Too Many Requests请求频率超限。检查点你是否在短时间内发送了太多请求免费套餐的QPS通常很低如1-2次/秒。行动立即降低并发数。在批量脚本中增加time.sleep(1)之类的延迟。考虑使用更高级的流控策略如令牌桶算法。500 Internal Server Error或502/503/504服务器端错误。检查点这可能是阿里云服务临时问题或者你的请求触发了某些内部错误。行动首先实现重试机制如上节所述。如果重试多次仍失败等待一段时间再试。检查阿里云的服务状态公告。任务一直处于PENDING或RUNNING状态异步任务卡住。检查点任务是否过于复杂服务端队列是否繁忙行动增加轮询的超时时间max_wait_seconds。对于非常重要的任务可以实现一个持久化的任务状态跟踪器即使脚本重启也能继续轮询。5.2 成本控制与额度监控AI绘图是计费服务自动化脚本一旦跑起来如果提示词列表很长可能不知不觉就消耗了大量额度。估算成本在阿里云控制台查看计费规则通常是按生成图片的尺寸和数量计费。在批量运行前先手动生成几张测试了解单张图片的成本再乘以任务数量进行估算。设置预算告警在阿里云费用中心设置预算告警当消费达到一定阈值时通过短信、邮件等方式通知你。代码层面限流在批量脚本中不仅限制并发数还可以增加每日/每月总额度检查。例如维护一个计数器每成功生成一张图片就加1当接近额度上限时自动停止脚本。使用免费额度充分利用新用户的免费额度进行开发和测试。5.3 生成结果的质检与后处理全自动化并不意味着完全放任不管。生成图片的质量需要抽查。自动抽样预览可以在批量脚本中每隔N张图片使用PIL库自动打开一张进行快速预览或者将缩略图保存到一个专门的“预览”目录。from PIL import Image import io def generate_and_preview(prompt, ...): saved_paths generate_image_async(prompt, ...) if saved_paths: # 打开第一张生成的图片 img Image.open(saved_paths[0]) img.thumbnail((400, 400)) # 缩略图 preview_path Path(./previews) / Path(saved_paths[0]).name img.save(preview_path) print(f预览图已保存: {preview_path}) return saved_paths后处理流水线生成的图片可能需要进行统一的后处理比如调整到固定尺寸、添加统一的水印、转换为WebP格式以节省空间等。可以将这些操作封装成函数集成到生成流程的末尾。元数据记录将每次生成的提示词、使用的参数、模型版本、生成时间、图片路径记录到一个日志文件或数据库中。这对于后续分析哪种提示词效果好、优化生成策略至关重要。5.4 应对API变更与版本升级云服务的API和模型更新是常态。你的脚本需要有应对变化的能力。隔离变化点将API的Endpoint、认证方式、请求/响应格式的解析逻辑封装在独立的函数或类中。当API升级时你只需要修改这一个地方。版本化配置在配置文件中注明当前脚本适配的API版本号。例如api_version: 2024-01-01。关注官方动态订阅阿里云相关的技术博客、公告或GitHub仓库及时了解服务更新、废弃通知和新功能发布。走到这一步你已经拥有了一个相当健壮的Python自动化图像创作工具。它不仅能处理单次请求还能进行安全的批量作业具备基本的错误恢复能力并且便于维护和扩展。从“能用”到“好用”这些工程化的思考和实践往往才是提升效率的关键。