基于Flux on G6e和 S3 Vector 的Icon 图片生成方案

在页面设计中，图标 (Icon) 至关重要，它不仅在界面中能节省空间、加速用户定位功能，更通过全球通用的视觉语言提升信息传达效率。和普通图片的生成不同，Icon的描述提示词简单，对精美度要求较高。另一方面，在大部分智能页面设计的ToC 应用中，所需要生成的Icon数量大、速度快。因此在这类场景中，调用文生图模型的SaaS API生成Icon在成本和端到端时延方面，都面临挑战。
本文将介绍如何基于Flux on G6e 和S3 Vector 构建一个具有高性价比的icon图片生成与检索方案，返回s3 url 用于满足智能UI设计中HTML或H5界面的Icon需求。相比于直接调用Flux API，该方案在ToC生产环境中具有成本优势，成本分析与临界值估算在本文的最后部分展开。

选型对比： Flux模型与SD 系列模型

Flux和Stable Diffusion虽然同属AI图像生成领域，图像生成性能各有千秋：

Flux：模型参数高达120亿，是目前已知最大的开源文生图模型之一，在处理复杂场景时能捕捉更多细节，在手指细节、复杂构图和动态光影表现上更精准。FLUX.1 [schnell] 是系列中速度最快的模型，4步推理即可生成高质量图像
Stable Diffusion: SD3.5版本有80亿参数，通过参数优化和商用授权策略吸引开发者。模型生成效果更强调真实感，风格一致性上更优。

基于上述差异，Flux常用于影视概念设计、复杂机械结构可视化、需要精准文字融合的广告创意等领域。对于ToC应用开发或需要将图像生成集成到AI工作流中的客户，FLUX.1-schnell模型的性价比是最优选择。

解决方案架构

方案流程如下：

先决条件：对于常用Icon，预先使用S3构建一个图片库，并对Icon 描述和S3 url进行向量化存储
向量查询：对于Html中需要加入Icon的地方，在向量库中Icon描述进行向量检索。
缓存命中：若图片库中存在符合的Icon图片，返回Icon所在的S3 url
按需生成：若图片库中无相应Icon，调用Flux API完成图片生成
库更新：定期对Flux API生成图片种类和数量进行分析，更新图片库。

其中，使用Lambda函数作为调度中心，负责协调向量检索、缓存命中判断和按需生成的完整流程。在性能方面，使用图片库向量检索做分流，在不考虑额外网络传输时延的情况下，将常见Icon的生成耗时从秒级缩短至毫秒级。使用S3 vector 而非Opensearch 作为向量库，则带来更大的成本优势。

关键模块实现

1. 构建基于S3 Vector的Icon向量库

首先构建能够描述Icon 图片库的json文件，包含Icon 类型、Icon描述和S3 url 存储地址。示例格式如下, 其中key、description、s3_url是必须字段。

  {
        "key": "arrow-right-001",
        "description": "向右箭头图标，用于导航和下一步操作，导航类图标",
        "s3_url": "https://icon-storage-bucket.s3.amazonaws.com/arrows/arrow-right.svg",
        "category": "navigation",
        "tags": ["箭头", "导航", "右", "下一步"]
    }

可以在控制台或使用SDK创建S3 Vector bucket，创建教程不再展开。本文的embedding使用Bedrock Titan Text Embeddings V2模型，向量索引中维度为1024，距离为余弦相似度。

index = s3_client.create_index(
    VectorBucketName='icon-vector-bucket',
    IndexName='icon-idx',
    DataType='float32',
    Dimension=1024,
    DistanceMetric='cosine'
)

使用Python SDK，通过读取JSON文件中的图标描述信息，调用Bedrock Titan模型生成文本嵌入向量。所有图标的向量数据和元数据被一次性写入S3 Vector Search索引，为后续的语义搜索提供基础数据支撑。

import boto3
import json

bedrock = boto3.client("bedrock-runtime", region_name="us-west-2")
s3vectors = boto3.client("s3vectors", region_name="us-west-2")

def index_all_icons(json_file_path):
    # 读取JSON文件
    with open(json_file_path, 'r', encoding='utf-8') as f:
        icons_data = json.load(f)
    
    # 生成所有embeddings
    embeddings = []
    for icon in icons_data:
        response = bedrock.invoke_model(
            modelId="amazon.titan-embed-text-v2:0",
            body=json.dumps({"inputText": icon["description"]})
        )
        embedding = json.loads(response["body"].read())["embedding"]
        embeddings.append(embedding)
    
    # 构建向量数据
    vectors = []
    for icon, embedding in zip(icons_data, embeddings):
        vectors.append({
            "key": icon["key"],
            "data": {"float32": embedding},
            "metadata": {
                "s3_url": icon["s3_url"],
                "description": icon["description"],
                "category": icon.get("category", ""),
                "tags": icon.get("tags", [])
            }
        })
    
    # 一次性写入所有向量
    s3vectors.put_vectors(
        vectorBucketName="icon-vector-bucket",
        indexName="icon-idx",
        vectors=vectors
    )

2. Icon 检索

对于基础检索功能，只需将所需的icon描述用同样的embedding模型进行向量化后，在向量库中进行搜索。为了过滤低质量匹配结果，引入similarity_threshold 阈值参数。当余弦距离小于阈值的时候，任务检索到的结果为低质量，不再返回。

def search_icons(query_text: str, top_k: int = 5, similarity_threshold: float = 0.7) -> List[Dict]:

    # 生成查询向量
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=json.dumps({"inputText": query_text})
    )
    embedding = json.loads(response["body"].read())["embedding"]
    
    # 执行向量搜索
    search_response = s3vectors.query_vectors(
        vectorBucketName="icon-vector-bucket",
        indexName="icon-idx",
        queryVector={"float32": embedding},
        topK=top_k,
        returnDistance=True,
        returnMetadata=True
    )
    
    # 过滤低相似度结果
    results = []
    for vector in search_response["vectors"]:
        similarity = 1 - vector["distance"]  # 余弦距离转相似度
        if similarity >= similarity_threshold:
            results.append({
                "key": vector["key"],
                "s3_url": vector["metadata"]["s3_url"],
                "description": vector["metadata"]["description"],
                "category": vector["metadata"].get("category", ""),
                "similarity": round(similarity, 3)
            })
    
    return results

若想在特定分类中进行检索，需要在参数中加入category，对应json文件中的类别描述。带分类过滤的搜索代码如下：

 search_response = s3vectors.query_vectors(
        vectorBucketName="icon-vector-bucket",
        indexName="icon-idx",
        queryVector={"float32": embedding},
        topK=top_k,
        #加入类别过滤
        filter={"category": category},
        returnDistance=True,
        returnMetadata=True
    )

3. 基于Fast API的Flux on G6e 部署

本文使用Fast API (https://fastapi.tiangolo.com/) 作为Flux 模型的调用接口，是因为FastAPI的高性能异步架构特别适合GPU密集型任务，能够高效处理并发的图像生成请求，避免GPU资源在等待I/O操作时闲置，能够最大化GPU利用率，在保持低延迟的同时支持更高的吞吐量。
Flux1.[Schnell] 参数量为12B，因此使用g6e.xlarge 部署。多进程架构使用 torch.multiprocessing 实现GPU任务分发：主进程运行FastAPI服务器，工作进程负责AI模型推理。

应用初始化

通过数据模型 (Pydantic) GenerationRequest 请求参数，GenerationResponse响应数据模型

from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from pydantic import BaseModel

app = FastAPI(title="FLUX.1-schnell Text-to-Image API with LoRA support")

MODEL_ID = "black-forest-labs/FLUX.1-schnell"

class GenerationRequest(BaseModel):
    prompt: str
    negative_prompt: Optional[str] = None
    num_inference_steps: Optional[int] = 4
    guidance_scale: Optional[float] = 3.5
    max_sequence_length: Optional[int] = 512
    seed: Optional[int] = None
    height: Optional[int] = 512
    width: Optional[int] = 512
  
class GenerationResponse(BaseModel):
    image: str  # base64 encoded image
    seed: int   # 返回使用的种子，便于复现

核心API端点

POST /generate: 主要图像生成接口，接受请求并将任务放入队列，使用非阻塞入队和时检查结果字典实现非阻塞等待

@app.post("/generate", response_model=GenerationResponse)
async def generate_image(request: GenerationRequest):
    # 生成唯一的任务 ID
    task_id = str(uuid.uuid4())
    # 将请求放入队列
    task_queue.put((task_id, request.dict()))
    print(f"Added task {task_id} to queue")
    max_wait_time = 300 
    start_time = time.time()
    while time.time() - start_time < max_wait_time:
        if task_id in result_dict:
            result = result_dict[task_id]
            # 删除结果以释放内存
            del result_dict[task_id]
            if "error" in result:
                raise HTTPException(status_code=500, detail=result["error"])
            return result
        # 短暂睡眠，避免忙等待
        await asyncio.sleep(0.5)
    raise HTTPException(status_code=504, detail="Request timed out")

POST /generate_view: 直接返回图像文件

@app.post("/generate_view")
async def generate_image_view(request: GenerationRequest):
    result = await generate_image(request)
    image_bytes = base64.b64decode(result["image"])
    return Response(content=image_bytes, media_type="image/png")

完整代码详见：https://github.com/CrazyCha/Flux-on-GPU

成本分析

Flux SaaS API 调用成本： FLux.schnell 的第三方平台定价 https://fal.ai/models/fal-ai/flux/schnell 每333次API call 成本约为 $1（即$0.003/次）。

常用Icon：假设将10,000个Icon 存入图片库，SVG格式约为10KB。S3数据存储的月成本可忽略不计，主要成本是向量检索。S3 Vector查询 $0.40/1000次请求，远小于Flux API调用。
新生成Icon：g6e.xlarge 在us-east-1 region的No upfront 单价为$1.1724/小时, 每天的成本约为$30。

基于上述的Flux on G6e.xlarge 部署方案，图片生成耗时如下：

并发请求数	成功数	总耗时（s）	平均耗时（s）
1	1 / 1	1.02	1.02
3	3 / 3	2.53	0.84
10	10 / 10	8.08	0.81
20	20 / 20	16.67	0.83

因此当并发量较小、且任务可等待的情况下，当日均生成新Icon的请求大于10000次时，相比于Flux Saas API，Flux on GPU 也更具有成本优势

*前述特定亚马逊云科技生成式人工智能相关的服务目前在亚马逊云科技海外区域可用。亚马逊云科技中国区域相关云服务由西云数据和光环新网运营，具体信息以中国区域官网为准。

本篇作者

AWS 架构师中心：云端创新的引领者

探索 AWS 架构师中心，获取经实战验证的最佳实践与架构指南，助您高效构建安全、可靠的云上应用

亚马逊AWS官方博客

基于Flux on G6e和 S3 Vector 的Icon 图片生成方案

选型对比： Flux模型与SD 系列模型

解决方案架构

关键模块实现

1. 构建基于S3 Vector的Icon向量库

2. Icon 检索

3. 基于Fast API的Flux on G6e 部署

成本分析

本篇作者

AWS 架构师中心：云端创新的引领者

了解

资源

开发人员

帮助

亚马逊AWS官方博客

基于Flux on G6e和 S3 Vector 的Icon 图片生成方案

选型对比： Flux模型与SD 系列模型

解决方案架构

关键模块实现

1. 构建基于S3 Vector的Icon向量库

2. Icon 检索

3. 基于Fast API的Flux on G6e 部署

成本分析

本篇作者

AWS 架构师中心： 云端创新的引领者

了解

资源

开发人员

帮助

AWS 架构师中心：云端创新的引领者