MCPcopy Create free account
hub / github.com/ZBayes/basic_rag / batch_test

Function batch_test

src/server/client.py:34–48  ·  view source on GitHub ↗
(query_list, process_num, request_num)

Source from the content-addressed store, hash-verified

32
33
34def batch_test(query_list, process_num, request_num):
35 # query_list:待请求query列表,process_num进程个数,request_num请求个数(每个进程)
36 pool = Pool(processes=process_num)
37 process_result = []
38 for i in range(process_num):
39 process_result.append(pool.apply_async(single_test, args=(query_list, request_num, str(i), )))
40 # processes.append(Process(target=single_test, args=(query_list, request_num, str(i), )))
41
42 pool.close()
43 pool.join()
44
45 time_list = []
46 for result in process_result:
47 time_list.extend(result.get())
48 return time_list
49
50
51# response = requests.post("http://127.0.0.1:9090/a", json.dumps({"query": "你好啊1"}))

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected