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

EmbeddingGemma-300m API开发指南：快速构建文本处理服务

news 2026/3/26 21:10:03

EmbeddingGemma-300m API开发指南：快速构建文本处理服务

1. 引言

文本嵌入技术正在改变我们处理和理解文本数据的方式。想象一下，你有一段文字，需要快速找到相似的文档、进行智能分类或者构建推荐系统，传统的关键词匹配方法已经无法满足需求。这时候，文本嵌入模型就能大显身手了。

EmbeddingGemma-300m作为Google推出的轻量级嵌入模型，虽然只有3亿参数，但在多语言文本理解方面表现出色。更重要的是，它足够小巧，可以在普通服务器甚至个人电脑上运行，这为开发者提供了极大的便利。

本文将带你从零开始，构建一个基于EmbeddingGemma-300m的RESTful API服务。无论你是想要为现有系统添加语义搜索功能，还是构建全新的文本处理应用，这篇指南都能帮你快速上手。

2. 环境准备与模型部署

2.1 系统要求与依赖安装

在开始之前，确保你的系统满足以下基本要求：

Python 3.8 或更高版本
至少4GB可用内存（模型本身约622MB）
网络连接用于下载模型

首先安装必要的Python依赖：

pip install flask ollama numpy requests

2.2 快速部署EmbeddingGemma-300m

使用Ollama可以极简地部署模型：

# 拉取模型 ollama pull embeddinggemma:300m # 运行模型服务 ollama serve

这个过程会自动下载模型并启动本地服务，默认监听在11434端口。你可以用以下命令测试服务是否正常：

curl http://localhost:11434/api/embeddings \ -d '{ "model": "embeddinggemma:300m", "input": "测试文本" }'

如果看到返回的向量数据，说明模型已经成功部署。

3. API服务设计与实现

3.1 基础Flask应用搭建

让我们从创建一个简单的Flask应用开始：

from flask import Flask, request, jsonify import requests import numpy as np app = Flask(__name__) OLLAMA_URL = "http://localhost:11434/api/embeddings" @app.route('/health', methods=['GET']) def health_check(): return jsonify({"status": "healthy"}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True)

这个基础应用已经可以运行，访问http://localhost:5000/health应该能看到服务状态。

3.2 核心嵌入接口实现

现在添加核心的文本嵌入功能：

@app.route('/embed', methods=['POST']) def embed_text(): try: data = request.get_json() text = data.get('text') if not text: return jsonify({"error": "No text provided"}), 400 # 调用Ollama服务 response = requests.post(OLLAMA_URL, json={ "model": "embeddinggemma:300m", "input": text }) if response.status_code == 200: result = response.json() return jsonify({ "embedding": result['embeddings'][0], "dimension": len(result['embeddings'][0]) }) else: return jsonify({"error": "Embedding service unavailable"}), 503 except Exception as e: return jsonify({"error": str(e)}), 500

这个接口接受JSON格式的文本输入，返回对应的768维向量。

3.3 批量处理与性能优化

单条处理效率较低，我们添加批量处理支持：

@app.route('/embed/batch', methods=['POST']) def embed_batch(): try: data = request.get_json() texts = data.get('texts', []) if not texts or not isinstance(texts, list): return jsonify({"error": "Texts must be a non-empty list"}), 400 # 批量调用 response = requests.post(OLLAMA_URL, json={ "model": "embeddinggemma:300m", "input": texts }) if response.status_code == 200: result = response.json() return jsonify({ "embeddings": result['embeddings'], "count": len(result['embeddings']) }) else: return jsonify({"error": "Embedding service unavailable"}), 503 except Exception as e: return jsonify({"error": str(e)}), 500

批量处理可以显著提升性能，特别是在处理大量文本时。

4. 高级功能实现

4.1 相似度计算接口

文本嵌入最常见的用途就是计算相似度：

from sklearn.metrics.pairwise import cosine_similarity @app.route('/similarity', methods=['POST']) def calculate_similarity(): try: data = request.get_json() text1 = data.get('text1') text2 = data.get('text2') if not text1 or not text2: return jsonify({"error": "Both text1 and text2 are required"}), 400 # 获取两个文本的嵌入 emb_response = requests.post(OLLAMA_URL, json={ "model": "embeddinggemma:300m", "input": [text1, text2] }) if emb_response.status_code == 200: embeddings = emb_response.json()['embeddings'] similarity = cosine_similarity( [embeddings[0]], [embeddings[1]] )[0][0] return jsonify({ "similarity": float(similarity), "text1": text1, "text2": text2 }) else: return jsonify({"error": "Embedding service unavailable"}), 503 except Exception as e: return jsonify({"error": str(e)}), 500

4.2 语义搜索功能

基于嵌入的语义搜索比关键词搜索更智能：

class VectorStore: def __init__(self): self.documents = [] self.embeddings = [] def add_document(self, text, embedding): self.documents.append(text) self.embeddings.append(embedding) def search(self, query_embedding, top_k=5): if not self.embeddings: return [] similarities = cosine_similarity( [query_embedding], self.embeddings )[0] # 获取最相似的文档索引 indices = np.argsort(similarities)[-top_k:][::-1] results = [] for idx in indices: results.append({ "text": self.documents[idx], "similarity": float(similarities[idx]) }) return results # 初始化向量存储 vector_store = VectorStore() @app.route('/search', methods=['POST']) def semantic_search(): try: data = request.get_json() query = data.get('query') top_k = data.get('top_k', 5) if not query: return jsonify({"error": "Query is required"}), 400 # 获取查询嵌入 response = requests.post(OLLAMA_URL, json={ "model": "embeddinggemma:300m", "input": query }) if response.status_code == 200: query_embedding = response.json()['embeddings'][0] results = vector_store.search(query_embedding, top_k) return jsonify({ "query": query, "results": results }) else: return jsonify({"error": "Embedding service unavailable"}), 503 except Exception as e: return jsonify({"error": str(e)}), 500

5. 性能优化与错误处理

5.1 连接池与超时设置

优化HTTP连接性能：

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 创建优化的会话对象 session = requests.Session() retry_strategy = Retry( total=3, backoff_factor=0.1, status_forcelist=[429, 500, 502, 503, 504] ) adapter = HTTPAdapter( max_retries=retry_strategy, pool_connections=10, pool_maxsize=10 ) session.mount("http://", adapter) session.mount("https://", adapter)

5.2 全面的错误处理

增强API的健壮性：

class APIError(Exception): def __init__(self, message, status_code=500): super().__init__(message) self.status_code = status_code def get_embedding(text): try: response = session.post( OLLAMA_URL, json={"model": "embeddinggemma:300m", "input": text}, timeout=30 ) response.raise_for_status() return response.json()['embeddings'][0] except requests.exceptions.Timeout: raise APIError("Embedding service timeout", 504) except requests.exceptions.ConnectionError: raise APIError("Embedding service unavailable", 503) except Exception as e: raise APIError(f"Embedding failed: {str(e)}", 500) @app.errorhandler(APIError) def handle_api_error(error): return jsonify({"error": str(error)}), error.status_code

5.3 缓存机制实现

减少重复计算：

from functools import lru_cache @lru_cache(maxsize=1000) def cached_embedding(text): return get_embedding(text) @app.route('/embed/cached', methods=['POST']) def embed_cached(): try: data = request.get_json() text = data.get('text') if not text: return jsonify({"error": "No text provided"}), 400 embedding = cached_embedding(text) return jsonify({ "embedding": embedding, "cached": True # 实际需要检查是否缓存命中 }) except APIError as e: return jsonify({"error": str(e)}), e.status_code

6. 部署与测试建议

6.1 生产环境部署

使用Gunicorn部署Flask应用：

pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 your_app:app

创建Dockerfile容器化部署：

FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 5000 CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]

6.2 API测试方案

使用Python编写自动化测试：

import unittest import requests class TestEmbeddingAPI(unittest.TestCase): BASE_URL = "http://localhost:5000" def test_health_check(self): response = requests.get(f"{self.BASE_URL}/health") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["status"], "healthy") def test_embedding(self): response = requests.post(f"{self.BASE_URL}/embed", json={ "text": "测试文本" }) self.assertEqual(response.status_code, 200) self.assertIn("embedding", response.json()) self.assertEqual(len(response.json()["embedding"]), 768) if __name__ == '__main__': unittest.main()