企业官网建设流程全解析

如何用 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_KEY；
• Python 环境中已安装openai和requests：

pip 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( model=BASE_MODEL, messages=[ {"role": "system", "content": SUPPORT_SYSTEM_PROMPT}, {"role": "user", "content": topic}, ], max_tokens=256, ) 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_url=BASE_URL, api_key=API_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 = f"customer-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_body=payload).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固定为 1024；params里指定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_body=payload).json() if not operation.get("dst"): raise RuntimeError(f"No destination dataset returned: {json.dumps(operation, indent=2)}") dst_dataset_id = operation["dst"][0]["id"] print(f"Operation: {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(f"Batch 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(f"Batch output dataset {output_dataset_id} exported no rows.") if written == 0: raise RuntimeError( "Batch output export succeeded, but no usable assistant replies were found. " f"Inspect {raw_export_path} for the returned schema." ) print(f"Curated {written} examples -> {output_path}") return output_path

几个过滤与提取规则值得注意：

• 一行记录能保留的前提是：能恢复出原始prompt、能提取出教师回答reply，且len(reply) >= 50字符。短回答会被直接丢弃，这决定了最终训练集的大小。
• extract_prompt_from_record优先用id_map（custom_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]}结构。如果提示词全部丢失或导出行为 0，step4会抛出带路径的错误信息，让你去检查对应的导出文件。

训练数据就绪后的下一步与排查

生成出的customer_curated_training.jsonl就是微调阶段的输入。同一 notebook 的后续步骤把它通过 OpenAI 兼容的 Files API 以purpose="fine-tune"上传，再对BASE_MODEL（meta-llama/Llama-3.1-8B-Instruct）发起 LoRA 微调任务，超参数为learning_rate=2e-5、n_epochs=2、lora=True、lora_r=16、lora_alpha=16、lora_dropout=0.05、packing=True。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