测试completions 流式 usage
()
| 106 | |
| 107 | |
| 108 | def test_non_chat_usage_stream(): |
| 109 | """测试completions 流式 usage""" |
| 110 | data = { |
| 111 | "prompt": "牛顿的三大运动定律是什么?", |
| 112 | "max_tokens": 50, |
| 113 | "stream": True, |
| 114 | "stream_options": {"include_usage": True, "continuous_usage_stats": True}, |
| 115 | "metadata": {"min_tokens": 10}, |
| 116 | } |
| 117 | completion_url = URL.replace("chat/completions", "completions") |
| 118 | |
| 119 | payload = build_request_payload(TEMPLATE, data) |
| 120 | |
| 121 | response = send_request(url=completion_url, payload=payload, stream=True) |
| 122 | chunks = get_stream_chunks(response) |
| 123 | # for idx, chunk in enumerate(chunks): |
| 124 | # print(f"\nchunk[{idx}]:\n{json.dumps(chunk, indent=2, ensure_ascii=False)}") |
| 125 | |
| 126 | usage = chunks[-1]["usage"] |
| 127 | total_tokens = usage["completion_tokens"] + usage["prompt_tokens"] |
| 128 | assert data["max_tokens"] >= usage["completion_tokens"], f"completion_tokens大于max_tokens, usage: {usage}" |
| 129 | assert ( |
| 130 | data["metadata"]["min_tokens"] <= usage["completion_tokens"] |
| 131 | ), f"completion_tokens小于min_tokens, usage: {usage}" |
| 132 | assert ( |
| 133 | usage["total_tokens"] == total_tokens |
| 134 | ), f"total_tokens不等于prompt_tokens + completion_tokens, usage: {usage}" |
| 135 | |
| 136 | |
| 137 | def test_non_chat_usage_non_stream(): |
nothing calls this directly
no test coverage detected