通义千问3-Reranker-0.6B详细步骤Gradio队列并发控制与限流配置如果你用过文本重排序模型可能会遇到这样的问题当多个用户同时访问你的服务时系统要么响应变慢要么直接崩溃。特别是像Qwen3-Reranker-0.6B这样的模型虽然推理速度快但在高并发场景下如果不做任何控制GPU内存很快就会爆掉。今天我就来分享一个实战方案如何为Qwen3-Reranker-0.6B的Gradio界面配置队列和限流让你的服务既能支持多人同时使用又能保持稳定运行。1. 为什么需要队列和限流在深入配置之前我们先搞清楚为什么要做这些控制。1.1 并发访问的挑战想象一下这样的场景你的Qwen3-Reranker服务部署好了同事们都觉得好用于是张三在排序搜索结果李四在优化文档推荐王五在测试问答匹配三个人同时点击开始排序会发生什么没有控制的情况三个请求同时到达GPUGPU内存瞬间被占满推理速度急剧下降最坏的情况显存溢出服务崩溃有控制的情况请求进入队列排队一次只处理一个请求其他请求等待系统稳定运行1.2 Qwen3-Reranker的资源特点Qwen3-Reranker-0.6B虽然参数少但在实际使用中每个请求需要约1-2GB显存推理时间约0.5-2秒取决于文本长度支持32K上下文长文本处理更耗资源如果不加控制10个并发请求就可能需要20GB显存这对大多数GPU来说都是不可承受的。2. Gradio队列基础配置Gradio自带了队列功能只需要简单配置就能启用。下面我们一步步来。2.1 修改启动脚本首先找到你的启动脚本通常位于/root/workspace/目录下。我们需要修改Gradio的启动参数。# 原来的简单启动方式 import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM import torch # 加载模型 MODEL_PATH /opt/qwen3-reranker/model/Qwen3-Reranker-0.6B tokenizer AutoTokenizer.from_pretrained(MODEL_PATH, padding_sideleft) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, device_mapauto ).eval() def rerank(query, documents, instructionNone): 重排序函数 # 处理逻辑... return sorted_results # 创建界面 iface gr.Interface( fnrerank, inputs[ gr.Textbox(label查询语句, placeholder输入您的问题...), gr.Textbox(label候选文档, placeholder每行一个文档..., lines10), gr.Textbox(label自定义指令可选, placeholder英文指令如Find relevant technical documents) ], outputsgr.JSON(label排序结果), examples[ [什么是机器学习, 机器学习是人工智能的一个分支\n深度学习是机器学习的一种\n传统编程需要明确规则, Find relevant technical documents] ] ) iface.launch(server_name0.0.0.0, server_port7860)现在我们添加队列配置# 添加队列配置的启动方式 iface.queue( # 关键配置启用队列 concurrency_count1, # 同时处理的任务数 max_size10, # 队列最大长度 api_openFalse # 是否开放API ).launch( server_name0.0.0.0, server_port7860, shareFalse )2.2 配置参数详解让我解释一下这些参数的作用concurrency_count1这是最重要的参数表示同时处理几个任务设为1表示一次只处理一个请求避免GPU过载如果你的GPU足够强大比如有24GB显存可以设为2或3max_size10队列能容纳的最大请求数当有10个请求在等待时新的请求会被拒绝防止队列无限增长导致内存耗尽api_openFalse是否开放API接口设为False更安全只通过Web界面访问2.3 验证队列是否生效启动服务后你可以这样测试队列效果打开两个浏览器标签都访问你的服务地址在两个标签中同时点击开始排序观察现象第一个请求立即开始处理第二个请求显示排队中...第一个完成后第二个自动开始你会在界面上看到这样的提示状态排队中位置2这表示队列正在正常工作。3. 高级限流配置队列解决了并发问题但还不够。我们还需要限流防止恶意请求或误操作。3.1 基于IP的限流Gradio本身不提供IP限流但我们可以用Python装饰器来实现from functools import wraps from collections import defaultdict import time from flask import request # 存储每个IP的请求记录 request_history defaultdict(list) def rate_limit(max_requests10, time_window60): 限流装饰器每分钟最多max_requests次请求 def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 获取客户端IP简化版实际可能需要更复杂的获取方式 client_ip request.remote_addr if hasattr(request, remote_addr) else unknown current_time time.time() # 清理过期的请求记录 request_history[client_ip] [ req_time for req_time in request_history[client_ip] if current_time - req_time time_window ] # 检查是否超过限制 if len(request_history[client_ip]) max_requests: return { error: 请求过于频繁, message: f每分钟最多{max_requests}次请求请稍后再试, wait_time: time_window - (current_time - request_history[client_ip][0]) } # 记录本次请求 request_history[client_ip].append(current_time) # 执行原函数 return func(*args, **kwargs) return wrapper return decorator # 应用到重排序函数 rate_limit(max_requests30, time_window60) # 每分钟最多30次 def rerank(query, documents, instructionNone): 带限流的重排序函数 # 原有的处理逻辑...3.2 基于令牌桶的限流对于更精细的控制可以使用令牌桶算法import threading import time class TokenBucket: 令牌桶限流器 def __init__(self, capacity, fill_rate): capacity: 桶容量 fill_rate: 每秒填充的令牌数 self.capacity float(capacity) self.tokens float(capacity) self.fill_rate float(fill_rate) self.last_time time.time() self.lock threading.Lock() def consume(self, tokens1): 消耗令牌返回是否成功 with self.lock: now time.time() # 计算这段时间应该填充的令牌 delta self.fill_rate * (now - self.last_time) self.tokens min(self.capacity, self.tokens delta) self.last_time now if self.tokens tokens: self.tokens - tokens return True return False # 创建全局限流器 # 假设桶容量10个令牌每秒填充2个令牌 rerank_limiter TokenBucket(capacity10, fill_rate2) def rerank_with_limiter(query, documents, instructionNone): 带令牌桶限流的重排序 if not rerank_limiter.consume(): return { error: 系统繁忙, message: 请求速率超过限制请稍后重试, suggestion: 可以尝试减少单次请求的文档数量 } # 原有的重排序逻辑...4. 完整配置示例下面是一个完整的、带队列和限流的配置示例#!/usr/bin/env python3 Qwen3-Reranker-0.6B 服务端配置 包含队列控制和限流功能 import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM import torch import time from functools import wraps from collections import defaultdict # 限流配置 class RateLimiter: 简单的频率限制器 def __init__(self, max_calls30, period60): self.max_calls max_calls self.period period self.calls [] def __call__(self, func): wraps(func) def wrapper(*args, **kwargs): now time.time() # 清理过期的调用记录 self.calls [call_time for call_time in self.calls if now - call_time self.period] if len(self.calls) self.max_calls: # 计算需要等待的时间 wait_time self.period - (now - self.calls[0]) return { status: error, message: f请求过于频繁请等待{wait_time:.1f}秒后重试, code: 429 } # 记录本次调用 self.calls.append(now) return func(*args, **kwargs) return wrapper # 创建限流器实例 limiter RateLimiter(max_calls30, period60) # 模型加载 print(正在加载Qwen3-Reranker-0.6B模型...) MODEL_PATH /opt/qwen3-reranker/model/Qwen3-Reranker-0.6B try: tokenizer AutoTokenizer.from_pretrained( MODEL_PATH, padding_sideleft, trust_remote_codeTrue ) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, device_mapauto, trust_remote_codeTrue ).eval() print(模型加载完成) except Exception as e: print(f模型加载失败: {e}) exit(1) # 重排序函数 limiter def rerank(query, documents, instructionNone): 文本重排序主函数 支持自定义指令优化 start_time time.time() if not query or not documents: return {error: 查询语句和候选文档不能为空} # 处理输入 docs [doc.strip() for doc in documents.split(\n) if doc.strip()] if not docs: return {error: 没有有效的候选文档} if len(docs) 50: return {error: 候选文档数量过多最多支持50个} # 默认指令 if not instruction: instruction Given a query, retrieve relevant passages results [] try: # 批量计算相关性分数 for i, doc in enumerate(docs): # 构建输入文本 text fInstruct: {instruction}\nQuery: {query}\nDocument: {doc} # Tokenize inputs tokenizer(text, return_tensorspt, truncationTrue, max_length8192) inputs inputs.to(model.device) # 推理 with torch.no_grad(): logits model(**inputs).logits[:, -1, :] # 计算相关性分数 score torch.softmax( logits[:, [tokenizer.convert_tokens_to_ids(no), tokenizer.convert_tokens_to_ids(yes)]], dim1 )[:, 1].item() results.append({ rank: i 1, document: doc[:200] ... if len(doc) 200 else doc, score: round(score, 4), relevance: 高 if score 0.7 else 中 if score 0.3 else 低 }) # 按分数排序 sorted_results sorted(results, keylambda x: x[score], reverseTrue) for i, item in enumerate(sorted_results): item[rank] i 1 # 计算处理时间 process_time time.time() - start_time return { status: success, query: query, total_documents: len(docs), processing_time: f{process_time:.2f}秒, results: sorted_results[:10] # 只返回前10个结果 } except Exception as e: return { status: error, message: f处理失败: {str(e)}, suggestion: 请检查输入文本长度或重试 } # Gradio界面 def create_interface(): 创建Gradio界面 # 自定义CSS改善界面显示 custom_css .gradio-container { max-width: 1200px !important; } .queue-status { padding: 10px; background: #f5f5f5; border-radius: 5px; margin-bottom: 15px; } with gr.Blocks(csscustom_css, titleQwen3-Reranker-0.6B 文本重排序) as demo: gr.Markdown( # Qwen3-Reranker-0.6B 文本重排序服务 **基于通义千问3的轻量级重排序模型支持100语言32K上下文** ) # 状态显示区域 with gr.Row(): status_box gr.Markdown( ### 服务状态 - **队列系统**: 已启用最大并发: 1队列长度: 10 - **限流保护**: 已启用30次/分钟 - **模型状态**: 已加载等待请求 ) # 输入区域 with gr.Row(): with gr.Column(scale1): query_input gr.Textbox( label 查询语句, placeholder例如什么是机器学习, lines3, max_lines5 ) instruction_input gr.Textbox( label⚙️ 自定义指令可选, placeholder英文指令如Find relevant technical documents about AI, valueGiven a query, retrieve relevant passages, lines2 ) submit_btn gr.Button( 开始排序, variantprimary) with gr.Column(scale2): documents_input gr.Textbox( label 候选文档每行一个, placeholder请输入候选文档每行一个...\n例如\n机器学习是人工智能的一个分支\n深度学习使用神经网络\n传统编程需要明确规则, lines15, max_lines20 ) # 输出区域 with gr.Row(): output_json gr.JSON( label 排序结果, show_labelTrue ) # 示例区域 with gr.Row(): gr.Examples( examples[ [ 如何学习Python编程, Python是一种高级编程语言\n学习Python需要掌握基础语法\n实践是最好的学习方式\n阅读官方文档很重要\n参加编程社区可以快速进步, Find educational resources for beginners ], [ 人工智能的未来发展趋势, AI将在医疗领域发挥重要作用\n自动驾驶技术正在快速发展\n自然语言处理取得突破\nAI伦理问题值得关注\nAI与物联网结合是趋势, Find technical insights about AI development ] ], inputs[query_input, documents_input, instruction_input], label 快速示例 ) # 事件绑定 submit_btn.click( fnrerank, inputs[query_input, documents_input, instruction_input], outputsoutput_json ) # 页面加载时的提示 demo.load( fnlambda: 页面加载完成请输入查询语句和候选文档开始排序。, outputsNone ) return demo # 主程序 if __name__ __main__: print(启动Qwen3-Reranker服务...) print(服务地址: http://0.0.0.0:7860) print(队列配置: concurrency_count1, max_size10) print(限流配置: 30次/分钟) demo create_interface() # 启动服务启用队列 demo.queue( concurrency_count1, # 同时处理1个请求 max_size10, # 队列最多10个请求 api_openFalse, # 不开放API default_concurrency_limit1 ).launch( server_name0.0.0.0, server_port7860, shareFalse, show_errorTrue, debugFalse )5. 生产环境优化建议上面的配置已经可以应对大多数场景但在生产环境中你可能还需要考虑以下几点5.1 监控与告警添加服务监控及时发现问题# 简单的监控装饰器 def monitor_performance(func): 监控函数性能 wraps(func) def wrapper(*args, **kwargs): start_time time.time() start_memory torch.cuda.memory_allocated() if torch.cuda.is_available() else 0 try: result func(*args, **kwargs) end_time time.time() end_memory torch.cuda.memory_allocated() if torch.cuda.is_available() else 0 # 记录性能指标 process_time end_time - start_time memory_used (end_memory - start_memory) / 1024 / 1024 # MB print(f[监控] 处理时间: {process_time:.2f}s, 显存使用: {memory_used:.1f}MB) # 如果处理时间过长记录警告 if process_time 5.0: print(f[警告] 处理时间过长: {process_time:.2f}s) return result except Exception as e: print(f[错误] 处理失败: {e}) raise return wrapper # 应用到重排序函数 monitor_performance def rerank(query, documents, instructionNone): # ...原有逻辑...5.2 动态调整队列参数根据GPU使用情况动态调整并发数import psutil import GPUtil def get_system_status(): 获取系统状态 status { cpu_percent: psutil.cpu_percent(), memory_percent: psutil.virtual_memory().percent, } if torch.cuda.is_available(): gpus GPUtil.getGPUs() if gpus: gpu gpus[0] status.update({ gpu_load: gpu.load * 100, gpu_memory_used: gpu.memoryUsed, gpu_memory_total: gpu.memoryTotal, gpu_memory_percent: gpu.memoryUtil * 100 }) return status def adjust_concurrency(): 根据系统状态调整并发数 status get_system_status() if gpu_memory_percent in status: memory_percent status[gpu_memory_percent] # 根据显存使用率调整并发数 if memory_percent 80: return 1 # 高负载只处理1个 elif memory_percent 60: return 2 # 中等负载处理2个 else: return 3 # 低负载处理3个 return 1 # 默认值5.3 错误处理与重试添加健壮的错误处理from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min4, max10) # 指数退避 ) def safe_rerank(query, documents, instructionNone): 带重试机制的重排序 try: return rerank(query, documents, instruction) except torch.cuda.OutOfMemoryError: # 显存不足清理缓存 torch.cuda.empty_cache() return { status: error, message: 显存不足请减少候选文档数量或文本长度, suggestion: 建议每次处理不超过20个文档每个文档不超过1000字 } except Exception as e: return { status: error, message: f处理失败: {str(e)} }6. 总结通过为Qwen3-Reranker-0.6B配置Gradio队列和限流我们实现了6.1 核心收获并发控制使用concurrency_count限制同时处理的任务数避免GPU过载队列管理通过max_size控制队列长度防止内存耗尽限流保护实现基于频率和令牌桶的限流防止滥用稳定运行添加监控、错误处理和动态调整确保服务稳定6.2 配置建议根据你的实际场景可以参考以下配置场景并发数队列长度限流策略个人测试1530次/分钟小团队使用210100次/分钟生产环境动态调整20综合策略6.3 实际效果配置完成后你的Qwen3-Reranker服务将具备稳定性即使多人同时使用也不会崩溃公平性请求按顺序处理不会饿死可监控随时了解系统状态和性能易维护出现问题可以快速定位和恢复记住好的配置不是一成不变的。你需要根据实际使用情况不断调整参数。开始时保守一点观察系统表现然后逐步优化。毕竟稳定的服务比最高的性能更重要。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
通义千问3-Reranker-0.6B详细步骤：Gradio队列并发控制与限流配置
通义千问3-Reranker-0.6B详细步骤Gradio队列并发控制与限流配置如果你用过文本重排序模型可能会遇到这样的问题当多个用户同时访问你的服务时系统要么响应变慢要么直接崩溃。特别是像Qwen3-Reranker-0.6B这样的模型虽然推理速度快但在高并发场景下如果不做任何控制GPU内存很快就会爆掉。今天我就来分享一个实战方案如何为Qwen3-Reranker-0.6B的Gradio界面配置队列和限流让你的服务既能支持多人同时使用又能保持稳定运行。1. 为什么需要队列和限流在深入配置之前我们先搞清楚为什么要做这些控制。1.1 并发访问的挑战想象一下这样的场景你的Qwen3-Reranker服务部署好了同事们都觉得好用于是张三在排序搜索结果李四在优化文档推荐王五在测试问答匹配三个人同时点击开始排序会发生什么没有控制的情况三个请求同时到达GPUGPU内存瞬间被占满推理速度急剧下降最坏的情况显存溢出服务崩溃有控制的情况请求进入队列排队一次只处理一个请求其他请求等待系统稳定运行1.2 Qwen3-Reranker的资源特点Qwen3-Reranker-0.6B虽然参数少但在实际使用中每个请求需要约1-2GB显存推理时间约0.5-2秒取决于文本长度支持32K上下文长文本处理更耗资源如果不加控制10个并发请求就可能需要20GB显存这对大多数GPU来说都是不可承受的。2. Gradio队列基础配置Gradio自带了队列功能只需要简单配置就能启用。下面我们一步步来。2.1 修改启动脚本首先找到你的启动脚本通常位于/root/workspace/目录下。我们需要修改Gradio的启动参数。# 原来的简单启动方式 import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM import torch # 加载模型 MODEL_PATH /opt/qwen3-reranker/model/Qwen3-Reranker-0.6B tokenizer AutoTokenizer.from_pretrained(MODEL_PATH, padding_sideleft) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, device_mapauto ).eval() def rerank(query, documents, instructionNone): 重排序函数 # 处理逻辑... return sorted_results # 创建界面 iface gr.Interface( fnrerank, inputs[ gr.Textbox(label查询语句, placeholder输入您的问题...), gr.Textbox(label候选文档, placeholder每行一个文档..., lines10), gr.Textbox(label自定义指令可选, placeholder英文指令如Find relevant technical documents) ], outputsgr.JSON(label排序结果), examples[ [什么是机器学习, 机器学习是人工智能的一个分支\n深度学习是机器学习的一种\n传统编程需要明确规则, Find relevant technical documents] ] ) iface.launch(server_name0.0.0.0, server_port7860)现在我们添加队列配置# 添加队列配置的启动方式 iface.queue( # 关键配置启用队列 concurrency_count1, # 同时处理的任务数 max_size10, # 队列最大长度 api_openFalse # 是否开放API ).launch( server_name0.0.0.0, server_port7860, shareFalse )2.2 配置参数详解让我解释一下这些参数的作用concurrency_count1这是最重要的参数表示同时处理几个任务设为1表示一次只处理一个请求避免GPU过载如果你的GPU足够强大比如有24GB显存可以设为2或3max_size10队列能容纳的最大请求数当有10个请求在等待时新的请求会被拒绝防止队列无限增长导致内存耗尽api_openFalse是否开放API接口设为False更安全只通过Web界面访问2.3 验证队列是否生效启动服务后你可以这样测试队列效果打开两个浏览器标签都访问你的服务地址在两个标签中同时点击开始排序观察现象第一个请求立即开始处理第二个请求显示排队中...第一个完成后第二个自动开始你会在界面上看到这样的提示状态排队中位置2这表示队列正在正常工作。3. 高级限流配置队列解决了并发问题但还不够。我们还需要限流防止恶意请求或误操作。3.1 基于IP的限流Gradio本身不提供IP限流但我们可以用Python装饰器来实现from functools import wraps from collections import defaultdict import time from flask import request # 存储每个IP的请求记录 request_history defaultdict(list) def rate_limit(max_requests10, time_window60): 限流装饰器每分钟最多max_requests次请求 def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 获取客户端IP简化版实际可能需要更复杂的获取方式 client_ip request.remote_addr if hasattr(request, remote_addr) else unknown current_time time.time() # 清理过期的请求记录 request_history[client_ip] [ req_time for req_time in request_history[client_ip] if current_time - req_time time_window ] # 检查是否超过限制 if len(request_history[client_ip]) max_requests: return { error: 请求过于频繁, message: f每分钟最多{max_requests}次请求请稍后再试, wait_time: time_window - (current_time - request_history[client_ip][0]) } # 记录本次请求 request_history[client_ip].append(current_time) # 执行原函数 return func(*args, **kwargs) return wrapper return decorator # 应用到重排序函数 rate_limit(max_requests30, time_window60) # 每分钟最多30次 def rerank(query, documents, instructionNone): 带限流的重排序函数 # 原有的处理逻辑...3.2 基于令牌桶的限流对于更精细的控制可以使用令牌桶算法import threading import time class TokenBucket: 令牌桶限流器 def __init__(self, capacity, fill_rate): capacity: 桶容量 fill_rate: 每秒填充的令牌数 self.capacity float(capacity) self.tokens float(capacity) self.fill_rate float(fill_rate) self.last_time time.time() self.lock threading.Lock() def consume(self, tokens1): 消耗令牌返回是否成功 with self.lock: now time.time() # 计算这段时间应该填充的令牌 delta self.fill_rate * (now - self.last_time) self.tokens min(self.capacity, self.tokens delta) self.last_time now if self.tokens tokens: self.tokens - tokens return True return False # 创建全局限流器 # 假设桶容量10个令牌每秒填充2个令牌 rerank_limiter TokenBucket(capacity10, fill_rate2) def rerank_with_limiter(query, documents, instructionNone): 带令牌桶限流的重排序 if not rerank_limiter.consume(): return { error: 系统繁忙, message: 请求速率超过限制请稍后重试, suggestion: 可以尝试减少单次请求的文档数量 } # 原有的重排序逻辑...4. 完整配置示例下面是一个完整的、带队列和限流的配置示例#!/usr/bin/env python3 Qwen3-Reranker-0.6B 服务端配置 包含队列控制和限流功能 import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM import torch import time from functools import wraps from collections import defaultdict # 限流配置 class RateLimiter: 简单的频率限制器 def __init__(self, max_calls30, period60): self.max_calls max_calls self.period period self.calls [] def __call__(self, func): wraps(func) def wrapper(*args, **kwargs): now time.time() # 清理过期的调用记录 self.calls [call_time for call_time in self.calls if now - call_time self.period] if len(self.calls) self.max_calls: # 计算需要等待的时间 wait_time self.period - (now - self.calls[0]) return { status: error, message: f请求过于频繁请等待{wait_time:.1f}秒后重试, code: 429 } # 记录本次调用 self.calls.append(now) return func(*args, **kwargs) return wrapper # 创建限流器实例 limiter RateLimiter(max_calls30, period60) # 模型加载 print(正在加载Qwen3-Reranker-0.6B模型...) MODEL_PATH /opt/qwen3-reranker/model/Qwen3-Reranker-0.6B try: tokenizer AutoTokenizer.from_pretrained( MODEL_PATH, padding_sideleft, trust_remote_codeTrue ) model AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtypetorch.float16, device_mapauto, trust_remote_codeTrue ).eval() print(模型加载完成) except Exception as e: print(f模型加载失败: {e}) exit(1) # 重排序函数 limiter def rerank(query, documents, instructionNone): 文本重排序主函数 支持自定义指令优化 start_time time.time() if not query or not documents: return {error: 查询语句和候选文档不能为空} # 处理输入 docs [doc.strip() for doc in documents.split(\n) if doc.strip()] if not docs: return {error: 没有有效的候选文档} if len(docs) 50: return {error: 候选文档数量过多最多支持50个} # 默认指令 if not instruction: instruction Given a query, retrieve relevant passages results [] try: # 批量计算相关性分数 for i, doc in enumerate(docs): # 构建输入文本 text fInstruct: {instruction}\nQuery: {query}\nDocument: {doc} # Tokenize inputs tokenizer(text, return_tensorspt, truncationTrue, max_length8192) inputs inputs.to(model.device) # 推理 with torch.no_grad(): logits model(**inputs).logits[:, -1, :] # 计算相关性分数 score torch.softmax( logits[:, [tokenizer.convert_tokens_to_ids(no), tokenizer.convert_tokens_to_ids(yes)]], dim1 )[:, 1].item() results.append({ rank: i 1, document: doc[:200] ... if len(doc) 200 else doc, score: round(score, 4), relevance: 高 if score 0.7 else 中 if score 0.3 else 低 }) # 按分数排序 sorted_results sorted(results, keylambda x: x[score], reverseTrue) for i, item in enumerate(sorted_results): item[rank] i 1 # 计算处理时间 process_time time.time() - start_time return { status: success, query: query, total_documents: len(docs), processing_time: f{process_time:.2f}秒, results: sorted_results[:10] # 只返回前10个结果 } except Exception as e: return { status: error, message: f处理失败: {str(e)}, suggestion: 请检查输入文本长度或重试 } # Gradio界面 def create_interface(): 创建Gradio界面 # 自定义CSS改善界面显示 custom_css .gradio-container { max-width: 1200px !important; } .queue-status { padding: 10px; background: #f5f5f5; border-radius: 5px; margin-bottom: 15px; } with gr.Blocks(csscustom_css, titleQwen3-Reranker-0.6B 文本重排序) as demo: gr.Markdown( # Qwen3-Reranker-0.6B 文本重排序服务 **基于通义千问3的轻量级重排序模型支持100语言32K上下文** ) # 状态显示区域 with gr.Row(): status_box gr.Markdown( ### 服务状态 - **队列系统**: 已启用最大并发: 1队列长度: 10 - **限流保护**: 已启用30次/分钟 - **模型状态**: 已加载等待请求 ) # 输入区域 with gr.Row(): with gr.Column(scale1): query_input gr.Textbox( label 查询语句, placeholder例如什么是机器学习, lines3, max_lines5 ) instruction_input gr.Textbox( label⚙️ 自定义指令可选, placeholder英文指令如Find relevant technical documents about AI, valueGiven a query, retrieve relevant passages, lines2 ) submit_btn gr.Button( 开始排序, variantprimary) with gr.Column(scale2): documents_input gr.Textbox( label 候选文档每行一个, placeholder请输入候选文档每行一个...\n例如\n机器学习是人工智能的一个分支\n深度学习使用神经网络\n传统编程需要明确规则, lines15, max_lines20 ) # 输出区域 with gr.Row(): output_json gr.JSON( label 排序结果, show_labelTrue ) # 示例区域 with gr.Row(): gr.Examples( examples[ [ 如何学习Python编程, Python是一种高级编程语言\n学习Python需要掌握基础语法\n实践是最好的学习方式\n阅读官方文档很重要\n参加编程社区可以快速进步, Find educational resources for beginners ], [ 人工智能的未来发展趋势, AI将在医疗领域发挥重要作用\n自动驾驶技术正在快速发展\n自然语言处理取得突破\nAI伦理问题值得关注\nAI与物联网结合是趋势, Find technical insights about AI development ] ], inputs[query_input, documents_input, instruction_input], label 快速示例 ) # 事件绑定 submit_btn.click( fnrerank, inputs[query_input, documents_input, instruction_input], outputsoutput_json ) # 页面加载时的提示 demo.load( fnlambda: 页面加载完成请输入查询语句和候选文档开始排序。, outputsNone ) return demo # 主程序 if __name__ __main__: print(启动Qwen3-Reranker服务...) print(服务地址: http://0.0.0.0:7860) print(队列配置: concurrency_count1, max_size10) print(限流配置: 30次/分钟) demo create_interface() # 启动服务启用队列 demo.queue( concurrency_count1, # 同时处理1个请求 max_size10, # 队列最多10个请求 api_openFalse, # 不开放API default_concurrency_limit1 ).launch( server_name0.0.0.0, server_port7860, shareFalse, show_errorTrue, debugFalse )5. 生产环境优化建议上面的配置已经可以应对大多数场景但在生产环境中你可能还需要考虑以下几点5.1 监控与告警添加服务监控及时发现问题# 简单的监控装饰器 def monitor_performance(func): 监控函数性能 wraps(func) def wrapper(*args, **kwargs): start_time time.time() start_memory torch.cuda.memory_allocated() if torch.cuda.is_available() else 0 try: result func(*args, **kwargs) end_time time.time() end_memory torch.cuda.memory_allocated() if torch.cuda.is_available() else 0 # 记录性能指标 process_time end_time - start_time memory_used (end_memory - start_memory) / 1024 / 1024 # MB print(f[监控] 处理时间: {process_time:.2f}s, 显存使用: {memory_used:.1f}MB) # 如果处理时间过长记录警告 if process_time 5.0: print(f[警告] 处理时间过长: {process_time:.2f}s) return result except Exception as e: print(f[错误] 处理失败: {e}) raise return wrapper # 应用到重排序函数 monitor_performance def rerank(query, documents, instructionNone): # ...原有逻辑...5.2 动态调整队列参数根据GPU使用情况动态调整并发数import psutil import GPUtil def get_system_status(): 获取系统状态 status { cpu_percent: psutil.cpu_percent(), memory_percent: psutil.virtual_memory().percent, } if torch.cuda.is_available(): gpus GPUtil.getGPUs() if gpus: gpu gpus[0] status.update({ gpu_load: gpu.load * 100, gpu_memory_used: gpu.memoryUsed, gpu_memory_total: gpu.memoryTotal, gpu_memory_percent: gpu.memoryUtil * 100 }) return status def adjust_concurrency(): 根据系统状态调整并发数 status get_system_status() if gpu_memory_percent in status: memory_percent status[gpu_memory_percent] # 根据显存使用率调整并发数 if memory_percent 80: return 1 # 高负载只处理1个 elif memory_percent 60: return 2 # 中等负载处理2个 else: return 3 # 低负载处理3个 return 1 # 默认值5.3 错误处理与重试添加健壮的错误处理from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min4, max10) # 指数退避 ) def safe_rerank(query, documents, instructionNone): 带重试机制的重排序 try: return rerank(query, documents, instruction) except torch.cuda.OutOfMemoryError: # 显存不足清理缓存 torch.cuda.empty_cache() return { status: error, message: 显存不足请减少候选文档数量或文本长度, suggestion: 建议每次处理不超过20个文档每个文档不超过1000字 } except Exception as e: return { status: error, message: f处理失败: {str(e)} }6. 总结通过为Qwen3-Reranker-0.6B配置Gradio队列和限流我们实现了6.1 核心收获并发控制使用concurrency_count限制同时处理的任务数避免GPU过载队列管理通过max_size控制队列长度防止内存耗尽限流保护实现基于频率和令牌桶的限流防止滥用稳定运行添加监控、错误处理和动态调整确保服务稳定6.2 配置建议根据你的实际场景可以参考以下配置场景并发数队列长度限流策略个人测试1530次/分钟小团队使用210100次/分钟生产环境动态调整20综合策略6.3 实际效果配置完成后你的Qwen3-Reranker服务将具备稳定性即使多人同时使用也不会崩溃公平性请求按顺序处理不会饿死可监控随时了解系统状态和性能易维护出现问题可以快速定位和恢复记住好的配置不是一成不变的。你需要根据实际使用情况不断调整参数。开始时保守一点观察系统表现然后逐步优化。毕竟稳定的服务比最高的性能更重要。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
