filtered_predictions = np.array(filtered_predictions)
时间: 2024-03-19 19:40:24 浏览: 81
这行代码的作用是将`filtered_predictions`转换为numpy数组。`filtered_predictions`是一个列表,通过`np.array()`函数将其转换为numpy数组,可以方便地进行一些数值计算和处理。最终,转换后的numpy数组被赋值给`filtered_predictions`变量。
相关问题
# unified_model_service.py import os import sys import logging import numpy as np from flask import Flask, request, jsonify from transformers import AutoTokenizer, AutoModel from FlagEmbedding import BGEM3FlagModel import torch from flask_cors import CORS # 配置路径 current_dir = os.path.dirname(os.path.abspath(__file__)) sys.path.append(os.path.join(current_dir, '..')) from config.config import MODEL_CONFIG # 初始化日志 log_file_path = "logs/model_service.log" os.makedirs(os.path.dirname(log_file_path), exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(log_file_path), logging.StreamHandler() ] ) logger = logging.getLogger("UnifiedModelService") app = Flask(__name__) CORS(app) # 全局模型变量 dense_model = None dense_tokenizer = None sparse_model = None def load_models(): global dense_model, dense_tokenizer, sparse_model # 加载稠密向量模型 (Qwen) if MODEL_CONFIG["dense_model"].get("enable", True): try: model_path = MODEL_CONFIG["dense_model"]["model_path"] logger.info(f"Loading dense vector model from: {model_path}") dense_tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True) dense_model = AutoModel.from_pretrained(model_path, trust_remote_code=True).eval() logger.info("Dense vector model loaded successfully") except Exception as e: logger.error(f"Dense model loading failed: {str(e)}", exc_info=True) if MODEL_CONFIG["dense_model"].get("required", True): raise # 加载稀疏向量模型 (BGE-M3) if MODEL_CONFIG["sparse_model"].get("enable", True): try: model_path = MODEL_CONFIG["sparse_model"]["model_path"] device = MODEL_CONFIG["sparse_model"].get("device", "cuda") use_fp16 = MODEL_CONFIG["sparse_model"].get("use_fp16", True) logger.info(f"Loading sparse vector model from: {model_path}") logger.info(f"Device: {device}, FP16: {use_fp16}") sparse_model = BGEM3FlagModel( model_path, device=device, use_fp16=use_fp16 ) logger.info("Sparse vector model loaded successfully") except Exception as e: logger.error(f"Sparse model loading failed: {str(e)}", exc_info=True) if MODEL_CONFIG["sparse_model"].get("required", True): raise def batch_encode_dense(texts): """批量生成归一化稠密向量""" global dense_tokenizer, dense_model vectors = [] batch_size = MODEL_CONFIG["dense_model"].get("batch_size", 32) max_length = MODEL_CONFIG["dense_model"].get("max_length", 512) for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] inputs = dense_tokenizer( batch, padding=True, truncation=True, max_length=max_length, return_tensors="pt" ) with torch.no_grad(): outputs = dense_model(**inputs) batch_vectors = outputs.last_hidden_state.mean(dim=1).numpy() vectors.append(batch_vectors) vectors = np.concatenate(vectors, axis=0) # 归一化 norms = np.linalg.norm(vectors, axis=1, keepdims=True) vectors = vectors / np.where(norms > 1e-12, norms, 1.0) return vectors def batch_encode_sparse(texts): """批量生成稀疏向量""" global sparse_model results = [] batch_size = MODEL_CONFIG["sparse_model"].get("batch_size", 8) max_length = MODEL_CONFIG["sparse_model"].get("max_length", 8192) for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] try: batch_result = sparse_model.encode( batch, batch_size=batch_size, max_length=max_length, return_dense=False, return_sparse=True ) logger.debug(f"Model output keys: {list(batch_result.keys())}") results.extend(batch_result["lexical_weights"]) except Exception as e: logger.error(f"Batch encoding error: {str(e)}", exc_info=True) # 为失败批次返回空字典 results.extend([{}] * len(batch)) # 转换格式以减小响应大小 formatted_vectors = [] for vec_dict in results: # 只保留非零权重项,并将numpy.float32转换为Python float filtered_vec = {token: float(weight) for token, weight in vec_dict.items() if weight > 1e-6} formatted_vectors.append(filtered_vec) return formatted_vectors @app.route('/encode', methods=['POST']) def encode(): try: data = request.json if not data or "texts" not in data: return jsonify({"error": "Missing 'texts' parameter"}), 400 texts = data["texts"] if not isinstance(texts, list): return jsonify({"error": "'texts' must be a list"}), 400 vector_type = data.get("type", "dense") logger.info(f"Received request for {vector_type} vectors, batch size: {len(texts)}") # 检查请求数量限制 max_batch_items = MODEL_CONFIG.get("max_batch_items", 100) if len(texts) > max_batch_items: return jsonify({ "error": f"Batch size exceeds limit ({max_batch_items})" }), 400 if vector_type == "dense": if not dense_model: return jsonify({"error": "Dense vector model not available"}), 503 vectors = batch_encode_dense(texts) return jsonify({"vectors": vectors.tolist(), "type": "dense"}) elif vector_type == "sparse": if not sparse_model: return jsonify({"error": "Sparse vector model not available"}), 503 vectors = batch_encode_sparse(texts) return jsonify({"vectors": vectors, "type": "sparse"}) else: return jsonify({"error": "Invalid vector type. Use 'dense' or 'sparse'"}), 400 except Exception as e: logger.error(f"Encoding request failed: {str(e)}", exc_info=True) return jsonify({"error": "Internal server error"}), 500 @app.route('/status', methods=['GET']) def status(): """服务状态检查端点""" return jsonify({ "status": "running", "dense_model_loaded": dense_model is not None, "sparse_model_loaded": sparse_model is not None }) if __name__ == '__main__': load_models() app.run( host=MODEL_CONFIG.get("host", "0.0.0.0"), port=MODEL_CONFIG.get("port", 8088), debug=MODEL_CONFIG.get("debug", False), threaded=True )这个模型服务如何通过curl测试输入多个texts
<think>首先,用户的问题是:“我想测试模型服务的编码功能 请问如何通过curl测试模型服务的多个文本输入功能”。这涉及到测试模型服务,特别是通过curl发送多个文本输入。
用户提到了“模型服务的编码功能”,可能指的是模型处理文本输入的能力,比如自然语言处理中的编码或嵌入功能。用户想用curl来测试这个功能,支持多个文本输入。
回顾系统指令:
- 所有行内数学表达式用$...$。
- 独立公式用$$...$$。
- LaTeX语法正确。
- 使用中文回答。
- 生成相关问题。
- 在回答中引用的段落末尾添加引用标识。
- 回答结构清晰,帮助用户逐步解决问题。
- 参考站内引用:[^1]和[^2]。
站内引用:
- [^1]: 关于部署模型到服务器,使用TensorFlow Serving提供RESTful API。
- [^2]: 关于测试Ollama服务器,使用curl和Python调用API。
用户的上一次问题可能是在讨论模型服务,但在这个对话中,我没有之前的上下文。系统说“参考用户上一次的问题和你的回答”,但在这个消息中,没有提供之前的对话。所以,我应该基于当前问题回答。
用户的问题:通过curl测试模型服务的多个文本输入功能。焦点是使用curl发送多个文本输入来测试编码功能。
关键点:
- 模型服务:可能是一个API端点,接受文本输入并返回编码(如嵌入向量)。
- curl:命令行工具,用于发送HTTP请求。
- 多个文本输入:用户想发送多个文本作为输入,而不是单个。
在API设计中,处理多个输入通常通过:
- 发送一个包含多个文本的数组或列表。
- 使用批处理端点。
从引用[^2],它显示了使用curl或Python测试API。例如,引用[^2]中有一个Python示例发送JSON数据:{"model": "deepseek-r1:1.5b", "prompt": "hello", "stream": False}。
对于多个文本输入,API可能需要接受一个数组。例如,{"inputs": ["text1", "text2", "text3"]}。
用户指定了“通过curl”,所以我应该提供curl命令示例。
结构回答:
1. 解释如何设置API请求。
2. 提供curl命令示例。
3. 确保使用中文。
4. 最后生成相关问题。
添加引用:由于引用[^1]和[^2]是关于部署和测试API的,我可以参考它们。
步骤:
- 确认API端点:用户需要知道模型服务的URL。例如,如果部署在localhost:8501,端点可能是/v1/models/bert:predict 或类似。
- 请求格式:通常POST请求,JSON body。
- 对于多个文本:在JSON中,使用一个键如"inputs"或"texts"包含一个数组。
例如,假设API设计为接受批处理:
```json
{
"instances": [
{"text": "text1"},
{"text": "text2"}
]
}
```
或直接:
```json
{
"texts": ["text1", "text2"]
}
```
这取决于具体实现。用户可能需要调整以匹配他们的模型服务。
从引用[^1],TensorFlow Serving使用gRPC或REST API。REST API通常有/predict端点。
从引用[^2],Ollama使用/api/generate,但那是生成文本,不是编码。
用户说“编码功能”,可能指的是获取文本嵌入,如BERT的嵌入向量。
所以,我应该提供一个通用的curl命令。
Curl命令示例:
```bash
curl -X POST http://localhost:8501/v1/models/bert:predict \
-H "Content-Type: application/json" \
-d '{
"instances": [
{"text": "This is the first text."},
{"text": "Here is another text."}
]
}'
```
这假设TensorFlow Serving风格。
在回答中:
- 逐步解释。
- 使用中文。
- 添加引用。
- 生成相关问题。
最后,生成相关问题。</think>### 如何通过curl测试模型服务的多个文本输入功能
要测试模型服务的编码功能(如文本嵌入或向量化),您需要通过curl发送包含多个文本输入的HTTP POST请求到模型服务的API端点。核心是构造一个包含多个文本的JSON数组作为请求体,并确保API支持批处理输入。以下是逐步指南:
#### 步骤1: 确认API端点和服务状态
- 首先,确保您的模型服务已部署并运行(如TensorFlow Serving或Ollama)。参考引用[^1],模型服务通常通过RESTful API提供端点,例如:
- TensorFlow Serving: `http://localhost:8501/v1/models/<model_name>:predict`
- Ollama: `http://localhost:11434/api/generate`(但需调整为编码功能,如嵌入端点)
- 测试服务是否可用:
```bash
curl http://localhost:8501/v1/models/bert_text_classification # 检查TensorFlow Serving模型状态[^1]
```
如果返回模型元数据(如版本和状态),说明服务正常运行。
#### 步骤2: 构造包含多个文本的curl请求
- 对于编码功能,API通常需要一个JSON请求体,其中包含一个文本数组(如`inputs`或`texts`键)。例如,发送两个文本输入:
```bash
curl -X POST http://localhost:8501/v1/models/bert_text_classification:predict \
-H "Content-Type: application/json" \
-d '{
"instances": [
{"text": "这是一个测试文本。"},
{"text": "另一个输入用于编码验证。"}
]
}'
```
**参数说明**:
- `-X POST`: 指定POST方法。
- `-H "Content-Type: application/json"`: 设置请求头为JSON格式。
- `-d '...'`: JSON请求体,`instances`是TensorFlow Serving的标准键(其他服务可能用`inputs`或`texts`),数组内每个对象包含一个文本字符串。
- 如果服务使用Ollama风格(引用[^2]),可能调整为:
```bash
curl -X POST http://localhost:11434/api/encode \ # 假设编码端点为/api/encode
-H "Content-Type: application/json" \
-d '{
"model": "deepseek-r1:1.5b",
"texts": ["文本1", "文本2", "文本3"], # 多个文本数组
"stream": false
}'
```
#### 步骤3: 解析响应和处理输出
- 响应通常是JSON格式,包含每个文本的编码结果(如向量数组)。例如:
```json
{
"predictions": [
[0.12, -0.45, ..., 0.78], // 第一个文本的嵌入向量
[0.34, 0.56, ..., -0.91] // 第二个文本的嵌入向量
]
}
```
- 使用工具如`jq`解析响应:
```bash
curl ... | jq '.predictions' # 提取编码向量
```
- **常见问题排查**:
- 如果返回错误(如`400 Bad Request`),检查JSON格式是否正确(使用在线验证工具)。
- 确保文本数量在服务批处理限制内(通常10-100个文本/请求)。
- 对于大文本集,分批次发送以避免超时。
#### 注意事项
- **API兼容性**:不同模型服务(如TensorFlow Serving vs. Ollama)的请求格式可能差异较大。参考服务文档调整`instances`/`texts`等键名。
- **性能测试**:使用多个文本测试时,监控响应时间和资源使用(如CPU/内存)。例如,发送10个文本:
```bash
curl ... -d '{"instances": [{"text":"文本1"}, ..., {"text":"文本10"}]}'
```
- **安全**:在公网部署时,添加认证(如API密钥),避免引用[^1]中的docker命令暴露端口风险。
通过此方法,您可以高效测试模型的编码能力,验证批处理输入是否返回一致的向量输出[^2]。
class YoloV5Detector: def __init__(self, weights, device): self.device = device self.model = torch.jit.load(weights).to(device) self.conf_thres = 0.35 self.iou_thres = 0.45 self.agnostic_nms = False self
### 使用 `YoloV5Detector` 类的实现细节
以下是关于如何构建和调试自定义的 `YoloV5Detector` 类的相关说明,该类实现了加载权重、设置设备以及调整置信度阈值 (`conf_thres`) 和交并比阈值 (`iou_thres`) 等功能。
#### 1. 加载权重
为了支持动态图到静态图的转换,可以利用 PyTorch 提供的 `torch.jit.load()` 方法加载经过 JIT 编译后的模型权重。这一步骤对于优化推理速度非常重要[^1]。
```python
import torch
class YoloV5Detector:
def __init__(self, model_path, device='cpu', conf_thres=0.4, iou_thres=0.5, agnostic_nms=False):
self.device = torch.device(device)
self.model = torch.jit.load(model_path).to(self.device) # Load the JIT compiled model
self.conf_thres = conf_thres
self.iou_thres = iou_thres
self.agnostic_nms = agnostic_nms
def detect(self, image_tensor):
"""
Perform object detection on a single input tensor.
Args:
image_tensor (Tensor): Input image as a Tensor of shape [batch_size, channels, height, width].
Returns:
List[Tensor]: Detected bounding boxes with confidence scores and class labels.
"""
predictions = self.model(image_tensor.to(self.device)) # Run inference
# Apply non-maximum suppression to filter out redundant detections
results = []
for pred in predictions:
filtered_pred = self._apply_nms(pred)
results.append(filtered_pred)
return results
def _apply_nms(self, prediction):
"""
Apply Non-Maximum Suppression (NMS).
Args:
prediction (Tensor): Raw output from the model before NMS is applied.
Returns:
Tensor: Filtered predictions after applying NMS.
"""
import torchvision.ops as ops
boxes = prediction[:, :4] # Extract box coordinates
scores = prediction[:, 4] # Confidence score
classes = prediction[:, 5:] # Class probabilities
selected_indices = ops.nms(boxes, scores, iou_threshold=self.iou_thres)
if not self.agnostic_nms:
final_boxes = []
for cls_id in range(classes.shape[1]):
cls_mask = classes[:, cls_id].argmax(dim=-1) == cls_id
cls_scores = scores * classes[:, cls_id][cls_mask]
cls_selected_indices = ops.nms(
boxes[cls_mask],
cls_scores,
iou_threshold=self.iou_thres
)
final_boxes.extend(cls_selected_indices.tolist())
else:
final_boxes = selected_indices.tolist()
return prediction[selected_indices]
```
#### 2. 设置参数
- **`device`**: 定义模型运行的硬件环境,默认为 CPU。如果 GPU 可用,则可以通过指定 `"cuda"` 来加速计算。
- **`conf_thres`**: 设定预测框的最低置信度分数,低于此值的预测会被过滤掉。
- **`iou_thres`**: 非极大抑制 (Non-Maximum Suppression, NMS) 中使用的 IoU 阈值,用于移除冗余边界框。
- **`agnostic_nms`**: 如果启用,忽略分类信息仅依据 IoU 进行筛选;否则按类别分别执行 NMS。
#### 3. 调试技巧
当遇到错误或性能瓶颈时,可尝试以下方法进行排查:
- **验证输入形状**:确保传入模型的张量大小符合预期,例如 `(1, 3, H, W)`。
- **检查 CUDA 兼容性**:确认所安装的 PyTorch 版本与显卡驱动匹配,并测试简单的矩阵运算是否正常工作。
- **打印中间变量**:在 `_apply_nms` 函数内部增加日志记录语句以便观察每步处理的结果。
#### 示例调用代码
假设我们已经有一个预训练好的 `.pt` 文件位于路径 `/path/to/model.pt` 下面是如何实例化对象并完成检测的过程:
```python
detector = YoloV5Detector('/path/to/model.pt', device='cuda', conf_thres=0.6, iou_thres=0.7)
# Prepare an example image tensor
from PIL import Image
from torchvision.transforms.functional import to_tensor
image = Image.open('test.jpg')
input_tensor = to_tensor(image).unsqueeze(0) # Add batch dimension
detections = detector.detect(input_tensor)
print(detections)
```
阅读全文
相关推荐















