import requests
import threading
import time
from queue import Queue
# 配置接口信息
url = "http://localhost:8000/v1/chat/completions"
headers = {
"Content-Type": "application/json"
}
data = {
"model": "/saves_model/qwen_32b-5.9",
"messages": [{"role": "user", "content": "六零空间有什么业务?"}]
# 用于存储结果的队列
result_queue = Queue()
# 定义请求函数
def make_request():
start_time = time.time()
try:
response = requests.post(url, headers=headers, json=data)
end_time = time.time()
if response.status_code == 200:
answer = response.json()["choices"][0]["message"]["content"]
tokens = response.json()["usage"]["completion_tokens"]
result_queue.put((end_time - start_time, tokens))
else:
print(f"请求失败，状态码: {response.status_code}")
except Exception as e:
print(f"请求发生错误: {e}")
# 定义测试函数
def test_concurrency(num_threads):
threads = []
for _ in range(num_threads):
thread = threading.Thread(target=make_request)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
# 统计结果
total_time = 0
total_tokens = 0
while not result_queue.empty():
time_taken, tokens = result_queue.get()
total_time += time_taken
total_tokens += tokens
# 计算延迟和吞吐量
avg_latency = total_time / num_threads if num_threads > 0 else 0
throughput = total_tokens / total_time if total_time > 0 else 0
print(f"并发数: {num_threads}")
print(f"平均延迟: {avg_latency:.2f} 秒")
print(f"吞吐量: {throughput:.2f} tokens/秒")
if __name__ == "__main__":
# 修改这个值来测试不同的并发数
num_threads = 75
test_concurrency(num_threads)
暂无数据 暂无数据 暂无数据 暂无数据
import requests
import threading
import time
from queue import Queue
# 配置接口信息
url = "http://localhost:8000/v1/chat/completions"
headers = {
"Content-Type": "application/json"
}
data = {
"model": "/saves_model/qwen_32b-5.9",
"messages": [{"role": "user", "content": "六零空间有什么业务?"}]
}
# 用于存储结果的队列
result_queue = Queue()
# 定义请求函数
def make_request():
start_time = time.time()
try:
response = requests.post(url, headers=headers, json=data)
end_time = time.time()
if response.status_code == 200:
answer = response.json()["choices"][0]["message"]["content"]
tokens = response.json()["usage"]["completion_tokens"]
result_queue.put((end_time - start_time, tokens))
else:
print(f"请求失败，状态码: {response.status_code}")
except Exception as e:
print(f"请求发生错误: {e}")
# 定义测试函数
def test_concurrency(num_threads):
threads = []
for _ in range(num_threads):
thread = threading.Thread(target=make_request)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
# 统计结果
total_time = 0
total_tokens = 0
while not result_queue.empty():
time_taken, tokens = result_queue.get()
total_time += time_taken
total_tokens += tokens
# 计算延迟和吞吐量
avg_latency = total_time / num_threads if num_threads > 0 else 0
throughput = total_tokens / total_time if total_time > 0 else 0
print(f"并发数: {num_threads}")
print(f"平均延迟: {avg_latency:.2f} 秒")
print(f"吞吐量: {throughput:.2f} tokens/秒")
if __name__ == "__main__":
# 修改这个值来测试不同的并发数
num_threads = 75
test_concurrency(num_threads)