From 88596c0c63770c06bdda47cfcf5df085a2cb91a6 Mon Sep 17 00:00:00 2001 From: Divano Date: Tue, 5 Aug 2025 10:24:12 +0800 Subject: [PATCH] Add more base chat cases (#3203) * add test base class * fix codestyle * fix codestyle * add base chat --- test/ce/server/core/__init__.py | 35 ++++++++++++--------- test/ce/server/core/utils.py | 25 ++++++++++++--- test/ce/server/test_base_chat.py | 54 ++++++++++++-------------------- 3 files changed, 61 insertions(+), 53 deletions(-) diff --git a/test/ce/server/core/__init__.py b/test/ce/server/core/__init__.py index 55c3d7103..95a56128c 100644 --- a/test/ce/server/core/__init__.py +++ b/test/ce/server/core/__init__.py @@ -9,10 +9,16 @@ from .logger import Logger base_logger = Logger(loggername="FDSentry", save_level="channel", log_path="./fd_logs").get_logger() base_logger.setLevel("INFO") -from .request_template import TEMPLATES -from .utils import build_request_payload, get_stream_chunks, send_request -__all__ = ["build_request_payload", "send_request", "TEMPLATES", "get_stream_chunks"] +from .request_template import TEMPLATES +from .utils import ( + build_request_payload, + get_stream_chunks, + get_token_list, + send_request, +) + +__all__ = ["build_request_payload", "send_request", "TEMPLATES", "get_stream_chunks", "get_token_list"] # 检查环境变量是否存在 URL = os.environ.get("URL") @@ -23,14 +29,15 @@ if not URL: missing_vars.append("URL") if not TEMPLATE: missing_vars.append("TEMPLATE") -if missing_vars: - if not URL: - msg = ( - f"❌ 缺少环境变量：{', '.join(missing_vars)}，请先设置，例如：\n" - f" export URL=http://localhost:8000/v1/chat/completions\n" - f" export TEMPLATE=TOKEN_LOGPROB" - ) - base_logger.error(msg) - sys.exit(33) # 终止程序 - if not TEMPLATE: - base_logger.warning("未启用请求模板，请在Case中自行设置请求模板") + +if not URL: + msg = ( + f"❌ 缺少环境变量：{', '.join(missing_vars)}，请先设置，例如：\n" + f" export URL=http://localhost:8000/v1/chat/completions\n" + f" export TEMPLATE=TOKEN_LOGPROB" + ) + base_logger.error(msg) + sys.exit(33) # 终止程序 + +if not TEMPLATE: + base_logger.warning("⚠️ 未设置 TEMPLATE，请确保在用例中显式传入请求模板。") diff --git a/test/ce/server/core/utils.py b/test/ce/server/core/utils.py index 45ef2b780..c12baf346 100644 --- a/test/ce/server/core/utils.py +++ b/test/ce/server/core/utils.py @@ -37,10 +37,6 @@ def send_request(url, payload, timeout=600, stream=False): Returns: response: 请求的响应结果，如果请求失败则返回None。 - - Raises: - None - """ headers = { "Content-Type": "application/json", @@ -60,7 +56,7 @@ def send_request(url, payload, timeout=600, stream=False): def get_stream_chunks(response): - """解析流式返回，生成chunk List[dict]""" + """解析流式返回，生成 chunk List[dict]""" chunks = [] if response.status_code == 200: @@ -82,3 +78,22 @@ def get_stream_chunks(response): base_logger.error("返回内容：", response.text) return chunks + + +def get_token_list(response): + """解析 response 中的 token 文本列表""" + token_list = [] + + try: + content_logprobs = response["choices"][0]["logprobs"]["content"] + except (KeyError, IndexError, TypeError) as e: + base_logger.error(f"解析失败：{e}") + return [] + + for token_info in content_logprobs: + token = token_info.get("token") + if token is not None: + token_list.append(token) + + base_logger.info(f"Token List:{token_list}") + return token_list diff --git a/test/ce/server/test_base_chat.py b/test/ce/server/test_base_chat.py index ae75b7808..8001ac878 100644 --- a/test/ce/server/test_base_chat.py +++ b/test/ce/server/test_base_chat.py @@ -9,7 +9,7 @@ some basic check for fd web api import json -from core import TEMPLATE, URL, build_request_payload, send_request +from core import TEMPLATE, URL, build_request_payload, get_token_list, send_request def test_stream_response(): @@ -76,9 +76,12 @@ def test_logprobs_enabled(): def test_stop_sequence(): data = { "stream": False, - "stop": ["果冻"], + "stop": ["。"], "messages": [ - {"role": "user", "content": "你要严格按照我接下来的话输出，输出冒号后面的内容，请输出：这是第一段。果冻这是第二段啦啦啦啦啦。"}, + { + "role": "user", + "content": "你要严格按照我接下来的话输出，输出冒号后面的内容，请输出：这是第一段。这是第二段啦啦啦啦啦。", + }, ], "max_tokens": 20, "top_p": 0, @@ -86,8 +89,11 @@ def test_stop_sequence(): payload = build_request_payload(TEMPLATE, data) resp = send_request(URL, payload).json() content = resp["choices"][0]["message"]["content"] + token_list = get_token_list(resp) print("截断输出:", content) assert "第二段" not in content + assert "第二段" not in token_list + assert "。" in token_list, "没有找到。符号" def test_sampling_parameters(): @@ -125,7 +131,7 @@ def test_multi_turn_conversation(): def test_bad_words_filtering(): - banned_tokens = ["和", "呀"] + banned_tokens = ["香"] data = { "stream": False, @@ -140,36 +146,14 @@ def test_bad_words_filtering(): payload = build_request_payload(TEMPLATE, data) response = send_request(URL, payload).json() - content = response["choices"][0]["message"]["content"] print("生成内容:", content) + token_list = get_token_list(response) for word in banned_tokens: - assert word not in content, f"bad_word '{word}' 不应出现在生成结果中" + assert word not in token_list, f"bad_word '{word}' 不应出现在生成结果中" - print("test_bad_words_filtering 通过：生成结果未包含被禁词") - - data = { - "stream": False, - "messages": [ - {"role": "system", "content": "你是一个助手，回答简洁清楚"}, - {"role": "user", "content": "请输出冒号后面的字，一模一样: 我爱吃果冻，苹果，香蕉，和荔枝呀呀呀"}, - ], - "top_p": 0, - "max_tokens": 69, - # "bad_words": banned_tokens, - } - - payload = build_request_payload(TEMPLATE, data) - response = send_request(URL, payload).json() - - content = response["choices"][0]["message"]["content"] - print("生成内容:", content) - - for word in banned_tokens: - assert word not in content, f"bad_word '{word}' 不应出现在生成结果中" - - print("test_bad_words_filtering 通过：生成结果未包含被禁词") + print("test_bad_words_filtering 正例验证通过") def test_bad_words_filtering1(): @@ -195,8 +179,10 @@ def test_bad_words_filtering1(): for word in banned_tokens: assert word not in content, f"bad_word '{word}' 不应出现在生成结果中" - print("test_bad_words_filtering 通过：生成结果未包含被禁词") - word = "呀呀" + print("test_bad_words_filtering1 通过：生成结果未包含被禁词") + + # 正例验证 + word = "呀" data = { "stream": False, "messages": [ @@ -212,7 +198,7 @@ def test_bad_words_filtering1(): content = response["choices"][0]["message"]["content"] print("生成内容:", content) + token_list = get_token_list(response) + assert word in token_list, f"'{word}' 应出现在生成结果中" - assert word in content, f" '{word}' 应出现在生成结果中" - - print("test_bad_words_filtering 通过：生成结果未包含被禁词") + print("test_bad_words_filtering1 正例验证通过")