# gte-base-zh Xinference API开发指南：Python requests调用嵌入接口完整示例 ## 1. 环境准备与快速部署 在开始使用gte-base-zh模型进行文本嵌入之前，我们需要先确保环境已经正确部署。这个模型由阿里巴巴达摩院训练，基于BERT框架，专门为中文文本嵌入优化，在信息检索、语义相似度计算等场景下表现优异。 **系统要求**： - Python 3.7+ - requests库（用于API调用） - 网络连接（用于访问Xinference服务） **快速检查服务状态**： 在开始编码前，先确认Xinference服务已经正常启动。通过以下命令查看服务日志： ```bash cat /root/workspace/model_server.log ``` 如果看到服务启动成功的提示信息，说明gte-base-zh模型已经准备就绪，可以通过API进行调用了。 ## 2. API接口基础理解 Xinference为gte-base-zh模型提供了标准的RESTful API接口，我们可以通过HTTP请求来调用文本嵌入功能。理解接口的基本结构是成功调用的关键。 **核心端点**： - 嵌入生成端点：`http://localhost:9997/v1/embeddings` - 请求方法：POST - 请求体：包含文本数据的JSON格式 **接口参数说明**： - `model`: 指定使用的模型名称（gte-base-zh） - `input`: 需要嵌入的文本内容（支持字符串或字符串数组） - `encoding_format`: 返回格式（默认为float） ## 3. Python requests调用完整示例 下面是一个完整的Python示例，展示如何使用requests库调用gte-base-zh的嵌入接口。 ### 3.1 基础调用示例 ```python import requests import json # Xinference服务地址 XINFERENCE_URL = "http://localhost:9997/v1/embeddings" def get_text_embedding(text): """ 获取单个文本的嵌入向量 """ headers = { "Content-Type": "application/json" } payload = { "model": "gte-base-zh", "input": text, "encoding_format": "float" } try: response = requests.post(XINFERENCE_URL, headers=headers, json=payload) response.raise_for_status() # 检查请求是否成功 result = response.json() embedding = result['data'][0]['embedding'] print(f"文本嵌入维度: {len(embedding)}") print(f"前10个维度值: {embedding[:10]}") return embedding except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None except KeyError as e: print(f"解析响应数据失败: {e}") return None # 测试调用 text = "自然语言处理是人工智能的重要分支" embedding = get_text_embedding(text) ``` ### 3.2 批量处理示例 在实际应用中，我们经常需要处理多个文本。gte-base-zh支持批量处理，可以显著提高效率。 ```python def get_batch_embeddings(texts): """ 批量获取多个文本的嵌入向量 """ headers = { "Content-Type": "application/json" } payload = { "model": "gte-base-zh", "input": texts, "encoding_format": "float" } try: response = requests.post(XINFERENCE_URL, headers=headers, json=payload) response.raise_for_status() result = response.json() embeddings = [item['embedding'] for item in result['data']] print(f"处理了 {len(embeddings)} 个文本") print(f"每个嵌入维度: {len(embeddings[0]) if embeddings else 0}") return embeddings except requests.exceptions.RequestException as e: print(f"批量请求失败: {e}") return None # 批量处理示例 texts = [ "机器学习是人工智能的核心技术", "深度学习在图像识别中表现优异", "自然语言处理让计算机理解人类语言" ] batch_embeddings = get_batch_embeddings(texts) ``` ### 3.3 错误处理与重试机制 在实际生产环境中，网络波动或服务暂时不可用是常见情况。添加适当的错误处理和重试机制很重要。 ```python import time from requests.exceptions import RequestException def robust_get_embedding(text, max_retries=3, retry_delay=2): """ 带重试机制的嵌入获取函数 """ headers = { "Content-Type": "application/json" } payload = { "model": "gte-base-zh", "input": text, "encoding_format": "float" } for attempt in range(max_retries): try: response = requests.post(XINFERENCE_URL, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json() return result['data'][0]['embedding'] except RequestException as e: if attempt == max_retries - 1: print(f"第{attempt+1}次尝试失败: {e}") raise else: print(f"第{attempt+1}次尝试失败，{retry_delay}秒后重试...") time.sleep(retry_delay) except KeyError as e: print(f"解析响应数据失败: {e}") return None return None # 使用带重试的调用 embedding = robust_get_embedding("这是一个测试文本", max_retries=3) ``` ## 4. 实际应用场景示例 了解了基础调用后，我们来看几个实际的应用场景，展示gte-base-zh嵌入向量的强大能力。 ### 4.1 语义相似度计算 ```python import numpy as np from numpy.linalg import norm def cosine_similarity(vec1, vec2): """ 计算两个向量的余弦相似度 """ return np.dot(vec1, vec2) / (norm(vec1) * norm(vec2)) def calculate_semantic_similarity(text1, text2): """ 计算两个文本的语义相似度 """ embedding1 = get_text_embedding(text1) embedding2 = get_text_embedding(text2) if embedding1 is not None and embedding2 is not None: similarity = cosine_similarity(embedding1, embedding2) print(f"文本1: {text1}") print(f"文本2: {text2}") print(f"语义相似度: {similarity:.4f}") return similarity else: print("获取嵌入失败") return None # 相似度计算示例 text_a = "我喜欢吃苹果" text_b = "苹果是一种水果" text_c = "我正在使用苹果手机" calculate_semantic_similarity(text_a, text_b) # 应该较高 calculate_semantic_similarity(text_a, text_c) # 应该较低 ``` ### 4.2 文本分类特征提取 ```python def extract_text_features(texts): """ 为文本分类任务提取特征向量 """ embeddings = get_batch_embeddings(texts) if embeddings: # 将嵌入向量转换为numpy数组，便于机器学习模型使用 features = np.array(embeddings) print(f"特征矩阵形状: {features.shape}") return features else: return None # 模拟文本分类数据 categories = { "科技": ["人工智能改变世界", "机器学习算法应用", "深度学习框架比较"], "体育": ["篮球比赛精彩瞬间", "足球运动员技术分析", "奥运会历史回顾"], "美食": ["中式烹饪技巧分享", "西餐制作方法", "健康饮食建议"] } all_texts = [] labels = [] for category, texts in categories.items(): all_texts.extend(texts) labels.extend([category] * len(texts)) # 提取特征向量 features = extract_text_features(all_texts) print(f"提取了 {len(features)} 个样本的特征，每个特征维度 {features.shape[1]}") ``` ### 4.3 智能搜索实现 ```python class SemanticSearch: def __init__(self): self.documents = [] self.embeddings = None def add_documents(self, documents): """ 添加文档到搜索库 """ self.documents = documents self.embeddings = get_batch_embeddings(documents) print(f"已索引 {len(documents)} 个文档") def search(self, query, top_k=3): """ 语义搜索 """ if self.embeddings is None: print("请先添加文档") return [] query_embedding = get_text_embedding(query) if query_embedding is None: return [] # 计算相似度 similarities = [] for doc_embedding in self.embeddings: similarity = cosine_similarity(query_embedding, doc_embedding) similarities.append(similarity) # 获取最相似的文档 indices = np.argsort(similarities)[::-1][:top_k] results = [] for idx in indices: results.append({ "document": self.documents[idx], "similarity": similarities[idx] }) return results # 使用示例 search_engine = SemanticSearch() documents = [ "Python是一种流行的编程语言", "机器学习需要大量的数据", "深度学习是机器学习的一个分支", "自然语言处理让计算机理解人类语言", "计算机视觉处理图像和视频数据" ] search_engine.add_documents(documents) # 执行搜索 query = "如何让计算机理解文本" results = search_engine.search(query) print(f"搜索查询: {query}") for i, result in enumerate(results): print(f"{i+1}. {result['document']} (相似度: {result['similarity']:.4f})") ``` ## 5. 性能优化与实践建议 在实际使用中，我们还需要考虑性能优化和最佳实践。 ### 5.1 批量处理优化 ```python def optimized_batch_processing(texts, batch_size=32): """ 优化批量处理，避免单次请求过大 """ all_embeddings = [] for i in range(0, len(texts), batch_size): batch_texts = texts[i:i+batch_size] print(f"处理批次 {i//batch_size + 1}: {len(batch_texts)} 个文本") batch_embeddings = get_batch_embeddings(batch_texts) if batch_embeddings: all_embeddings.extend(batch_embeddings) # 添加短暂延迟，避免对服务造成过大压力 time.sleep(0.1) return all_embeddings # 处理大量文本 large_text_collection = ["文本示例"] * 100 # 模拟100个文本 embeddings = optimized_batch_processing(large_text_collection, batch_size=20) ``` ### 5.2 缓存机制实现 ```python from functools import lru_cache @lru_cache(maxsize=1000) def cached_get_embedding(text): """ 带缓存的嵌入获取，避免重复计算相同文本 """ return get_text_embedding(text) # 使用缓存 text = "重复使用的文本" embedding1 = cached_get_embedding(text) # 第一次调用，实际请求 embedding2 = cached_get_embedding(text) # 第二次调用，使用缓存 ``` ### 5.3 异步处理示例 对于需要高性能的应用，可以考虑使用异步请求。 ```python import aiohttp import asyncio async def async_get_embeddings(session, texts): """ 异步获取多个文本的嵌入 """ url = "http://localhost:9997/v1/embeddings" payload = { "model": "gte-base-zh", "input": texts, "encoding_format": "float" } async with session.post(url, json=payload) as response: if response.status == 200: result = await response.json() return [item['embedding'] for item in result['data']] else: print(f"请求失败: {response.status}") return None async def main_async(): """ 异步处理主函数 """ texts = ["文本1", "文本2", "文本3"] # 替换为实际文本 async with aiohttp.ClientSession() as session: embeddings = await async_get_embeddings(session, texts) print(f"异步获取到 {len(embeddings)} 个嵌入向量") # 运行异步示例 # asyncio.run(main_async()) ``` ## 6. 总结 通过本指南，我们全面了解了如何使用Python requests库调用gte-base-zh模型的Xinference API接口。从基础调用到高级应用，从单个文本处理到批量优化，我们覆盖了实际开发中需要的各种场景。 **关键要点回顾**： - gte-base-zh提供了高质量的中文文本嵌入能力 - Xinference的RESTful API接口简单易用 - 批量处理和错误处理是生产环境中的必备功能 - 嵌入向量可以用于语义搜索、文本分类等多种应用 **实践建议**： 1. 始终添加适当的错误处理和重试机制 2. 对于重复文本，考虑实现缓存来提高效率 3. 批量处理时注意控制请求大小，避免给服务造成过大压力 4. 根据实际应用场景选择合适的相似度计算方法 现在你已经掌握了gte-base-zh API调用的完整技能，可以开始在你的项目中实现强大的文本嵌入功能了。记得先从简单的示例开始，逐步扩展到复杂的应用场景。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。