gte-base-zh开发者实操手册：Python调用Xinference Embedding API代码实例

1. 快速了解gte-base-zh嵌入模型

gte-base-zh是由阿里巴巴达摩院训练的中文文本嵌入模型，基于BERT框架构建。这个模型在一个包含大量相关文本对的大规模语料库上进行训练，涵盖了广泛的领域和场景。

模型的核心能力：

• 将中文文本转换为高维向量表示
• 计算文本之间的语义相似度
• 支持信息检索、语义匹配、文本重排序等应用
• 专门针对中文语言特点优化

模型本地地址为：/usr/local/bin/AI-ModelScope/gte-base-zh

2. 环境准备与模型部署

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.py

2.3 验证服务状态

检查模型服务是否启动成功：

cat /root/workspace/model_server.log

如果看到类似下面的输出，说明服务启动成功：

模型加载完成，服务已启动
等待接收请求...

3. Python调用Embedding API实战

3.1 安装必要的Python库

首先确保安装了所需的Python依赖：

pip install requests numpy

3.2 基础文本嵌入代码示例

下面是一个简单的Python示例，展示如何调用gte-base-zh的嵌入API：

import requests
import json
import numpy as np

class GTEEmbeddingClient:
    def __init__(self, base_url="http://localhost:9997"):
        self.base_url = base_url
        self.embedding_url = f"{base_url}/v1/embeddings"
    
    def get_embedding(self, text, model="gte-base-zh"):
        """获取单个文本的嵌入向量"""
        payload = {
            "model": model,
            "input": text
        }
        
        try:
            response = requests.post(self.embedding_url, json=payload)
            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, model="gte-base-zh"):
        """批量获取多个文本的嵌入向量"""
        payload = {
            "model": model,
            "input": texts
        }
        
        try:
            response = requests.post(self.embedding_url, json=payload)
            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(key=lambda x: x[1], reverse=True)
        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_k=5):
        """搜索相关文档"""
        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(key=lambda x: x[1], reverse=True)
        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_k=3)
print("搜索结果:")
for result in results:
    print(f"相似度: {result['score']:.4f}")
    print(f"文档: {result['document']}")
    print("---")

4.2 文本聚类应用

def text_clustering(texts, n_clusters=3):
    """对文本进行聚类分析"""
    from sklearn.cluster import KMeans
    
    # 获取所有文本的嵌入
    embeddings = client.get_embeddings_batch(texts)
    if not embeddings:
        return None
    
    # 使用KMeans进行聚类
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    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_clusters=2)
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_size=32):
    """处理大规模文本数据集"""
    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_embeddings

5.2 错误处理与重试机制

增强代码的健壮性：

import time
from requests.exceptions import RequestException

class RobustEmbeddingClient(GTEEmbeddingClient):
    def __init__(self, base_url="http://localhost:9997", max_retries=3):
        super().__init__(base_url)
        self.max_retries = max_retries
    
    def get_embedding_with_retry(self, text, model="gte-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 None

5.3 嵌入向量缓存

减少重复计算，提高性能：

from functools import lru_cache

class CachedEmbeddingClient(GTEEmbeddingClient):
    @lru_cache(maxsize=1000)
    def get_embedding_cached(self, text, model="gte-base-zh"):
        """带缓存的嵌入获取"""
        return self.get_embedding(text, model)

6. 常见问题与解决方案

6.1 服务连接问题

如果遇到连接问题，可以检查以下方面：

def check_service_status(base_url="http://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_file="embeddings.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星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。