如何用 Nebius Data Lab 的 70B 教师模型批量推理生成客服微调训练数据【免费下载链接】Nebius-CookbookA collection of projects showcasing RAG, agents, workflows, and other AI use cases项目地址: https://gitcode.com/GitHub_Trending/ne/Nebius-Cookbook如果你要给一个客服模型做微调缺的不是训练代码而是一批高质量的问答训练样本。Nebius-Cookbook 的 fine_tuning/customer_support_datalab 示例给出了一条完整路径先用 8B 基线模型生成一轮客服对话把原始数据集上传到 Nebius Data Lab再让meta-llama/Llama-3.3-70B-Instruct这个 70B 教师模型对整份数据集做批量推理batch inference最后把教师模型的输出导出、过滤、改写成可直接用于 LoRA 微调的对话式 JSONL。本文覆盖其中“生成训练数据”的四个步骤。整个流程由 customer_support_datalab_finetuning_tutorial.ipynb 支撑所有命令、payload 和过滤逻辑都以该 notebook 为准。运行前需要准备什么Notebook 的 Prerequisites 一节列出了三项前提Nebius API key导出为环境变量NEBIUS_API_KEYPython 环境中已安装openai和requestspip install openai requests账号有权限创建 Data Lab 数据集、微调任务和私有自定义模型。流程中用到的关键常量均来自 notebook 初始化单元格API_KEY os.environ[NEBIUS_API_KEY] BASE_URL https://api.tokenfactory.nebius.com/v1/ CONTROL_URL https://api.tokenfactory.nebius.com BASE_MODEL meta-llama/Llama-3.1-8B-Instruct # 基线学生模型 TEACHER_MODEL meta-llama/Llama-3.3-70B-Instruct # 70B 教师模型 DATA_LAB_FOLDER /demo/customer-support # Data Lab 中数据集所在文件夹 BATCH_COMPLETION_WINDOW 24h # 批量推理完成窗口中间产物会写入nebius-datalab-pipeline/scripts/artifacts/目录notebook 还会把各步骤产出的 dataset ID、operation ID 等写入customer_notebook_state.json这样中断后可以从后续步骤重跑而不丢 ID。notebook 通过向上查找nebius-datalab-pipeline/scripts/customer.py来定位仓库根目录找不到时就以当前工作目录为根独立运行时产物同样落在上述相对路径下。示例场景使用两套系统提示词SUPPORT_SYSTEM_PROMPT描述 Northwind Gadgets 的退换、退款、保修等政策会写入每条训练样本TEACHER_SYSTEM_PROMPT则让教师模型以“资深客服政策专家”口吻作答。领域问题由DOMAIN_TOPICS列表提供共 10 条如耳机到货即损坏、未使用商品退货、礼品卡退款等换自己的业务时替换这个列表和两套提示词即可。第一步用 8B 基线模型生成初始客服对话批量推理的输入不是凭空写的 prompt 列表而是先用较小的BASE_MODEL对每个领域问题生成一次回答得到prompt/completion对。这一步让后续教师-学生蒸馏的原始数据集更接近真实客服场景。def step1_generate_inference_logs(topics: list[str]) - list[dict[str, Any]]: logs: list[dict[str, Any]] [] for topic in topics: resp client.chat.completions.create( modelBASE_MODEL, messages[ {role: system, content: SUPPORT_SYSTEM_PROMPT}, {role: user, content: topic}, ], max_tokens256, ) logs.append({ prompt: topic, completion: resp.choices[0].message.content, model: BASE_MODEL, }) print(f✓ {topic[:70]}) return logs logs step1_generate_inference_logs(DOMAIN_TOPICS)其中client是OpenAI(base_urlBASE_URL, api_keyAPI_KEY)创建的 OpenAI 兼容客户端。每完成一个 topic 会打印一行确认。跑完后logs里应包含与DOMAIN_TOPICS等量的prompt/completion记录。第二步把原始数据集上传到 Data Lab这一步把基线对话写成结构化 JSONL并以带 schema 的 dataset 形式上传到 Data Lab。每行保留custom_id后续把教师输出对回原始 prompt 用、原始prompt、给教师模型的messages以及基线回答base_completion以便追溯。def teacher_messages(prompt: str) - list[dict[str, str]]: return [ {role: system, content: TEACHER_SYSTEM_PROMPT}, {role: user, content: prompt}, ] def step2_upload_raw_dataset(logs: list[dict[str, Any]]): path ARTIFACT_DIR / customer_raw_dataset.jsonl dataset_name fcustomer-support-raw-{uuid.uuid4().hex[:8]} rows, id_map [], {} with path.open(w) as handle: for entry in logs: record { custom_id: uuid.uuid4().hex, prompt: entry[prompt], messages: teacher_messages(entry[prompt]), base_completion: entry[completion] or , base_model: entry[model], } handle.write(json.dumps(record) \n) rows.append(record) id_map[record[custom_id]] entry[prompt] payload { name: dataset_name, schema: [ {name: custom_id, type: {name: string}}, {name: prompt, type: {name: string}}, {name: messages, type: {name: json}}, {name: base_completion, type: {name: string}}, {name: base_model, type: {name: string}}, ], folder: DATA_LAB_FOLDER, rows: rows, } dataset datalab_request(POST, /v1/datasets, json_bodypayload).json() print(Raw dataset ID:, dataset[id]) print(Raw dataset version:, dataset[current_version]) return dataset[id], dataset[current_version], id_map raw_dataset_id, raw_dataset_version, id_map step2_upload_raw_dataset(logs)datalab_request是对控制面 API 的统一封装CONTROL_URL Bearer 鉴权超时 60 秒非 2xx 会抛错。判断这一步成功的依据打印出Raw dataset ID和Raw dataset version。批量推理必须绑定 dataset 的id和version这两个值要保存下来——notebook 会把它们写入状态文件。本地同时落盘的customer_raw_dataset.jsonl里id_map依赖的custom_id与prompt映射是第四步恢复原始问题的来源。第三步用 70B 教师模型跑批量推理拿到 dataset 的 ID 和 version 后向 Data Lab 提交一个batch_inference类型的 operation。核心 payload 如下src指向刚上传的数据集mapping指定messages和custom_id两列、并把max_tokens固定为 1024params里指定TEACHER_MODEL和 24 小时的完成窗口。def step3_run_batch_inference(source_dataset_id, source_dataset_version): payload { type: batch_inference, src: [ { id: source_dataset_id, version: source_dataset_version, mapping: { type: text_messages, messages: {type: column, name: messages}, custom_id: {type: column, name: custom_id}, max_tokens: {type: text, value: 1024}, }, } ], dst: [], params: { model: TEACHER_MODEL, completion_window: BATCH_COMPLETION_WINDOW, }, } operation datalab_request(POST, /v1/operations, json_bodypayload).json() if not operation.get(dst): raise RuntimeError(fNo destination dataset returned: {json.dumps(operation, indent2)}) dst_dataset_id operation[dst][0][id] print(fOperation: {operation[id]} status{operation[status]}) print(Output dataset ID:, dst_dataset_id) while True: time.sleep(POLL_SECONDS) operation datalab_request(GET, f/v1/operations/{operation[id]}).json() print(Batch status:, operation[status]) if operation[status] in {succeeded, failed, cancelled, unknown}: break if operation[status] ! succeeded: raise RuntimeError(fBatch inference ended with status{operation[status]}) return operation[id], dst_dataset_id operation_id, output_dataset_id step3_run_batch_inference(raw_dataset_id, raw_dataset_version)批量推理是异步任务notebook 明确提示operation 会在queued或running状态停留一段时间才结束。代码每 30 秒轮询一次GET /v1/operations/{id}直到状态落入succeeded、failed、cancelled、unknown之一只有succeeded才继续其余状态直接抛出RuntimeError。成功后拿到operation_id和输出数据集output_dataset_id教师模型的结果就存在这个新的输出数据集里。第四步导出教师输出并改写成微调训练数据这是整个数据生成流程中价值最高的一步把教师模型的批量输出导出为 JSONL按custom_id对回原始 prompt丢弃弱回答最终写成每条样本都带 system / user / assistant 三个角色的对话式训练文件。def step4_download_and_curate(output_dataset_id: str, id_map: dict[str, str]) - Path: export_response datalab_request( GET, f/v1/datasets/{output_dataset_id}/export, params{format: jsonl}, ) raw_export_path ARTIFACT_DIR / customer_batch_output.jsonl raw_export_path.write_text(export_response.text) output_path ARTIFACT_DIR / customer_curated_training.jsonl records [json.loads(line) for line in export_response.text.strip().splitlines() if line.strip()] written 0 with output_path.open(w) as handle: for rec in records: prompt extract_prompt_from_record(rec, id_map) reply extract_reply_from_record(rec) if not prompt or not reply or len(reply) 50: continue sample { messages: [ {role: system, content: SUPPORT_SYSTEM_PROMPT}, {role: user, content: prompt}, {role: assistant, content: reply}, ] } handle.write(json.dumps(sample) \n) written 1 if not records: raise RuntimeError(fBatch output dataset {output_dataset_id} exported no rows.) if written 0: raise RuntimeError( Batch output export succeeded, but no usable assistant replies were found. fInspect {raw_export_path} for the returned schema. ) print(fCurated {written} examples - {output_path}) return output_path几个过滤与提取规则值得注意一行记录能保留的前提是能恢复出原始prompt、能提取出教师回答reply且len(reply) 50字符。短回答会被直接丢弃这决定了最终训练集的大小。extract_prompt_from_record优先用id_mapcustom_id→ prompt对回原始问题对不上时再从记录的messages/completed_dialogue/prompt字段里找 user 消息。extract_reply_from_record会依次尝试completion、assistant_response、output_text、text等字段以及response.body.choices这类 OpenAI 风格结构最后回退到completed_dialogue/messages里最后一条 assistant 消息。这套多路提取逻辑正是为了兼容批量推理返回的不同 schema——这也是为什么原始导出文件要完整落盘。成功的判断方式很直接终端打印Curated {written} examples - {output_path}且nebius-datalab-pipeline/scripts/artifacts/customer_curated_training.jsonl中每行都是{messages: [system, user, assistant]}结构。如果提示词全部丢失或导出行为 0step4会抛出带路径的错误信息让你去检查对应的导出文件。训练数据就绪后的下一步与排查生成出的customer_curated_training.jsonl就是微调阶段的输入。同一 notebook 的后续步骤把它通过 OpenAI 兼容的 Files API 以purposefine-tune上传再对BASE_MODELmeta-llama/Llama-3.1-8B-Instruct发起 LoRA 微调任务超参数为learning_rate2e-5、n_epochs2、loraTrue、lora_r16、lora_alpha16、lora_dropout0.05、packingTrue。notebook 特别提醒微调调用必须使用当前的 LoRA 字段名lora、lora_r、lora_alpha、lora_dropout旧字段名如lora_rank不会产生可部署的 LoRA 任务。之后再部署 adapter 并做 smoke test完整代码见 notebook 原文。数据链路出了问题时notebook 的 Troubleshooting 一节给出的排查方式随时用load_state()重新打开状态文件从断点步骤继续不必从头重跑调试数据问题时检查scripts/artifacts/下的三个文件customer_raw_dataset.jsonl上传内容、customer_batch_output.jsonl教师模型原始输出、customer_curated_training.jsonl最终训练集三者正好对应第四步的输入与输出若批量推理最终状态不是succeeded先看轮询打印的Batch status变化再对照operation_id在 Data Lab 侧排查若跳过前面步骤直接跑第四步id_map可以从落盘的customer_raw_dataset.jsonl里重新加载load_id_map_from_artifact()但该文件不存在时会提示先重跑第一步和第二步。这条流程的边界也要清楚示例只覆盖一个客服场景训练集大小受 50 字符过滤规则限制批量推理有 24 小时完成窗口且任务会排队。想适配账单、保修、欺诈或内部 IT helpdesk 等其他助手notebook 说明只需替换DOMAIN_TOPICS和两套策略提示词其余 Data Lab 上传、70B 批量推理与数据改写逻辑保持不变。【免费下载链接】Nebius-CookbookA collection of projects showcasing RAG, agents, workflows, and other AI use cases项目地址: https://gitcode.com/GitHub_Trending/ne/Nebius-Cookbook创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考