当前位置：首页 > news >正文

基于EmbeddingGemma-300m的语义搜索系统开发实战

news 2026/5/12 10:06:18

基于EmbeddingGemma-300m的语义搜索系统开发实战

1. 引言

想象一下，你正在开发一个电商平台，用户输入"适合夏天穿的轻薄透气运动鞋"，传统的关键词搜索可能只能匹配到包含这些词汇的商品，但无法理解"轻薄透气"的实际含义。而语义搜索却能理解用户的真实意图，找到那些材质透气、设计轻便的运动鞋，即使用户的查询语句和商品描述用词完全不同。

这就是语义搜索的魅力所在。今天我们要介绍的EmbeddingGemma-300m，是Google推出的轻量级嵌入模型，虽然只有3亿参数，但在语义理解方面的表现却相当出色。更重要的是，它足够轻量，可以在普通笔记本电脑上流畅运行，让中小团队也能轻松构建高质量的语义搜索系统。

本文将手把手带你搭建一个完整的语义搜索系统，从数据准备到查询处理，再到结果优化，每个环节都会提供可运行的代码示例。无论你是想要改进现有搜索功能，还是从零开始构建智能搜索，这篇文章都能给你实用的指导。

2. 环境准备与模型部署

2.1 安装必要依赖

首先确保你的Python环境是3.8或更高版本，然后安装所需的包：

pip install ollama numpy pandas sentence-transformers scikit-learn

2.2 部署EmbeddingGemma模型

EmbeddingGemma通过Ollama来部署非常简单：

# 拉取模型 ollama pull embeddinggemma:300m # 验证模型是否可用 ollama list

如果看到embeddinggemma:300m在列表中，说明模型已经准备就绪。

2.3 基础嵌入生成测试

让我们先测试一下模型的基本功能：

import ollama # 生成文本嵌入 response = ollama.embed( model='embeddinggemma:300m', input='为什么天空是蓝色的？' ) print(f'嵌入向量维度: {len(response.embeddings)}') print(f'前10个维度值: {response.embeddings[:10]}')

这段代码会输出一个768维的向量，这就是模型对输入文本的数学表示。语义搜索的核心就是比较这些向量的相似度。

3. 构建语义搜索系统

3.1 数据准备与预处理

假设我们有一个商品数据集，包含商品标题和描述：

import pandas as pd # 示例商品数据 products = [ {"id": 1, "title": "夏季透气运动鞋", "description": "网面设计，轻盈透气，适合跑步和日常穿着"}, {"id": 2, "title": "冬季保暖登山靴", "description": "加厚内衬，防滑鞋底，适合户外登山活动"}, {"id": 3, "title": "休闲帆布鞋", "description": "经典款式，舒适百搭，适合日常休闲场合"}, {"id": 4, "title": "专业篮球鞋", "description": "高帮设计，缓震科技，适合篮球运动"}, {"id": 5, "title": "轻便健步鞋", "description": "超轻材质，弹性鞋底，适合中老年人健步走"} ] df = pd.DataFrame(products)

3.2 批量生成嵌入向量

为了提高效率，我们可以批量处理文本生成嵌入：

def generate_embeddings_batch(texts, batch_size=32): """批量生成文本嵌入""" all_embeddings = [] for i in range(0, len(texts), batch_size): batch_texts = texts[i:i+batch_size] try: response = ollama.embed( model='embeddinggemma:300m', input=batch_texts ) all_embeddings.extend(response.embeddings) except Exception as e: print(f"处理批次 {i//batch_size + 1} 时出错: {e}") # 如果批量失败，尝试逐个处理 for text in batch_texts: try: response = ollama.embed( model='embeddinggemma:300m', input=text ) all_embeddings.append(response.embeddings[0]) except Exception as e2: print(f"处理文本失败: {text[:50]}...") all_embeddings.append([0]*768) # 填充零向量 return all_embeddings # 为所有商品生成嵌入 product_texts = [f"{row['title']} {row['description']}" for _, row in df.iterrows()] df['embedding'] = generate_embeddings_batch(product_texts) print("嵌入生成完成，前3个商品的嵌入维度:", [len(emb) for emb in df['embedding'][:3]])

3.3 构建向量索引

为了快速搜索，我们需要构建一个向量索引：

import numpy as np from sklearn.metrics.pairwise import cosine_similarity class VectorIndex: def __init__(self): self.vectors = [] self.metadata = [] def add_item(self, vector, metadata): self.vectors.append(vector) self.metadata.append(metadata) def search(self, query_vector, top_k=5): """搜索最相似的项目""" if not self.vectors: return [] # 计算余弦相似度 similarities = cosine_similarity([query_vector], self.vectors)[0] # 获取最相似的项目 indices = np.argsort(similarities)[::-1][:top_k] results = [] for idx in indices: results.append({ 'metadata': self.metadata[idx], 'similarity': float(similarities[idx]) }) return results # 构建索引 index = VectorIndex() for _, row in df.iterrows(): index.add_item(row['embedding'], { 'id': row['id'], 'title': row['title'], 'description': row['description'] })

4. 实现搜索功能

4.1 基本搜索实现

现在让我们实现完整的搜索流程：

def semantic_search(query, top_k=5): """语义搜索函数""" # 生成查询嵌入 try: response = ollama.embed( model='embeddinggemma:300m', input=query ) query_embedding = response.embeddings[0] except Exception as e: print(f"生成查询嵌入失败: {e}") return [] # 执行搜索 results = index.search(query_embedding, top_k) return results # 测试搜索 test_query = "想要一双透气轻便的运动鞋" results = semantic_search(test_query) print(f"查询: '{test_query}'") print("搜索结果:") for i, result in enumerate(results): print(f"{i+1}. {result['metadata']['title']} (相似度: {result['similarity']:.3f})") print(f" 描述: {result['metadata']['description']}")

4.2 高级搜索功能

为了提升搜索体验，我们可以添加一些高级功能：

def advanced_semantic_search(query, filters=None, min_similarity=0.3, top_k=10): """带过滤的高级语义搜索""" if filters is None: filters = {} # 生成查询嵌入 try: response = ollama.embed( model='embeddinggemma:300m', input=query ) query_embedding = response.embeddings[0] except Exception as e: print(f"生成查询嵌入失败: {e}") return [] # 执行搜索 raw_results = index.search(query_embedding, top_k * 2) # 获取更多结果用于过滤 # 应用过滤器 filtered_results = [] for result in raw_results: if result['similarity'] < min_similarity: continue # 这里可以添加更多过滤逻辑 # 例如按类别、价格范围等过滤 filtered_results.append(result) if len(filtered_results) >= top_k: break return filtered_results[:top_k]

5. 性能优化与实践建议

5.1 批量处理优化

对于大量数据，批量处理可以显著提升效率：

def optimize_batch_processing(texts, optimal_batch_size=16): """优化批量处理参数""" # EmbeddingGemma-300m在批量大小为16时通常表现最佳 embeddings = [] for i in range(0, len(texts), optimal_batch_size): batch = texts[i:i+optimal_batch_size] try: response = ollama.embed( model='embeddinggemma:300m', input=batch ) embeddings.extend(response.embeddings) except Exception as e: print(f"批量处理失败，尝试减小批量大小: {e}") # 失败时尝试单个处理 for text in batch: try: response = ollama.embed( model='embeddinggemma:300m', input=text ) embeddings.append(response.embeddings[0]) except Exception as e2: print(f"处理单个文本失败: {text[:50]}...") embeddings.append([0]*768) return embeddings

5.2 缓存机制

实现简单的缓存来避免重复计算：

import hashlib import json from functools import lru_cache class EmbeddingCache: def __init__(self, max_size=1000): self.cache = {} self.max_size = max_size def get_key(self, text): """生成缓存键""" return hashlib.md5(text.encode()).hexdigest() def get(self, text): """获取缓存值""" key = self.get_key(text) return self.cache.get(key) def set(self, text, embedding): """设置缓存值""" if len(self.cache) >= self.max_size: # 简单的LRU策略：移除最早的一个项目 self.cache.pop(next(iter(self.cache))) key = self.get_key(text) self.cache[key] = embedding # 使用缓存的搜索函数 cache = EmbeddingCache() def cached_semantic_search(query, top_k=5): """带缓存的语义搜索""" # 检查查询缓存 cached_embedding = cache.get(query) if cached_embedding is not None: print("使用缓存嵌入") query_embedding = cached_embedding else: # 生成新嵌入 try: response = ollama.embed( model='embeddinggemma:300m', input=query ) query_embedding = response.embeddings[0] cache.set(query, query_embedding) # 缓存结果 except Exception as e: print(f"生成查询嵌入失败: {e}") return [] return index.search(query_embedding, top_k)

5.3 实时索引更新

对于需要频繁更新的场景：

class DynamicVectorIndex(VectorIndex): def __init__(self, update_threshold=100): super().__init__() self.update_counter = 0 self.update_threshold = update_threshold self.vector_matrix = None def add_item(self, vector, metadata): super().add_item(vector, metadata) self.update_counter += 1 # 定期优化索引结构 if self.update_counter % self.update_threshold == 0: self.optimize_index() def optimize_index(self): """优化索引性能""" if self.vectors: self.vector_matrix = np.array(self.vectors) def search(self, query_vector, top_k=5): """重写搜索方法以提高性能""" if not self.vectors: return [] if self.vector_matrix is None: self.optimize_index() # 使用矩阵运算提高计算效率 query_vector = np.array(query_vector).reshape(1, -1) similarities = cosine_similarity(query_vector, self.vector_matrix)[0] indices = np.argsort(similarities)[::-1][:top_k] results = [] for idx in indices: results.append({ 'metadata': self.metadata[idx], 'similarity': float(similarities[idx]) }) return results

6. 实际应用场景

6.1 电商商品搜索

# 模拟电商搜索场景 def ecommerce_search_example(): queries = [ "适合跑步的轻便鞋子", "冬季户外防滑靴子", "日常休闲穿的舒适鞋", "打篮球用的专业运动鞋" ] for query in queries: print(f"\n 搜索: {query}") results = semantic_search(query, top_k=3) for i, result in enumerate(results): print(f" {i+1}. {result['metadata']['title']} " f"(相似度: {result['similarity']:.3f})") print("-" * 50) # 运行示例 ecommerce_search_example()

6.2 内容检索系统

# 构建内容检索系统 class ContentRetrievalSystem: def __init__(self): self.index = DynamicVectorIndex() self.content_map = {} def add_content(self, content_id, text, metadata=None): """添加内容到检索系统""" if metadata is None: metadata = {} # 生成嵌入 response = ollama.embed( model='embeddinggemma:300m', input=text ) embedding = response.embeddings[0] # 存储到索引 self.index.add_item(embedding, { 'content_id': content_id, 'text': text, **metadata }) self.content_map[content_id] = { 'text': text, 'metadata': metadata } def search_content(self, query, top_k=5): """搜索相关内容""" results = semantic_search(query, top_k) return results def get_related_content(self, content_id, top_k=3): """获取相关内容""" if content_id not in self.content_map: return [] # 使用自身内容作为查询 text = self.content_map[content_id]['text'] return self.search_content(text, top_k)