1. 先搞清楚这个 API 到底能做什么

如果你正在找一种能通过代码直接生成图片的方法，Claude Code 的图片生成 API 值得先了解清楚。这类功能最怕的就是文档不全、环境复杂、跑通一次要折腾半天。我一般会先看它到底解决的是通用图片生成，还是特定场景的配图需求。

从标题和搜索材料看，这应该是 Claude Code 工具集里的一个子功能，专门用来处理图片生成任务。和常见的文字生成图片工具不同，它更偏向开发者集成，需要你通过代码调用 API 接口来完成生成。这种方式的优势是能嵌入到自己的应用或自动化流程里，但前提是得先搞清楚输入输出格式、资源要求和常见坑点。

实际使用时，最关键的不是功能列表有多长，而是能不能在你的开发环境里稳定跑起来。我建议先确认几个基础问题：是否需要额外申请 API 密钥、支持哪些图片格式输入、输出分辨率范围、单次生成耗时大概多少、有没有并发限制。这些信息决定了你能不能直接拿来用，还是需要额外做适配。

2. 环境准备和前置检查

在开始写代码调用 API 之前，得先把环境理顺。很多调用失败的问题，其实出在环境配置环节。

2.1 确认 Claude Code 基础环境

Claude Code 本身可能有多种安装方式：桌面版、命令行工具、IDE 插件或者纯 API 服务。你需要先确认自己用的是哪种形式。如果是桌面版，通常已经内置了基础环境；如果是命令行或插件形式，可能需要单独配置 API 端点。

我一般会按这个顺序检查：

1. 先运行claude --version或类似命令，确认 Claude Code 主程序能正常响应
2. 查看帮助文档里有没有图片生成相关的子命令或参数
3. 确认当前网络环境能访问所需的 API 服务（有些服务可能需要特定网络条件）

如果还没有安装 Claude Code，建议先走官方推荐的安装流程。搜索材料里提到了 Windows、Ubuntu 等不同系统的安装方法，选择和你开发环境匹配的那个。安装完成后不要急着调用复杂功能，先用一个最简单的文本生成任务测试基础连通性。

2.2 API 密钥和权限配置

图片生成 API 通常需要认证。从搜索材料看，可能会遇到api error: 402 insufficient balance（余额不足）或login failed. check api token（令牌错误）这类问题。这说明调用前需要先获取有效的 API 密钥并正确配置。

具体步骤一般是：

1. 在 Claude Code 的控制台或官网申请 API 密钥
2. 在代码或配置文件中设置这个密钥
3. 测试密钥是否有效

密钥的配置方式因环境而异。如果是命令行环境，可能需要在~/.claude/config这样的配置文件中设置；如果是代码调用，通常通过环境变量或直接写在初始化参数里。我建议先用环境变量方式，这样更安全也更灵活：

export CLAUDE_API_KEY="your_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": f"Bearer {api_key}"}) print(f"API 连接测试: {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, size="1024x1024"): api_key = os.getenv("CLAUDE_API_KEY") if not api_key: return {"error": "API密钥未设置"} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {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", headers=headers, data=json.dumps(data), timeout=30 ) if response.status_code == 200: return response.json() else: return {"error": f"API返回错误: {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, indent=2, ensure_ascii=False))

3.2 处理 API 响应和图片保存

API 调用成功后，需要正确处理返回的图片数据。常见的返回格式有两种：

1. 图片 URL：API 生成图片后返回一个临时可访问的 URL
2. Base64 编码：图片数据直接编码在 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_ok=True) # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"image_{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 第一次调用的常见问题排查

第一次调用很容易遇到各种问题，我一般按这个顺序排查：

1. 认证失败：检查 API 密钥是否正确设置，是否有足够的余额或调用额度
2. 参数错误：确认请求的 JSON 格式正确，特别是模型名称、尺寸等参数是否支持
3. 网络超时：适当增加 timeout 值，检查网络连接稳定性
4. 额度限制：查看 API 调用配额和频率限制

如果遇到api error: 400这类错误，仔细看错误信息中的具体描述。比如搜索材料中提到的this model's 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, size="1024x1024", quality="standard", style="vivid"): """支持更多参数的生成函数""" 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_retries=3, retry_delay=5): """带重试机制的图片生成""" for attempt in range(max_retries): try: result = generate_image(prompt) if "error" in result: error_msg = result["error"] print(f"第 {attempt+1} 次尝试失败: {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"第 {attempt+1} 次尝试异常: {str(e)}") if attempt < max_retries - 1: time.sleep(retry_delay) return {"error": f"经过 {max_retries} 次尝试后仍失败"} # 使用重试机制 result = robust_image_generation("复杂的奇幻场景", max_retries=3)

5. 批量生成和性能考虑

单张图片生成稳定后，接下来要考虑批量处理的需求。这时候不能简单用循环调用，需要更细致的性能和管理策略。

5.1 实现安全的批量生成

直接使用 for 循环调用 API 很容易触发频率限制。更好的做法是控制并发数和添加间隔：

import time from concurrent.futures import ThreadPoolExecutor, as_completed def batch_generate_images(prompts, max_workers=2, delay_between_calls=1.0): """批量生成图片，控制并发和频率""" results = [] def generate_with_delay(prompt, index): """单个生成任务，带延迟控制""" if index > 0: time.sleep(delay_between_calls) return prompt, generate_image(prompt) with ThreadPoolExecutor(max_workers=max_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_workers=2)

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_ok=True) # 创建元数据文件 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 = f"image_{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", encoding="utf-8") as f: json.dump(self.metadata, f, indent=2, ensure_ascii=False) 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"进度: {i+1}/{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, success=True): """结束一次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(interval=1) 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_attempts=3): """带完整错误处理的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"第{attempt+1}次尝试失败: {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_length=2000): """验证和预处理提示词""" 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(f"API调用失败: {result['error']}") else: print("生成成功！")

6.3 网络异常和超时处理

网络问题在实际使用中很常见，需要有相应的处理机制：

import socket import requests.adapters from urllib3.util.retry import Retry def create_robust_session(retries=3, backoff_factor=0.3): """创建具有重试机制的会话""" session = requests.Session() # 配置重试策略 retry_strategy = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE"] ) adapter = requests.adapters.HTTPAdapter(max_retries=retry_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, session=None): """使用稳健会话的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": f"Bearer {api_key}", "Content-Type": "application/json"}, data=json.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": f"HTTP错误: {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), mimetype='image/png', as_attachment=True, download_name='generated_image.png' ) return jsonify({"error": "生成失败"}), 500 if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)

7.2 批量处理和数据流水线

对于需要处理大量图片生成的场景，可以构建完整的数据流水线：

import pandas as pd from queue import Queue import threading class ImageGenerationPipeline: """图片生成流水线""" def __init__(self, input_csv, output_dir, batch_size=10): 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_ok=True) 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(f"Worker {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_workers=2): """运行流水线""" 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(target=self.worker, args=(i+1,)) 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'), index=False) 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_workers=2)

7.3 质量