diff --git a/openseek/competition/LongContext-ICL-Annotation/READMD_cn.md b/openseek/competition/LongContext-ICL-Annotation/READMD_cn.md deleted file mode 100644 index ffee7ca0..00000000 --- a/openseek/competition/LongContext-ICL-Annotation/READMD_cn.md +++ /dev/null @@ -1,71 +0,0 @@ -# 超长长上下文场景中LLM自动数据标注挑战赛 - ---- - -## 消息 - -- **[2026-01-20] `发布`：** 赛事信息已在 **Kaggle** 正式上线。详情见：[FlagOS Open Computing Global Challenge](https://www.kaggle.com/competitions/flag-os-open-computing-global-challenge). -- **[2026-01-06] `发布`：** 由 **众智 FlagOS 社区**、**北京智源人工智能研究院（BAAI）** 与 **CCF ODTC** 联合主办的综合性大赛 **FlagOS 开放计算全球挑战赛** 正式发布。详情见： - [FlagOS开放计算全球挑战赛- AI赛事通 | 数据算法赛](https://www.competehub.dev/zh/competitions/modelscope180) - - ---- - - -## 快速开始 -### 1. 环境 - -```bash -openai -torch -flagScale -``` - -### 2. 下载模型权重 -```bash -hf download Qwen/Qwen3-4B --local-dir Qwen3-4B -# or -modelscope download --model Qwen/Qwen3-4B -``` -### 3. 长文本配置 -在`Qwen3-4B/config.json`将原有配置替换为： -```json -"rope_scaling": { - "rope_type": "yarn", - "factor": 4.0, - "original_max_position_embeddings": 32768 -} -``` -### 4. 模型部署 - -请根据实际需求，配置 `llm_config.yaml` 文件。启动配置 - -```bash -cd FlagScale -python run.py --config-path .. --config-name llm_config action=run -``` - -在模型服务启动后，可通过以下方式测试本地 API： - -```bash -python api_test.py -``` - -如需停止服务，请执行： - -```bash -python run.py --config-path .. --config-name llm_config action=stop -``` - -### 5. 运行/改进基线方法（Baseline） - -启动如下命令开始模型标注 -```bash -python main.py -``` - -实现新的标注方法，请修改`method.py`文件。你可以在该文件中： -* 定义新的指令模板、 -* 定义新的上下文示例选择策略 -* 定义新的模型推理、标注方案 -* 添加自定义后处理逻辑 diff --git a/openseek/competition/LongContext-ICL-Annotation/README.md b/openseek/competition/LongContext-ICL-Annotation/README.md index c52275b6..cc037199 100644 --- a/openseek/competition/LongContext-ICL-Annotation/README.md +++ b/openseek/competition/LongContext-ICL-Annotation/README.md @@ -26,7 +26,7 @@ flagScale ### 2. Download Model Weights ```bash -hf download Qwen/Qwen3-4B --local-dir Qwen3-4B +hf download Qwen/Qwen3-4B --local-dir ~/.cache/modelscope/hub/models/Qwen/Qwen3-4B # or modelscope download --model Qwen/Qwen3-4B ``` @@ -48,6 +48,7 @@ In `Qwen3-4B/config.json`, replace the original configuration with the following Configure the `llm_config.yaml` file according to your actual requirements. Then start the service with: ```bash +git clone https://gitee.com/flagos-ai/FlagScale cd FlagScale python run.py --config-path .. --config-name llm_config action=run ``` @@ -69,7 +70,8 @@ python run.py --config-path .. --config-name llm_config action=stop Start the baseline annotation pipeline with: ```bash -python main.py +cd src +python main.py --task_id=0 --task_step=0 ``` To implement a new annotation method, modify the `method.py` file. Within this file, you may: @@ -78,3 +80,18 @@ To implement a new annotation method, modify the `method.py` file. Within this f - Design new context example selection strategies - Implement alternative model inference and annotation pipelines - Add custom post-processing logic + + +The table below is the release compatibility matrix for vLLM Ascend release. +```csv +vLLM Ascend vLLM Python CANN PyTorch/torch_npu Triton Ascend Recommend +v0.14.0rc1 v0.14.1 >= 3.10, < 3.12 8.5.0 2.9.0 / 2.9.0 3.2.0 Y +v0.13.0 v0.13.0 >= 3.10, < 3.12 8.5.0 2.9.0 / 2.8.0.post2 3.2.0 N +v0.13.0rc2 v0.13.0 >= 3.10, < 3.12 8.5.0 2.8.0 / 2.8.0.post1 3.2.0 N +v0.13.0rc1 v0.13.0 >= 3.10, < 3.12 8.3.RC2 2.8.0 / 2.8.0 3.2.0 Y +v0.12.0rc1 v0.12.0 >= 3.10, < 3.12 8.3.RC2 2.8.0 / 2.8.0 3.2.0 Y +v0.11.0 v0.11.0 >= 3.9 , < 3.12 8.3.RC2 2.7.1 / 2.7.1.post1 3.2.0 N +v0.11.0rc3 v0.11.0 >= 3.9, < 3.12 8.3.RC2 2.7.1 / 2.7.1.post1 3.2.0 N +v0.11.0rc2 v0.11.0 >= 3.9, < 3.12 8.3.RC2 2.7.1 / 2.7.1 3.2.0 Y +v0.11.0rc1 v0.11.0 >= 3.9, < 3.12 8.3.RC1 2.7.1 / 2.7.1 3.2.0 Y +``` diff --git a/openseek/competition/LongContext-ICL-Annotation/README_cn.md b/openseek/competition/LongContext-ICL-Annotation/README_cn.md new file mode 100644 index 00000000..b28d40d9 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/README_cn.md @@ -0,0 +1,167 @@ +# 超长长上下文场景中LLM自动数据标注挑战赛 + +--- + +## 消息 + +- **[2026-01-20] `发布`：** 赛事信息已在 **Kaggle** 正式上线。详情见：[FlagOS Open Computing Global Challenge](https://www.kaggle.com/competitions/flag-os-open-computing-global-challenge). +- **[2026-01-06] `发布`：** 由 **众智 FlagOS 社区**、**北京智源人工智能研究院（BAAI）** 与 **CCF ODTC** 联合主办的综合性大赛 **FlagOS 开放计算全球挑战赛** 正式发布。详情见： + [FlagOS开放计算全球挑战赛- AI赛事通 | 数据算法赛](https://www.competehub.dev/zh/competitions/modelscope180) + + +--- + + +## 快速开始 +### 1. 环境 + +```bash +openai +torch +flagScale +``` + +### 2. 下载模型权重 +```bash +hf download Qwen/Qwen3-4B --local-dir /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B +# or +modelscope download --model Qwen/Qwen3-4B +``` + +### 3. 长文本配置 +在`Qwen3-4B/config.json`将原有配置替换为： +```json +"rope_scaling": { + "rope_type": "yarn", + "factor": 4.0, + "original_max_position_embeddings": 32768 +} +``` + +### 4. 模型部署 +安装环境和下载模型 +```bash +pip install -r requirements.txt +bash create_env_ascend.sh +``` + +请根据实际需求，配置 `llm_config.yaml` 文件，启动服务 +```bash +bash start-qwen3_4_serve.sh +``` + +在模型服务启动后，可通过以下方式测试本地 API： +```bash +python api_test.py +``` + +如需停止服务，请执行： +```bash +bash stop-qwen3_4_serve.sh +``` + +### 5. 运行/改进基线方法（Baseline） +如需开始模型标注，启动如下命令 +```bash +bash start-all-tasks_on_4_serve.sh +``` + +如需结束模型标注，启动如下命令 +```bash +bash stop-all-tasks.sh +``` + +### 6. 安装环境补充说明 +```sh +# 系统：比赛官方提供的云服务器 (为了加速实验 - 建议用更多的卡数，推荐用8个) +# GPU：Ascend 910C x 2 卡 +# CPU：20核 +# 磁盘：200G +# 内存：80G +# NPU：CANN8.3.RC2 + +# 检查基础环境： +cat /usr/local/Ascend/ascend-toolkit/latest/version.cfg +cat /usr/local/Ascend/ascend-toolkit/set_env.sh + +# 6.1、安装环境（flagscale所需CANN、vllm、vllm-ascend版本依赖请按实际环境调整） +DIR=your-project-path/OpenSeek/openseek/competition/LongContext-ICL-Annotation/src +cd $DIR +pip install -r $DIR/requirements.txt +bash create_env_ascend.sh + +# 6.2、启动服务（启动服务后需要耐心等一小段时间，让模型服务完全就绪，可用npu-smi或nputop查看状态） +DIR=your-project-path/OpenSeek/openseek/competition/LongContext-ICL-Annotation/src +cd $DIR +# 单机多卡启动 ，若超过2张NPU，请调整llm_config.yaml的参数：tensor_parallel_size数值与ASCEND_RT_VISIBLE_DEVICES数值 +VLLM_PLUGINS=fl flagscale server qwen3 --config $DIR/llm_config.yaml +# 或者 +bash start-qwen3-server.sh +``` + +### 7. 执行命令补充说明 +#### 7.1) 参数 +```bash +--task_id=0， 遍历处理所有任务 +--task_id=$id，(one of [1,2,3,4,5,6,7,8])，则只处理对应的一个任务 +--task_step=1，则利用ICL样本提取关键操作Operation +--task_step=2，则利用Operation获取预测结果Result +--task_step=0，则首先利用ICL提取Operation，然后利用Operation获取Result +``` + +#### 7.2) 入口文件 +```bash +main.py 总是每次同步提交一个批次，便于前台调试查看过程, + 例如: + DEBUG_PRINT_STREAM=1 python main.py --task_id=5 --task_step=1 + DEBUG_PRINT_STREAM=2 python main.py --task_id=5 --task_step=1 +main_batch.py 可以每次异步并发提交多个批次，用于后台命令完成任务，与服务端张量和数据并行计算协同，加速实验进度 + 例如: (不要设置DEBUG_PRINT_STREAM=1，因为多个异步并发请求，实时输出是混乱无序的) + REQ_CONCURRENCY_NUM=8 DEBUG_PRINT_STREAM=0 python main_batch.py --task_id=5 --task_step=1 + DEBUG_PRINT_STREAM=2 python main_batch.py --task_id=5 --task_step=1 +``` + +#### 7.3) 如果你采用了更优的自定义配置，只需设置以下环境变量（可按实际修改默认值）： +```bash +export SERVE_API_KEY="" +export SERVE_BASE_URL="http://localhost:2026/v1" +export SERVE_MODEL_NAME="Qwen/Qwen3-4B" +export SERVE_TOKENIZER_PATH=~/.cache/modelscope/hub/models/Qwen/Qwen3-4B +export REQ_CONCURRENCY_NUM=2 +export ICL_REPEAT_N=1 +# +# REQ_CONCURRENCY_NUM 默认2，客户端API并发请求数，如果卡很多，可以增加该数值，与服务端张量和数据并行计算协同，加快实验进度 +# ICL_REPEAT_N 默认1，在提取Opration时对ICL的重复次数，如果卡很多，可以增加该数值，提高Opration的完整性 +# +``` + +#### 7.4) 执行命令分析解读：需要考虑这8个任务数据kv缓存的相互干扰因素 +DIR=your-project-path/OpenSeek/openseek/competition/LongContext-ICL-Annotation/src +##### 7.4.1）最佳执行命令（推荐） +在卡少时可以用a或b方式，本人`单机2卡部署成4个服务(每卡2个服务)`使用c方式 (为了在并发时让不同任务KV缓存隔离，未使用max_num_seqs>1 tensor_parallel_size>1的方案) +```bash +# (a方式) 1个服务，逐个任务串行执行，8个任务的kv缓存相互干扰较小 +cd $DIR && bash start-qwen3_1_serve.sh +cd $DIR && bash start-all-tasks_on_1_serve.sh +# 注：也可以在每处理一个后，重启服务清除缓存排除干扰，但是有点麻烦与耗时。若有人感兴趣，可以尝试 + +# (b方式) 2个服务，两阶段分步执行（干扰比a少，速度比a快） +cd $DIR && bash start-qwen3_2_serve.sh + +# 首先在第一个服务：执行第一阶段 +cd $DIR && bash start-all-tasks_on_2_serve-step_1.sh +# 然后在第二个服务：启动自动循环监视，在发现第一步中出现第一个文件生成时，就自动开始第二阶段, 在第一阶段完成后，自动接管其服务用于处理第二阶段的其他任务 +cd $DIR && nohup bash start-all-tasks_on_2_serve-step_2.sh > s2-wait-s1.log 2>&1 & + +# (c方式) 4个服务，8个任务KV缓存在不同服务，多任务并行执行，干扰很小速度较快 +cd $DIR && bash start-qwen3_4_serve.sh +cd $DIR && bash start-all-tasks_on_4_serve.sh + +# (d方式) 8个服务，多个任务并行执行，因为8个任务的kv缓存在8个不同服务，没有相互干扰（最佳之最推荐） +# 仅作参考 +cd $DIR && bash start-all-tasks_on_8_serves.sh +``` + +##### 7.4.2) 待处理问题：证明KV缓存干扰是否明显影响稳定性与准确率） +在单个服务多个任务并行执行时，其kv缓存相互干扰因素，是否会严重影响数据结果的稳定性和准确率？ +若影响越小，则标注方案部署越容易。本人没有时间证明这个问题，若有人感兴趣，可以尝试证明一下。 diff --git a/openseek/competition/LongContext-ICL-Annotation/outputs/result.zip b/openseek/competition/LongContext-ICL-Annotation/outputs/result.zip index 9c5de265..175966f0 100644 Binary files a/openseek/competition/LongContext-ICL-Annotation/outputs/result.zip and b/openseek/competition/LongContext-ICL-Annotation/outputs/result.zip differ diff --git a/openseek/competition/LongContext-ICL-Annotation/requirements.txt b/openseek/competition/LongContext-ICL-Annotation/requirements.txt new file mode 100644 index 00000000..9e5d714b --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/requirements.txt @@ -0,0 +1,11 @@ +requests +asyncio +aiohttp +openai +argparse +tqdm +transformers +scikit-build-core>=0.11 +pybind11 +ninja +cmake diff --git a/openseek/competition/LongContext-ICL-Annotation/src/api_client.py b/openseek/competition/LongContext-ICL-Annotation/src/api_client.py new file mode 100644 index 00000000..943d3d20 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/api_client.py @@ -0,0 +1,187 @@ +import os +import re +import json +import time +import requests +import asyncio +import aiohttp + +from openai import OpenAI, AsyncOpenAI, APIConnectionError, AuthenticationError, APIError, BadRequestError + +from const import CONST_DEBUG_SHOW_STREAM, CONST_DEBUG_SHOW_ANSWER + + +serve_api_key = os.environ.get('SERVE_API_KEY', '') +serve_base_url = os.environ.get('SERVE_BASE_URL', 'http://localhost:2026/v1') +serve_model_name = os.environ.get('SERVE_MODEL_NAME', 'Qwen/Qwen3-4B') +try: + debug_print_stream = int(os.environ.get('DEBUG_PRINT_STREAM', 0)) +except ValueError: + debug_print_stream = 0 + + +# 文件目录: {your-project-path}/OpenSeek/openseek/competition/LongContext-ICL-Annotation +LongContext_ICL_Annotation_DIR = os.path.dirname(os.path.dirname(__file__)) + +llm_visit_history_filepath = os.path.join(LongContext_ICL_Annotation_DIR, 'llm_visit_history.log') + +# 记录ICL提取执行进度 +def log_llm_visit_history(*args): + with open(llm_visit_history_filepath, 'a') as f: + f.write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' :: ' + json.dumps(args)+'\n') + + +# 使用CHAT模版请求模型服务端 +class ChatClient: + + __client = AsyncOpenAI(base_url = serve_base_url.rstrip('/'), api_key = serve_api_key, timeout = 3600) + + # 多个并发请求模型服务（流式） + @staticmethod + def batch_request_llm_api_stream(prompts: list, max_tokens=10240, temperature=0.6, enable_thinking=True): + answers = [] + texts = asyncio.run(ChatClient.__batch_request_chat_api_stream(prompts, max_tokens, temperature, enable_thinking)) + log_llm_visit_history('batch_request_llm_api_stream-p1', '__batch_request_chat_api_stream', 'state', [item[1] for item in texts], 'max_tokens', max_tokens) + for i, (text, state) in enumerate(texts): + if debug_print_stream == str(CONST_DEBUG_SHOW_ANSWER) or debug_print_stream == CONST_DEBUG_SHOW_ANSWER: + print('text, state = ', text, state) + + # 若格式不符合要求，则多请求几次尽量提高成功率 + if state == False: + time.sleep(3) + text, state = asyncio.run(ChatClient.__request_chat_api_stream(prompts[i], max_tokens, temperature, enable_thinking, 256)) + log_llm_visit_history('batch_request_llm_api_stream-p2', '__request_chat_api_stream', 'state', state, 'max_tokens', max_tokens) + if debug_print_stream == str(CONST_DEBUG_SHOW_ANSWER) or debug_print_stream == CONST_DEBUG_SHOW_ANSWER: + print('text, state = ', text, state) + + answers.append((text, state)) + + return answers + + + # 单个请求模型服务（流式） + @staticmethod + def request_llm_api_stream(prompt: str, max_tokens=10240, temperature=0.7, enable_thinking=True): + text, state = asyncio.run(ChatClient.__request_chat_api_stream(prompt, max_tokens, temperature, enable_thinking)) + log_llm_visit_history('request_llm_api_stream-p1', '__request_chat_api_stream', 'state', state, 'max_tokens', max_tokens) + if debug_print_stream == str(CONST_DEBUG_SHOW_ANSWER) or debug_print_stream == CONST_DEBUG_SHOW_ANSWER: + print('text, state = ', text, state) + + # 若格式不符合要求，则多请求一次尽量提高成功率 + if state == False: + time.sleep(3) + text, state = asyncio.run(ChatClient.__request_chat_api_stream(prompt, max_tokens, temperature, enable_thinking, 256)) + log_llm_visit_history('request_llm_api_stream-p2', '__request_chat_api_stream', 'state', state, 'max_tokens', max_tokens) + if debug_print_stream == str(CONST_DEBUG_SHOW_ANSWER) or debug_print_stream == CONST_DEBUG_SHOW_ANSWER: + print('text, state = ', text, state) + + return text, state + + + # 并发请求模型服务（流式） + @staticmethod + async def __batch_request_chat_api_stream(prompts: list, max_tokens=10240, temperature=0.6, enable_thinking=True): + tasks = [] + for prompt in prompts: + tasks.append(ChatClient.__request_chat_api_stream(prompt, max_tokens, temperature, enable_thinking)) + + return await asyncio.gather(*tasks) + + + # 模型服务流式输出 + @staticmethod + async def __request_chat_api_stream(prompt: str, max_tokens=10240, temperature=0.6, enable_thinking=True, len_for_exception_response=0): + messages = [ + { + "role": "system", + "content": "你是一名无所不知无所不能的全能高手，请仔细理解任务描述和任务目标，展开所有联想找到最佳策略和方法，尽你所能生成用户期望的最终结果，返回结果的格式也须符合用户使用要求。" + }, + { + "role": "user", + "content": str(prompt) if enable_thinking else str(prompt) + '/no_think', # 必须强制转一次字符串 + } + ] + + try: + response = await ChatClient.__client.chat.completions.create( + model = serve_model_name, + messages = messages, + temperature = temperature, + max_tokens = max_tokens, + stream = True, + ) + + # 粗略估计字符串长度限制 + max_len = 4 * max_tokens + full_len = 0 + full_content = "" + async with response: + async for chunk in response: + content = None + if chunk.choices: + delta = chunk.choices[0].delta + if getattr(delta, 'reasoning_content', None): + content = delta.reasoning_content + elif getattr(delta, 'content', None): + content = delta.content + + if content != None: + full_content += content + # 利用 env:DEBUG_PRINT_STREAM 判断是否实时打印（调试查看模型的输出内容） + if debug_print_stream == str(CONST_DEBUG_SHOW_STREAM) or int(debug_print_stream) == CONST_DEBUG_SHOW_STREAM: + print(content, end="", flush=True) + + # 若发现长度异常，则主动终止连接提前截断输出，在这里粗略估计即可 + full_len += len(content) + if full_len > max_len: + break + + text = full_content.strip() + # print("Response-full_text:", text) + + if text.startswith(''): + idx = text.find('') + if idx > -1: + # 去掉think内容 + text = text[idx + 8 : None].strip() + else: + # 处理异常数据：胡言乱语、无限重复等等导致输出不完整 + if len_for_exception_response <= 0: + return None, False + else: + return text[ - len_for_exception_response : None ], False + + # print("Response-text:", text) + return text, True + + except APIConnectionError as e: + print(f"连接失败: {e}") + print("请检查 base_url 是否配置正确，或者网络是否正常。") + raise e + except AuthenticationError as e: + print(f"鉴权失败: {e}") + print("请检查 API Key 是否填写正确。") + raise e + except APIError as e: + error = str(e) + if "inappropriate" in error: + return 'Output data may contain inappropriate content', False + else: + raise e + except BadRequestError as e: + # Error code: 400 - Input data may contain inappropriate content. + error = str(e) + if "inappropriate" in error: + return 'Input data may contain inappropriate content', False + else: + raise e + except Exception as e: + error = str(e) + if "inappropriate" in error: + return 'Output data may contain inappropriate content', False + else: + raise e + +if __name__=="__main__": + debug_print_stream = 1 + ChatClient.request_llm_api_stream("请列举Golang中GC的原理和案例") diff --git a/openseek/competition/LongContext-ICL-Annotation/src/const.py b/openseek/competition/LongContext-ICL-Annotation/src/const.py new file mode 100644 index 00000000..6348fd71 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/const.py @@ -0,0 +1,59 @@ +# 限制输出长度 +CONST_LIMIT_PROMPT_MAX_TOKENS = 8192 + +# 限制输出长度 +CONST_LIMIT_OUTPUT_MAX_TOKENS = 10240 + +# 调试实时打印流式输出 +CONST_DEBUG_SHOW_STREAM = 1 + +# 调试打印正文结果（排除think后的answer) +CONST_DEBUG_SHOW_ANSWER = 2 + +# 任务数据文件名称 +CONST_TASK_FILES = { + 1: 'openseek-1_closest_integers.json', + 2: 'openseek-2_count_nouns_verbs.json', + 3: 'openseek-3_collatz_conjecture.json', + 4: 'openseek-4_conala_concat_strings.json', + 5: 'openseek-5_semeval_2018_task1_tweet_sadness_detection.json', + 6: 'openseek-6_mnli_same_genre_classification.json', + 7: 'openseek-7_jeopardy_answer_generation_all.json', + 8: 'openseek-8_kernel_generation.json', +} + +# 模型输出tokens长度限制 +CONST_TASK_OUTPUT_MAX_TOKENS_SIZES = { + 1: 8192 + 8, + 2: 8192 + 8, + 3: 8192 + 128, + 4: 8192 + 128, + 5: 8192 + 16, + 6: 8192 + 16, + 7: 8192 + 128, + 8: 8192 + 2048, +} + +# 输出答案的字符串长度范围（注：既不包含think长度，也不是tokens长度） +CONST_TASK_CHECK_STRLEN_SIZES = { + 1: { 'min': 1, 'max': 8 }, + 2: { 'min': 1, 'max': 8 }, + 3: { 'min': 1, 'max': 256 }, + 4: { 'min': 1, 'max': 256 }, + 5: { 'min': 1, 'max': 32 }, + 6: { 'min': 1, 'max': 4 }, + 7: { 'min': 1, 'max': 256 }, + 8: { 'min': 384, 'max': 10240 }, +} + +# TEST预测返回类型 +CONST_TASK_TEST_RETURN_DATA_TYPES = { + 1: 'number', # 例如：12, 3, 5, -1, -0.36 等数字 + 2: 'number', + 3: 'array', # 例如: [1,2,3], ['a','b', 'c'] 等数组 + 4: 'text', # 任意英文字符串 + 5: 'text', + 6: 'text', + 7: 'phrase', # 英文短语 + 8: 'code', # 编程代码 +} diff --git a/openseek/competition/LongContext-ICL-Annotation/src/create_env_ascend.sh b/openseek/competition/LongContext-ICL-Annotation/src/create_env_ascend.sh new file mode 100644 index 00000000..51652f35 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/create_env_ascend.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +# vllm +VLLM_TARGET_DEVICE=empty pip install vllm==0.13.0 --extra-index https://download.pytorch.org/whl/cpu +# 忽略vllm依赖torch的版本提示 +pip install vllm_ascend==0.13.0rc1 +pip install vllm-plugin-fl==0.1.0+vllm0.13.0 --extra-index-url https://resource.flagos.net/repository/flagos-pypi-hosted/simple + +# FlagScale +git clone https://github.com/flagos-ai/FlagScale +cd FlagScale +pip install setuptools==82.0.0 scikit-build-core==0.11 pybind11==3.0.2 ninja==1.13.0 cmake==4.2.3 +pip install ".[ascend-serve]" -v --no-build-isolation +cd .. + +# 常用工具 +pip install ascend-nputop +pip install modelscope + +# 下载Qwen3-4B +pip install "setuptools<82.0.0" +modelscope download --model Qwen/Qwen3-4B +sed -i 's#"rope_scaling": null#"rope_scaling": {"rope_type": "yarn", "factor": 4.0, "original_max_position_embeddings": 40960}#g' ~/.cache/modelscope/hub/models/Qwen/Qwen3-4B/config.json +cat ~/.cache/modelscope/hub/models/Qwen/Qwen3-4B/config.json + +# 环境变量 +export HCCL_CONNECT_TIMEOUT=180 +export HCCL_IF_MTU=10240 +export VLLM_PLUGINS=fl +export VLLM_ASCEND_ENABLE_CONTEXT_PARALLEL=1 +export VLLM_WORKER_MULTIPROC_METHOD=spawn +export VLLM_ASCEND_USE_TORCHAIR=0 +export ASCEND_RT_VISIBLE_DEVICES=0,1 +export ASCEND_VISIBLE_DEVICES=0,1 +export VLLM_USE_MODELSCOPE=true +export TRITON_ALL_BLOCKS_PARALLEL=1 +export PYTORCH_NPU_ALLOC_CONF=max_split_size_mb:256 + +export VLLM_ASCEND_ENABLE_FUSED_MC2=1 # 启用融合算子，支持TND +export VLLM_ASCEND_ENABLE_FLASHCOMM=1 # 多卡并行时启用FlashComm +export TASK_QUEUE_ENABLE=2 # EP (Enqueued Processing) 模式 diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config.yaml index 94e8a2cd..9d38e030 100644 --- a/openseek/competition/LongContext-ICL-Annotation/src/llm_config.yaml +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config.yaml @@ -1,14 +1,27 @@ serve: -- serve_id: vllm_model - engine: vllm - engine_args: - model: ../Qwen3-4B - host: 0.0.0.0 - uvicorn_log_level: warning - port: 2026 - gpu_memory_utilization: 0.9 - trust_remote_code: true - no_enable_prefix_caching: true + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2026 + trust_remote_code: true + gpu_memory_utilization: 0.9 + pipeline_parallel_size: 1 + tensor_parallel_size: 2 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 40960 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + distributed_executor_backend: mp experiment: exp_name: qwen3_4b @@ -20,11 +33,11 @@ experiment: deploy: use_fs_serve: false envs: - CUDA_VISIBLE_DEVICES: 0 - CUDA_DEVICE_MAX_CONNECTIONS: 1 + ASCEND_RT_VISIBLE_DEVICES: 0,1 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 action: run hydra: run: - dir: ${experiment.exp_dir}/hydra \ No newline at end of file + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_0.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_0.yaml new file mode 100644 index 00000000..a64d3591 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_0.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2026 + trust_remote_code: true + gpu_memory_utilization: 0.9 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 40960 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s2_0/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 0 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_1.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_1.yaml new file mode 100644 index 00000000..b32e43ce --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s2_1.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2027 + trust_remote_code: true + gpu_memory_utilization: 0.9 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 40960 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s2_1/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 1 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_0.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_0.yaml new file mode 100644 index 00000000..fcf13b64 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_0.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2026 + trust_remote_code: true + gpu_memory_utilization: 0.4 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 20480 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s4_0/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 0 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_1.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_1.yaml new file mode 100644 index 00000000..62ba17d5 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_1.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2027 + trust_remote_code: true + gpu_memory_utilization: 0.4 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 20480 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s4_1/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 0 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_2.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_2.yaml new file mode 100644 index 00000000..63719d37 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_2.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2028 + trust_remote_code: true + gpu_memory_utilization: 0.4 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 20480 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s4_2/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 1 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_3.yaml b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_3.yaml new file mode 100644 index 00000000..f19a5132 --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/llm_config_s4_3.yaml @@ -0,0 +1,42 @@ +serve: + - serve_id: vllm_model + engine: vllm + engine_args: + model: /root/.cache/modelscope/hub/models/Qwen/Qwen3-4B + host: 0.0.0.0 + served-model-name: Qwen/Qwen3-4B + uvicorn_log_level: warning + port: 2029 + trust_remote_code: true + gpu_memory_utilization: 0.4 + pipeline_parallel_size: 1 + tensor_parallel_size: 1 + block_size: 128 + max_num_seqs: 1 + max_model_len: 20480 + max_num_batched_tokens: 20480 + dtype: bfloat16 + enforce_eager: true + enable_prefix_caching: true + enable_chunked_prefill: true + disable_log_stats: true + async_scheduling: true + +experiment: + exp_name: qwen3_4b + exp_dir: s4_3/${experiment.exp_name} + task: + type: serve + runner: + hostfile: null + deploy: + use_fs_serve: false + envs: + ASCEND_RT_VISIBLE_DEVICES: 1 + ASCEND_RT_DEVICE_MAX_CONNECTIONS: 8 + +action: run + +hydra: + run: + dir: ${experiment.exp_dir}/hydra diff --git a/openseek/competition/LongContext-ICL-Annotation/src/main.py b/openseek/competition/LongContext-ICL-Annotation/src/main.py index c2949795..17aefa48 100644 --- a/openseek/competition/LongContext-ICL-Annotation/src/main.py +++ b/openseek/competition/LongContext-ICL-Annotation/src/main.py @@ -1,92 +1,196 @@ -import json, os, argparse +import os +import re +import time +import json +import random +import argparse +random.seed(123) + from tqdm import tqdm, trange from transformers import AutoTokenizer -# from method import build_prompt, select_examples, annotate +from method import select_examples, select_operations, refer_get_operation_for_icl, refer_get_result_for_test +from const import CONST_TASK_FILES, CONST_TASK_OUTPUT_MAX_TOKENS_SIZES, CONST_LIMIT_PROMPT_MAX_TOKENS, CONST_LIMIT_OUTPUT_MAX_TOKENS, CONST_TASK_TEST_RETURN_DATA_TYPES -from method import build_prompt, select_examples -from method import annotate_nvidia as annotate # For Nvidia GPU -# from method import annotate_ascend as annotate # For Huawei Ascend +# 文件目录: {your-project-path}/OpenSeek/openseek/competition/LongContext-ICL-Annotation +LongContext_ICL_Annotation_DIR = os.path.dirname(os.path.dirname(__file__)) -TASK_FILES = { - 1: './data/openseek-1_closest_integers.json', - 2: './data/openseek-2_count_nouns_verbs.json', - 3: './data/openseek-3_collatz_conjecture.json', - 4: './data/openseek-4_conala_concat_strings.json', - 5: './data/openseek-5_semeval_2018_task1_tweet_sadness_detection.json', - 6: './data/openseek-6_mnli_same_genre_classification.json', - 7: './data/openseek-7_jeopardy_answer_generation_all.json', - 8: '../data/openseek-8_kernel_generation.json', -} +data_folder = 'data' +operation_folder = 'operations_tmp' +result_folder = 'outputs' -def parser_args(): - parser = argparse.ArgumentParser() - parser.add_argument('--task_id', type=int, required=True, - help='Task ID to evaluate, should be in [1, 7].') - parser.add_argument('--max_input_length', type=int, default=10_000, - help='Maximum input length for the model.') - parser.add_argument('--log_path_prefix', type=str, - default='../outputs/', - help='Prefix path to save the evaluation logs.') - parser.add_argument('--tokenizer_path', type=str, - default='/share/project/wuhaiming/spaces/data_agent/OpenSeek-main/openseek/competition/LongContext-ICL-Annotation/src/Qwen3-4B') - args = parser.parse_args() - return args - -def evaluate(task_id:int, - qwen_tokenizer:AutoTokenizer, - max_input_length:int=128_000, - log_path_prefix:str='./outputs/' - )->float: - assert task_id in [i for i in range(1, 9)],\ - f"task_id should be in [1, 8], but got {task_id}." - - task_file = TASK_FILES[task_id] +get_task_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, data_folder, CONST_TASK_FILES[task_id]) +get_operation_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, operation_folder, f'operation-{task_id}.json') +get_result_filepath = lambda task_id, version: os.path.join(LongContext_ICL_Annotation_DIR, result_folder, f'openseek-{task_id}-v{version}.jsonl') + +get_icl_progress_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, f'icl_progress-{task_id}.log') +get_test_progress_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, f'test_progress-{task_id}.log') + + +# 记录ICL提取执行进度 +def print_icl_progress(task_id, *args): + print(args) + +# 记录TEST执行进度 +def print_test_progress(task_id, *args): + print(args) + + +# 利用ICL提取特定问题的关键操作Operation，可以重复执行多轮，每一轮使用不同的ICL样本组合批次 +def processing( + task_id:int, + qwen_tokenizer:AutoTokenizer, + start=0, + stop=-1 +): + assert 1 <= task_id <= 8, f"task_id should be in [1,2,3,4,5,6,7,8], but got {task_id}." + assert stop < 0 or stop >= start, f"must be `stop < 0 or stop >= start`, stop error: {stop}" + + # 保存Operation + operation_file = get_operation_filepath(task_id) + print_icl_progress(task_id, 'operation_file', operation_file) + + task_operations = set() + operation_dir = os.path.dirname(operation_file) + os.makedirs(operation_dir, mode=0o755, exist_ok=True) + + task_file = get_task_filepath(task_id) + print_icl_progress(task_id, 'task_file', task_file) with open(task_file, 'r') as f: task_dict = json.load(f) - + + task_name = task_dict['task_name'] + task_description = task_dict['Definition'][0] + icl_examples = task_dict['examples'] + print_icl_progress(task_id, 'task_name', task_name) + print_icl_progress(task_id, 'icl_examples-size', len(icl_examples)) + + # 遍历所有示例样本，自动计算合适长度的批次 + start = start if start >= 0 else 0 + stop = stop if stop >= start else len(icl_examples) - 1 + while start <= stop: + + icl_batch_data, start_next = select_examples(icl_examples, qwen_tokenizer, start, stop, CONST_LIMIT_OUTPUT_MAX_TOKENS) + + print("\n") + print_icl_progress(task_id, 'icl_batch_data-strlen', len(icl_batch_data), "start", start, "batch_size", start_next - start, "task", task_id, task_name) + start = start_next + + operations, state = refer_get_operation_for_icl(task_description, icl_batch_data, CONST_LIMIT_OUTPUT_MAX_TOKENS) + + print("\n--------------------------------------------------------------------------------------------------------------------") + if operations == None: + print_icl_progress(task_id, f"The batch get operations is None, state: {state}", ) + else: + print_icl_progress(task_id, f"The batch get operations is Success, state: {state}", operations) + for v in operations: + task_operations.add(v) + + # 写入持久化文件 + operations_list = [v for v in task_operations] + print_icl_progress(task_id, 'operations_list-size', len(operations_list)) + + if len(operations_list) > 0: + with open(operation_file, 'w') as f: + json.dump(operations_list, f) + + print_icl_progress(task_id, "*"*50) + + +# 利用特定问题的Operation获取测试数据的推理结果 +def evaluate( + task_id:int, + qwen_tokenizer:AutoTokenizer, + start=0, + stop=-1 +): + assert 1 <= task_id <= 8, f"task_id should be in [1,2,3,4,5,6,7,8], but got {task_id}." + assert stop < 0 or stop >= start, f"must be `stop < 0 or stop >= start`, stop error: {stop}" + + # 加载Operation + operation_file = get_operation_filepath(task_id) + print_test_progress(task_id, 'operation_file', operation_file) + with open(operation_file, 'r') as f: + task_operations = json.load(f) + + # 无论有多少关键操作，都只提交推理一次测试，所以也只遍历一次关键操作，简单获取足量的关键操作即可 + random.shuffle(task_operations) + core_operations, _ = select_operations(task_operations, qwen_tokenizer, 0, len(task_operations) - 1, CONST_LIMIT_PROMPT_MAX_TOKENS) + print_test_progress(task_id, 'core_operations-strlen', len(core_operations)) + + # 加载任务文件 + task_file = get_task_filepath(task_id) + with open(task_file, 'r') as f: + task_dict = json.load(f) + task_name = task_dict['task_name'] task_description = task_dict['Definition'][0] - icl_examples = task_dict['examples'][:100] test_samples = task_dict['test_samples'] - + print_test_progress(task_id, 'test_samples-size', len(test_samples), "task", task_id, task_name) + + # 测试结果保存位置 version = 1 - output_file = f'{log_path_prefix}openseek-{task_id}-v{version}.jsonl' - output_path = os.path.dirname(output_file) - os.makedirs(output_path, exist_ok=True) - while os.path.exists(output_file): + result_file = get_result_filepath(task_id, version) + result_dir = os.path.dirname(result_file) + os.makedirs(result_dir, mode=0o755, exist_ok=True) + while os.path.exists(result_file): version += 1 - output_file = f'{log_path_prefix}openseek-{task_id}-v{version}.jsonl' - with open(output_file, 'w') as f: + result_file = get_result_filepath(task_id, version) + with open(result_file, 'w') as f: pass - - examples_str = None - for test_sample in tqdm(test_samples, desc=f'Evaluation on Task {task_id}: {task_name}'): - test_record = dict() - - test_sample_id = test_sample['id'] - test_record['test_sample_id'] = test_sample_id - - - text2annotate = test_sample['input'] - prompt = build_prompt(task_description, text2annotate) - if examples_str is None: - examples_str = select_examples(icl_examples, task_description, text2annotate) - input_prompt = prompt.replace("[[EXAMPLES]]\n\n", examples_str+'\n\n') - - # tokenized_input = qwen_tokenizer(input_prompt, return_tensors="pt") - # if tokenized_input['input_ids'].shape[1] > max_input_length: - # test_record['prediction'] = None - # else: - # prediction = annotate(input_prompt) - # test_record['prediction'] = prediction - prediction = annotate(input_prompt) - test_record['prediction'] = prediction - with open(output_file, 'a') as f: + + # 提交测试 + start = start if start >= 0 else 0 + stop = stop if stop >= start else len(test_samples) - 1 + for i in range(start, stop + 1): + print(f'Evaluation item={i} on Task {task_id}: {task_name}') + + test_sample_id = test_samples[i]['id'] + test_input = test_samples[i]['input'] + prediction, state = refer_get_result_for_test(task_description, test_input, core_operations, CONST_TASK_TEST_RETURN_DATA_TYPES[task_id], CONST_TASK_OUTPUT_MAX_TOKENS_SIZES[task_id]) + + print("\n--------------------------------------------------------------------------------------------------------------------") + print_test_progress(task_id, f"Answer state={state} item={i}:", prediction) + + # 保存测试结果 + test_record = {'test_sample_id': test_sample_id, "state": state, 'reps': 0, 'prediction': prediction} + with open(result_file, 'a') as f: f.write(json.dumps(test_record)+'\n') + print_test_progress(task_id, "*"*50) + + +# 命令参数解析 +def parser_args(): + parser = argparse.ArgumentParser() + + parser.add_argument('--task_id', type=int, required=True, + help='Task ID, one of [1,2,3,4,5,6,7,8]') + parser.add_argument('--task_step', type=int, required=True, + help='Task Step, one of [1,2], 1 for processing step, 2 for evaluate step') + + # 可定位某个ICL批次或某个TEST测试 + parser.add_argument('--start', type=int, default=0, help='Last Stop Task Row Index[0,len-1].') + parser.add_argument('--stop', type=int, default=-1, help='Next Stop Task Row Index[0,len-1].') + + # 分词器参数位置路径 + default_tokenizer_path = os.path.join(os.path.expanduser('~'), '.cache/modelscope/hub/models/Qwen/Qwen3-4B') + parser.add_argument('--tokenizer_path', type=str, default=default_tokenizer_path, help='Model path for serve.') + + return parser.parse_args() + + if __name__ == '__main__': args = parser_args() - qwen_tokenizer = AutoTokenizer.from_pretrained(args.tokenizer_path) - evaluate(args.task_id, qwen_tokenizer, args.max_input_length, args.log_path_prefix) \ No newline at end of file + + tokenizer_path = os.environ.get('SERVE_TOKENIZER_PATH', args.tokenizer_path) + qwen_tokenizer = AutoTokenizer.from_pretrained(tokenizer_path) + + # 首先提取Operation + if args.task_step == 1: + processing(args.task_id, qwen_tokenizer, args.start, args.stop) + + # 然后提交测试 + if args.task_step == 2: + evaluate(args.task_id, qwen_tokenizer, args.start, args.stop) diff --git a/openseek/competition/LongContext-ICL-Annotation/src/main_batch.py b/openseek/competition/LongContext-ICL-Annotation/src/main_batch.py new file mode 100644 index 00000000..3e9261bd --- /dev/null +++ b/openseek/competition/LongContext-ICL-Annotation/src/main_batch.py @@ -0,0 +1,379 @@ +import os +import re +import json +import time +import random +import shutil +import argparse +random.seed(123) + +from tqdm import tqdm, trange +from transformers import AutoTokenizer + +# 必须强制 DEBUG_PRINT_STREAM=0 防止在后台运行时实时打印模型的输出内容，会拖慢速度，而且实时打印是胡乱无序的 +os.environ['DEBUG_PRINT_STREAM'] = '0' + +from method import select_examples, select_operations, parse_result, batch_refer_get_operation_for_icl, batch_refer_get_result_for_test +from const import CONST_TASK_FILES, CONST_TASK_OUTPUT_MAX_TOKENS_SIZES, CONST_LIMIT_PROMPT_MAX_TOKENS, CONST_LIMIT_OUTPUT_MAX_TOKENS, CONST_TASK_CHECK_STRLEN_SIZES, CONST_TASK_TEST_RETURN_DATA_TYPES + + +# 提取轮次（若NPU卡较少，为了加速实验可选择默认N=1，若NPU卡较多 可以尝试多跑几遍） +try: + N = int(os.environ.get('ICL_REPEAT_N', 1)) +except ValueError: + N = 0 + +# 异步并发数（与服务器支持的批次并行相近即可, 协同服务端张量和数据并行计算, 可加速实验进度） +try: + req_concurrency_size = int(os.environ.get('REQ_CONCURRENCY_NUM', 2)) +except ValueError: + req_concurrency_size = 0 + +if N <=0: + raise Exception('ICL_REPEAT_N must be a integer and bigger than 0') +if req_concurrency_size <= 0: + raise Exception('REQ_CONCURRENCY_NUM must be a integer and bigger than 0') + + +# 文件目录: {your-project-path}/OpenSeek/openseek/competition/LongContext-ICL-Annotation +LongContext_ICL_Annotation_DIR = os.path.dirname(os.path.dirname(__file__)) + +data_folder = 'data' +operation_folder = 'operations_tmp' +result_folder = 'outputs' + +get_task_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, data_folder, CONST_TASK_FILES[task_id]) +get_operation_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, operation_folder, f'operation-{task_id}.json') +get_result_filepath = lambda task_id, version: os.path.join(LongContext_ICL_Annotation_DIR, result_folder, f'openseek-{task_id}-v{version}.jsonl') + +get_icl_progress_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, f'icl_progress-{task_id}.log') +get_test_progress_filepath = lambda task_id: os.path.join(LongContext_ICL_Annotation_DIR, f'test_progress-{task_id}.log') + + +# 记录ICL提取执行进度 +def log_icl_progress(task_id, *args): + with open(get_icl_progress_filepath(task_id), 'a') as f: + f.write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' :: ' + json.dumps(args)+'\n') + + +# 记录TEST执行进度 +def log_test_progress(task_id, *args): + with open(get_test_progress_filepath(task_id), 'a') as f: + f.write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' :: ' + json.dumps(args)+'\n') + + +# 利用ICL提取特定问题的关键操作Operation，可以重复执行多轮，每一轮使用不同的ICL样本组合批次 +def processing(task_id:int, qwen_tokenizer:AutoTokenizer, start = 0, stop = -1): + assert 1 <= task_id <= 8, f"task_id should be in [1, 8], but got {task_id}." + assert stop < 0 or stop >= start, f"must be `stop < 0 or stop >= start`, stop error: {stop}" + + # 保存Operation + operation_file = get_operation_filepath(task_id) + log_icl_progress(task_id, 'operation_file', operation_file) + + task_operations = set() + + operation_dir = os.path.dirname(operation_file) + os.makedirs(operation_dir, mode=0o755, exist_ok=True) + + if os.path.exists(operation_file): + if N == 1: + log_icl_progress(task_id, 'not need repeat processing for N=1') + return + # 与上一轮的去重合并 + with open(operation_file, 'r') as f: + old_operations_list = json.load(f) + task_operations.update(old_operations_list) + + task_file = get_task_filepath(task_id) + log_icl_progress(task_id, 'task_file', task_file) + with open(task_file, 'r') as f: + task_dict = json.load(f) + + task_name = task_dict['task_name'] + task_description = task_dict['Definition'][0] + icl_examples = task_dict['examples'] + log_icl_progress(task_id, 'task_name', task_name) + log_icl_progress(task_id, 'icl_examples-size', len(icl_examples)) + + for i in range(N): + log_icl_progress(task_id, f"第 {i + 1} 轮次ICL遍历") + + # 在重复采样时，打乱ICL的顺序，每轮即可得到不同样本组合的批次，让样本表现为多样性 + if i > 0: + random.shuffle(icl_examples) + + # 遍历所有示例样本，自动计算合适长度的批次 + icl_batch_data_list = [] + start = start if start >= 0 else 0 + stop = stop if stop >= start else len(icl_examples) - 1 + while start <= stop: + icl_batch_data, start_next = select_examples(icl_examples, qwen_tokenizer, start, stop, CONST_LIMIT_PROMPT_MAX_TOKENS) + log_icl_progress(task_id, 'icl_batch_data-strlen:', len(icl_batch_data), f"N_i={i}", "start:", start, "icl_batch_size:", start_next - start, "task:", task_id, task_name) + start = start_next + + icl_batch_data_list.append(icl_batch_data) + if len(icl_batch_data_list) == req_concurrency_size or start >= stop: + operations_list = batch_refer_get_operation_for_icl(task_description, icl_batch_data_list, CONST_LIMIT_OUTPUT_MAX_TOKENS) + icl_batch_data_list = [] + + for operations, state in operations_list: + if operations == None: + log_icl_progress(task_id, f"The N_i={i} req_concurrency_size: {req_concurrency_size} get operations is None, state: {state}") + else: + log_icl_progress(task_id, f"The N_i={i} req_concurrency_size: {req_concurrency_size} get operations is Success, state: {state}", operations) + for v in operations: + task_operations.add(v) + + # 写入持久化文件 + operations_list = [v for v in task_operations] + log_icl_progress(task_id, 'operations_list-size', len(operations_list)) + + if len(operations_list) > 0: + with open(operation_file, 'w') as f: + json.dump(operations_list, f) + + log_icl_progress(task_id, "*"*50) + + +# 利用特定问题的Operation获取测试数据的推理结果 +def evaluate(task_id:int, qwen_tokenizer:AutoTokenizer, start = 0, stop = -1): + assert 1 <= task_id <= 8, f"task_id should be in [1, 8], but got {task_id}." + assert stop < 0 or stop >= start, f"must be `stop < 0 or stop >= start`, stop error: {stop}" + + # 加载Operation + operation_file = get_operation_filepath(task_id) + log_test_progress(task_id, 'operation_file', operation_file) + with open(operation_file, 'r') as f: + task_operations = json.load(f) + + # 无论有多少关键操作，都只提交推理一次测试，所以也只遍历一次关键操作，简单获取足量的关键操作即可 + random.shuffle(task_operations) + core_operations, _ = select_operations(task_operations, qwen_tokenizer, 0, len(task_operations) - 1, CONST_LIMIT_PROMPT_MAX_TOKENS) + log_test_progress(task_id, 'core_operations-strlen', len(core_operations)) + + # 加载任务文件 + task_file = get_task_filepath(task_id) + with open(task_file, 'r') as f: + task_dict = json.load(f) + + task_name = task_dict['task_name'] + task_description = task_dict['Definition'][0] + test_samples = task_dict['test_samples'] + log_test_progress(task_id, 'test_samples-size', len(test_samples), "task", task_id, task_name) + + # 测试结果保存位置 + version = 1 + result_file = get_result_filepath(task_id, version) + result_dir = os.path.dirname(result_file) + os.makedirs(result_dir, mode=0o755, exist_ok=True) + while os.path.exists(result_file): + version += 1 + result_file = get_result_filepath(task_id, version) + with open(result_file, 'w') as f: + pass + + # 提交测试 + start = start if start >= 0 else 0 + stop = stop if stop >= start else len(test_samples) - 1 + for i in range(start, stop + 1, req_concurrency_size): + log_test_progress(task_id, f'Evaluation items({i}:{i+req_concurrency_size}) on Task {task_id}: {task_name}') + + test_sample_id_list = [] + test_input_list = [] + for test_sample in test_samples[i: min(i + req_concurrency_size, stop + 1)]: + test_sample_id_list.append(test_sample['id']) + test_input_list.append(test_sample['input']) + + prediction_list = batch_refer_get_result_for_test(task_description, test_input_list, core_operations, CONST_TASK_TEST_RETURN_DATA_TYPES[task_id], CONST_TASK_OUTPUT_MAX_TOKENS_SIZES[task_id]) + + j = 0 + for test_sample_id, (prediction, state) in zip(test_sample_id_list, prediction_list): + log_test_progress(task_id, f"Answer state={state} item={i+j}:", prediction) + j += 1 + + # 保存测试结果 + test_record = {'test_sample_id': test_sample_id, 'state': state, 'reps': 0, 'prediction': prediction} + with open(result_file, 'a') as f: + f.write(json.dumps(test_record)+'\n') + + log_test_progress(task_id, "*"*50) + + +# 自动检查异常数据：可能是服务端为正常返回，也可能是KV缓存，也可能是模型不稳定 +# 像这样的异常数据，可用通过重新请求服务，最终得到正常输出 +def check_evaluate_retry(task_id:int, qwen_tokenizer:AutoTokenizer, reps = 0): + assert 1 <= task_id <= 8, f"task_id should be in [1, 8], but got {task_id}." + + # 测试结果保存位置 + version = 1 + result_file = get_result_filepath(task_id, version) + if not os.path.exists(result_file): + # 没有结果文件，无需检查 + print(f'task_id={task_id}暂无对应结果文件，无需检查') + return + + min_len, max_len = CONST_TASK_CHECK_STRLEN_SIZES[task_id]['min'], CONST_TASK_CHECK_STRLEN_SIZES[task_id]['max'] + checks = {} + outputs = [] + with open(result_file, 'r') as file: + # 逐行读取并解析JSON + i = 0 + for line in file: + # 解析每行的JSON数据 + data = json.loads(line) + outputs.append(data) + # 检查是否满足特定条件 + data_len = len(data['prediction']) if data['prediction'] != None else 0 + if data['state'] == False or data_len < min_len or data_len > max_len: + checks[data['test_sample_id']] = i + # next + i += 1 + + if len(checks.values()) == 0: + # 没有异常数据，无需检查 + print(f'task_id={task_id}没有异常数据，无需检查') + return + + # 备份 + backup_result_file = result_file + '_backup' + shutil.copy(result_file, backup_result_file) + + # 加载Operation + operation_file = get_operation_filepath(task_id) + log_test_progress(task_id, 'operation_file', operation_file) + with open(operation_file, 'r') as f: + task_operations = json.load(f) + + # 无论有多少关键操作，都只提交推理一次测试，所以也只遍历一次关键操作，简单获取足量的关键操作即可 + random.shuffle(task_operations) + core_operations, _ = select_operations(task_operations, qwen_tokenizer, 0, len(task_operations) - 1, CONST_LIMIT_PROMPT_MAX_TOKENS) + log_test_progress(task_id, 'core_operations-strlen', len(core_operations)) + + # 加载任务文件 + task_file = get_task_filepath(task_id) + with open(task_file, 'r') as f: + task_dict = json.load(f) + + task_name = task_dict['task_name'] + task_description = task_dict['Definition'][0] + test_samples = task_dict['test_samples'] + log_test_progress(task_id, 'test_samples-size', len(test_samples), "task", task_id, task_name) + + check_ids = checks.keys() + indexs = [ i for i, item in enumerate(test_samples) if item['id'] in check_ids ] + for k in range(0, len(indexs), req_concurrency_size): + sample_idxs = indexs[k : k + req_concurrency_size] + + test_sample_id_list = [] + test_input_list = [] + for i in sample_idxs: + test_sample_id_list.append(test_samples[i]['id']) + test_input_list.append(test_samples[i]['input']) + + log_test_progress(task_id, f"Retry items={sample_idxs}") + prediction_list = batch_refer_get_result_for_test(task_description, test_input_list, core_operations, CONST_TASK_TEST_RETURN_DATA_TYPES[task_id], CONST_TASK_OUTPUT_MAX_TOKENS_SIZES[task_id]) + + j = 0 + for test_sample_id, (prediction, state) in zip(test_sample_id_list, prediction_list): + raw = prediction + if state == False: + prediction, _ = parse_result(prediction, CONST_TASK_TEST_RETURN_DATA_TYPES[task_id]) + + log_test_progress(task_id, f"Correct state={state} item={sample_idxs[j]}:", prediction) + j += 1 + + # 保存测试结果 + outputs[checks[test_sample_id]] = {'test_sample_id': test_sample_id, 'state': state, 'reps': reps, 'prediction': prediction} + if raw != prediction: + outputs[checks[test_sample_id]]['raw'] = raw + + new_result_file = get_result_filepath(task_id, version + 1) + with open(new_result_file, 'w') as f: + for data in outputs: + f.write(json.dumps(data) + '\n') + + # 覆盖旧文件 + os.replace(new_result_file, result_file) + log_test_progress(task_id, "*"*50) + + +# 命令参数解析 +def parser_args(): + parser = argparse.ArgumentParser() + + parser.add_argument( + '--task_id', + type=str, + required=True, + help='Task ID, examples: task_id=1 or task_id=[1,2] or task_id=0, all task if task_id=0 or task_id is empty, one task if task_id>=1 and task_id<=8, multi task if task_id is list' + ) + parser.add_argument( + '--task_step', + type=str, + required=True, + help='Task Step, examples: task_step=1 or task_step=[1,2] or task_step=0, all if task_step=0 or task_step is empty, processing if task_step=1, evaluate if task_step=2, check_evaluate_retry if task_step=3' + ) + + # 可定位某个ICL批次或某个TEST测试 + parser.add_argument('--start', type=int, default=0, help='Last Stop Task Row Index[0,len-1].') + parser.add_argument('--stop', type=int, default=-1, help='Next Stop Task Row Index[0,len-1].') + + # 分词器参数位置路径 + default_tokenizer_path = os.path.join(os.path.expanduser('~'), '.cache/modelscope/hub/models/Qwen/Qwen3-4B') + parser.add_argument('--tokenizer_path', type=str, default=default_tokenizer_path, help='Model path for serve.') + + return parser.parse_args() + + +def parse_args_task_id_or_step(arg_value: str, default_list: list): + arg_value = arg_value.strip() + if not arg_value or arg_value == '0': + return default_list + + cleaned = arg_value.lstrip('[').rstrip(']') + try: + return [int(x.strip()) for x in cleaned.split(',') if x.strip()] + except ValueError: + raise ValueError(f"Invalid list of integers: {arg_value}") + + +if __name__ == '__main__': + args = parser_args() + + task_step_list = parse_args_task_id_or_step(args.task_step, [1, 2, 3]) + for task_step in task_step_list: + if not task_step in [1,2,3]: + raise Exception(f'Task Step error: {task_step_list}') + + task_id_list = parse_args_task_id_or_step(args.task_id, [1, 2, 3, 4, 5, 6, 7, 8]) + for task_id in task_id_list: + if not task_id in [1,2,3,4,5,6,7,8]: + raise Exception(f'Task ID error: {task_id_list}') + + # tokenizer + tokenizer_path = os.environ.get('SERVE_TOKENIZER_PATH', args.tokenizer_path) + qwen_tokenizer = AutoTokenizer.from_pretrained(tokenizer_path) + + # 串行处理任务 + for task_id in task_id_list: + + # 首先提取Operation + if 1 in task_step_list: + processing(task_id, qwen_tokenizer, args.start, args.stop) + args.start = 0 + args.stop = -1 + + # 然后提交测试 + if 2 in task_step_list: + evaluate(task_id, qwen_tokenizer, args.start, args.stop) + args.start = 0 + args.stop = -1 + + # 最后检查Result异常数据：可能是GPU负荷太高、GPU高温降频等无法正常完成，也可能是KV缓存干扰，也可能是模型输出不稳定，可能是网络问题等等 + if 3 in task_step_list: + # 检查3遍 + for i in range(3): + reps = i + 1 + check_evaluate_retry(task_id, qwen_tokenizer, reps) + time.sleep(30 * reps) diff --git a/openseek/competition/LongContext-ICL-Annotation/src/method.py b/openseek/competition/LongContext-ICL-Annotation/src/method.py index 386daf22..e3f07a5d 100644 --- a/openseek/competition/LongContext-ICL-Annotation/src/method.py +++ b/openseek/competition/LongContext-ICL-Annotation/src/method.py @@ -1,277 +1,357 @@ - import re + from collections import Counter from transformers import AutoTokenizer -""" Here is an example of implementation of Long-Context Data Annotation. """ +from api_client import ChatClient +from const import CONST_LIMIT_OUTPUT_MAX_TOKENS -def build_prompt____(task_description: str, text2annotate: str) -> str: - """ - Build a high-precision English prompt for long-context data annotation (optimized for Qwen3-4B). - Core requirement: Final answer MUST be wrapped in