本文详细介绍如何使用 Python 异步编程和 AI 大模型构建一个智能商品标签生成系统,包括图像筛选、Google Lens 搜索、标签生成等核心功能。

📋 目录

  • 一、项目概述

  • 二、技术栈

  • 三、项目架构

  • 四、核心功能

  • 五、环境配置

  • 六、代码详解

  • 七、使用示例

  • 八、性能优化

  • 九、常见问题

  • 十、总结


一、项目概述

1.1 项目简介

service_ai_pipe 是一个基于 AI 大模型的商品图像分析和标签生成服务,主要功能包括:

  • ✅ 智能图片筛选:使用 Qwen 视觉模型自动筛选高质量商品图片

  • ✅ 多阶段处理:Step 1 初筛 → Step 1.5 剔除多主体 → Step 2 精选图片并初打标签

  • ✅ Google Lens 搜索:对选出的图片进行反向图片搜索,获取相似商品信息

  • ✅ 综合标签生成:结合原商品信息和 Google Lens 结果,生成最终标签和商品简介

  • ✅ 消息队列集成:从 RabbitMQ 消费任务,处理完成后发送消息通知

  • ✅ 优雅中断:支持 Ctrl+C 优雅退出,确保数据安全

  • ✅ 断点续传:支持中断后从断点继续处理

1.2 项目结构

service_ai_pipe/
├── __init__.py                          # 包初始化
├── main.py                              # 主程序入口
├── ai_processor.py                      # AI 处理核心逻辑
├── db_service.py                        # 数据库操作服务
├── pre_download_images.py               # 图片预下载脚本(可选)
├── supplement_google_lens_and_step4.py  # 补回 Google Lens 和 Step4 脚本
├── diagnose_performance.py              # 性能诊断工具
├── analyze_fallback_performance.py      # 性能分析工具
├── apply_performance_optimization.py    # 性能优化应用
├── README.md                            # 项目文档
├── 配置说明.md                          # 配置说明
├── CONFIG.md                            # 配置参考
└── CSDN博客教程.md                      # 本文档

二、技术栈

2.1 核心技术

技术 版本 用途
Python 3.8+ 编程语言
asyncio 内置 异步编程框架
Qwen (通义千问) 最新 AI 视觉模型和文本生成
Google Lens API Decodo 反向图片搜索
SQLAlchemy 2.0+ ORM 数据库操作
RabbitMQ 最新 消息队列
loguru 最新 日志记录

2.2 AI 模型配置

  • Step 1/1.5/2:使用 qwen3-vl-flash(快速模型,适合大量图片筛选)

  • Step 4(最终标签):使用 qwen3-max(高精度模型,适合最终标签生成)


三、项目架构

3.1 架构图

┌─────────────────────────────────────────────────────────┐
│                  service_ai_pipe                         │
├─────────────────────────────────────────────────────────┤
│                                                           │
│  ┌──────────────┐    ┌──────────────┐                  │
│  │   main.py    │───▶│ai_processor  │                  │
│  │  (入口)      │    │  (核心逻辑)  │                  │
│  └──────────────┘    └──────────────┘                  │
│         │                    │                           │
│         │                    ▼                           │
│         │            ┌──────────────┐                  │
│         │            │  Qwen API    │                  │
│         │            │  Google Lens │                  │
│         │            └──────────────┘                  │
│         │                    │                           │
│         │                    ▼                           │
│         │            ┌──────────────┐                  │
│         └───────────▶│ db_service   │                  │
│                      │  (数据库)    │                  │
│                      └──────────────┘                  │
│                                                           │
└─────────────────────────────────────────────────────────┘
         │                    │                    │
         ▼                    ▼                    ▼
    ┌─────────┐         ┌─────────┐         ┌─────────┐
    │RabbitMQ │         │  MySQL  │         │ 图片URL │
    │  Queue  │         │Database │         │  存储   │
    └─────────┘         └─────────┘         └─────────┘

3.2 数据流

RabbitMQ (product.new 消息)
    ↓
获取待处理商品
    ↓
选择图片(主图3 + 质检图3 + SKU图3)
    ↓
Step 1: 初筛(剔除垃圾图)
    ↓
Step 1.5: 剔除多主体/拼接图
    ↓
Step 2: 精选 1-3 张最佳图片 + 初打标签
    ↓
Google Lens 搜索(对选出的第一张图片)
    ↓
Step 4: 综合标签生成(结合原商品信息 + Google Lens 结果)
    ↓
更新数据库(tags, pic_url, introduce)
    ↓
RabbitMQ (product.labeled 消息)

3.3 核心流程

开始
  ↓
从 RabbitMQ 消费任务
  ↓
获取商品数据
  ↓
选择图片(最多9张)
  ↓
Step 1: 初筛(剔除垃圾图)
  ↓
Step 1.5: 剔除多主体/拼接图
  ↓
Step 2: 精选 1-3 张最佳图片 + 初打标签
  ↓
是否紧急模式?
  ├─ 是 → 直接使用 Step2 结果,跳过 Google Lens 和 Step4
  └─ 否 → 继续
      ↓
      Google Lens 搜索
      ↓
      Step 4: 综合标签生成
      ↓
更新数据库
  ↓
发送消息到 RabbitMQ
  ↓
完成

四、核心功能

4.1 图片选择

从商品的 image_urls 中选择最多 9 张图片:

  • 主图:最多 3 张

  • 质检图(QC图):最多 3 张

  • SKU图:最多 3 张

关键代码

def select_images(image_urls: Dict[str, List[str]], max_per_type: int = 3) -> List[str]:
    """选择图片"""
    selected = []
    
    # 主图
    main_images = image_urls.get("main_images", [])[:max_per_type]
    selected.extend(main_images)
    
    # 质检图
    qc_images = image_urls.get("qc_images", [])[:max_per_type]
    selected.extend(qc_images)
    
    # SKU图
    sku_images = image_urls.get("sku_images", [])[:max_per_type]
    selected.extend(sku_images)
    
    return selected[:9]  # 最多9张

4.2 Step 1: 初筛(剔除垃圾图)

使用 Qwen 视觉模型剔除垃圾图片(模糊、无关、低质量等)。

Prompt 示例

请分析这组图片,剔除以下类型的图片:
1. 模糊、不清晰
2. 与商品无关
3. 低质量、损坏
4. 重复图片

返回保留的图片索引列表。

4.3 Step 1.5: 剔除多主体/拼接图

剔除包含多个主体或拼接的图片,确保每张图片只包含一个商品主体。

Prompt 示例

请分析这组图片,剔除以下类型的图片:
1. 包含多个商品主体
2. 拼接图(多个商品拼在一起)
3. 商品列表图

返回保留的图片索引列表(每张图片应该只包含一个商品主体)。

4.4 Step 2: 精选图片并初打标签

从剩余图片中精选 1-3 张最佳图片,并提取初步标签(分类、品牌、适用人群、季节、使用场景等)。

Prompt 示例

请从这组图片中精选 1-3 张最佳图片(清晰、完整展示商品),并提取以下信息:
1. 分类(从预定义列表中选择)
2. 品牌
3. 适用人群(Men/Women/Unisex Adult/Kids/Baby/Maternity/General)
4. 季节(Spring/Summer/Autumn/Winter/...)
5. 使用场景(Daily Casual/Business/Home/...)

返回选中的图片索引和标签信息(包含置信度)。

4.5 Google Lens 搜索

对选出的第一张图片进行 Google Lens 反向图片搜索,获取前 10 条相似商品信息。

关键代码

async def search_google_lens(self, image_url: str) -> List[str]:
    """Google Lens 搜索"""
    async with self.google_lens_semaphore:
        _stage_inc("google_lens")
        try:
            titles = await self.google_lens_client.search_by_image_url(
                image_url=image_url,
                max_results=10
            )
            return titles
        finally:
            _stage_dec("google_lens")

4.6 Step 4: 综合标签生成

结合原商品信息和 Google Lens 搜索结果,生成最终标签和商品简介。

标签选择策略

  • 当匹配分数 >= 81 时,使用 Step 2 初打标签的值(value

  • 当匹配分数 < 81 时,使用 Google Lens 提取的值(instand

Prompt 示例

请结合以下信息生成最终标签和商品简介:

原商品信息:
- 名称:{product_name}
- 描述:{product_description}

Step 2 初打标签:
- 分类:{classification}(置信度:{confidence})
- 品牌:{brand}(置信度:{confidence})
- ...

Google Lens 搜索结果:
{titles}

请生成:
1. 最终分类(从预定义列表中选择)
2. 最终属性(品牌、适用人群、季节、使用场景、型号)
3. 商品简介(50-100字,突出商品特点和卖点)

注意:当 Step 2 置信度 >= 81 时,优先使用 Step 2 的值;否则使用 Google Lens 的结果。

4.7 紧急模式

紧急模式下会跳过 Google Lens 和 Step4,直接使用 Step2 的结果。

启用方式

AI_PIPE_SKIP_GOOGLE_LENS_AND_STEP4=true

特点

  • 处理速度更快(跳过 Google Lens 和 Step4)

  • 商品 introduce 字段会被设置为 "000" 作为标记

  • 后续可以通过补回脚本补充 Google Lens 和 Step4 处理

4.8 补回 Google Lens 和 Step4

对所有符合条件的商品执行 Google Lens 和 Step4 处理,即使之前已经处理过也会重新更新。

识别条件

  • tags 不为空

  • pic_url 不为空

  • status = 0(未删除)

运行方式

python3 -m service_ai_pipe.supplement_google_lens_and_step4

五、环境配置

5.1 安装依赖

# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate  # Linux/Mac
# 或
venv\Scripts\activate  # Windows

# 安装依赖
pip install sqlalchemy aiomysql loguru httpx pydantic-settings
pip install aiohttp pika

5.2 环境变量配置

创建 .env 文件:

# Qwen API 配置
QWEN_API_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
QWEN_API_KEY=sk-xxx  # 替换为你的实际 API Key

# Qwen 模型配置
QWEN_MODEL_STEP1=qwen3-vl-flash      # Step 1 初筛(快速模型)
QWEN_MODEL_STEP1_5=qwen3-vl-flash   # Step 1.5 剔除多主体(快速模型)
QWEN_MODEL_STEP2=qwen3-vl-flash      # Step 2 精选图片(快速模型)
QWEN_MODEL_FINAL=qwen3-max          # 综合标签生成(高精度模型)
QWEN_MAX_RETRIES=3
QWEN_TIMEOUT=120

# Google Lens API 配置(Decodo API)
GOOGLE_LENS_API_KEY=Basic xxx  # 替换为你的实际 API Key(注意:需要包含 "Basic " 前缀)
DECODO_API_ENDPOINT=https://scraper-api.decodo.com/v2/scrape
DECODO_MAX_RESULTS=10
DECODO_TIMEOUT=120
DECODO_REQUEST_INTERVAL=0.05

# 图片下载配置
IMAGES_TEMP_DIR=temp_images              # 临时图片保存目录
MAX_IMAGES_PER_TYPE=3                    # 每种类型最多选择几张(主图/质检/SKU)
MAX_INPUT_IMAGES_PER_PRODUCT=9          # 每个商品最多输入几张(3+3+3=9)

# 标签选择阈值
TAG_CONFIDENCE_THRESHOLD=81  # >= 81 使用 value(初打标签),< 81 使用 instand(Google Lens)

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=findqc_db

# RabbitMQ 配置
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

# 日志级别
LOG_LEVEL=INFO

# 紧急模式(跳过 Google Lens 和 Step4)
AI_PIPE_SKIP_GOOGLE_LENS_AND_STEP4=false

# 并发控制
AI_PIPE_STEP1_MAX_CONCURRENT=5          # Step 1 并发数
AI_PIPE_STEP1_5_MAX_CONCURRENT=5        # Step 1.5 并发数
AI_PIPE_STEP2_MAX_CONCURRENT=5          # Step 2 并发数
AI_PIPE_GOOGLE_LENS_MAX_CONCURRENT=3    # Google Lens 并发数
AI_PIPE_STEP4_MAX_CONCURRENT=3          # Step 4 并发数

5.3 API Key 获取

5.3.1 Qwen API Key
  1. 访问 阿里云 DashScope 控制台

  2. 创建 API Key

  3. 格式:sk-xxxxxxxxxxxxx

5.3.2 Google Lens API Key (Decodo)
  1. 访问 Decodo API 官网

  2. 注册账号并获取 API Key

  3. 格式:Basic xxxxxxxxxxxxxx(注意:需要包含 "Basic " 前缀)


六、代码详解

6.1 AI 处理器(ai_processor.py)

6.1.1 步骤级并发控制

使用 asyncio.Semaphore 控制每个步骤的并发数,避免 API 限流。

class AIProcessor:
    """AI 处理器(支持步骤级并发控制)"""
    
    def __init__(
        self,
        step1_semaphore: Optional[asyncio.Semaphore] = None,
        step1_5_semaphore: Optional[asyncio.Semaphore] = None,
        step2_semaphore: Optional[asyncio.Semaphore] = None,
        google_lens_semaphore: Optional[asyncio.Semaphore] = None,
        step4_semaphore: Optional[asyncio.Semaphore] = None,
    ):
        self.step1_semaphore = step1_semaphore or asyncio.Semaphore(5)
        self.step1_5_semaphore = step1_5_semaphore or asyncio.Semaphore(5)
        self.step2_semaphore = step2_semaphore or asyncio.Semaphore(5)
        self.google_lens_semaphore = google_lens_semaphore or asyncio.Semaphore(3)
        self.step4_semaphore = step4_semaphore or asyncio.Semaphore(3)
6.1.2 Step 1: 初筛
async def step1_filter_images(self, image_urls: List[str]) -> List[str]:
    """Step 1: 初筛(剔除垃圾图)"""
    async with self.step1_semaphore:
        _stage_inc("step1")
        try:
            prompt = """
            请分析这组图片,剔除以下类型的图片:
            1. 模糊、不清晰
            2. 与商品无关
            3. 低质量、损坏
            4. 重复图片
            
            返回保留的图片索引列表(从0开始)。
            """
            
            response = await self.qwen_client.analyze_images(
                image_urls=image_urls,
                prompt=prompt,
                model=settings.qwen_model_step1
            )
            
            # 解析响应,获取保留的图片索引
            kept_indices = self._parse_indices(response)
            kept_images = [image_urls[i] for i in kept_indices if i < len(image_urls)]
            
            return kept_images
        finally:
            _stage_dec("step1")
6.1.3 Step 2: 精选图片并初打标签

python

复制

下载

async def step2_select_and_tag(
    self,
    image_urls: List[str],
    product_name: str,
    product_description: str
) -> Tuple[List[str], Dict[str, Any]]:
    """Step 2: 精选图片并初打标签"""
    async with self.step2_semaphore:
        _stage_inc("step2")
        try:
            prompt = f"""
            请从这组图片中精选 1-3 张最佳图片(清晰、完整展示商品),并提取以下信息:
            
            商品名称:{product_name}
            商品描述:{product_description}
            
            请提取:
            1. 分类(从以下列表中选择):{CATEGORIES}
            2. 品牌
            3. 适用人群(从以下列表中选择):{TARGET_AUDIENCE_OPTIONS}
            4. 季节(从以下列表中选择):{SEASON_OPTIONS}
            5. 使用场景(从以下列表中选择):{ENVIRONMENT_OPTIONS}
            
            返回格式:
            {{
                "selected_indices": [0, 1],
                "classification": {{"value": "Tops", "confidence": 85}},
                "attributes": {{
                    "brand": {{"value": "Nike", "confidence": 90}},
                    "target_audience": {{"value": "Men", "confidence": 95}},
                    "season": {{"value": "Spring", "confidence": 80}},
                    "environment": {{"value": "Daily Casual", "confidence": 85}}
                }}
            }}
            """
            
            response = await self.qwen_client.analyze_images(
                image_urls=image_urls,
                prompt=prompt,
                model=settings.qwen_model_step2
            )
            
            # 解析响应
            result = self._parse_step2_response(response)
            selected_indices = result.get("selected_indices", [])
            selected_images = [image_urls[i] for i in selected_indices if i < len(image_urls)]
            
            return selected_images, result
        finally:
            _stage_dec("step2")
6.1.4 Step 4: 综合标签生成

python

复制

下载

async def step4_generate_final_tags(
    self,
    product_name: str,
    product_description: str,
    step2_classification: Dict[str, Any],
    step2_attributes: Dict[str, Any],
    google_lens_titles: List[str]
) -> Tuple[Dict[str, Any], str]:
    """Step 4: 综合标签生成"""
    async with self.step4_semaphore:
        _stage_inc("step4")
        try:
            prompt = f"""
            请结合以下信息生成最终标签和商品简介:
            
            原商品信息:
            - 名称:{product_name}
            - 描述:{product_description}
            
            Step 2 初打标签:
            - 分类:{step2_classification}
            - 属性:{step2_attributes}
            
            Google Lens 搜索结果:
            {chr(10).join(google_lens_titles)}
            
            请生成:
            1. 最终分类(从以下列表中选择):{CATEGORIES}
            2. 最终属性(品牌、适用人群、季节、使用场景、型号)
            3. 商品简介(50-100字,突出商品特点和卖点)
            
            注意:
            - 当 Step 2 置信度 >= {settings.tag_confidence_threshold} 时,优先使用 Step 2 的值
            - 否则使用 Google Lens 的结果
            """
            
            response = await self.qwen_client.generate_text(
                prompt=prompt,
                model=settings.qwen_model_final
            )
            
            # 解析响应
            final_tags, introduce = self._parse_step4_response(response)
            
            return final_tags, introduce
        finally:
            _stage_dec("step4")

6.2 主程序(main.py)

6.2.1 任务处理流程

python

复制

下载

async def process_product_task(task: TaskProduct):
    """处理单个商品任务"""
    try:
        # 1. 获取商品数据
        async with db.async_session_maker() as session:
            product = await AIPipeDBService.get_product(session, task.product_id)
            if not product:
                logger.warning(f"商品不存在: product_id={task.product_id}")
                return
        
        # 2. 选择图片
        image_urls = product.image_urls or {}
        selected_images = select_images(image_urls, max_per_type=3)
        
        if not selected_images:
            logger.warning(f"商品没有图片: findqc_id={product.findqc_id}")
            return
        
        # 3. AI 处理
        result = await ai_processor.process_product(
            findqc_id=product.findqc_id,
            product_id=product.id,
            product_name=product.name or "",
            product_description=product.description or "",
            image_urls_list=selected_images
        )
        
        # 4. 更新数据库
        async with db.async_session_maker() as session:
            await AIPipeDBService.update_product_ai_results(
                session=session,
                product_id=product.id,
                tags=result["tags"],
                pic_url=result["pic_url"],
                introduce=result["introduce"]
            )
            await session.commit()
        
        # 5. 发送消息到 RabbitMQ
        await send_product_labeled_message(product.id, product.findqc_id)
        
    except Exception as e:
        logger.error(f"处理商品任务失败: {e}")
        _log_failed_product(...)
6.2.2 主循环

python

复制

下载

async def main_loop():
    """主循环:从数据库轮询待处理任务"""
    while not shutdown_manager.is_shutdown_requested():
        try:
            # 查询待处理任务
            async with db.async_session_maker() as session:
                tasks = await AIPipeDBService.get_pending_tasks(
                    session=session,
                    limit=10,
                    require_mainimg=_require_mainimg
                )
            
            if not tasks:
                # 没有任务,等待一段时间
                await asyncio.sleep(5)
                continue
            
            # 并发处理任务
            semaphore = asyncio.Semaphore(10)  # 最多10个并发任务
            
            async def process_task_safe(task):
                async with semaphore:
                    await process_product_task(task)
            
            await asyncio.gather(
                *[process_task_safe(task) for task in tasks],
                return_exceptions=True
            )
            
        except Exception as e:
            logger.error(f"主循环异常: {e}")
            await asyncio.sleep(5)

6.3 数据库服务(db_service.py)

6.3.1 获取待处理任务
@staticmethod
async def get_pending_tasks(
    session: AsyncSession,
    limit: int = 10,
    require_mainimg: bool = False
) -> List[TaskProduct]:
    """获取待处理任务"""
    stmt = (
        select(TaskProduct)
        .join(Product, TaskProduct.findqc_id == Product.findqc_id)
        .where(
            TaskProduct.status == 0,  # 未处理
            Product.status == 0,      # 未删除
            or_(
                Product.tags.is_(None),
                Product.tags == {}
            )  # 还没有标签
        )
    )
    
    if require_mainimg:
        stmt = stmt.where(Product.mainImg.isnot(None), Product.mainImg != '')
    
    stmt = stmt.limit(limit)
    
    result = await session.execute(stmt)
    return list(result.scalars().all())
6.3.2 更新 AI 处理结果
@staticmethod
async def update_product_ai_results(
    session: AsyncSession,
    product_id: int,
    tags: Optional[Dict[str, Any]],
    pic_url: Optional[str],
    introduce: Optional[str],
) -> bool:
    """更新商品的 AI 处理结果"""
    try:
        stmt = (
            update(Product)
            .where(Product.id == product_id)
            .values(
                tags=tags,
                pic_url=pic_url,
                introduce=introduce,
                last_update=datetime.now(CHINA_TZ)
            )
        )
        await session.execute(stmt)
        await session.flush()
        return True
    except Exception as e:
        logger.error(f"更新商品失败: {e}")
        return False

七、使用示例

7.1 启动服务

bash

复制

下载

# 激活虚拟环境
source venv/bin/activate

# 启动 AI 处理管道服务
python -m service_ai_pipe.main

7.2 紧急模式

bash

复制

下载

# 设置环境变量
export AI_PIPE_SKIP_GOOGLE_LENS_AND_STEP4=true

# 启动服务
python -m service_ai_pipe.main

7.3 补回 Google Lens 和 Step4

# 对所有符合条件的商品执行 Google Lens 和 Step4 处理
python3 -m service_ai_pipe.supplement_google_lens_and_step4

7.4 图片预下载(可选)

# 预先下载所有需要的图片到本地
python3 -m service_ai_pipe.pre_download_images

7.5 查看日志

# 控制台输出(实时)
# 日志会自动输出到控制台

# 文件日志(按天轮转)
tail -f logs/ai_pipe_2024-05-20.log

八、性能优化

8.1 步骤级并发控制

使用 asyncio.Semaphore 控制每个步骤的并发数,避免 API 限流:

# 配置建议
AI_PIPE_STEP1_MAX_CONCURRENT=5          # Step 1 并发数
AI_PIPE_STEP1_5_MAX_CONCURRENT=5        # Step 1.5 并发数
AI_PIPE_STEP2_MAX_CONCURRENT=5          # Step 2 并发数
AI_PIPE_GOOGLE_LENS_MAX_CONCURRENT=3    # Google Lens 并发数(API 限流较严格)
AI_PIPE_STEP4_MAX_CONCURRENT=3          # Step 4 并发数(使用高精度模型,较慢)

8.2 模型选择

  • Step 1/1.5/2:使用快速模型 qwen3-vl-flash,适合大量图片筛选

  • Step 4:使用高精度模型 qwen3-max,适合最终标签生成

8.3 图片选择优化

  • 限制每种类型最多选择 3 张图片

  • 每个商品最多输入 9 张图片(3+3+3)

  • 减少 API 调用次数,提升处理速度

8.4 数据库优化

  • 使用索引优化查询性能

  • 批量提交事务,减少数据库压力

  • 使用连接池管理数据库连接

8.5 紧急模式

紧急模式下跳过 Google Lens 和 Step4,处理速度提升 2-3 倍:

AI_PIPE_SKIP_GOOGLE_LENS_AND_STEP4=true

后续可以通过补回脚本补充 Google Lens 和 Step4 处理。


九、常见问题

9.1 Qwen API 调用失败

问题Qwen API 调用失败: 401 Unauthorized

解决

  • 检查 .env 文件中的 QWEN_API_KEY 是否正确

  • 确认 API Key 是否有效(访问 DashScope 控制台 检查)

9.2 Google Lens API 调用失败

问题Google Lens API 调用失败: 401 Unauthorized

解决

  • 检查 .env 文件中的 GOOGLE_LENS_API_KEY 是否正确

  • 确认 API Key 格式是否正确(需要包含 "Basic " 前缀)

  • 检查 Decodo API 账户余额

9.3 图片下载失败

问题图片下载失败: Connection timeout

解决

  • 检查网络连接

  • 增加超时时间(修改 DECODO_TIMEOUT 配置)

  • 使用图片预下载脚本预先下载图片

9.4 处理速度慢

问题:处理速度太慢

解决

  • 增加并发数(修改 AI_PIPE_*_MAX_CONCURRENT 配置)

  • 使用紧急模式(跳过 Google Lens 和 Step4)

  • 使用快速模型(qwen3-vl-flash)进行图片筛选

9.5 标签质量不高

问题:生成的标签不准确

解决

  • 使用高精度模型(qwen3-max)进行最终标签生成

  • 启用 Google Lens 搜索(提供更多参考信息)

  • 调整 TAG_CONFIDENCE_THRESHOLD 阈值

9.6 内存占用过高

问题:内存占用过高

解决

  • 减少并发数(降低 AI_PIPE_*_MAX_CONCURRENT 配置)

  • 限制图片数量(减少 MAX_IMAGES_PER_TYPE 配置)

  • 及时清理临时文件


十、总结

10.1 项目亮点

  1. 智能图片筛选:使用 AI 视觉模型自动筛选高质量商品图片

  2. 多阶段处理:Step 1 → Step 1.5 → Step 2 → Google Lens → Step 4,确保标签质量

  3. 高性能:使用异步编程和步骤级并发控制,大幅提升处理效率

  4. 灵活配置:支持紧急模式、补回处理等多种运行模式

  5. 稳定性:优雅中断、错误处理、日志记录等机制确保系统稳定运行

10.2 适用场景

  • 电商平台商品标签自动生成

  • 商品图像分析和分类

  • 反向图片搜索应用

  • AI 驱动的商品推荐系统

10.3 后续优化方向

  • 支持更多 AI 模型(GPT-4V、Claude 等)

  • 添加标签质量评估和反馈机制

  • 优化图片选择算法

  • 支持批量处理模式

  • 添加监控和统计功能


📚 参考资料


作者:MadPrinter
最后更新:2025-12-26
项目地址GitHub


💡 提示:如果本文对你有帮助,欢迎点赞、收藏、转发!如有问题,欢迎在评论区留言讨论。

更多推荐