测试流式响应开启 logprobs 后，首个 token 的概率信息是否正确。
()
| 68 | |
| 69 | |
| 70 | def test_stream_with_logprobs(): |
| 71 | """ |
| 72 | 测试流式响应开启 logprobs 后，首个 token 的概率信息是否正确。 |
| 73 | """ |
| 74 | data = { |
| 75 | "stream": True, |
| 76 | "messages": [ |
| 77 | {"role": "system", "content": "You are a helpful assistant."}, |
| 78 | {"role": "user", "content": "牛顿的三大运动定律是什么？"}, |
| 79 | ], |
| 80 | "max_tokens": 3, |
| 81 | } |
| 82 | |
| 83 | payload = build_request_payload(TEMPLATE, data) |
| 84 | response = send_request(URL, payload) |
| 85 | |
| 86 | # 解析首个包含 content 的流式 chunk |
| 87 | result_chunk = {} |
| 88 | for line in response.iter_lines(): |
| 89 | if not line: |
| 90 | continue |
| 91 | decoded = line.decode("utf-8").removeprefix("data: ") |
| 92 | if decoded == "[DONE]": |
| 93 | break |
| 94 | |
| 95 | chunk = json.loads(decoded) |
| 96 | content = chunk["choices"][0]["delta"].get("content") |
| 97 | if content: |
| 98 | result_chunk = chunk |
| 99 | print(json.dumps(result_chunk, indent=2, ensure_ascii=False)) |
| 100 | break |
| 101 | |
| 102 | # 校验概率字段 |
| 103 | assert result_chunk["choices"][0]["delta"]["content"] == "牛顿" |
| 104 | assert result_chunk["choices"][0]["logprobs"]["content"][0]["token"] == "牛顿" |
| 105 | assert result_chunk["choices"][0]["logprobs"]["content"][0]["logprob"] == -0.031025361269712448 |
| 106 | assert result_chunk["choices"][0]["logprobs"]["content"][0]["top_logprobs"][0] == { |
| 107 | "token": "牛顿", |
| 108 | "logprob": -0.031025361269712448, |
| 109 | "bytes": [231, 137, 155, 233, 161, 191], |
| 110 | } |
| 111 | |
| 112 | |
| 113 | def test_stream_without_logprobs(): |
no test coverage detected