测试流式响应关闭 logprobs 后,确认响应中不包含 logprobs 字段。
()
| 111 | |
| 112 | |
| 113 | def test_stream_without_logprobs(): |
| 114 | """ |
| 115 | 测试流式响应关闭 logprobs 后,确认响应中不包含 logprobs 字段。 |
| 116 | """ |
| 117 | data = { |
| 118 | "stream": True, |
| 119 | "logprobs": False, |
| 120 | "top_logprobs": None, |
| 121 | "messages": [ |
| 122 | {"role": "system", "content": "You are a helpful assistant."}, |
| 123 | {"role": "user", "content": "牛顿的三大运动定律是什么?"}, |
| 124 | ], |
| 125 | "max_tokens": 3, |
| 126 | } |
| 127 | |
| 128 | payload = build_request_payload(TEMPLATE, data) |
| 129 | response = send_request(URL, payload) |
| 130 | |
| 131 | # 解析首个包含 content 的流式 chunk |
| 132 | result_chunk = {} |
| 133 | for line in response.iter_lines(): |
| 134 | if not line: |
| 135 | continue |
| 136 | decoded = line.decode("utf-8").removeprefix("data: ") |
| 137 | if decoded == "[DONE]": |
| 138 | break |
| 139 | |
| 140 | chunk = json.loads(decoded) |
| 141 | content = chunk["choices"][0]["delta"].get("content") |
| 142 | if content: |
| 143 | result_chunk = chunk |
| 144 | print(json.dumps(result_chunk, indent=2, ensure_ascii=False)) |
| 145 | break |
| 146 | |
| 147 | # 校验 logprobs 字段不存在 |
| 148 | assert result_chunk["choices"][0]["delta"]["content"] == "牛顿" |
| 149 | assert result_chunk["choices"][0]["logprobs"] is None |
| 150 | |
| 151 | |
| 152 | def test_stream_with_temp_scaled_logprobs(): |
no test coverage detected