你们单位的deepseek能支持几个人同时提问？

通过昨天的并发测试已经搞清楚了 ollama 的排队运行原理。当多人使用本地部署的 deepseek 时，使用的是先问先答的排队机制。

deepseek 回答问题的时间是固定的，不会因为问的人多变得结巴。但同时问的人多，deepseek 会选择一个一个处理，后面的人会处于等待状态。测试过程可以看我昨天的文章：

多并发场景 deepseek 答案生成速度会变慢吗？

回到主题，本地部署的deepseek能支持多少人同时问问题？

如果你家的deepseek是通过ollama来运行的，ollama 自身有排队机制。理论上只要有人问，deepseek就会回答，不存在只问不答的情况。但后问的人会排队，当排队时间达到人能忍受的极限就没法再用了。

这个极限在互动对话中实时性要求比较高，一般1分钟以内还能接受，超1分钟就没法忍了。你每问一下都先卡1分钟，再开始思考，一轮下来要3分钟。等待1分钟，回答用时2分钟，全部完毕3分钟。这是人的忍耐极限了。

这个放在 API 调用场景人的忍耐极限可以再长一点，一般3~5分钟都可以接受。只要把该生成的内容生成好即可，等待的同时还可以干别的。与对话不同，对话节奏慢会打断思路。API 调用主要目的是生成格式文本，比如报告、病例等。

通过预计一次完整回答需要多长时间，就可以估计出 deepseek 服务器能承受的最大并发量。

前面铺垫完毕，后面通过脚本测试并发量。通过 /api/generate API 接口模拟多人在线的情况。

$ pyhton strees.py 10

程序逻辑是使用多线程，保证一直有10个会话在队列中，通过 /api/generate 向 deepseek 服务器发起请求：

def send_request():  
  url = "http://127.0.0.1:11434/api/generate"  
    data = {       
     "model": "deepseek-r1:32b",   
         "prompt": "解释一下量子计算",     
            "stream": False   
         }  
      response = requests.post(url, json=data, timeout=None)

通过死循环保证一直有10个会话在运行，此时有人再向deepseek提问，一定会排在这10个会话之外。相当于保证了系统一直有10个人在同时问问题。

这种情况模拟了并发量为10的情况，此时如果在 open webui 上通过对话方式问问题，会出现卡顿情况：

在 ollama 后台可以看到每个会话的响应时间：

当并发量为10时，每个请求大概需要3分钟到4分钟。其中上面的 chat 就是通过 open webui 提问的会话，其它 generate 请求是通过 API 压力测试的请求。

当使用脚本进行压力测试时，10个并发量就使所有人的提问加等待时间上升到3分钟左右，使用体验变差。

通过观察压测脚本输出，也可以看到其中一个会话反馈时间为 190s 左右。

通过 python stress.py 10 传参的形式来调整并发量，使用 ctrl+c 可以中止测试。

压力测试脚本如下：

import requests
import threading
import sys
import time

def send_request():   
 url = "http://127.0.1.1:11434/api/generate"  
   data = {      
     "model": "deepseek-r1:32b",  
     "prompt": "解释一下量子计算",        
     "stream": False    
     }    
     while True:        
     try:            
     response = requests.post(url, json=data, timeout=None)            
     print(f"Status Code: {response.status_code}, Time: {response.elapsed.total_seconds()}s")        
     except Exception as e:           
      print(f"Request failed: {e}")

try:    
concurrent_workers = int(sys.argv[1])
    # 启动工作线程    
    for _ in range(concurrent_workers):       
     t = threading.Thread(target=send_request)       
     t.daemon = True        
     t.start()
    # 添加线程数监控    
    while True:        
    # 计算实际工作线程数（总线程数 - 主线程）        
    current_workers = threading.active_count() - 1        
    print(f"\n[监控] 目标线程数: {concurrent_workers} | 实际运行线程数: {current_workers}")        
    time.sleep(3)  # 每3秒输出一次监控信息
except KeyboardInterrupt:    
     print("\n[!] 已终止持续并发请求")

打赏