Claude Code图片生成API开发指南：从环境配置到批量处理

1. 先搞清楚这个 API 到底能做什么如果你正在找一种能通过代码直接生成图片的方法Claude Code 的图片生成 API 值得先了解清楚。这类功能最怕的就是文档不全、环境复杂、跑通一次要折腾半天。我一般会先看它到底解决的是通用图片生成还是特定场景的配图需求。从标题和搜索材料看这应该是 Claude Code 工具集里的一个子功能专门用来处理图片生成任务。和常见的文字生成图片工具不同它更偏向开发者集成需要你通过代码调用 API 接口来完成生成。这种方式的优势是能嵌入到自己的应用或自动化流程里但前提是得先搞清楚输入输出格式、资源要求和常见坑点。实际使用时最关键的不是功能列表有多长而是能不能在你的开发环境里稳定跑起来。我建议先确认几个基础问题是否需要额外申请 API 密钥、支持哪些图片格式输入、输出分辨率范围、单次生成耗时大概多少、有没有并发限制。这些信息决定了你能不能直接拿来用还是需要额外做适配。2. 环境准备和前置检查在开始写代码调用 API 之前得先把环境理顺。很多调用失败的问题其实出在环境配置环节。2.1 确认 Claude Code 基础环境Claude Code 本身可能有多种安装方式桌面版、命令行工具、IDE 插件或者纯 API 服务。你需要先确认自己用的是哪种形式。如果是桌面版通常已经内置了基础环境如果是命令行或插件形式可能需要单独配置 API 端点。我一般会按这个顺序检查先运行claude --version或类似命令确认 Claude Code 主程序能正常响应查看帮助文档里有没有图片生成相关的子命令或参数确认当前网络环境能访问所需的 API 服务有些服务可能需要特定网络条件如果还没有安装 Claude Code建议先走官方推荐的安装流程。搜索材料里提到了 Windows、Ubuntu 等不同系统的安装方法选择和你开发环境匹配的那个。安装完成后不要急着调用复杂功能先用一个最简单的文本生成任务测试基础连通性。2.2 API 密钥和权限配置图片生成 API 通常需要认证。从搜索材料看可能会遇到api error: 402 insufficient balance余额不足或login failed. check api token令牌错误这类问题。这说明调用前需要先获取有效的 API 密钥并正确配置。具体步骤一般是在 Claude Code 的控制台或官网申请 API 密钥在代码或配置文件中设置这个密钥测试密钥是否有效密钥的配置方式因环境而异。如果是命令行环境可能需要在~/.claude/config这样的配置文件中设置如果是代码调用通常通过环境变量或直接写在初始化参数里。我建议先用环境变量方式这样更安全也更灵活export CLAUDE_API_KEYyour_actual_key_here然后写一个最简单的验证脚本来测试密钥是否有效import os import requests api_key os.getenv(CLAUDE_API_KEY) if not api_key: print(请先设置 CLAUDE_API_KEY 环境变量) exit(1) # 这里应该是实际的 API 端点需要查看最新文档 response requests.get(https://api.claude.com/v1/models, headers{Authorization: fBearer {api_key}}) print(fAPI 连接测试: {response.status_code})2.3 依赖包和网络条件图片生成通常需要额外的依赖库。除了基础的 requests 库可能还需要处理图像格式的 Pillow 库、处理异步任务的 asyncio 等。先创建一个干净的虚拟环境然后按官方文档安装推荐版本的依赖包。网络条件也很关键。图片生成涉及的数据传输量比纯文本大需要稳定的网络连接。如果遇到unable to connect to api (connectionrefused)或超时问题先检查防火墙是否放行了 API 端口的出入站连接代理设置是否正确如果所在网络需要代理DNS 解析是否正常3. 第一个图片生成任务环境准备好后不要直接写复杂的批量生成代码先确保单次调用能正常工作。3.1 理解 API 的基本调用格式图片生成 API 的请求通常包含几个关键参数提示词prompt描述要生成什么图片模型选择model使用哪个图片生成模型尺寸size输出图片的分辨率数量n一次生成几张图片质量quality标准版或高清版一个最基础的请求体例如下import requests import json import os def generate_image(prompt, size1024x1024): api_key os.getenv(CLAUDE_API_KEY) if not api_key: return {error: API密钥未设置} headers { Content-Type: application/json, Authorization: fBearer {api_key} } data { model: claude-image-gen, # 实际模型名需查文档 prompt: prompt, size: size, n: 1, response_format: url # 或 b64_json 直接返回base64编码 } try: response requests.post( https://api.claude.com/v1/images/generations, headersheaders, datajson.dumps(data), timeout30 ) if response.status_code 200: return response.json() else: return {error: fAPI返回错误: {response.status_code}, details: response.text} except requests.exceptions.Timeout: return {error: 请求超时请检查网络或调整超时时间} except Exception as e: return {error: f请求异常: {str(e)}} # 测试调用 result generate_image(一只在沙发上睡觉的橘猫阳光明媚的下午) print(json.dumps(result, indent2, ensure_asciiFalse))3.2 处理 API 响应和图片保存API 调用成功后需要正确处理返回的图片数据。常见的返回格式有两种图片 URLAPI 生成图片后返回一个临时可访问的 URLBase64 编码图片数据直接编码在 JSON 响应中URL 方式适合快速测试但链接通常有有效期。Base64 方式更可靠适合生产环境。保存图片的补充代码import base64 from datetime import datetime def save_image_from_response(api_response, save_dir./generated_images): 根据API响应保存图片 if error in api_response: print(f错误: {api_response[error]}) return None os.makedirs(save_dir, exist_okTrue) # 生成文件名 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) filename fimage_{timestamp}.png filepath os.path.join(save_dir, filename) if data in api_response and len(api_response[data]) 0: image_data api_response[data][0] if url in image_data: # 从URL下载图片 url image_data[url] img_response requests.get(url) if img_response.status_code 200: with open(filepath, wb) as f: f.write(img_response.content) print(f图片已保存: {filepath}) return filepath elif b64_json in image_data: # 从Base64解码保存 b64_data image_data[b64_json] image_bytes base64.b64decode(b64_data) with open(filepath, wb) as f: f.write(image_bytes) print(f图片已保存: {filepath}) return filepath print(未找到有效的图片数据) return None # 完整测试流程 result generate_image(简约风格的科技公司logo蓝色调) if error not in result: saved_path save_image_from_response(result) if saved_path: print(f生成成功图片位置: {saved_path})3.3 第一次调用的常见问题排查第一次调用很容易遇到各种问题我一般按这个顺序排查认证失败检查 API 密钥是否正确设置是否有足够的余额或调用额度参数错误确认请求的 JSON 格式正确特别是模型名称、尺寸等参数是否支持网络超时适当增加 timeout 值检查网络连接稳定性额度限制查看 API 调用配额和频率限制如果遇到api error: 400这类错误仔细看错误信息中的具体描述。比如搜索材料中提到的this models maximum context length is 1048565 tokens说明输入过长the supported api model names are deepseek-v4-pro or deepseek说明模型名称需要调整。4. 优化提示词和参数调整单次调用能工作后就要考虑如何提高生成质量。图片生成的效果很大程度上取决于提示词的质量和参数设置。4.1 编写有效的图片提示词好的提示词应该包含主体描述明确要生成什么风格要求写实、卡通、水彩、简约等细节特征颜色、光线、角度、背景负面提示不希望出现的内容提示词编写示例对比# 效果可能一般的提示词 poor_prompt 一只猫 # 效果更好的提示词 good_prompt 一只橘色虎斑猫趴在窗台上阳光从侧面照射毛发光泽柔和背景是模糊的室内环境照片级真实感4K画质 # 带风格要求的提示词 style_prompt 赛博朋克风格的城市街景霓虹灯招牌雨天街道反射灯光未来主义建筑数字艺术风格 prompts_to_test [ (基础版, 一只猫), (详细版, good_prompt), (风格版, style_prompt) ] for name, prompt in prompts_to_test: print(f\n测试提示词: {name}) result generate_image(prompt) if error not in result: save_image_from_response(result, f./test_{name})4.2 调整生成参数不同的参数组合会对结果产生显著影响def generate_with_parameters(prompt, size1024x1024, qualitystandard, stylevivid): 支持更多参数的生成函数 api_key os.getenv(CLAUDE_API_KEY) data { model: claude-image-gen, prompt: prompt, size: size, n: 1, quality: quality, # standard 或 hd style: style, # vivid 或 natural response_format: b64_json } # ... 其余请求代码类似前面示例 # 测试不同参数组合 test_cases [ {size: 256x256, desc: 小尺寸}, {size: 1024x1024, desc: 标准尺寸}, {size: 1024x1024, quality: hd, desc: 高清质量}, {size: 1024x1024, style: natural, desc: 自然风格} ] base_prompt 宁静的山水风景画 for params in test_cases: desc params.pop(desc) print(f\n生成测试: {desc}) result generate_with_parameters(base_prompt, **params) if error not in result: save_image_from_response(result, f./param_test_{desc})4.3 处理生成失败和重试机制图片生成不一定每次都能成功需要有适当的错误处理和重试逻辑def robust_image_generation(prompt, max_retries3, retry_delay5): 带重试机制的图片生成 for attempt in range(max_retries): try: result generate_image(prompt) if error in result: error_msg result[error] print(f第 {attempt1} 次尝试失败: {error_msg}) # 如果是额度不足或认证问题重试没用 if 402 in error_msg or 401 in error_msg: break # 其他错误可以重试 if attempt max_retries - 1: print(f等待 {retry_delay} 秒后重试...) time.sleep(retry_delay) continue else: return result # 成功则返回 except Exception as e: print(f第 {attempt1} 次尝试异常: {str(e)}) if attempt max_retries - 1: time.sleep(retry_delay) return {error: f经过 {max_retries} 次尝试后仍失败} # 使用重试机制 result robust_image_generation(复杂的奇幻场景, max_retries3)5. 批量生成和性能考虑单张图片生成稳定后接下来要考虑批量处理的需求。这时候不能简单用循环调用需要更细致的性能和管理策略。5.1 实现安全的批量生成直接使用 for 循环调用 API 很容易触发频率限制。更好的做法是控制并发数和添加间隔import time from concurrent.futures import ThreadPoolExecutor, as_completed def batch_generate_images(prompts, max_workers2, delay_between_calls1.0): 批量生成图片控制并发和频率 results [] def generate_with_delay(prompt, index): 单个生成任务带延迟控制 if index 0: time.sleep(delay_between_calls) return prompt, generate_image(prompt) with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_prompt { executor.submit(generate_with_delay, prompt, i): prompt for i, prompt in enumerate(prompts) } # 收集结果 for future in as_completed(future_to_prompt): prompt future_to_prompt[future] try: result_prompt, result future.result() results.append((result_prompt, result)) print(f完成: {result_prompt[:30]}...) except Exception as e: print(f生成失败: {prompt} - {str(e)}) results.append((prompt, {error: str(e)})) return results # 批量生成示例 prompt_list [ 清晨的森林雾气弥漫阳光透过树叶, 未来城市的空中交通飞行汽车穿梭, 抽象几何艺术蓝色和金色的组合, 复古咖啡馆内部温暖灯光书本散落 ] batch_results batch_generate_images(prompt_list, max_workers2)5.2 管理生成结果和元数据批量生成时需要系统化地管理输出结果import json from datetime import datetime class ImageBatchManager: 批量图片生成管理器 def __init__(self, base_output_dir./batch_output): self.base_dir base_output_dir self.session_id datetime.now().strftime(%Y%m%d_%H%M%S) self.output_dir os.path.join(base_output_dir, self.session_id) os.makedirs(self.output_dir, exist_okTrue) # 创建元数据文件 self.metadata_file os.path.join(self.output_dir, metadata.json) self.metadata { session_id: self.session_id, start_time: datetime.now().isoformat(), images: [] } def save_image_with_metadata(self, prompt, api_response, image_index): 保存图片和对应的元数据 if error in api_response: # 记录错误信息 image_meta { index: image_index, prompt: prompt, error: api_response[error], timestamp: datetime.now().isoformat(), success: False } else: # 保存图片并记录元数据 filename fimage_{image_index:04d}.png filepath os.path.join(self.output_dir, filename) saved_path save_image_from_response(api_response, self.output_dir) if saved_path: image_meta { index: image_index, prompt: prompt, filename: filename, filepath: saved_path, timestamp: datetime.now().isoformat(), success: True } else: image_meta { index: image_index, prompt: prompt, error: 图片保存失败, timestamp: datetime.now().isoformat(), success: False } self.metadata[images].append(image_meta) return image_meta def finalize_batch(self): 完成批量生成保存元数据 self.metadata[end_time] datetime.now().isoformat() self.metadata[total_images] len(self.metadata[images]) self.metadata[success_count] sum(1 for img in self.metadata[images] if img.get(success, False)) with open(self.metadata_file, w, encodingutf-8) as f: json.dump(self.metadata, f, indent2, ensure_asciiFalse) print(f批量生成完成成功 {self.metadata[success_count]}/{self.metadata[total_images]}) print(f结果保存在: {self.output_dir}) # 使用管理器进行批量生成 manager ImageBatchManager() for i, prompt in enumerate(prompt_list): result generate_image(prompt) meta manager.save_image_with_metadata(prompt, result, i) print(f进度: {i1}/{len(prompt_list)} - {成功 if meta[success] else 失败}) manager.finalize_batch()5.3 性能监控和优化建议在批量生成过程中需要关注几个性能指标import time import psutil # 需要安装: pip install psutil class PerformanceMonitor: 简单的性能监控 def __init__(self): self.start_time None self.call_count 0 self.success_count 0 self.total_response_time 0 def start_call(self): 开始一次API调用 self.call_count 1 return time.time() def end_call(self, start_time, successTrue): 结束一次API调用 response_time time.time() - start_time self.total_response_time response_time if success: self.success_count 1 return response_time def get_stats(self): 获取统计信息 if self.call_count 0: return 暂无调用数据 avg_time self.total_response_time / self.call_count success_rate (self.success_count / self.call_count) * 100 # 获取系统资源信息 memory_usage psutil.virtual_memory().percent cpu_usage psutil.cpu_percent(interval1) return { 总调用次数: self.call_count, 成功次数: self.success_count, 成功率: f{success_rate:.1f}%, 平均响应时间: f{avg_time:.2f}秒, 内存使用率: f{memory_usage:.1f}%, CPU使用率: f{cpu_usage:.1f}% } # 在批量生成中加入性能监控 monitor PerformanceMonitor() optimized_prompts [ 测试图片1, 测试图片2, 测试图片3 # 简化的测试提示词 ] for prompt in optimized_prompts: start_time monitor.start_call() result generate_image(prompt) response_time monitor.end_call(start_time, error not in result) print(f生成完成 - 耗时: {response_time:.2f}s - 状态: {成功 if error not in result else 失败}) time.sleep(1) # 避免过快调用 print(\n性能统计:) stats monitor.get_stats() for key, value in stats.items(): print(f{key}: {value})6. 错误处理和边界情况在实际使用中会遇到各种预期之外的情况。完善的错误处理能让你的代码更健壮。6.1 常见的 API 错误码处理从搜索材料看Claude Code API 可能返回各种错误码需要有对应的处理策略def handle_api_error(error_response, prompt, attempt_count): 根据错误类型采取不同的处理策略 error_msg error_response.get(error, ) error_handlers { 400: lambda: handle_400_error(error_msg, prompt), 401: lambda: handle_auth_error(error_msg), 402: lambda: handle_quota_error(error_msg), 429: lambda: handle_rate_limit(error_msg, attempt_count), 500: lambda: handle_server_error(error_msg, attempt_count), connection: lambda: handle_connection_error(error_msg, attempt_count) } # 识别错误类型 error_type connection for code in [400, 401, 402, 429, 500]: if code in error_msg: error_type code break handler error_handlers.get(error_type, lambda: f未知错误: {error_msg}) return handler() def handle_400_error(error_msg, prompt): 处理参数错误 if maximum context length in error_msg: # 提示词过长 suggested_length len(prompt) * 0.8 # 建议缩短20% return { action: retry, suggestion: f提示词过长建议缩短到约{int(suggested_length)}字符, modify_prompt: True } elif model in error_msg and supported in error_msg: # 模型名称错误 return { action: retry, suggestion: 检查并更正模型名称, modify_params: True } else: return { action: stop, suggestion: 检查请求参数格式, need_human_review: True } def handle_quota_error(error_msg): 处理额度不足 return { action: stop, suggestion: API额度不足需要充值或等待重置, urgent: True } def handle_rate_limit(error_msg, attempt_count): 处理频率限制 if attempt_count 3: wait_time 60 * (2 ** attempt_count) # 指数退避 return { action: retry, suggestion: f触发频率限制等待{wait_time}秒后重试, wait_seconds: wait_time } else: return { action: stop, suggestion: 多次触发频率限制建议降低调用频率 } # 使用错误处理 def robust_api_call(prompt, max_attempts3): 带完整错误处理的API调用 for attempt in range(max_attempts): result generate_image(prompt) if error in result: error_handling handle_api_error(result, prompt, attempt) if error_handling[action] retry: if attempt max_attempts - 1: wait_time error_handling.get(wait_seconds, 5) print(f第{attempt1}次尝试失败: {error_handling[suggestion]}) time.sleep(wait_time) continue else: # stop 或其他动作 print(f不可重试错误: {error_handling[suggestion]}) break else: return result # 成功 return {error: 所有重试尝试均失败}6.2 输入验证和预处理很多 API 错误其实可以通过输入验证来预防def validate_and_preprocess_prompt(prompt, max_length2000): 验证和预处理提示词 if not prompt or not prompt.strip(): return {valid: False, error: 提示词不能为空} prompt prompt.strip() # 长度检查 if len(prompt) max_length: return { valid: False, error: f提示词过长({len(prompt)}字符)最大支持{max_length}字符, suggestion: prompt[:max_length-100] ...[自动截断] } # 敏感词检查基础版 sensitive_words [暴力, 违法, 侵权] # 根据实际需求扩展 found_sensitive [word for word in sensitive_words if word in prompt] if found_sensitive: return { valid: False, error: f提示词包含敏感内容: {, .join(found_sensitive)}, suggestion: 请修改提示词内容 } # 预处理移除多余空格确保格式整洁 processed_prompt .join(prompt.split()) return { valid: True, processed_prompt: processed_prompt, original_length: len(prompt), processed_length: len(processed_prompt) } # 使用输入验证 def safe_generate_image(raw_prompt): 安全的图片生成函数 validation validate_and_preprocess_prompt(raw_prompt) if not validation[valid]: return {error: validation[error], validation: validation} actual_prompt validation[processed_prompt] print(f使用处理后的提示词({validation[processed_length]}字符): {actual_prompt[:100]}...) return generate_image(actual_prompt) # 测试各种边界情况 test_prompts [ , # 空提示词 , # 只有空格 a * 3000, # 超长提示词 生成一幅美丽的山水画, # 正常提示词 生成包含暴力内容的图片 # 敏感内容 ] for test_prompt in test_prompts: print(f\n测试提示词: {test_prompt[:50]}...) result safe_generate_image(test_prompt) if error in result and validation in result: print(f输入验证失败: {result[error]}) elif error in result: print(fAPI调用失败: {result[error]}) else: print(生成成功)6.3 网络异常和超时处理网络问题在实际使用中很常见需要有相应的处理机制import socket import requests.adapters from urllib3.util.retry import Retry def create_robust_session(retries3, backoff_factor0.3): 创建具有重试机制的会话 session requests.Session() # 配置重试策略 retry_strategy Retry( totalretries, backoff_factorbackoff_factor, status_forcelist[429, 500, 502, 503, 504], allowed_methods[HEAD, GET, POST, PUT, DELETE, OPTIONS, TRACE] ) adapter requests.adapters.HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 设置合理的超时 session.request lambda method, url, **kwargs: requests.Session.request( session, method, url, timeout(3.05, 30), **kwargs ) return session # 使用稳健的会话 def reliable_api_call(prompt, sessionNone): 使用稳健会话的API调用 if session is None: session create_robust_session() api_key os.getenv(CLAUDE_API_KEY) data { model: claude-image-gen, prompt: prompt, size: 1024x1024, n: 1, response_format: b64_json } try: response session.post( https://api.claude.com/v1/images/generations, headers{Authorization: fBearer {api_key}, Content-Type: application/json}, datajson.dumps(data) ) response.raise_for_status() # 如果状态码不是200抛出异常 return response.json() except requests.exceptions.Timeout: return {error: 请求超时建议检查网络连接或增加超时时间} except requests.exceptions.ConnectionError: return {error: 网络连接错误请检查网络设置} except requests.exceptions.HTTPError as e: return {error: fHTTP错误: {e.response.status_code} - {e.response.text}} except Exception as e: return {error: f未知错误: {str(e)}} # 测试稳健调用 robust_session create_robust_session() result reliable_api_call(测试网络稳健性的图片, robust_session) print(f稳健调用结果: {成功 if error not in result else result[error]})7. 实际应用场景和进阶用法基础功能稳定后可以探索更实际的应用场景和进阶用法。7.1 集成到现有项目中将图片生成 API 集成到 Web 应用、自动化脚本或其他项目中from flask import Flask, request, jsonify, send_file import io import base64 app Flask(__name__) app.route(/generate-image, methods[POST]) def generate_image_api(): 提供图片生成API接口 data request.get_json() if not data or prompt not in data: return jsonify({error: 缺少prompt参数}), 400 prompt data[prompt] size data.get(size, 1024x1024) # 输入验证 validation validate_and_preprocess_prompt(prompt) if not validation[valid]: return jsonify({error: validation[error]}), 400 # 生成图片 result reliable_api_call(validation[processed_prompt]) if error in result: return jsonify({error: result[error]}), 500 # 返回图片数据 if data in result and len(result[data]) 0: image_data result[data][0] if b64_json in image_data: # 直接返回Base64数据 return jsonify({ success: True, image_data: image_data[b64_json], prompt: prompt }) return jsonify({error: 生成失败}), 500 app.route(/download-image, methods[POST]) def download_image_api(): 生成并下载图片文件 data request.get_json() prompt data.get(prompt, ) result reliable_api_call(prompt) if error in result: return jsonify({error: result[error]}), 500 if data in result and result[data]: image_b64 result[data][0][b64_json] image_bytes base64.b64decode(image_b64) return send_file( io.BytesIO(image_bytes), mimetypeimage/png, as_attachmentTrue, download_namegenerated_image.png ) return jsonify({error: 生成失败}), 500 if __name__ __main__: app.run(debugTrue, host0.0.0.0, port5000)7.2 批量处理和数据流水线对于需要处理大量图片生成的场景可以构建完整的数据流水线import pandas as pd from queue import Queue import threading class ImageGenerationPipeline: 图片生成流水线 def __init__(self, input_csv, output_dir, batch_size10): self.input_csv input_csv self.output_dir output_dir self.batch_size batch_size self.task_queue Queue() self.results [] self.lock threading.Lock() os.makedirs(output_dir, exist_okTrue) def load_tasks(self): 从CSV加载生成任务 df pd.read_csv(self.input_csv) for _, row in df.iterrows(): self.task_queue.put({ id: row[id], prompt: row[prompt], size: row.get(size, 1024x1024), category: row.get(category, general) }) def worker(self, worker_id): 工作线程函数 session create_robust_session() while not self.task_queue.empty(): try: task self.task_queue.get_nowait() except: break print(fWorker {worker_id} 处理任务: {task[id]}) result reliable_api_call(task[prompt], session) with self.lock: if error not in result and data in result: # 保存成功的图片 filename f{task[id]}_{task[category]}.png save_image_from_response(result, self.output_dir) self.results.append({ id: task[id], prompt: task[prompt], status: success, filename: filename }) else: self.results.append({ id: task[id], prompt: task[prompt], status: failed, error: result.get(error, unknown) }) self.task_queue.task_done() time.sleep(1) # 避免过快调用 def run(self, num_workers2): 运行流水线 self.load_tasks() total_tasks self.task_queue.qsize() print(f开始处理 {total_tasks} 个任务使用 {num_workers} 个工作线程) threads [] for i in range(num_workers): thread threading.Thread(targetself.worker, args(i1,)) thread.start() threads.append(thread) # 等待所有任务完成 self.task_queue.join() # 等待所有线程结束 for thread in threads: thread.join() # 保存结果报告 results_df pd.DataFrame(self.results) results_df.to_csv(os.path.join(self.output_dir, generation_report.csv), indexFalse) success_count len(results_df[results_df[status] success]) print(f流水线完成成功: {success_count}/{total_tasks}) return results_df # 使用流水线处理批量任务 # 先创建任务CSV文件包含id,prompt,size,category列 pipeline ImageGenerationPipeline(tasks.csv, ./pipeline_output) results pipeline.run(num_workers2)7.3 质量