测试payload seed参数
()
| 9 | |
| 10 | |
| 11 | def test_seed_stream(): |
| 12 | """测试payload seed参数""" |
| 13 | data = { |
| 14 | "messages": [ |
| 15 | {"role": "system", "content": "You are a helpful assistant."}, |
| 16 | {"role": "user", "content": "牛顿的三大运动定律是什么？"}, |
| 17 | ], |
| 18 | "seed": 26, |
| 19 | "max_tokens": 50, |
| 20 | "stream": True, |
| 21 | "stream_options": {"include_usage": True, "continuous_usage_stats": True}, |
| 22 | } |
| 23 | |
| 24 | payload = build_request_payload(TEMPLATE, data) |
| 25 | response_1 = send_request(url=URL, payload=payload, stream=True) |
| 26 | # print(response_1.text) |
| 27 | chunks_1 = get_stream_chunks(response_1) |
| 28 | # print(chunks_1) |
| 29 | # for idx, chunk in enumerate(chunks_1): |
| 30 | # print(f"\nchunk[{idx}]:\n{json.dumps(chunk, indent=2, ensure_ascii=False)}") |
| 31 | resul_1 = "".join([x["choices"][0]["delta"]["content"] for x in chunks_1[:-1]]) |
| 32 | logprobs_1 = [json.dumps(x["choices"][0]["logprobs"]["content"][0], ensure_ascii=False) for x in chunks_1[1:-1]] |
| 33 | # print(resul_1) |
| 34 | # print(logprobs_1, type(logprobs_1[0])) |
| 35 | |
| 36 | response_2 = send_request(url=URL, payload=payload, stream=True) |
| 37 | chunks_2 = get_stream_chunks(response_2) |
| 38 | resul_2 = "".join([x["choices"][0]["delta"]["content"] for x in chunks_2[:-1]]) |
| 39 | logprobs_2 = [json.dumps(x["choices"][0]["logprobs"]["content"][0], ensure_ascii=False) for x in chunks_2[1:-1]] |
| 40 | # print(resul_2) |
| 41 | |
| 42 | assert resul_1 == resul_2, "top_p=0, 固定seed, 两次请求结果不一致" |
| 43 | for idx, (l1, l2) in enumerate(zip(logprobs_1, logprobs_2)): |
| 44 | assert l1 == l2, f"top_p=0, 固定seed, logprobs[{idx}]不一致" |
| 45 | |
| 46 | |
| 47 | def test_chat_usage_stream(): |
no test coverage detected