AI计算机视觉实践笔记

计算机视觉图像处理 CV

作者: 醉月思发布日期: 2023-07-25 分类: 技术实践

像 Google Vision API、阿里云内容审核这些，确实能用，但有几个问题：

成本问题：商用 API 都要钱，量大之后账单会让你怀疑人生
数据隐私：公司内部的敏感图片，不想传到第三方
定制化需求：我们的违规标准跟通用的不太一样，需要自定义检测逻辑
性能要求：需要实时响应，而 API 调用有网络延迟

所以我们决定自己搭建一个 CV pipeline。

从一个真实的需求开始

最近在做一个内容审核系统，需要自动识别用户上传的图片中是否包含违规内容。刚开始以为这事很简单，随便找个现成的模型调用一下就完了。结果折腾了一圈才发现，从"图像输入"到"认知判断"之间，有一个很长的 pipeline 需要自己搭建。

这篇文章就是记录我在搭建这个 AI 计算机视觉 pipeline 过程中踩过的坑，以及一些实用的解决方案。如果你也在做类似的图像处理项目，希望这些经验能帮你省点时间。

为什么需要自己的 pipeline

直接调用 API 不好吗？像 Google Vision API、阿里云内容审核这些，确实能用，但有几个问题：

成本问题：商用 API 都要钱，量大之后账单会让你怀疑人生
数据隐私：公司内部的敏感图片，不想传到第三方
定制化需求：我们的违规标准跟通用的不太一样，需要自定义检测逻辑
性能要求：需要实时响应，而 API 调用有网络延迟

所以我们决定自己搭建一个 CV pipeline。

整体架构

先看下最终的架构图：

graph TD A[图片上传] --> B[图像预处理] B --> C[多模型推理] C --> D[后处理与融合] D --> E[规则引擎] E --> F[结果输出] C -->|并行| C1[NSFW 检测] C -->|并行| C2[文字识别 OCR] C -->|并行| C3[物体检测] C -->|并行| C4[人脸识别]

这个 pipeline 的核心思想是：不要试图用一个模型解决所有问题，而是针对不同任务用专门的模型，最后用规则引擎做决策。

实现细节

1. 图像预处理

这一步容易被忽视，但其实很重要。原始图片可能质量很差，直接影响后续的检测效果。

import cv2
import numpy as np

def preprocess_image(image_path, max_size=2048):
    """图像预处理：调整大小、去噪、增强对比度"""
    img = cv2.imread(image_path)
    if img is None:
        raise ValueError("无法读取图片")

    # 保持宽高比调整大小
    h, w = img.shape[:2]
    scale = min(max_size / max(h, w), 1.0)
    img = cv2.resize(img, (int(w * scale), int(h * scale)))

    # 去噪
    img = cv2.fastNlMeansDenoisingColored(img, None, 10, 10, 7, 21)

    # 增强对比度
    lab = cv2.cvtColor(img, cv2.COLOR_BGR2LAB)
    l, a, b = cv2.split(lab)
    l = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)).apply(l)
    img = cv2.cvtColor(cv2.merge([l, a, b]), cv2.COLOR_LAB2BGR)

    return img

踩坑：一开始我直接把原图送进模型，结果有些超大尺寸的图片把 GPU 显存撑爆了。后来加了个尺寸限制，但要注意保持宽高比，否则物体变形会影响检测精度。

2. 多模型并行推理

不同的任务用不同的模型，但可以并行处理来提高吞吐量：

from concurrent.futures import ThreadPoolExecutor
import torch

class CVModelPool:
    """模型池，管理多个 CV 模型的推理"""
    def __init__(self):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.models = {}
        self._load_models()

    def _load_models(self):
        """加载所有需要的模型"""
        # NSFW 检测模型
        self.models['nsfw'] = self._load_nsfw_model()

        # OCR 模型
        self.models['ocr'] = self._load_ocr_model()

        # 物体检测模型
        self.models['detection'] = self._load_detection_model()

    def infer(self, image, model_name):
        """指定模型推理"""
        return self.models[model_name].predict(image)

    def infer_all(self, image):
        """所有模型并行推理"""
        results = {}
        with ThreadPoolExecutor(max_workers=len(self.models)) as executor:
            futures = {
                name: executor.submit(self.infer, image, name)
                for name in self.models.keys()
            }
            for name, future in futures.items():
                results[name] = future.result()
        return results

踩坑：多线程在 PyTorch 中要注意，CUDA 操作不是线程安全的。我用 ThreadPoolExecutor 只是为了在 CPU 部分并行，真正的推理还是串行的。如果要真正并行推理，需要用多进程。

3. 后处理与结果融合

不同模型的结果需要统一格式，然后融合判断：

class ResultFusion:
    """结果融合器"""
    def __init__(self):
        self.weights = {
            'nsfw': 0.8,
            'ocr': 0.9,
            'detection': 0.7,
            'face': 0.6
        }

    def fuse_results(self, raw_results):
        """融合多个模型的推理结果"""
        final_score = 0.0
        reasons = []

        # NSFW 检测
        if raw_results['nsfw']['nsfw_score'] > 0.7:
            final_score += self.weights['nsfw']
            reasons.append("疑似违规内容")

        # OCR 文字检测
        forbidden_words = self._check_forbidden_words(
            raw_results['ocr']['text']
        )
        if forbidden_words:
            final_score += self.weights['ocr']
            reasons.append(f"检测到敏感词: {', '.join(forbidden_words)}")

        # 物体检测
        forbidden_objects = self._check_forbidden_objects(
            raw_results['detection']['objects']
        )
        if forbidden_objects:
            final_score += self.weights['detection']
            reasons.append(f"检测到违禁物品: {', '.join(forbidden_objects)}")

        # 人脸识别
        if raw_results['face']['person_count'] > 1:
            final_score += self.weights['face'] * 0.3
            reasons.append("多人场景")

        return {
            'score': min(final_score, 1.0),
            'reasons': reasons,
            'action': self._decide_action(final_score)
        }

    def _decide_action(self, score):
        """根据分数决定后续动作"""
        if score >= 0.8:
            return 'reject'
        elif score >= 0.5:
            return 'manual_review'
        else:
            return 'approve'

踩坑：一开始是简单叠加各个模型的分数，但这样会忽略了不同模型的置信度差异。后来给每个模型加了权重，并且用阈值做决策，效果好了很多。

4. 性能优化

这个 pipeline 性能是个大问题，特别是要处理大量图片时。

几个优化点：

模型量化：把 FP32 模型量化到 INT8，推理速度提升 2-3 倍，精度损失很小
批量推理：把多张图片打包成 batch，一次推理处理
模型剪枝：去掉模型中不重要的通道，减少计算量
缓存机制：对相同内容的图片缓存结果，避免重复计算

import hashlib
import pickle
from functools import lru_cache

class CachedInference:
    """带缓存的推理"""
    def __init__(self, model, cache_dir='./cache'):
        self.model = model
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def _get_image_hash(self, image):
        """计算图片的哈希值用于缓存"""
        return hashlib.md5(image.tobytes()).hexdigest()

    def predict(self, image):
        """带缓存的预测"""
        img_hash = self._get_image_hash(image)
        cache_file = os.path.join(self.cache_dir, f"{img_hash}.pkl")

        # 尝试从缓存读取
        if os.path.exists(cache_file):
            with open(cache_file, 'rb') as f:
                return pickle.load(f)

        # 缓存未命中，进行推理
        result = self.model.predict(image)

        # 存入缓存
        with open(cache_file, 'wb') as f:
            pickle.dump(result, f)

        return result

踩坑：缓存不是万能的。有些用户会修改图片的一两个像素来绕过检测，这时候缓存反而会帮助作弊。所以对缓存要设置过期时间，并且定期清理。

部署与监控

使用 Docker 部署

为了保证环境一致性，我用 Docker 来部署这个 pipeline：

FROM pytorch/pytorch:2.0.1-cuda11.7-cudnn8-runtime

RUN apt-get update && apt-get install -y \
    libgl1-mesa-glx \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY src/ /app/src/
WORKDIR /app

CMD ["python", "src/main.py"]

监控指标

部署后需要监控这些指标：

from prometheus_client import Counter, Histogram, Gauge

# 推理次数
inference_counter = Counter('cv_inference_total', 'Total CV inferences', ['model', 'status'])

# 推理耗时
inference_latency = Histogram('cv_inference_duration_seconds', 'CV inference duration', ['model'])

# GPU 使用率
gpu_usage = Gauge('gpu_usage_percent', 'GPU usage percentage')

# 内存使用
memory_usage = Gauge('memory_usage_bytes', 'Memory usage in bytes')

踩坑：监控很重要。有次模型推理时间突然变长，从 200ms 涨到 2s。通过监控发现是 GPU 被其他进程占满了，排查后解决了资源竞争问题。

效果总结

最后说说效果：

准确率：召回率 95%，精确率 92%，基本满足需求
性能：单张图片平均处理时间 300ms（包括所有模型推理）
成本：自建后每月成本只有 API 调用的 1/10
可控性：可以根据业务需求灵活调整检测规则

结语

搭建 CV pipeline 这一路踩了不少坑，但也学到了很多。核心感悟是：

不要迷信单一模型：组合多个专用模型比一个全能模型效果好
数据质量比模型复杂度更重要：好的预处理能提升整体效果
性能优化是个系统工程：不是改一两个参数就能解决的
监控和日志很重要：出现问题时要能快速定位

如果你也在做类似的 CV 项目，希望这些经验对你有帮助。计算机视觉不只是调调模型参数，更像是在搭积木——要把很多不同的组件组合起来，才能构建出一个完整的系统。

版权声明: 本文首发于指尖魔法屋-AI计算机视觉实践笔记（https://blog.thinkmoon.cn/post/326-ai-computer-vision-pipeline-image-cognition/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

数据增强：简单变换不够用了之后

数据增强：简单变换不够用了之后一旦进项目，好看的架构图就没那么管用了。