nli-MiniLM2-L6-H768实操手册：批量API调用限流与异步结果回调实现-编程阁

nli-MiniLM2-L6-H768实操手册：批量API调用限流与异步结果回调实现

1. 工具概述

nli-MiniLM2-L6-H768是一款基于cross-encoder/nli-MiniLM2-L6-H768轻量级NLI模型开发的本地零样本文本分类工具。它无需任何微调训练，只需输入文本和自定义标签，即可一键完成文本分类任务。该工具支持可视化概率展示，兼容CPU/GPU环境，具有极速推理能力，并能完全离线运行。

1.1 核心优势

零样本学习：无需标注数据或模型微调
轻量高效：小模型体量，加载速度快，推理迅速
隐私安全：纯本地运行，无数据外传风险
灵活易用：支持任意自定义标签，操作简单直观

2. 批量API调用实现

2.1 基础API接口

from transformers import AutoModelForSequenceClassification, AutoTokenizer import torch model = AutoModelForSequenceClassification.from_pretrained('cross-encoder/nli-MiniLM2-L6-H768') tokenizer = AutoTokenizer.from_pretrained('cross-encoder/nli-MiniLM2-L6-H768') def classify_text(text, candidate_labels): inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True) with torch.no_grad(): outputs = model(**inputs) scores = torch.softmax(outputs.logits, dim=1) return {label: float(score) for label, score in zip(candidate_labels, scores[0])}

2.2 批量处理实现

对于需要处理大量文本的场景，我们可以通过以下方式实现批量处理：

from concurrent.futures import ThreadPoolExecutor import queue class BatchClassifier: def __init__(self, max_workers=4): self.executor = ThreadPoolExecutor(max_workers=max_workers) self.request_queue = queue.Queue() def submit_task(self, text, labels, callback=None): future = self.executor.submit(self._process_single, text, labels) if callback: future.add_done_callback(callback) return future def _process_single(self, text, labels): return classify_text(text, labels)

3. 限流机制实现

3.1 令牌桶限流算法

为了防止API被过度调用导致系统资源耗尽，我们实现了一个简单的令牌桶限流机制：

import time from threading import Lock class RateLimiter: def __init__(self, rate, capacity): self.rate = rate # 每秒允许的请求数 self.capacity = capacity # 桶的容量 self.tokens = capacity self.last_time = time.time() self.lock = Lock() def acquire(self): with self.lock: now = time.time() elapsed = now - self.last_time self.tokens = min(self.capacity, self.tokens + elapsed * self.rate) self.last_time = now if self.tokens >= 1: self.tokens -= 1 return True return False

3.2 集成限流的分类器

将限流器集成到批量分类器中：

class RateLimitedClassifier(BatchClassifier): def __init__(self, max_workers=4, rate=10): super().__init__(max_workers) self.rate_limiter = RateLimiter(rate, rate) def submit_task(self, text, labels, callback=None): while not self.rate_limiter.acquire(): time.sleep(0.1) return super().submit_task(text, labels, callback)

4. 异步结果回调实现

4.1 回调函数设计

def result_callback(future): try: result = future.result() print(f"分类结果: {result}") # 这里可以添加结果处理逻辑，如存储到数据库等 except Exception as e: print(f"处理失败: {e}")

4.2 完整使用示例

if __name__ == "__main__": classifier = RateLimitedClassifier(max_workers=4, rate=5) texts = ["人工智能正在改变世界", "足球比赛非常精彩", "这家餐厅的服务很差"] labels = ["科技", "体育", "餐饮", "情感积极", "情感消极"] for text in texts: classifier.submit_task(text, labels, result_callback) # 等待所有任务完成 classifier.executor.shutdown(wait=True)

5. 性能优化建议

5.1 批处理推理

对于大量小文本，可以使用模型的批处理能力提高效率：

def batch_classify(texts, candidate_labels): inputs = tokenizer(texts, return_tensors="pt", truncation=True, padding=True, max_length=128) with torch.no_grad(): outputs = model(**inputs) scores = torch.softmax(outputs.logits, dim=1) return [{label: float(score) for label, score in zip(candidate_labels, row)} for row in scores]

5.2 GPU加速

如果使用GPU，可以优化如下：

device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = model.to(device) def classify_text_gpu(text, candidate_labels): inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True).to(device) with torch.no_grad(): outputs = model(**inputs) scores = torch.softmax(outputs.logits.cpu(), dim=1) return {label: float(score) for label, score in zip(candidate_labels, scores[0])}