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

gte-base-zh实战教程：Python调用Xinference API实现批量文本向量化

news 2026/5/12 5:44:23

gte-base-zh实战教程：Python调用Xinference API实现批量文本向量化

1. 教程概述

今天给大家分享一个实用的技术方案：如何使用Python调用Xinference API来实现批量文本向量化。如果你正在处理大量中文文本数据，需要将它们转换为向量表示，那么这个教程就是为你准备的。

学完本教程，你将掌握：

如何部署和启动gte-base-zh嵌入模型
如何使用Python代码调用Xinference API
如何实现批量文本的向量化处理
如何处理实际应用中的常见问题

前置要求：

基本的Python编程知识
了解什么是文本嵌入和向量化
有Linux服务器操作经验更佳

这个方案特别适合需要处理大量中文文本的场景，比如搜索引擎、推荐系统、语义相似度计算等。让我们开始吧！

2. 环境准备与模型部署

2.1 了解gte-base-zh模型

gte-base-zh是由阿里巴巴达摩院训练的中文文本嵌入模型。它基于BERT框架，专门针对中文文本优化，能够将文本转换为高质量的向量表示。

这个模型在大量相关文本对的大规模语料库上训练，覆盖了广泛的领域和场景。这意味着它能够很好地理解中文语义，适用于信息检索、语义文本相似性、文本重排序等多种任务。

2.2 模型部署步骤

首先，我们需要确认模型已经正确部署。gte-base-zh模型的本地地址为：

/usr/local/bin/AI-ModelScope/gte-base-zh

启动Xinference服务的命令如下：

xinference-local --host 0.0.0.0 --port 9997

模型启动脚本地址为：

/usr/local/bin/launch_model_server.py

2.3 验证模型服务

启动服务后，我们需要确认模型是否正常运行。可以通过以下命令查看日志：

cat /root/workspace/model_server.log

如果看到类似"模型加载成功"或"服务启动完成"的信息，说明模型已经准备就绪。

3. Python调用API实战

3.1 安装必要的Python库

在开始编写代码前，我们需要安装一些必要的Python库：

pip install requests numpy pandas tqdm

requests：用于发送HTTP请求到Xinference API
numpy：处理数值计算和向量操作
pandas：处理表格数据，方便批量操作
tqdm：显示进度条，让批量处理更直观

3.2 基础API调用示例

让我们先写一个最简单的API调用示例，了解基本的工作流程：

import requests import json # Xinference服务器的地址 XINFERENCE_URL = "http://localhost:9997/v1/embeddings" def get_embedding_single(text): """获取单个文本的向量表示""" headers = { "Content-Type": "application/json" } data = { "model": "gte-base-zh", "input": text } try: response = requests.post(XINFERENCE_URL, headers=headers, json=data) response.raise_for_status() # 检查请求是否成功 result = response.json() embedding = result['data'][0]['embedding'] 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_embedding_single(text) print(f"文本向量维度: {len(embedding)}") print(f"前10个向量值: {embedding[:10]}")

3.3 批量处理实现

单个文本处理效率太低，我们来实现批量处理功能：

import numpy as np from tqdm import tqdm import time def get_embedding_batch(texts, batch_size=32, delay=0.1): """ 批量获取文本向量 :param texts: 文本列表 :param batch_size: 每批处理的文本数量 :param delay: 批处理之间的延迟（秒），避免服务器过载 :return: 向量列表 """ all_embeddings = [] # 使用tqdm显示进度条 for i in tqdm(range(0, len(texts), batch_size), desc="处理进度"): batch_texts = texts[i:i + batch_size] headers = { "Content-Type": "application/json" } data = { "model": "gte-base-zh", "input": batch_texts } try: response = requests.post(XINFERENCE_URL, headers=headers, json=data) response.raise_for_status() result = response.json() batch_embeddings = [item['embedding'] for item in result['data']] all_embeddings.extend(batch_embeddings) # 添加延迟，避免服务器压力过大 time.sleep(delay) except Exception as e: print(f"处理批次 {i//batch_size + 1} 时出错: {e}") # 对于失败的批次，使用单个文本重试 for text in batch_texts: try: embedding = get_embedding_single(text) if embedding is not None: all_embeddings.append(embedding) else: all_embeddings.append([0] * 768) # 填充零向量 except: all_embeddings.append([0] * 768) return all_embeddings # 示例：批量处理文本 sample_texts = [ "人工智能是未来的发展趋势", "机器学习需要大量的数据训练", "深度学习在图像识别领域表现出色", "自然语言处理让计算机理解人类语言", "计算机视觉可以识别图像中的物体" ] embeddings = get_embedding_batch(sample_texts) print(f"成功处理了 {len(embeddings)} 个文本的向量化")

4. 实际应用案例

4.1 处理大型文本文件

在实际项目中，我们通常需要处理大量的文本数据。下面是一个处理文本文件的完整示例：

import pandas as pd def process_text_file(input_file, output_file, text_column="text"): """ 处理包含文本的CSV文件，为每行文本生成向量 :param input_file: 输入CSV文件路径 :param output_file: 输出文件路径 :param text_column: 包含文本的列名 """ # 读取数据 print("正在读取数据...") df = pd.read_csv(input_file) texts = df[text_column].tolist() # 生成向量 print("开始生成文本向量...") embeddings = get_embedding_batch(texts) # 保存结果 print("正在保存结果...") result_df = df.copy() result_df['embedding'] = embeddings result_df.to_csv(output_file, index=False) print(f"处理完成！共处理 {len(texts)} 个文本") return result_df # 使用示例 # result = process_text_file("input.csv", "output_with_embeddings.csv")

4.2 语义相似度计算

有了文本向量，我们可以计算文本之间的相似度：

from sklearn.metrics.pairwise import cosine_similarity def calculate_similarity(text1, text2): """计算两个文本的语义相似度""" # 获取向量 embedding1 = get_embedding_single(text1) embedding2 = get_embedding_single(text2) if embedding1 is None or embedding2 is None: return None # 计算余弦相似度 similarity = cosine_similarity([embedding1], [embedding2])[0][0] return similarity def find_similar_texts(query_text, texts, top_n=5): """在文本列表中查找与查询文本最相似的文本""" # 获取查询文本的向量 query_embedding = get_embedding_single(query_text) if query_embedding is None: return [] # 获取所有文本的向量 text_embeddings = get_embedding_batch(texts) # 计算相似度 similarities = cosine_similarity([query_embedding], text_embeddings)[0] # 获取最相似的文本 similar_indices = similarities.argsort()[-top_n:][::-1] results = [] for idx in similar_indices: results.append({ 'text': texts[idx], 'similarity': similarities[idx] }) return results # 使用示例 query = "机器学习算法" texts_to_search = [ "深度学习神经网络", "监督学习训练模型", "数据挖掘技术", "人工智能应用", "计算机编程基础" ] similar_results = find_similar_texts(query, texts_to_search) for result in similar_results: print(f"相似度: {result['similarity']:.4f} - 文本: {result['text']}")

5. 性能优化与实用技巧

5.1 批量处理优化

为了提高处理效率，我们可以采用一些优化策略：

def optimized_batch_processing(texts, max_workers=4, timeout=30): """ 使用多线程优化批量处理 :param texts: 文本列表 :param max_workers: 最大线程数 :param timeout: 请求超时时间（秒） """ from concurrent.futures import ThreadPoolExecutor, as_completed import threading # 线程安全的计数器 processed_count = 0 lock = threading.Lock() def process_single(text): nonlocal processed_count headers = { "Content-Type": "application/json" } data = { "model": "gte-base-zh", "input": text } try: response = requests.post( XINFERENCE_URL, headers=headers, json=data, timeout=timeout ) response.raise_for_status() result = response.json() embedding = result['data'][0]['embedding'] with lock: processed_count += 1 if processed_count % 100 == 0: print(f"已处理 {processed_count} 个文本") return embedding except Exception as e: print(f"处理文本失败: {e}") return None # 使用线程池处理 embeddings = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_text = { executor.submit(process_single, text): text for text in texts } for future in tqdm(as_completed(future_to_text), total=len(texts), desc="多线程处理"): try: embedding = future.result() if embedding is not None: embeddings.append(embedding) else: embeddings.append([0] * 768) # 填充零向量 except Exception as e: print(f"处理异常: {e}") embeddings.append([0] * 768) return embeddings

5.2 错误处理与重试机制

在实际应用中，网络请求可能会失败，我们需要健壮的错误处理：

def robust_embedding_request(text, max_retries=3): """带重试机制的向量请求""" for attempt in range(max_retries): try: headers = { "Content-Type": "application/json" } data = { "model": "gte-base-zh", "input": text } response = requests.post( XINFERENCE_URL, headers=headers, json=data, timeout=10 ) response.raise_for_status() result = response.json() return result['data'][0]['embedding'] except requests.exceptions.Timeout: print(f"请求超时，第 {attempt + 1} 次重试") time.sleep(2 ** attempt) # 指数退避 except requests.exceptions.ConnectionError: print(f"连接错误，第 {attempt + 1} 次重试") time.sleep(2 ** attempt) except Exception as e: print(f"其他错误: {e}") break print(f"经过 {max_retries} 次重试后仍然失败") return None

5.3 内存优化技巧

处理大量文本时，内存管理很重要：

def process_large_dataset(texts, chunk_size=1000, output_file="embeddings.npy"): """ 处理超大数据集，分块处理并保存到文件 :param texts: 文本列表 :param chunk_size: 每个块的大小 :param output_file: 输出文件路径 """ all_embeddings = [] for i in range(0, len(texts), chunk_size): chunk_texts = texts[i:i + chunk_size] print(f"处理第 {i//chunk_size + 1} 个块，共 {len(chunk_texts)} 个文本") chunk_embeddings = get_embedding_batch(chunk_texts) # 保存当前块的向量 np.save(f"temp_chunk_{i//chunk_size}.npy", chunk_embeddings) # 清空内存 del chunk_embeddings all_embeddings = None print(f"第 {i//chunk_size + 1} 个块处理完成") # 合并所有块（如果需要） print("开始合并所有块...") # 这里可以根据需要实现块的合并逻辑 print("处理完成！")