Files
gangyan/langchain-chat/tests/test_Qwen72B_32B.py

80 lines
10 KiB
Python
Raw Normal View History

import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# 模型配置
model_config = {
"Qwen2-72B-Instruct": {
"model_name": "Qwen2-72B-Instruct",
"api_base_url": "http://192.168.56.123:8821/v1",
"api_key": "fake",
},
"Qwen1.5-32B-Chat": {
"model_name": "Qwen1.5-32B-Chat",
"api_base_url": "http://192.168.56.123:8822/v1",
"api_key": "fake",
}
}
# 选择要测试的模型
selected_model = "Qwen2-72B-Instruct"
API_URL = model_config[selected_model]["api_base_url"] + "/completions"
API_KEY = model_config[selected_model]["api_key"]
MODEL_NAME = model_config[selected_model]["model_name"]
# 测试输入
TEST_PROMPT_short = "现在开始成语接龙:"
TEST_PROMPT_long = "总结这段内容:各州、市人民政府省直各委、办、厅、局为贯彻落实《国务院关于印发“十二五”期间深化医药卫生体制改革规划暨实施方案的通知》国发201211号、《国务院办公厅印发关于县级公立医院综合改革试点意见的通知》国办发201233号等文件要求加快推进我省县级公立医院改革试点工作结合我省实际经省人民政府同意现提出如下实施意见一、指导思想以科学发展观为指导按照保基本、强基层、建机制的要求遵循上下联动、内增活力、外加推力的原则围绕政事分开、管办分开、医药分开、营利性和非营利性分开的改革要求明确县级公立医院的功能定位改革县级公立医院补偿机制和运行机制完善医疗服务体系健全行业监管机制努力提高县级医疗机构的技术水平和服务能力切实缓解“看病难、看病贵”问题让广大人民群众看病就医更方便、更安全。二、基本原则——政府主导惠民利民。强化政府对县级公立医院在规划、政策、投入和监管等方面的责任维护县级公立医院公益性质努力减轻人民群众医疗费用负担实现人人享有基本医疗卫生服务的目标。——创新机制综合推进。建立县级公立医院运行新机制。通过调整医疗服务收费、增加财政对县级公立医院改革的投入、改革医保支付方式、合理降低医药费用等综合措施确保改革有力推进。——积极探索稳步发展。各地结合自身实际制定县级公立医院综合改革试点方案探索改革的有效形式和办法。切实做到人民群众得实惠医务人员受鼓舞公立医院得发展。三、改革目标以破除“以药补医”机制为关键环节以改革补偿机制和落实医院自主经营管理权为切入点统筹推进县级公立医院管理体制、补偿机制、人事分配、价格机制、医保支付制度、采购机制、监管机制等综合改革建立起维护公益性、调动积极性、保障可持续的县级公立医院运行机制。坚持以改革促发展加强以人才、技术、临床重点专科为核心的县级公立医院能力建设统筹县域医疗卫生体系发展实现“小病不出乡镇大病不出县城预防在基层”90%左右的患者在县域内就诊的改革目标。四、实施步骤2012年10月1日起在16个州的30个县市、区实施县级公立医院综合改革试点。力争2013年总结评估形成基本路子为2015年实现县级公立医院阶段性改革目标打好基础。五、改革任务制定县级公立医院发展规划。县市、区人民政府要针对县域群众主要健康问题根据人口数量和分布、地理交通等因素制定县域卫生规划和医疗机构设置规划合理确定县域内医院的数量、布局、功能、规模和标准。在每个县市、区重点办好1—2所县级公立医院含中医医院。按照“填平补齐”原则完成县级医院标准化建设30万以上人口的县市、区至少有1所县级综合公立医院达到二级甲等水平。以县级公立医院为中心完善县域急救服务体系建立县域院前急救体系。严格控制县级公立医院建设规模和大型设备配置。鼓励有条件的地区探索对医疗资源进行整合、重组和改制优化资源配置。鼓励资源集约化探索成立检查检验中心推行检查检验结果医疗机构互认以及后勤服务外包等。落实支持和引导社会资本办医政策。明确县级公立医院功能定位。明确县级公立医院是县域内的医疗卫生中心和农村三级医疗卫生服务网络的龙头并与城市大医院分工协作。主要为县域居民提供基本医疗服务包括运用适宜医疗技术和药物
def make_request(api_url, api_key, prompt, model_name, max_tokens):
headers = {"Authorization": f"Bearer {api_key}"}
data = {
"model": model_name,
"prompt": prompt,
"max_tokens": max_tokens
}
start_time = time.time()
response = requests.post(api_url, headers=headers, json=data)
end_time = time.time()
response_time = end_time - start_time
return response.status_code, response_time
def test_average_response_time(api_url, api_key, prompt, model_name, max_tokens, num_requests):
response_times = []
with ThreadPoolExecutor(max_workers=num_requests) as executor:
futures = [executor.submit(make_request, api_url, api_key, prompt, model_name, max_tokens) for _ in range(num_requests)]
for future in as_completed(futures):
status_code, response_time = future.result()
if status_code == 200:
response_times.append(response_time)
average_response_time = sum(response_times) / len(response_times) if response_times else 0
return average_response_time, response_times
def print_csv(headers, rows):
header_line = ",".join(headers)
print(header_line)
for row in rows:
row_line = ",".join(map(str, row))
print(row_line)
# 主程序
if __name__ == "__main__":
max_tokens_list = [512, 1024, 2048, 4096]
num_requests = 100 # 要发起的请求数量
for max_tokens in max_tokens_list:
# 测试短提示
average_response_time_short, response_times_short = test_average_response_time(API_URL, API_KEY, TEST_PROMPT_short, MODEL_NAME, max_tokens, num_requests)
headers = ["请求数量", "max_tokens", "平均响应时间 (秒)"]
rows_short = [[num_requests, max_tokens, f"{average_response_time_short:.2f}"]]
print(f"\n100个请求的平均响应时间测试结果 (短提示, max_tokens={max_tokens}):")
print_csv(headers, rows_short)
# 测试长提示
average_response_time_long, response_times_long = test_average_response_time(API_URL, API_KEY, TEST_PROMPT_long, MODEL_NAME, max_tokens, num_requests)
rows_long = [[num_requests, max_tokens, f"{average_response_time_long:.2f}"]]
print(f"\n100个请求的平均响应时间测试结果 (长提示, max_tokens={max_tokens}):")
print_csv(headers, rows_long)