果蔬识别系统性能优化之路（五）

前情提要

果蔬识别系统性能优化之路（四）
剩下标题

• 新建store_feature表，关联storeCode和featureId表，对数据库举行规范化，创建一个新的表来映射storeCode与feature的关系，从而可以利用简朴的WHERE条件来充实利用索引
  
• 实现对特征向量ivf的增删改查
  

办理方案

新建storeFeature表

• 新建store表，storeFeature表
  

1. import { Entity, PrimaryGeneratedColumn, Column, OneToMany } from 'typeorm';
2. import { StoreFeature } from '../../feature/entities/store-feature.entity';
4. @Entity()
5. export class Store {
6.   @PrimaryGeneratedColumn()
7.   id: number;
9.   @Column({ unique: true })
10.   storeCode: string;
12.   @Column({ nullable: true })
13.   storeName: string;
15.   @OneToMany(() => StoreFeature, (storeFeature) => storeFeature.store)
16.   storeFeatures: StoreFeature[];
17. }

复制代码

1. import { Entity, ManyToOne, JoinColumn, PrimaryGeneratedColumn } from 'typeorm';
2. import { Store } from '../../store/entities/store.entity';
3. import { Feature } from './feature.entity';
5. @Entity()
6. export class StoreFeature {
7.   @PrimaryGeneratedColumn()
8.   id: number;
10.   @ManyToOne(() => Store, { onDelete: 'CASCADE' })
11.   @JoinColumn({ name: 'storeCode', referencedColumnName: 'storeCode' })
12.   store: Store;
14.   @ManyToOne(() => Feature, { onDelete: 'CASCADE' })
15.   @JoinColumn({ name: 'featureId', referencedColumnName: 'id' })
16.   feature: Feature;
17. }

复制代码

storeFeature表关联store表和feature表

• feature.service大改造
  

1. import { Injectable } from '@nestjs/common';
2. import { CreateFeatureDto } from './dto/create-feature.dto';
3. import { Feature } from './entities/feature.entity';
4. import { InjectRepository } from '@nestjs/typeorm';
5. import { Repository, In } from 'typeorm';
6. import { RedisService } from '../redis/redis.service';
7. import { HttpService } from '@nestjs/axios';
8. import { firstValueFrom } from 'rxjs';
9. import * as FormData from 'form-data';
10. import { Img } from '../img/entities/img.entity';
11. import { Store } from '../store/entities/store.entity';
12. import { StoreFeature } from './entities/store-feature.entity';
14. @Injectable()
15. export class FeatureService {
16.   constructor(
17.     @InjectRepository(Feature)
18.     private readonly featureRepository: Repository<Feature>,
19.     @InjectRepository(Img)
20.     private readonly imgRepository: Repository<Img>,
21.     @InjectRepository(Store)
22.     private readonly storeRepository: Repository<Store>,
23.     @InjectRepository(StoreFeature)
24.     private readonly storeFeatureRepository: Repository<StoreFeature>,
25.     private readonly httpService: HttpService,
26.     private readonly redisService: RedisService,
27.   ) {
28.   }
30.   /**
31.    * 创建
32.    * @param file
33.    * @param createFeatureDto
34.    * @param needSync //是否需要同步redis，默认为true
35.    */
36.   async create(file: Express.Multer.File, createFeatureDto: CreateFeatureDto, needSync: boolean = true): Promise<Feature> {
37.     const img = this.imgRepository.create({
38.       img: file.buffer,
39.     });
40.     await this.imgRepository.save(img);
41.     const [feature, store] = await Promise.all([
42.       new Promise(async (resolve) => {
43.         const feature: Feature = this.featureRepository.create({
44.           ...createFeatureDto,
45.           imgId: img.id,
46.         });
47.         await this.featureRepository.save(feature);
48.         resolve(feature);
49.       }),
50.       new Promise(async (resolve) => {
51.         let store = await this.storeRepository.findOne({ where: { storeCode: createFeatureDto.storeCode } });
52.         if (!store) {
53.           store = this.storeRepository.create({
54.             storeCode: createFeatureDto.storeCode,
55.             storeName: createFeatureDto.storeName,
56.           });
57.           await this.storeRepository.save(store);
58.         }
59.         resolve(store);
60.       }),
61.     ]);
62.     const storeFeature = this.storeFeatureRepository.create({
63.       feature,
64.       store,
65.     });
66.     await this.storeFeatureRepository.save(storeFeature);
67.     needSync && await this.syncRedis(createFeatureDto.storeCode);
68.     return feature as Feature;
69.   }
71.   /**
72.    * 同步redis
73.    * @param storeCode
74.    */
75.   async syncRedis(storeCode: string) {
76.     const url = 'http://localhost:5000/sync'; // Python 服务的 URL
77.     const s = Date.now();
78.     const response = await firstValueFrom(this.httpService.post(url, { storeCode }));
79.     const { ids } = response.data;
80.     await this.redisService.set(`${storeCode}-featureDatabase`, JSON.stringify(ids));
81.     const e = Date.now();
82.     console.log(`门店：${storeCode}，同步redis耗时：${e - s}ms`);
83.   }
85.   /**
86.    * 查询所有
87.    * @param storeCode
88.    * @param selectP
89.    */
90.   async findAll(storeCode: string, selectP?: string[]) {
91.     return await this.featureRepository
92.       .createQueryBuilder('feature')
93.       .select(selectP)
94.       .innerJoin(StoreFeature, 'storeFeature', 'feature.id = storeFeature.featureId')
95.       .innerJoin(Store, 'store', 'storeFeature.storeCode = store.storeCode')
96.       .where('store.storeCode = :storeCode', { storeCode })
97.       .getMany();
98.   }
100.   /**
101.    * 查询特性及其关联的图像
102.    * @param storeCode
103.    */
104.   async findAllWithImage(storeCode: string): Promise<Feature[]> {
105.     return await this.featureRepository.createQueryBuilder('feature')
106.       .leftJoinAndSelect('feature.img', 'img')
107.       .innerJoin(StoreFeature, 'storeFeature', 'feature.id = storeFeature.featureId')
108.       .innerJoin(Store, 'store', 'storeFeature.storeCode = store.storeCode')
109.       .where('store.storeCode = :storeCode', { storeCode })
110.       .getMany();
111.   }
113.   /**
114.    * 删除门店所有数据
115.    * @param storeCode
116.    */
117.   async removeAll(storeCode: string): Promise<void> {
118.     const store = await this.storeRepository.findOne({ where: { storeCode }, relations: ['storeFeatures'] });
119.     if (!store) {
120.       return;
121.     }
122.     // 批量删除 storeFeatures 和 store
123.     if (store.storeFeatures.length > 0) {
124.       await this.storeFeatureRepository
125.         .query('DELETE FROM store_feature WHERE id IN (?)', [store.storeFeatures.map(sf => sf.id)]);
126.     }
127.     await this.storeRepository.remove(store);  // 删除 store
128.     const unreferencedFeatures = await this.featureRepository
129.       .createQueryBuilder('feature')
130.       .leftJoinAndSelect('feature.img', 'img')
131.       .leftJoin('feature.storeFeatures', 'storeFeature')
132.       .where('storeFeature.id IS NULL') // 这里的条件确保我们只选择那些没有其他引用的 feature
133.       .getMany();
134.     // 批量删除未引用的 features
135.     if (unreferencedFeatures.length > 0) {
136.       for (const feature of unreferencedFeatures) {
137.         await this.remove(feature);
138.       }
139.     }
140.     await this.redisService.del(`${storeCode}-featureDatabase`);
141.     await this.syncRedis(storeCode);
142.   }
144.   /**
145.    * 预测
146.    * @param file
147.    * @param num
148.    * @param storeCode
149.    * @param justPredict
150.    * @param needList
151.    */
152.   async predict(
153.     file: Express.Multer.File,
154.     num: string = '5',
155.     storeCode: string,
156.     justPredict: string = 'false',
157.     needList: boolean = false,
158.   ) {
159.     const PYTHON_SERVICE_URL = 'http://localhost:5000/predict'; // Python service URL
160.     const REDIS_KEY_PREFIX = '-featureDatabase';
161.     const startTime = Date.now();
162.     const numInt = parseInt(num);
163.     const isJustPredict = justPredict === 'true';
165.     try {
166.       // Prepare form data
167.       const formData = new FormData();
168.       formData.append('file', file.buffer, file.originalname);
169.       formData.append('storeCode', storeCode);
170.       formData.append('justPredict', justPredict);
172.       // Send request to Python service
173.       const response = await firstValueFrom(this.httpService.post(PYTHON_SERVICE_URL, formData));
174.       const { features, index, predictTime } = response.data;
176.       if (isJustPredict) {
177.         return this.buildResponse([], features, predictTime, startTime, numInt);
178.       }
180.       // Retrieve feature database from Redis
181.       const featureDatabaseStr = await this.redisService.get(`${storeCode}${REDIS_KEY_PREFIX}`);
182.       if (!featureDatabaseStr) {
183.         return this.buildResponse([], features, predictTime, startTime, numInt);
184.       }
186.       // Parse the Redis result and filter the IDs
187.       const featureDatabase = JSON.parse(featureDatabaseStr);
188.       const ids = index
189.         .map((idx: number) => featureDatabase[idx]);
191.       if (!ids.length) {
192.         return this.buildResponse([], features, predictTime, startTime, numInt);
193.       }
195.       // Query for features in the database
196.       const featureList = await this.featureRepository.createQueryBuilder('feature')
197.         .where('feature.id IN (:...ids)', { ids })
198.         .orderBy(`FIELD(feature.id, ${ids.map((id: any) => `'${id}'`).join(', ')})`, 'ASC')
199.         .getMany();
201.       // Filter to ensure unique labels
202.       const uniqueList = this.filterUniqueFeatures(featureList, numInt);
204.       const result = this.buildResponse(uniqueList, features, predictTime, startTime, numInt);
205.       return needList ? { ...result, featureList: featureList.map(({ features, ...rest }) => rest) } : result;
206.     } catch (error) {
207.       throw new Error(`Prediction failed: ${error.message}`);
208.     }
209.   }
211.   private filterUniqueFeatures(featureList: any[], limit: number) {
212.     const uniqueList = [];
213.     for (const feature of featureList) {
214.       if (!uniqueList.some(f => f.label === feature.label)) {
215.         uniqueList.push(feature);
216.       }
217.       if (uniqueList.length === limit) break;
218.     }
219.     return uniqueList;
220.   }
222.   private buildResponse(list: any[], features: any, predictTime: string, startTime: number, num: number) {
223.     const totalTime = `${Date.now() - startTime}ms`;
224.     return {
225.       predictTime,
226.       [`top${num}`]: list.map(({ features, ...rest }) => rest),
227.       features,
228.       totalTime,
229.     };
230.   }
232.   /**
233.    * 计算余弦相似度
234.    * @param vecA
235.    * @param vecB
236.    */
237.   cosineSimilarity(vecA: number[], vecB: number[]): number {
238.     if (vecA.length !== vecB.length) {
239.       throw new Error('Vectors must be of the same length');
240.     }
241.     const dotProduct = vecA.reduce((sum, value, index) => sum + value * vecB[index], 0);
242.     const magnitudeA = Math.sqrt(vecA.reduce((sum, value) => sum + value * value, 0));
243.     const magnitudeB = Math.sqrt(vecB.reduce((sum, value) => sum + value * value, 0));
244.     return dotProduct / (magnitudeA * magnitudeB);
245.   }
247.   /**
248.    * 查找相似
249.    * @param inputFeatures
250.    * @param num
251.    * @param storeCode
252.    */
253.   async findTopNSimilar(inputFeatures: number[], num: number, storeCode: string): Promise<{
254.     label: string;
255.     similarity: number
256.   }[]> {
257.     const featureDatabaseStr = await this.redisService.get(`${storeCode}-featureDatabase`);
258.     if (!featureDatabaseStr) {
259.       return [];
260.     }
261.     const featureDatabase = JSON.parse(featureDatabaseStr);
262.     const similarities = featureDatabase.map(({ features, label }) => {
263.       let similarity = 0;
264.       if (features) {
265.         similarity = this.cosineSimilarity(inputFeatures, features);
266.       }
267.       return { label: label as string, similarity: similarity as number };
268.     });
270.     similarities.sort((a: { similarity: number; }, b: { similarity: number; }) => b.similarity - a.similarity);
272.     const uniqueLabels = new Set<string>();
273.     const topNUnique: { label: string; similarity: number; }[] = [];
274.     for (const item of similarities) {
275.       if (!uniqueLabels.has(item.label as string)) {
276.         uniqueLabels.add(item.label);
277.         item.similarity = Math.round(item.similarity * 100) / 100;
278.         topNUnique.push(item);
279.         if (topNUnique.length === num) break;
280.       }
281.     }
282.     return topNUnique;
283.   }
285.   /**
286.    * 根据名称查询
287.    * @param label
288.    * @param storeCode
289.    */
290.   async getByName(label: string, storeCode: string): Promise<Feature[]> {
291.     return await this.featureRepository
292.       .createQueryBuilder('feature')
293.       .leftJoinAndSelect('feature.img', 'img')
294.       .innerJoin(StoreFeature, 'storeFeature', 'feature.id = storeFeature.featureId')
295.       .innerJoin(Store, 'store', 'storeFeature.storeCode = store.storeCode')
296.       .where('store.storeCode = :storeCode', { storeCode })
297.       .andWhere('feature.label = :label', { label })
298.       .getMany();
299.   }
301.   /**
302.    * 根据名称向量个数查询
303.    * @param label
304.    * @param storeCode
305.    */
306.   async getCountByLabel(label: string, storeCode: string): Promise<number> {
307.     return await this.featureRepository
308.       .createQueryBuilder('feature')
309.       .leftJoinAndSelect('feature.img', 'img')
310.       .innerJoin(StoreFeature, 'storeFeature', 'feature.id = storeFeature.featureId')
311.       .innerJoin(Store, 'store', 'storeFeature.storeCode = store.storeCode')
312.       .where('store.storeCode = :storeCode', { storeCode })
313.       .andWhere('feature.label = :label', { label })
314.       .getCount();
315.   }
317.   /**
318.    * 批量学习
319.    * @param files
320.    * @param createFeatureDto
321.    */
322.   async batchStudy(files: Express.Multer.File[], createFeatureDto: CreateFeatureDto) {
323.     const list = [];
324.     for (const file of files) {
325.       try {
326.         const { features: f } = await this.predict(file, '5', createFeatureDto.storeCode, 'true');
327.         const feature = await this.create(file, {
328.           ...createFeatureDto,
329.           features: f,
330.         }, false);
331.         // 创建一个副本，不包含 `features` 属性
332.         const { features, ...featureWithoutFeatures } = feature;
333.         // 将不包含 `features` 属性的对象推送到数组中
334.         list.push(featureWithoutFeatures);
335.       } catch (e) {
336.         console.error(e);
337.       }
338.     }
339.     await this.syncRedis(createFeatureDto.storeCode);
340.     return list;
341.   }
343.   /**
344.    * 删除门店的特征值数据
345.    * @param feature
346.    */
347.   async remove(feature: Feature) {
348.     await this.featureRepository.remove(feature);
349.     await this.imgRepository.remove(feature.img);
350.   }
352.   /**
353.    * 批量删除
354.    * @param ids
355.    * @param storeCode
356.    */
357.   async batchRemove(ids: string, storeCode: string) {
358.     const list = ids.split(',').map(id => +id);
359.     // 批量查询所有相关的 Feature
360.     const features = await this.featureRepository.find({
361.       where: { id: In(list) },
362.       relations: ['img', 'storeFeatures'],
363.     });
364.     for (const feature of features) {
365.       feature && await this.remove(feature);
366.       await this.storeFeatureRepository.remove(feature.storeFeatures);
367.     }
368.     await this.syncRedis(storeCode);
369.   }
371.   /**
372.    * 导入数据
373.    * @param storeCode
374.    * @param sourceStoreCode
375.    * @param storeName
376.    */
377.   async importData(storeCode: string, sourceStoreCode?: string, storeName?: string) {
378.     let storeFeatures = [];
379.     // 第一步：查询指定 storeCode 关联的所有 featureId
380.     const storeFeatureIds = await this.storeFeatureRepository
381.       .createQueryBuilder('storeFeature')
382.       .select('storeFeature.featureId')
383.       .where('storeFeature.storeCode = :storeCode', { storeCode })
384.       .getRawMany();
386.     // 提取出 featureId 列表
387.     const featureIdsToExclude = storeFeatureIds.map(row => row.featureId);
388.     let distinctFeatureIds = [];
389.     if (featureIdsToExclude.length === 0) {
390.       distinctFeatureIds = await this.storeFeatureRepository
391.         .createQueryBuilder('storeFeature')
392.         .select('DISTINCT storeFeature.featureId')  // 确保 featureId 唯一
393.         .getRawMany();
394.     } else {
395.       // 第二步：排除这些 featureId，并确保 featureId 唯一
396.       distinctFeatureIds = await this.storeFeatureRepository
397.         .createQueryBuilder('storeFeature')
398.         .select('DISTINCT storeFeature.featureId')  // 确保 featureId 唯一
399.         .where('storeFeature.featureId NOT IN (:...featureIdsToExclude)', { featureIdsToExclude })  // 排除 featureId
400.         .getRawMany();
401.     }
402.     const featureIds = distinctFeatureIds.map(record => record.featureId);
403.     if (!sourceStoreCode) {
404.       storeFeatures = await this.featureRepository
405.         .createQueryBuilder('feature')
406.         .leftJoinAndSelect('feature.img', 'img')
407.         .whereInIds(featureIds)
408.         .getMany();
409.     } else {
410.       storeFeatures = await this.featureRepository
411.         .createQueryBuilder('feature')
412.         .leftJoinAndSelect('feature.img', 'img')
413.         .innerJoin('feature.storeFeatures', 'storeFeatures')
414.         .whereInIds(featureIds)
415.         .andWhere('storeFeatures.storeCode = :storeCode', { storeCode: sourceStoreCode })  // 使用参数化查询
416.         .getMany();
417.     }
418.     let targetStore = await this.storeRepository.findOne({ where: { storeCode: storeCode } });
419.     if (!targetStore) {
420.       targetStore = this.storeRepository.create({
421.         storeCode: storeCode,
422.         storeName: storeName,
423.       });
424.       await this.storeRepository.save(targetStore);
425.     }
426.     // Create new StoreFeature records for the target storeCode
427.     const newStoreFeatures = storeFeatures.map((feature: Feature) => ({
428.       store: targetStore,
429.       feature, // Reuse the existing feature
430.     }));
431.     // Save new StoreFeature records
432.     const storeFeatureInstances = this.storeFeatureRepository.create(newStoreFeatures);
433.     await this.storeFeatureRepository.save(storeFeatureInstances);
434.     await this.syncRedis(storeCode);
435.     return `同步完成，共导入${storeFeatures.length}条数据`;
436.   }
438.   async init() {
439.     const distinctStoreCodes = await this.storeRepository
440.       .createQueryBuilder('store')
441.       .select('store.storeCode')
442.       .distinct(true)
443.       .getRawMany();
444.     const syncList = [];
445.     for (const row of distinctStoreCodes) {
446.       const storeCode = row.store_storeCode;
447.       syncList.push(this.syncRedis(storeCode));
448.     }
449.     await Promise.all(syncList);
450.     console.log('初始化完成');
451.   }
452. }

复制代码

• 效果：并没有提升多少，但幸亏关系更清楚，为之后的拓展打了底子
  

实现ivf的动态增删改查

• 结论：ivf无法在不训练只增长的情况下举行新增向量的识别，所以每次新增向量必须重新举行训练和添加
  
• python端ivf改造
  detect.py（识别和同步方法）
  

1. import tensorflow as tf
2. from tensorflow.keras.applications import MobileNetV2
3. import numpy as np
4. import time
5. import gc
6. from ivf import IVFPQ
7. from feature import get_feature_by_store_code
8. import orjson
9. from concurrent.futures import ThreadPoolExecutor
11. # 加载预训练的 MobileNetV2 模型，不包含顶部的分类层
12. model = MobileNetV2(input_shape=(224, 224, 3), weights='imagenet', include_top=False, pooling='avg')
15. class MainDetect:
16.     # 初始化
17.     def __init__(self):
18.         super().__init__()
19.         # 模型初始化
20.         self.image_id = None
21.         self.image_features = None
22.         self.model = tf.keras.models.load_model("models/custom/my-model.h5")
23.         self.ivfObj = {}
25.     def classify_image(self, image_data, store_code, just_predict):
26.         # Load and preprocess image
27.         img = tf.image.decode_image(image_data, channels=3)
28.         img = tf.image.resize(img, [224, 224])
29.         img = tf.expand_dims(img, axis=0)  # Add batch dimension
31.         # Run model prediction
32.         start_time = time.time()
33.         outputs = model.predict(img)
34.         # outputs = self.model.predict(outputs)
35.         # prediction = tf.divide(outputs, tf.norm(outputs))
36.         i = []
37.         if just_predict == "false":
38.             if store_code + '-featureDatabase' in self.ivfObj:
39.                 i = self.ivfObj[store_code + '-featureDatabase'].search(outputs)
40.                 i = i.flatten().tolist()
42.         end_time = time.time()
44.         # Calculate elapsed time
45.         elapsed_time = end_time - start_time
47.         # Flatten the outputs and return them
48.         # output_data = prediction.numpy().flatten().tolist()
49.         output_data = outputs.flatten().tolist()
51.         # Force garbage collection to free up memory
52.         del img, outputs, end_time, start_time  # Ensure variables are deleted
53.         gc.collect()
55.         return {"outputs": output_data, "time": f"{elapsed_time * 1000:.2f}ms", "index": i}
57.     def sync(self, store_code):
58.         if store_code + '-featureDatabase' in self.ivfObj:
59.             del self.ivfObj[store_code + '-featureDatabase']
60.         data = get_feature_by_store_code(store_code)
62.         if len(data) == 0:
63.             return []
64.         else:
65.             def parse_features(item):
66.                 return orjson.loads(item['features'])
68.             with ThreadPoolExecutor() as executor:
69.                 features_list = list(executor.map(parse_features, data))
70.             # 提取所有特征并转换为 NumPy 数组
71.             features = np.array(features_list, dtype=np.float32)
72.             self.ivfObj[store_code + '-featureDatabase'] = IVFPQ(features)
73.             ids = [item['id'] for item in data]
74.             return ids

复制代码

ivf.py（ivf构造）

1. import faiss
2. import numpy as np
4. num_threads = 8
5. faiss.omp_set_num_threads(num_threads)
8. class IVFPQ:
9.     def __init__(self, features, nlist=100, m=16, n_bits=8):
10.         d = features.shape[1]
11.         # 创建量化器
12.         quantizer = faiss.IndexFlatL2(d)  # 使用L2距离进行量化
13.         self.index = faiss.IndexIVFFlat(quantizer, d, nlist)
14.         # self.index = faiss.IndexIVFPQ(quantizer, d, nlist, m, n_bits)
15.         # 训练索引
16.         count = 3900
17.         if features.size >= count * d:
18.             self.index.train(features)
19.             if features.size > 1000 * d:
20.                 batch_size = 1000  # 每次处理1000个特征
21.                 for i in range(0, len(features), batch_size):
22.                     self.index.add(features[i:i + batch_size])
23.             else:
24.                 self.index.add(features)
25.         else:
26.             points = int(count - features.size / d)
27.             np.random.seed(points)
28.             xb = np.random.random((points, d)).astype('float32')  # 模拟数据库中的特征向量
29.             combined_features = np.vstack((features, xb))  # Stack them vertically
30.             # 训练索引
31.             self.index.train(combined_features)
32.             self.index.add(combined_features)  # 将特征向量添加到索引中
34.     def search(self, xq, k=100):
35.         d, i = self.index.search(xq, k)
36.         return i
38.     def add(self, xb):
39.         self.index.add(xb)
41.     def train(self, xb):
42.         self.index.train(xb)
44.     def sync(self, features):
45.         for i in range(len(features)):
46.             self.add(features[i])

复制代码

结语

这个项目优化到这差不多告一段落了，后续还有啥优化点会继续跟进，稍后会把整个架构图和功能点都梳理一遍

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。