gte-base-zh开发者实操手册Python调用Xinference Embedding API代码实例1. 快速了解gte-base-zh嵌入模型gte-base-zh是由阿里巴巴达摩院训练的中文文本嵌入模型基于BERT框架构建。这个模型在一个包含大量相关文本对的大规模语料库上进行训练涵盖了广泛的领域和场景。模型的核心能力将中文文本转换为高维向量表示计算文本之间的语义相似度支持信息检索、语义匹配、文本重排序等应用专门针对中文语言特点优化模型本地地址为/usr/local/bin/AI-ModelScope/gte-base-zh2. 环境准备与模型部署2.1 启动Xinference服务首先需要启动Xinference服务使用以下命令xinference-local --host 0.0.0.0 --port 9997这个命令会在本地启动一个推理服务监听9997端口。2.2 启动gte-base-zh模型服务使用专门的启动脚本发布模型服务python /usr/local/bin/launch_model_server.py2.3 验证服务状态检查模型服务是否启动成功cat /root/workspace/model_server.log如果看到类似下面的输出说明服务启动成功模型加载完成服务已启动 等待接收请求...3. Python调用Embedding API实战3.1 安装必要的Python库首先确保安装了所需的Python依赖pip install requests numpy3.2 基础文本嵌入代码示例下面是一个简单的Python示例展示如何调用gte-base-zh的嵌入APIimport requests import json import numpy as np class GTEEmbeddingClient: def __init__(self, base_urlhttp://localhost:9997): self.base_url base_url self.embedding_url f{base_url}/v1/embeddings def get_embedding(self, text, modelgte-base-zh): 获取单个文本的嵌入向量 payload { model: model, input: text } try: response requests.post(self.embedding_url, jsonpayload) response.raise_for_status() result response.json() return result[data][0][embedding] except Exception as e: print(f获取嵌入向量失败: {e}) return None def get_embeddings_batch(self, texts, modelgte-base-zh): 批量获取多个文本的嵌入向量 payload { model: model, input: texts } try: response requests.post(self.embedding_url, jsonpayload) response.raise_for_status() result response.json() return [item[embedding] for item in result[data]] except Exception as e: print(f批量获取嵌入向量失败: {e}) return None # 使用示例 if __name__ __main__: client GTEEmbeddingClient() # 单个文本嵌入 text 自然语言处理是人工智能的重要领域 embedding client.get_embedding(text) print(f文本嵌入向量维度: {len(embedding)}) print(f前10个维度值: {embedding[:10]}) # 批量文本嵌入 texts [ 机器学习算法, 深度学习模型, 计算机视觉应用 ] embeddings client.get_embeddings_batch(texts) print(f批量处理了 {len(embeddings)} 个文本)3.3 语义相似度计算示例基于嵌入向量计算文本相似度def cosine_similarity(vec1, vec2): 计算两个向量的余弦相似度 vec1 np.array(vec1) vec2 np.array(vec2) return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2)) def calculate_similarities(client, query_text, candidate_texts): 计算查询文本与候选文本的相似度 # 获取所有文本的嵌入 all_texts [query_text] candidate_texts embeddings client.get_embeddings_batch(all_texts) if embeddings: query_embedding embeddings[0] candidate_embeddings embeddings[1:] similarities [] for i, candidate_embedding in enumerate(candidate_embeddings): similarity cosine_similarity(query_embedding, candidate_embedding) similarities.append((candidate_texts[i], similarity)) # 按相似度排序 similarities.sort(keylambda x: x[1], reverseTrue) return similarities return None # 使用示例 client GTEEmbeddingClient() query 人工智能技术 candidates [ 机器学习算法, 深度学习框架, 自然语言处理, 计算机视觉, 数据分析方法 ] results calculate_similarities(client, query, candidates) print(相似度排序结果:) for text, score in results: print(f{text}: {score:.4f})4. 高级应用场景示例4.1 文档检索系统构建一个简单的文档检索系统class DocumentRetrievalSystem: def __init__(self, client): self.client client self.documents [] self.embeddings [] def add_documents(self, documents): 添加文档到检索系统 self.documents.extend(documents) new_embeddings self.client.get_embeddings_batch(documents) if new_embeddings: self.embeddings.extend(new_embeddings) def search(self, query, top_k5): 搜索相关文档 query_embedding self.client.get_embedding(query) if not query_embedding: return [] similarities [] for i, doc_embedding in enumerate(self.embeddings): similarity cosine_similarity(query_embedding, doc_embedding) similarities.append((i, similarity)) # 按相似度排序并返回top_k结果 similarities.sort(keylambda x: x[1], reverseTrue) results [] for idx, score in similarities[:top_k]: results.append({ document: self.documents[idx], score: score, index: idx }) return results # 使用示例 retrieval_system DocumentRetrievalSystem(client) # 添加示例文档 documents [ 机器学习是人工智能的一个分支专注于开发能够从数据中学习的算法, 深度学习使用多层神经网络来处理复杂的数据模式, 自然语言处理使计算机能够理解、解释和生成人类语言, 计算机视觉让机器能够识别和理解图像和视频内容, 强化学习通过试错来学习最优决策策略 ] retrieval_system.add_documents(documents) # 执行搜索 results retrieval_system.search(如何让计算机理解人类语言, top_k3) print(搜索结果:) for result in results: print(f相似度: {result[score]:.4f}) print(f文档: {result[document]}) print(---)4.2 文本聚类应用def text_clustering(texts, n_clusters3): 对文本进行聚类分析 from sklearn.cluster import KMeans # 获取所有文本的嵌入 embeddings client.get_embeddings_batch(texts) if not embeddings: return None # 使用KMeans进行聚类 kmeans KMeans(n_clustersn_clusters, random_state42) clusters kmeans.fit_predict(embeddings) # 组织聚类结果 results {} for i, cluster_id in enumerate(clusters): if cluster_id not in results: results[cluster_id] [] results[cluster_id].append(texts[i]) return results # 使用示例 tech_texts [ 神经网络和深度学习, 卷积神经网络在图像识别中的应用, 循环神经网络处理序列数据, 支持向量机分类算法, 随机森林集成学习方法, 梯度下降优化算法 ] clusters text_clustering(tech_texts, n_clusters2) for cluster_id, texts in clusters.items(): print(f聚类 {cluster_id 1}:) for text in texts: print(f - {text}) print()5. 性能优化与最佳实践5.1 批量处理优化当需要处理大量文本时使用批量接口可以显著提高效率def process_large_dataset(texts, batch_size32): 处理大规模文本数据集 all_embeddings [] for i in range(0, len(texts), batch_size): batch_texts texts[i:i batch_size] batch_embeddings client.get_embeddings_batch(batch_texts) if batch_embeddings: all_embeddings.extend(batch_embeddings) print(f已处理 {min(i batch_size, len(texts))}/{len(texts)} 个文本) return all_embeddings5.2 错误处理与重试机制增强代码的健壮性import time from requests.exceptions import RequestException class RobustEmbeddingClient(GTEEmbeddingClient): def __init__(self, base_urlhttp://localhost:9997, max_retries3): super().__init__(base_url) self.max_retries max_retries def get_embedding_with_retry(self, text, modelgte-base-zh): 带重试机制的嵌入获取 for attempt in range(self.max_retries): try: return self.get_embedding(text, model) except RequestException as e: if attempt self.max_retries - 1: raise e print(f请求失败第 {attempt 1} 次重试...) time.sleep(2 ** attempt) # 指数退避 return None5.3 嵌入向量缓存减少重复计算提高性能from functools import lru_cache class CachedEmbeddingClient(GTEEmbeddingClient): lru_cache(maxsize1000) def get_embedding_cached(self, text, modelgte-base-zh): 带缓存的嵌入获取 return self.get_embedding(text, model)6. 常见问题与解决方案6.1 服务连接问题如果遇到连接问题可以检查以下方面def check_service_status(base_urlhttp://localhost:9997): 检查服务状态 try: response requests.get(f{base_url}/v1/models) if response.status_code 200: print(服务正常运行) models response.json() print(f可用模型: {models}) return True else: print(f服务异常状态码: {response.status_code}) return False except Exception as e: print(f服务连接失败: {e}) return False # 检查服务状态 check_service_status()6.2 内存使用优化处理大量嵌入向量时注意内存使用def process_with_memory_optimization(texts, output_fileembeddings.npy): 内存优化的批量处理 with open(output_file, wb) as f: for i in range(0, len(texts), 100): # 分批处理 batch_texts texts[i:i 100] embeddings client.get_embeddings_batch(batch_texts) if embeddings: # 将嵌入向量保存到文件避免内存积累 for embedding in embeddings: np.save(f, np.array(embedding)) print(f处理进度: {min(i 100, len(texts))}/{len(texts)})7. 总结通过本教程我们学习了如何使用Python调用Xinference部署的gte-base-zh嵌入模型API。主要内容包括核心知识点gte-base-zh模型的基本特性和应用场景Xinference服务的部署和启动方法使用Python requests库调用嵌入API的基本操作文本嵌入向量的获取和处理技巧实用技能单个文本和批量文本的嵌入处理基于嵌入向量的语义相似度计算文档检索系统的构建方法文本聚类应用的实际实现性能优化批量处理提高效率的方法错误处理和重试机制内存使用优化策略嵌入向量缓存技术gte-base-zh作为一个专门针对中文优化的文本嵌入模型在语义理解、文本匹配、信息检索等场景中表现出色。通过Xinference提供的标准化API开发者可以轻松地将这个强大的模型集成到自己的应用中。在实际使用中建议根据具体需求调整批量处理大小合理使用缓存机制并注意错误处理和服务的监控以确保应用的稳定性和性能。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。