Spaces:
Sleeping
Sleeping
| """生成 SFT 训练数据（mlx-lm chat 格式 jsonl）。 | |
| 四类场景： | |
| 1. 行业知识问答 —— 由知识库切块 + 本地 LLM 生成 Q/A（教师模型：Ollama / mlx 后端任意） | |
| 2. 客服问答 + 工具调用 —— 程序化模板 + 真实工具执行结果（保证工具调用轨迹 100% 正确） | |
| 3. 单据/地址抽取 —— 程序化合成地址 → 规则解析器给标签 | |
| 4. 路径/调度 —— 程序化模板 + 真实工具结果 | |
| 输出：data/sft/train.jsonl, data/sft/valid.jsonl | |
| 用法：uv run python scripts/gen_sft.py --n-kb 200 --n-tool 300 --n-extract 150 --n-route 100 | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import json | |
| import random | |
| from pathlib import Path | |
| from logicllm import config | |
| from logicllm.agent.prompts import build_system, openai_tools | |
| from logicllm.llm.client import ChatClient | |
| from logicllm.rag.kb import split_markdown | |
| from logicllm.tools import all_specs, call_tool | |
| from logicllm.tools.geo import CITY_PROVINCE, CITY_DISTRICTS, province_full | |
| from logicllm.tools.waybill import init_db, _conn | |
| OUT = config.DATA_DIR / "sft" | |
| rnd = random.Random(7) | |
| SYSTEM = build_system() | |
| TOOLS = openai_tools(all_specs()) | |
| SYSTEM_TOOLS, SYSTEM_PLAIN = "tools", "plain" # 标记：是否附带 tools | |
| NAMES = ["张伟", "王芳", "李娜", "刘强", "陈静", "杨洋", "赵磊", "黄敏", "周杰", "吴丹", "徐丽", "孙浩", "马超", "朱琳", "胡军"] | |
| STREETS = ["人民路", "解放路", "中山路", "建设大道", "科技园路", "文三路", "南京东路", "天府大道", "金融街", "滨江路"] | |
| ITEMS = [("笔记本电脑", 2.5), ("衣服", 1.2), ("书籍", 3.0), ("零食", 1.8), ("化妆品", 0.6), ("手机", 0.4), ("茶叶", 1.0), ("鞋子", 1.5), ("文件", 0.2), ("小家电", 5.0)] | |
| _KB = None | |
| def _context_for(question: str) -> str | None: | |
| """与线上 Agent 相同的检索逻辑，把知识库上下文注入 system，保证训练/推理分布一致。""" | |
| global _KB | |
| from logicllm.rag.kb import KnowledgeBase | |
| if _KB is None: | |
| _KB = KnowledgeBase() | |
| if not _KB.available: | |
| return None | |
| hits = _KB.search(question) | |
| return KnowledgeBase.format(hits) if hits else None | |
| def sample(mode: str, turns: list[dict], rag_prob: float = 1.0) -> dict: | |
| """turns 里 assistant 的工具调用用 tc()、工具返回用 tr() 构造。 | |
| 线上推理几乎总会命中检索（min_score 0.35），所以所有样本默认都注入上下文——否则模型会学成"有上下文就不调工具"。""" | |
| q = turns[0]["content"] | |
| ctx = _context_for(q) if (mode == SYSTEM_PLAIN or rnd.random() < rag_prob) else None | |
| # 线上推理始终带工具，训练样本也一律带 | |
| return {"messages": [{"role": "system", "content": build_system(ctx)}, *turns], "tools": TOOLS} | |
| def tc(call: dict) -> dict: | |
| return {"role": "assistant", "content": "", "tool_calls": [{"type": "function", "function": {"name": call["name"], "arguments": call["arguments"]}}]} | |
| def tr(result: dict) -> dict: | |
| return {"role": "tool", "content": json.dumps(result, ensure_ascii=False)} | |
| # ---------- 1. 知识问答（教师模型生成） ---------- | |
| async def gen_kb(n: int, client: ChatClient) -> list[dict]: | |
| chunks = [] | |
| for p in sorted(config.KB_DIR.glob("*.md")): | |
| chunks.extend(split_markdown(p)) | |
| out = [] | |
| sem = asyncio.Semaphore(4) | |
| async def one(chunk): | |
| prompt = ( | |
| "你是物流培训讲师。根据下面的资料，生成 3 个用户可能提出的问题及其专业、准确、简洁的回答。" | |
| "问题要多样（客户口吻、员工口吻、术语解释等），回答必须只依据资料，不要编造。" | |
| "严格输出 JSON 数组：[{\"q\":\"...\",\"a\":\"...\"}]，不要输出其他内容。\n\n" | |
| f"资料标题：{chunk.title}\n资料：\n{chunk.text}" | |
| ) | |
| async with sem: | |
| try: | |
| txt = (await client.chat([{"role": "user", "content": prompt}], temperature=0.7, max_tokens=1500)).content | |
| except Exception as e: # noqa: BLE001 | |
| print("teacher error:", e) | |
| return [] | |
| s, e = txt.find("["), txt.rfind("]") | |
| try: | |
| arr = json.loads(txt[s:e + 1]) | |
| except Exception: # noqa: BLE001 | |
| return [] | |
| return [sample(SYSTEM_PLAIN, [{"role": "user", "content": x["q"]}, {"role": "assistant", "content": x["a"]}]) | |
| for x in arr if isinstance(x, dict) and x.get("q") and x.get("a")] | |
| while len(out) < n: | |
| batch = rnd.sample(chunks, min(len(chunks), max(1, (n - len(out)) // 3 + 1))) | |
| res = await asyncio.gather(*(one(c) for c in batch)) | |
| for r in res: | |
| out.extend(r) | |
| print(f" kb: {len(out)}/{n}") | |
| if not any(res): | |
| break | |
| return out[:n] | |
| # ---------- 2. 客服 + 工具调用 ---------- | |
| NO_SVC_NOTE = "您未指明服务类型，按默认的标准快递计算（如需特快/经济请告诉我）。\n" | |
| def _svc_prefix(svc_in_text: bool, svc: str) -> str: | |
| """用户说了服务类型就明确复述"按您指定的 X 计算"，没说才用默认说明——两种情形都给模型明确的开头，避免乱套模板。""" | |
| return f"按您指定的{svc}计算：\n" if svc_in_text else NO_SVC_NOTE | |
| EXC_TIPS = {"破损": "请保留外包装并拍照（外包装+内件），如已保价按声明价值赔付，未保价按实际损失赔付，上限为运费 3 倍。", | |
| "丢失": "轨迹停更超过 7 天可认定丢失；未保价按运费 3 倍赔偿（最高 300 元）并退还运费，已保价按声明价值赔付。", | |
| "延误": "特快超承诺时效 1 天以上可退还特快与标准快递的差价。", | |
| "地址错误": "派送前可联系修改，同城免费；跨城需补差价。"} | |
| def _query_answer(no: str, res: dict) -> str: | |
| exc = f"，当前标记有异常「{res['exception']}」，我们会尽快处理" if res.get("exception") else "" | |
| return (f"运单 {no} 的情况如下：\n- 路线：{res['origin']} → {res['destination']}\n- 服务：{res['service']}，重量 {res['weight_kg']}kg，运费 {res['fee']} 元\n" | |
| f"- 当前状态：**{res['status']}**{exc}\n- 下单时间：{res['created_at'].replace('T', ' ')}\n如需查看详细轨迹，我可以继续为您查询。") | |
| def _freight_answer(o: str, d: str, w: float, res: dict) -> str: | |
| return (f"{o} → {d}（{res['zone']}），计费重量 {res['billable_weight_kg']}kg" | |
| + (f"（体积重 {res['volume_weight_kg']}kg 大于实重）" if res["volume_weight_kg"] > w else "") | |
| + f"：\n- {res['breakdown']}\n- **合计 {res['total_fee']} 元**") | |
| def _exception_answer(no: str, typ: str, res: dict) -> str: | |
| return f"非常抱歉给您带来不便。已为运单 {no} 登记「{typ}」异常，工单号 **{res['ticket_no']}**，客服将在 48 小时内回访处理。\n{EXC_TIPS[typ]}" | |
| async def gen_tool(n: int) -> list[dict]: | |
| init_db() | |
| c = _conn() | |
| wbs = [dict(r) for r in c.execute("SELECT * FROM waybill").fetchall()] | |
| c.close() | |
| out = [] | |
| for _ in range(n): | |
| kind = rnd.choice(["query", "track", "freight", "eta", "search", "search", "exception", "ask_no", "create", | |
| "combo", "followup_svc", "ask_then_no", "complain_then_no", | |
| "eta_date", "exc_status", "phone_mask", "no_field"]) | |
| if kind == "query": | |
| w = rnd.choice(wbs) | |
| u = rnd.choice([f"帮我查一下运单 {w['no']}", f"{w['no']} 这个单子现在什么情况", f"查询 {w['no']}", f"我的快递 {w['no']} 到哪了"]) | |
| call = {"name": "query_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(**{"name": call["name"], "args": call["arguments"]}) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": _query_answer(w["no"], res)}])) | |
| elif kind == "track": | |
| w = rnd.choice(wbs) | |
| u = rnd.choice([f"{w['no']} 的物流轨迹", f"看看 {w['no']} 的运输记录", f"运单{w['no']}现在在哪个环节"]) | |
| call = {"name": "track_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| lines = "\n".join(f"- {e['ts'].replace('T', ' ')} {e['location']}：{e['event']}" for e in res["events"]) | |
| a = f"运单 {w['no']} 的轨迹：\n{lines}\n\n最新状态：{res['current']['event']}（{res['current']['location']}）。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "freight": | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| item, w = rnd.choice(ITEMS) | |
| w = round(w * rnd.uniform(0.6, 2.5), 1) | |
| dims = rnd.choice([None, (rnd.randint(20, 60), rnd.randint(15, 45), rnd.randint(10, 40))]) | |
| svc = rnd.choice(["标准快递", "特快", "经济"]) | |
| svc_in_text = rnd.random() > 0.3 | |
| if not svc_in_text: | |
| svc = "标准快递" | |
| dv = rnd.choice([0, 0, 500, 1000, 3000]) | |
| args = {"origin": o, "destination": d, "weight_kg": w, "service": svc} | |
| utxt = f"从{o}寄{item}到{d}，{w}kg" | |
| if dims: | |
| args.update(length_cm=dims[0], width_cm=dims[1], height_cm=dims[2]) | |
| utxt += f"，箱子 {dims[0]}×{dims[1]}×{dims[2]}cm" | |
| if dv: | |
| args["declared_value"] = dv | |
| utxt += f"，保价{dv}元" | |
| utxt += (f"，{svc}" if svc_in_text else "") + rnd.choice(["多少钱？", "运费怎么算？", "要花多少运费"]) | |
| call = {"name": "calc_freight", "arguments": args} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = _svc_prefix(svc_in_text, svc) + _freight_answer(o, d, w, res) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": utxt}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "eta": | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| svc = rnd.choice(["标准快递", "特快", "经济"]) | |
| svc_in_text = rnd.random() > 0.35 | |
| if not svc_in_text: | |
| svc = "标准快递" # 用户未指明时不得臆测，按默认标准快递并说明 | |
| u = rnd.choice([f"{o}到{d}{svc}几天能到？", f"今天从{o}发{svc}到{d}，什么时候到"]) if svc_in_text else rnd.choice([f"{o}寄{d}要多久", f"{o}到{d}几天能到"]) | |
| call = {"name": "estimate_eta", "arguments": {"origin": o, "destination": d, "service": svc}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = ("" if svc_in_text else "按默认的标准快递估算（特快可再快 1 天）：") + f"{o} → {d} 约 {res['distance_km']}km，{svc}预计 **{res['est_days']} 天**送达（约 {res['est_delivery_date']}）。" + (f"{res['note']}。" if res["note"] else "") + "实际以揽收后轨迹为准，恶劣天气可能延误。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "search": | |
| w = rnd.choice(wbs) | |
| ph = rnd.choice([w["sender_phone"], w["receiver_phone"]]) | |
| u = rnd.choice([f"我手机号 {ph}，帮我看看有哪些快递", f"用 {ph} 查一下我的运单"]) | |
| call = {"name": "search_waybills", "arguments": {"phone": ph}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| lines = "\n".join(f"- {x['no']}：{x['origin']} → {x['destination']}，{x['status']}" for x in res["waybills"]) | |
| a = f"手机号 {ph} 名下共 {res['count']} 个运单：\n{lines}\n需要查看某一单的详细轨迹请告诉我运单号。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "exception": | |
| w = rnd.choice(wbs) | |
| typ = rnd.choice(["破损", "丢失", "延误", "地址错误"]) | |
| first = {"破损": f"运单 {w['no']} 收到时外箱破损了，要投诉", "丢失": f"运单 {w['no']} 好多天没更新，应该是丢了，要投诉", | |
| "延误": f"运单 {w['no']} 超过承诺时效还没到，要投诉", "地址错误": f"运单 {w['no']} 地址填错了，要改"}[typ] | |
| u = rnd.choice([first, f"{w['no']} {typ}，怎么处理", f"我要给 {w['no']} 登记{typ}"]) | |
| call = {"name": "report_exception", "arguments": {"waybill_no": w["no"], "type": typ, "description": f"用户反馈{typ}"}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": _exception_answer(w["no"], typ, res)}])) | |
| elif kind == "ask_no": | |
| u = rnd.choice(["我的快递到哪了？", "帮我查下快递", "我的包裹怎么还没到", "查一下物流"]) | |
| a = "好的，请提供您的运单号（LL 开头的 12 位编号），或者告诉我寄/收件人手机号，我来帮您查询。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, {"role": "assistant", "content": a}])) | |
| elif kind == "combo": # 一句话同时问运费和时效：两个工具都要调，先算钱再算时效 | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| item, w = rnd.choice(ITEMS) | |
| w = round(w * rnd.uniform(0.6, 3.0), 1) | |
| svc = rnd.choice(["标准快递", "特快", "经济"]) | |
| svc_in_text = rnd.random() > 0.4 | |
| if not svc_in_text: | |
| svc = "标准快递" | |
| svc_txt = f"{svc}" if svc_in_text else "" | |
| u = rnd.choice([f"{o}寄{w}公斤{item}到{d}，{svc_txt}多少钱几天能到", f"从{o}发{item}到{d}，{w}kg，{svc_txt}运费和时效分别是多少", | |
| f"{o}到{d} {w}kg {item}，{svc_txt}要多少钱、什么时候能到"]) | |
| c1 = {"name": "calc_freight", "arguments": {"origin": o, "destination": d, "weight_kg": w, "service": svc}} | |
| r1 = await call_tool(c1["name"], c1["arguments"]) | |
| c2 = {"name": "estimate_eta", "arguments": {"origin": o, "destination": d, "service": svc}} | |
| r2 = await call_tool(c2["name"], c2["arguments"]) | |
| a = (_svc_prefix(svc_in_text, svc) + f"{o} → {d}（{r1['zone']}），计费重量 {r1['billable_weight_kg']}kg：\n" | |
| f"- 运费：{r1['breakdown']}，**合计 {r1['total_fee']} 元**\n" | |
| f"- 时效：约 {r2['distance_km']}km，{svc}预计 **{r2['est_days']} 天**送达（约 {r2['est_delivery_date']}）" + (f"，{r2['note']}" if r2["note"] else "") | |
| + "\n实际以揽收后轨迹为准。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(c1), tr(r1), tc(c2), tr(r2), {"role": "assistant", "content": a}])) | |
| elif kind == "followup_svc": # 多轮：先按默认标准快递报价，追问换服务类型要重新调工具 | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| item, w = rnd.choice(ITEMS) | |
| w = round(w * rnd.uniform(0.6, 2.5), 1) | |
| u1 = rnd.choice([f"从{o}寄{w}公斤{item}到{d}多少钱？", f"{o}到{d}，{item} {w}kg，运费多少"]) | |
| r1 = await call_tool("calc_freight", {"origin": o, "destination": d, "weight_kg": w, "service": "标准快递"}) | |
| a1 = NO_SVC_NOTE + _freight_answer(o, d, w, r1) | |
| new_svc = rnd.choice(["特快", "经济"]) | |
| u2 = rnd.choice([f"改成{new_svc}呢？", f"{new_svc}多少钱", f"换{new_svc}的话", f"那{new_svc}要多少"]) | |
| c2 = {"name": "calc_freight", "arguments": {"origin": o, "destination": d, "weight_kg": w, "service": new_svc}} | |
| r2 = await call_tool(c2["name"], c2["arguments"]) | |
| diff = round(r2["total_fee"] - r1["total_fee"], 2) | |
| a2 = (f"换成{new_svc}的话，{o} → {d}，计费重量 {r2['billable_weight_kg']}kg：\n- {r2['breakdown']}\n- **合计 {r2['total_fee']} 元**" | |
| + (f"，比标准快递贵 {diff} 元" if diff > 0 else f"，比标准快递便宜 {-diff} 元") + "。") | |
| # 历史轮次只保留最终答复文本（与线上会话历史一致，不含工具轨迹） | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u1}, {"role": "assistant", "content": a1}, | |
| {"role": "user", "content": u2}, tc(c2), tr(r2), {"role": "assistant", "content": a2}])) | |
| elif kind == "ask_then_no": # 多轮：先没给单号，追问后给了单号再查 | |
| w = rnd.choice(wbs) | |
| u1 = rnd.choice(["帮我查个运单", "查一下我的快递", "我想查物流", "帮我看看包裹到哪了"]) | |
| a1 = "好的，请提供您的运单号（LL 开头的 12 位编号），或者告诉我寄/收件人手机号，我来帮您查询。" | |
| u2 = rnd.choice([w["no"], f"运单号是 {w['no']}", f"{w['no']}，麻烦了", f"单号 {w['no']}"]) | |
| call = {"name": "query_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u1}, {"role": "assistant", "content": a1}, | |
| {"role": "user", "content": u2}, tc(call), tr(res), {"role": "assistant", "content": _query_answer(w["no"], res)}])) | |
| elif kind == "complain_then_no": # 多轮：先描述问题，再给单号登记异常 | |
| w = rnd.choice(wbs) | |
| typ = rnd.choice(["破损", "丢失", "延误"]) | |
| u1 = {"破损": rnd.choice(["我的快递收到的时候摔坏了，怎么办", "包裹外箱破了，里面东西也坏了"]), | |
| "丢失": rnd.choice(["我的快递好多天没动静了，是不是丢了", "包裹一直不更新物流，怎么办"]), | |
| "延误": rnd.choice(["我寄的特快已经超时了还没到", "快递比承诺时间晚了两天了"])}[typ] | |
| a1 = f"{EXC_TIPS[typ]}\n请提供运单号，我帮您登记「{typ}」异常工单，客服会在 48 小时内回访。" | |
| u2 = rnd.choice([f"运单号是 {w['no']}，帮我登记投诉", f"{w['no']}，麻烦登记一下", f"单号 {w['no']}，要投诉"]) | |
| call = {"name": "report_exception", "arguments": {"waybill_no": w["no"], "type": typ, "description": f"用户反馈{typ}"}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u1}, {"role": "assistant", "content": a1}, | |
| {"role": "user", "content": u2}, tc(call), tr(res), {"role": "assistant", "content": _exception_answer(w["no"], typ, res)}])) | |
| elif kind == "eta_date": # 用户给了寄出日期：必须把 ship_date 传给工具，送达日按该日期推算，而不是按"今天" | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| svc = rnd.choice(["标准快递", "特快", "经济"]) | |
| svc_in_text = rnd.random() > 0.4 | |
| if not svc_in_text: | |
| svc = "标准快递" | |
| month = rnd.randint(9, 12) | |
| day = rnd.randint(1, 28) | |
| ship = f"2026-{month:02d}-{day:02d}" | |
| date_txt = rnd.choice([f"{month} 月 {day} 日", f"{month}月{day}号", ship, f"2026 年 {month} 月 {day} 日"]) | |
| svc_txt = svc if svc_in_text else "" | |
| u = rnd.choice([f"{date_txt}从{o}寄{svc_txt}到{d}，哪天能到", f"我{date_txt}发{svc_txt}件，{o}到{d}，预计几号送达", | |
| f"{o}寄{d}，{date_txt}寄出，{svc_txt}什么时候到"]) | |
| call = {"name": "estimate_eta", "arguments": {"origin": o, "destination": d, "service": svc, "ship_date": ship}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = (("" if svc_in_text else "按默认的标准快递估算（特快可再快 1 天）：") | |
| + f"{date_txt}从{o}寄出，{o} → {d} 约 {res['distance_km']}km，{svc}预计 **{res['est_days']} 天**送达，即 **{res['est_delivery_date']}** 前后。" | |
| + (f"{res['note']}。" if res["note"] else "") + "实际以揽收后轨迹为准，恶劣天气可能延误。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "exc_status": # 只是问"有没有异常"：查运单即可，绝不能登记新异常（report_exception 只在用户明确要投诉/登记时用） | |
| w = rnd.choice(wbs) | |
| u = rnd.choice([f"{w['no']} 这单有没有异常", f"运单 {w['no']} 是不是出问题了", f"{w['no']} 有异常记录吗", f"帮我看看 {w['no']} 有没有什么问题"]) | |
| call = {"name": "query_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| if res.get("exception"): | |
| a = (f"运单 {w['no']} 当前标记有异常「{res['exception']}」，状态 {res['status']}（{res['origin']} → {res['destination']}，{res['service']}）。" | |
| f"网点正在处理；如需登记投诉工单或了解处理进度，请告诉我。") | |
| else: | |
| a = (f"运单 {w['no']} 目前没有异常记录，状态正常：**{res['status']}**（{res['origin']} → {res['destination']}，{res['service']}）。" | |
| f"如果您实际遇到了破损、丢失或延误等问题，告诉我可以帮您登记工单。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "phone_mask": # 问电话：查到后必须脱敏（隐藏中间四位、姓名部分字），并说明依据 | |
| w = rnd.choice(wbs) | |
| who = rnd.choice(["收件人", "寄件人"]) | |
| u = rnd.choice([f"{w['no']} {who}的电话是多少", f"查一下运单 {w['no']} 的{who}手机号", f"{w['no']} {who}联系方式", f"运单 {w['no']} 的{who}怎么联系"]) | |
| call = {"name": "query_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| ph = res["receiver_phone"] if who == "收件人" else res["sender_phone"] | |
| nm = res["receiver"] if who == "收件人" else res["sender"] | |
| masked, nm_masked = ph[:3] + "****" + ph[7:], nm[0] + "*" * (len(nm) - 1) | |
| a = (f"运单 {w['no']} 的{who}是 {nm_masked}，联系电话 {masked}。" | |
| f"按《个人信息保护法》和公司隐私规定，手机号中间四位需脱敏展示，完整号码请在有权限的内部系统中查看。" | |
| f"该运单当前状态：{res['status']}。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": a}])) | |
| elif kind == "no_field": # 问系统里根本没有的字段：如实说明没有，不编造原因 | |
| w = rnd.choice(wbs) | |
| ask, fld = rnd.choice([("司机是谁", "司机信息"), ("司机电话", "司机电话"), ("车牌号", "车牌号"), ("派送员的电话", "派送员电话"), | |
| ("快递员叫什么", "快递员姓名"), ("航班号", "航班/车次信息"), ("用的哪家航空公司", "承运航空公司")]) | |
| u = rnd.choice([f"{w['no']} 这单的{ask}", f"运单 {w['no']} 的{ask}是多少", f"帮我查 {w['no']} 的{ask}"]) | |
| call = {"name": "query_waybill", "arguments": {"waybill_no": w["no"]}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = (f"抱歉，运单系统里不记录{fld}，我无法提供。能查到的是运单 {w['no']} 的当前状态：**{res['status']}**（{res['origin']} → {res['destination']}，{res['service']}）。" | |
| f"如需联系派送环节，建议联系{res['destination']}的派送网点，或者我帮您登记一条咨询工单。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": a}])) | |
| else: # create | |
| o, d = rnd.sample(list(CITY_PROVINCE), 2) | |
| s, r = rnd.sample(NAMES, 2) | |
| sp, rp = f"13{rnd.randint(100000000, 999999999)}", f"15{rnd.randint(100000000, 999999999)}" | |
| item, w = rnd.choice(ITEMS) | |
| addr = f"{d}市{rnd.choice(['高新区', '中心区', '新华区'])}{rnd.choice(STREETS)}{rnd.randint(1, 300)}号" | |
| u = f"帮我下单：寄件人{s} {sp}，{o}；收件人{r} {rp}，{addr}；{item} {w}kg，标准快递" | |
| args = {"sender": s, "sender_phone": sp, "origin": o, "receiver": r, "receiver_phone": rp, | |
| "destination": d, "address": addr, "weight_kg": w, "service": "标准快递"} | |
| call = {"name": "create_waybill", "arguments": args} | |
| # 不真的写库，构造一个合理结果 | |
| fee = await call_tool("calc_freight", {"origin": o, "destination": d, "weight_kg": w}) | |
| res = {"waybill_no": f"LL2026{rnd.randint(100000, 999999)}{rnd.randint(10, 99)}", "fee": fee["total_fee"], "status": "已揽收", "created_at": "2026-08-22T10:00"} | |
| a = f"下单成功！运单号 **{res['waybill_no']}**，{o} → {d}，运费 {res['fee']} 元，当前状态：已揽收。快递员会在揽收时段上门取件，请保持电话畅通。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| return out | |
| # ---------- 3. 抽取 ---------- | |
| def gen_extract(n: int) -> list[dict]: | |
| from logicllm.tools.address import parse_address | |
| out = [] | |
| for _ in range(n): | |
| city = rnd.choice(list(CITY_DISTRICTS)) | |
| prov = CITY_PROVINCE[city] | |
| name = rnd.choice(NAMES) | |
| phone = f"1{rnd.choice('3589')}{rnd.randint(100000000, 999999999)}" | |
| # 电话写法变体：+86 / 空格 / 连字符分组，标签一律归一为 11 位连续数字 | |
| phone_txt = rnd.choice([phone, phone, phone, f"+86 {phone[:3]} {phone[3:7]} {phone[7:]}", f"{phone[:3]}-{phone[3:7]}-{phone[7:]}", f"{phone[:3]} {phone[3:7]} {phone[7:]}"]) | |
| dist = rnd.choice(CITY_DISTRICTS[city]) | |
| detail = f"{rnd.choice(STREETS)}{rnd.randint(1, 500)}号{rnd.choice(['', '3栋2单元501', 'A座1201', '科技园T3栋'])}" | |
| provtxt = "" if prov == city else rnd.choice([province_full(prov), prov, ""]) | |
| citytxt = f"{city}市" if rnd.random() < 0.7 else city | |
| item, w = rnd.choice(ITEMS) | |
| layout = rnd.choice([ | |
| f"收件人：{name} {phone_txt} {provtxt}{citytxt}{dist}{detail}", | |
| f"{provtxt}{citytxt}{dist}{detail}，{name}，{phone_txt}", | |
| f"{name} {phone_txt}\n{provtxt}{citytxt}{dist}{detail}\n物品：{item} {w}kg", | |
| f"请发到{provtxt}{citytxt}{dist}{detail}，收件人{name}，电话{phone_txt}，寄的是{item}", | |
| ]) | |
| u = rnd.choice(["提取收件信息：", "把下面的地址解析成结构化字段：", "帮我抽取这段文本中的收件人信息，输出JSON：", ""]) + layout | |
| label = {"name": name, "phone": phone, "province": prov, "city": city, "district": dist, "detail": detail, | |
| "item": item if item in layout else "", "weight_kg": w if f"{w}kg" in layout else ""} | |
| answer = "```json\n" + json.dumps(label, ensure_ascii=False, indent=2) + "\n```" | |
| if rnd.random() < 0.7: | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, {"role": "assistant", "content": answer}])) | |
| else: # 先调规则解析器再输出 JSON | |
| call = {"name": "parse_address", "arguments": {"text": layout}} | |
| res = parse_address(layout) | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": answer}])) | |
| return out | |
| # ---------- 4. 路径/调度 ---------- | |
| async def gen_route(n: int) -> list[dict]: | |
| out = [] | |
| cities = list(CITY_PROVINCE) | |
| for _ in range(n): | |
| if rnd.random() < 0.6: | |
| o, d = rnd.sample(cities, 2) | |
| via = rnd.sample([c for c in cities if c not in (o, d)], 1) if rnd.random() < 0.3 else None | |
| u = f"{o}到{d}的干线怎么走" + (f"，要经过{via[0]}" if via else "") + rnd.choice(["？", "，大概多少公里", "，需要开多久"]) | |
| args = {"origin": o, "destination": d} | |
| if via: | |
| args["via"] = via | |
| call = {"name": "plan_route", "arguments": args} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| if "error" in res: | |
| continue | |
| a = f"推荐路线：{' → '.join(res['path'])}\n- 总里程约 {res['distance_km']} km，{res['legs']} 段干线\n- 按 70km/h 均速估算行驶约 {res['est_drive_hours']} 小时，单司机需 {max(1, round(res['est_drive_hours'] / 8 + 0.49))} 天（每日驾驶 ≤ 8 小时）。" | |
| else: | |
| depot = rnd.choice(cities) | |
| stops = rnd.sample([c for c in cities if c != depot], rnd.randint(3, 6)) | |
| no_return = rnd.random() < 0.3 | |
| u = (f"从{depot}出发，要给{'、'.join(stops)}送货，怎么排顺序最省路？" if not no_return | |
| else rnd.choice([f"从{depot}出发送{'、'.join(stops)}，送完不回{depot}，怎么排", f"{depot}发车给{'、'.join(stops)}送货，单程不返回，顺序怎么排最省"])) | |
| call = {"name": "optimize_delivery_order", "arguments": {"depot": depot, "stops": stops, **({"return_to_depot": False} if no_return else {})}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| if "error" in res: | |
| continue | |
| a = (f"建议配送顺序：{' → '.join(res['sequence'])}" + ("（单程，不返回）" if no_return else "") | |
| + f"\n- 总里程约 {res['total_km']} km，行驶约 {res['est_drive_hours']} 小时\n- 采用最近邻 + 2-opt 优化；若各点有收货时间窗，请告诉我以便进一步调整。") | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), | |
| tr(res), {"role": "assistant", "content": a}])) | |
| return out | |
| # ---------- 5. 失败与边界轨迹 ---------- | |
| async def gen_failures(n: int) -> list[dict]: | |
| init_db() | |
| out = [] | |
| unknown_cities = ["洛阳", "三亚", "桂林", "唐山", "襄阳", "赣州", "遵义", "绵阳"] | |
| foreign_cities = ["纽约", "伦敦", "东京", "洛杉矶", "新加坡", "巴黎", "首尔", "悉尼"] | |
| for _ in range(n): | |
| kind = rnd.choice(["bad_no", "bad_no_track", "bad_city_eta", "bad_city_freight", "bad_city_route", "no_result_phone", "intl_freight"]) | |
| if kind in ("bad_no", "bad_no_track"): | |
| no = f"LL{rnd.randint(2027000000, 2029999999)}{rnd.randint(10, 99)}" | |
| tool_name = "query_waybill" if kind == "bad_no" else "track_waybill" | |
| u = rnd.choice([f"帮我查下 {no}", f"{no} 到哪了", f"查一下运单 {no} 的物流"]) | |
| call = {"name": tool_name, "arguments": {"waybill_no": no}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = f"抱歉，系统里没有找到运单 {no}。请核对运单号是否正确（本系统运单号为 LL 开头加 10 位数字），也可以告诉我寄/收件人手机号，我用手机号帮您查。" | |
| elif kind == "bad_city_eta": | |
| c1 = rnd.choice(unknown_cities); c2 = rnd.choice(list(CITY_PROVINCE)) | |
| u = f"{c1}到{c2}几天能到？" | |
| call = {"name": "estimate_eta", "arguments": {"origin": c1, "destination": c2, "service": "标准快递"}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| near = rnd.sample(list(CITY_PROVINCE), 2) | |
| a = f"抱歉，{c1} 暂不在系统收录的城市范围内，无法直接估算时效。您可以确认一下城市名，或者选择邻近的已开通城市（如 {near[0]}、{near[1]}）再帮您估算。" | |
| elif kind == "bad_city_freight": | |
| c1 = rnd.choice(list(CITY_PROVINCE)); c2 = rnd.choice(unknown_cities) | |
| item, w = rnd.choice(ITEMS) | |
| u = f"从{c1}寄{item}到{c2}，{w}kg 多少钱？" | |
| call = {"name": "calc_freight", "arguments": {"origin": c1, "destination": c2, "weight_kg": w}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = f"抱歉，{c2} 暂不在已开通的计费城市范围内，算不了这条线路的运费。请确认城市名是否有误，或告诉我具体想寄到哪个已开通城市。" | |
| elif kind == "intl_freight": # 境外目的地：工具返回未收录，不得编造总价 | |
| c1 = rnd.choice(list(CITY_PROVINCE)); c2 = rnd.choice(foreign_cities) | |
| item, w = rnd.choice(ITEMS) | |
| u = rnd.choice([f"从{c1}寄{item}到{c2}多少钱？", f"{c1}发{w}kg{item}去{c2}，运费怎么算", f"寄东西到{c2}要多少运费"]) | |
| args = {"origin": c1, "destination": c2, "weight_kg": w} if c1 in u else {"origin": "上海", "destination": c2, "weight_kg": w} | |
| call = {"name": "calc_freight", "arguments": args} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = (f"抱歉，{c2} 是境外城市，本系统的运费计算只覆盖国内已开通城市，无法直接报价。" | |
| "国际件请走国际业务：国际快递（DHL/FedEx/UPS）约 3–7 天，适合文件和样品；空运 5–10 天适合批量货；海运最便宜但需 20–45 天。" | |
| "告诉我货物类型、重量和时效要求，我可以帮您选合适的方式。") | |
| elif kind == "bad_city_route": | |
| c1 = rnd.choice(unknown_cities); c2 = rnd.choice(list(CITY_PROVINCE)) | |
| u = f"{c1}到{c2}的干线怎么走？" | |
| call = {"name": "plan_route", "arguments": {"origin": c1, "destination": c2}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = f"抱歉，路网数据里没有收录 {c1}，无法规划这条干线。如果您指的是其他城市请更正一下；也可以从收录的邻近枢纽城市出发规划。" | |
| else: # no_result_phone | |
| ph = f"13{rnd.randint(100000000, 999999999)}" | |
| u = rnd.choice([f"手机号 {ph} 名下有哪些运单？", f"用 {ph} 帮我查下快递"]) | |
| call = {"name": "search_waybills", "arguments": {"phone": ph}} | |
| res = await call_tool(call["name"], call["arguments"]) | |
| a = f"手机号 {ph} 名下没有查到运单。请确认号码是否为下单时预留的手机号，或直接提供运单号（LL 开头 12 位）我来查询。" | |
| out.append(sample(SYSTEM_TOOLS, [{"role": "user", "content": u}, tc(call), tr(res), {"role": "assistant", "content": a}])) | |
| return out | |
| # ---------- 6. 数值套档（无城市，纯规则问答） ---------- | |
| # 答案全部由与 estimate_eta / calc_freight 相同的规则算出；分界点附近加密采样，专治"1600km 答 3 天"这类套错档 | |
| ETA_TIERS = [(500, 2, "≤500km"), (1500, 3, "500–1500km"), (2500, 4, "1500–2500km"), (float("inf"), 5, ">2500km")] | |
| ETA_ADJ = {"标准快递": 0, "特快": -1, "经济": 2, "零担": 3} | |
| def _n(x: float) -> str: | |
| return f"{x:g}" if float(x).is_integer() else f"{round(x, 2):g}" | |
| def _eta_days(d: int, svc: str, remote: bool) -> tuple[int, int, str]: | |
| base, label = next((days, lab) for ub, days, lab in ETA_TIERS if d <= ub) | |
| return max(1, base + (2 if remote else 0) + ETA_ADJ[svc]), base, label | |
| def _tier_chain(d: int) -> str: | |
| """把里程与 500/1500/2500 逐个比较写出来，让模型先比大小再套档，而不是凭感觉猜档。""" | |
| return "，".join(f"{d} {'>' if d > b else '≤'} {b}" for b in (500, 1500, 2500)) | |
| def _near_boundary(bounds: list[int], lo: int, hi: int) -> int: | |
| if rnd.random() < 0.7: | |
| b = rnd.choice(bounds) | |
| return max(lo, b + rnd.choice([0, 0, -1, 1]) * rnd.choice([10, 20, 50, 80, 100, 150])) | |
| return rnd.randint(lo, hi) // 10 * 10 | |
| def gen_numeric(n: int) -> list[dict]: | |
| from logicllm.tools.pricing import ADD_KG, FIRST_KG, SERVICE_MULT, VOLUME_DIVISOR | |
| import math | |
| out = [] | |
| for _ in range(n): | |
| kind = rnd.choice(["eta", "eta", "eta", "eta", "eta_cmp", "zone", "zone", "freight", "freight", "volume", "insurance", "oversize", "alcohol", "alcohol", "ltl", "ltl"]) | |
| if kind == "eta": | |
| d = _near_boundary([500, 1500, 2500], 60, 4000) | |
| svc = rnd.choice(["", "", "标准快递", "特快", "经济", "零担"]) | |
| remote = rnd.random() < 0.2 | |
| days, base, label = _eta_days(d, svc or "标准快递", remote) | |
| where = rnd.choice(["新疆", "西藏", "青海", "内蒙古", "甘肃", "宁夏"]) if remote else "" | |
| q = rnd.choice([f"{d} 公里{svc}大概几天到？", f"{d}km 的线路{svc}要几天？", f"两地相距 {d} 公里，{svc or '快递'}几天能到", | |
| f"寄到 {d} 公里外{svc}多久能收到？"]) | |
| if remote: | |
| q = q.rstrip("？") + f"，收件地在{where}？" | |
| steps = [f"里程 {d}km 逐档比较：" + _tier_chain(d) + f" → {label} 档，标准快递 {base} 天"] | |
| if remote: | |
| steps.append(f"{where}属偏远六省区，+2 天") | |
| if svc and svc != "标准快递": | |
| steps.append({"特快": "特快 −1 天（最少 1 天）", "经济": "经济 +2 天", "零担": "零担 +3 天"}[svc]) | |
| a = ("；".join(steps) + f" → **{svc or '标准快递'}约 {days} 天**。" | |
| + ("" if svc else "（未指明服务类型，按标准快递计算；特快可快 1 天。）") | |
| + ("边界值归低一档：500/1500/2500km 整分别是 2/3/4 天。" if any(abs(d - b) <= 150 for b in (500, 1500, 2500)) else "")) | |
| elif kind == "eta_cmp": | |
| b = rnd.choice([500, 1500, 2500]) | |
| d1, d2 = b - rnd.choice([20, 50, 100, 150]), b + rnd.choice([20, 50, 100, 150]) | |
| if rnd.random() < 0.5: | |
| d1, d2 = d2, d1 | |
| (t1, _, l1), (t2, _, l2) = _eta_days(d1, "标准快递", False), _eta_days(d2, "标准快递", False) | |
| q = rnd.choice([f"{d1} 公里和 {d2} 公里时效一样吗？", f"{d1}km 跟 {d2}km 标准快递差几天？"]) | |
| a = (f"分界线是 {b}km：{d1} {'>' if d1 > b else '≤'} {b} → {l1} 档 {t1} 天；{d2} {'>' if d2 > b else '≤'} {b} → {l2} 档 {t2} 天。" | |
| f"**不一样，相差 {abs(t1 - t2)} 天**（{min(d1, d2)}km {min(t1, t2)} 天，{max(d1, d2)}km {max(t1, t2)} 天）。") | |
| elif kind == "zone": | |
| d = _near_boundary([800], 100, 3000) | |
| zone = "邻近" if d <= 800 else "跨区" | |
| q = rnd.choice([f"跨省寄件，两地 {d} 公里，按什么区域计费？", f"{d}km 跨省算邻近还是跨区？"]) | |
| a = (f"跨省件看 800km 分界线：{d} {'>' if d > 800 else '≤'} 800 → **按{zone}计费**" | |
| f"（{'首重 12 元、续重 6 元/kg' if zone == '跨区' else '首重 10 元、续重 4 元/kg'}）。" | |
| f"规则：跨省且里程 ≤800km 为邻近，>800km 为跨区，800km 整算邻近；同省不论距离都按同省（首重 8 元、续重 2 元/kg）。") | |
| elif kind in ("freight", "volume"): | |
| zone = rnd.choice(list(FIRST_KG)) | |
| svc = rnd.choice(["标准快递", "标准快递", "特快", "经济"]) | |
| w = round(rnd.uniform(0.3, 12), 1) | |
| dims = (rnd.randint(20, 70), rnd.randint(15, 50), rnd.randint(10, 45)) if (kind == "volume" or rnd.random() < 0.4) else None | |
| vol = dims[0] * dims[1] * dims[2] / VOLUME_DIVISOR if dims else 0 | |
| bill = math.ceil(max(w, vol) * 2) / 2 | |
| base = (FIRST_KG[zone] + max(0, bill - 1) * ADD_KG[zone]) * SERVICE_MULT[svc] | |
| box = f"，箱子 {dims[0]}×{dims[1]}×{dims[2]}cm" if dims else "" | |
| if kind == "volume": | |
| q = rnd.choice([f"{w}kg 的包裹{box}，计费重量是多少？", f"实重 {w}kg{box}，按多少公斤收费？"]) | |
| a = (f"计费重量 **{_n(bill)}kg**。体积重 = {dims[0]}×{dims[1]}×{dims[2]}÷6000 ≈ {round(vol, 2)}kg，" | |
| f"取实重与体积重较大者 {round(max(w, vol), 2)}kg，再向上取整到 0.5kg。") | |
| else: | |
| q = rnd.choice([f"{zone}件 {w}kg{box}，{svc}运费多少？", f"按{zone}计费，{w}公斤{box}，{svc}要多少钱？"]) | |
| steps = [] | |
| if dims: | |
| steps.append(f"体积重 {dims[0]}×{dims[1]}×{dims[2]}÷6000 ≈ {round(vol, 2)}kg") | |
| steps.append(f"计费重量 {_n(bill)}kg（取实重与体积重较大者，向上取整到 0.5kg）") | |
| steps.append(f"{zone}：首重 {_n(FIRST_KG[zone])} 元 + 续重 {_n(max(0, bill - 1))}kg×{_n(ADD_KG[zone])} 元" | |
| + (f"，×{SERVICE_MULT[svc]}（{svc}）" if svc != "标准快递" else "")) | |
| a = "；".join(steps) + f"，**合计 {_n(round(base, 2))} 元**（不含偏远、保价等附加费）。" | |
| elif kind == "insurance": | |
| v = rnd.choice([50, 100, 150, 199, 200, 201, 300, 500, 800, 1000, 2000, 3000, 5000, 8000, 10000]) | |
| fee = max(1.0, v * 0.005) | |
| q = rnd.choice([f"保价 {v} 元要交多少保价费？", f"声明价值 {v} 元，保价费怎么算？"]) | |
| a = (f"规则是声明价值 × 0.5%，最低 1 元：{v}×0.5% = {_n(round(v * 0.005, 2))} 元" | |
| + ("，不足 1 元按 1 元收" if v * 0.005 < 1 else "") + f" → **保价费 {_n(round(fee, 2))} 元**。") | |
| elif kind == "alcohol": # 酒精度三档：≤24 不限 / 24–70 限寄 / >70 禁寄；酒精制品（非酒类饮品）≥24% 直接按易燃易爆禁寄 | |
| deg = rnd.choice([10, 15, 20, 23, 24, 25, 30, 38, 42, 45, 50, 52, 53, 56, 60, 65, 68, 70, 71, 72, 75, 80, 90, 95, 99]) | |
| is_drink = rnd.random() < 0.7 | |
| thing = rnd.choice(["酒", "白酒", "米酒", "黄酒", "洋酒", "威士忌"]) if is_drink else rnd.choice(["酒精", "医用酒精", "酒精消毒液", "酒精喷雾"]) | |
| q = rnd.choice([f"{deg} 度的{thing}能寄吗？", f"{deg}% 的{thing}可以走快递吗？", f"酒精浓度 {deg}% 的{thing}寄得了吗", f"{thing}，{deg} 度，能不能寄"]) | |
| cmp24, cmp70 = (">" if deg > 24 else "≤"), (">" if deg > 70 else "≤") | |
| chain = f"{deg} {cmp24} 24，{deg} {cmp70} 70" | |
| if not is_drink and deg >= 24: | |
| a = (f"{thing}是酒精制品而不是酒类饮品，规则是酒精浓度 ≥24% 即按易燃易爆品禁寄：{deg} ≥ 24 → **不能寄**" | |
| + ("；而且任何超过 70% 的酒精液体都禁寄" if deg > 70 else "") + "。") | |
| elif deg <= 24: | |
| a = f"酒精度分三档（≤24 不限；24–70 限寄；>70 禁寄）。{chain} → ≤24% 档 → **能寄，不限量**，密封防漏、正常防碎包装即可。" | |
| elif deg <= 70: | |
| a = (f"酒精度分三档（≤24 不限；24–70 限寄；>70 禁寄）。{chain} → 24%–70% 档 → **能寄，有条件**：每件不超过 5L、只能陆运、需防碎包装。" | |
| + ("70 度整仍在区间内，超过 70 才禁寄。" if deg == 70 else "")) | |
| else: | |
| a = f"酒精度分三档（≤24 不限；24–70 限寄；>70 禁寄）。{chain} → >70% 档 → **不能寄**，属高浓度易燃液体，禁寄。" | |
| elif kind == "ltl": # 零担门槛 ≥30kg | |
| w = rnd.choice([5, 10, 15, 20, 25, 28, 29, 29.5, 30, 31, 35, 40, 50, 80, 120, 200]) | |
| q = rnd.choice([f"{_n(w)} 公斤的货能走零担吗？", f"{_n(w)}kg 可以发零担吗", f"{_n(w)} 公斤走零担还是快递？"]) | |
| if w >= 30: | |
| a = f"零担（不足整车、多票拼车）在本公司的起运门槛是单票 ≥30kg：{_n(w)} ≥ 30 → **能走零担**，按标准快递运费 ×0.5 计收，时效比标准快递慢 3 天，适合不急的大件。" | |
| else: | |
| a = f"零担（不足整车、多票拼车）在本公司的起运门槛是单票 ≥30kg：{_n(w)} < 30 → **不能走零担**，请走快递：标准快递，或特快（×1.6，快 1 天）、经济（×0.8，慢 2 天）。" | |
| else: # oversize | |
| if rnd.random() < 0.5: | |
| w = rnd.choice([45, 48, 49.5, 50, 50.5, 51, 52, 55, 60]) | |
| hit = w > 50 | |
| q = rnd.choice([f"单件 {_n(w)} 公斤要加超重操作费吗？", f"{_n(w)}kg 的货算超重吗？"]) | |
| a = (f"超重线是单件 > 50kg（另有单边 > 150cm 的超长线）：{_n(w)} {'>' if hit else '≤'} 50 → " | |
| + ("**要加收 30 元操作费**。" if hit else "**不收超重费**（恰好 50kg 不算超）。")) | |
| else: | |
| L = rnd.choice([120, 140, 149, 150, 151, 155, 160, 180]) | |
| hit = L > 150 | |
| q = rnd.choice([f"箱子最长边 {L}cm 要加钱吗？", f"单边 {L} 厘米算超长吗？"]) | |
| a = (f"超长线是单边 > 150cm（另有单件 > 50kg 的超重线）：{L} {'>' if hit else '≤'} 150 → " | |
| + ("**要加收 30 元操作费**。" if hit else "**不收超长费**（恰好 150cm 不算超）。")) | |
| out.append(sample(SYSTEM_PLAIN, [{"role": "user", "content": q}, {"role": "assistant", "content": a}])) | |
| return out | |
| async def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--n-kb", type=int, default=200) | |
| ap.add_argument("--n-tool", type=int, default=300) | |
| ap.add_argument("--n-extract", type=int, default=150) | |
| ap.add_argument("--n-route", type=int, default=100) | |
| ap.add_argument("--n-fail", type=int, default=60) | |
| ap.add_argument("--n-numeric", type=int, default=0, help="数值套档问答（时效/计费区域/运费/体积重/保价/超重）") | |
| ap.add_argument("--valid-ratio", type=float, default=0.08) | |
| ap.add_argument("--seed", type=int, default=7, help="随机种子；换种子即得到一批新样本") | |
| ap.add_argument("--out", default=str(OUT), help="输出目录") | |
| ap.add_argument("--skip-kb", action="store_true", help="不用教师模型生成知识问答") | |
| ap.add_argument("--kb-cache", default=str(config.DATA_DIR / "seed" / "teacher_kb_qa.json"), help="已生成的知识问答缓存，存在则直接复用") | |
| a = ap.parse_args() | |
| rnd.seed(a.seed) | |
| # report_exception 会写库：在临时的全新演示库上生成，既不污染 data/waybills.sqlite，也避免历次生成的异常轨迹累积进样本 | |
| import tempfile | |
| from logicllm.tools import waybill | |
| waybill.DB_PATH = Path(tempfile.mkdtemp()) / "waybills.sqlite" | |
| out_dir = Path(a.out) | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| data = [] | |
| print("生成工具调用样本…"); data += await gen_tool(a.n_tool) | |
| print("生成抽取样本…"); data += gen_extract(a.n_extract) | |
| print("生成路径样本…"); data += await gen_route(a.n_route) | |
| print("生成失败/边界样本…"); data += await gen_failures(a.n_fail) | |
| if a.n_numeric: | |
| print("生成数值套档样本…"); data += gen_numeric(a.n_numeric) | |
| cache = Path(a.kb_cache) | |
| if not a.skip_kb and a.n_kb > 0 and cache.exists(): | |
| pairs = json.load(open(cache, encoding="utf-8")) | |
| print(f"复用缓存知识问答 {len(pairs)} 条：{cache}") | |
| data += [sample(SYSTEM_PLAIN, [{"role": "user", "content": x["q"]}, {"role": "assistant", "content": x["a"]}]) for x in pairs] | |
| elif not a.skip_kb and a.n_kb > 0: | |
| client = ChatClient() | |
| if await client.healthy(): | |
| print(f"用教师模型 {client.model} 生成知识问答…") | |
| kb_samples = await gen_kb(a.n_kb, client) | |
| data += kb_samples | |
| json.dump([{"q": x["messages"][1]["content"], "a": x["messages"][2]["content"]} for x in kb_samples], | |
| open(cache, "w", encoding="utf-8"), ensure_ascii=False, indent=0) | |
| else: | |
| print("⚠️ LLM 后端不可用，跳过知识问答生成") | |
| await client.aclose() | |
| # 手写高质量种子 | |
| for p in (config.DATA_DIR / "seed").glob("*.jsonl"): | |
| for l in p.read_text(encoding="utf-8").splitlines(): | |
| if l.strip(): | |
| d = json.loads(l) | |
| data.append(sample(SYSTEM_PLAIN, d["messages"][1:])) # 手写种子也注入检索上下文 | |
| # 会话级去重（避免同一轨迹既进训练又进验证） | |
| seen, uniq = set(), [] | |
| for d in data: | |
| key = json.dumps(d["messages"], ensure_ascii=False, sort_keys=True) | |
| if key not in seen: | |
| seen.add(key) | |
| uniq.append(d) | |
| print(f"会话去重：{len(data)} → {len(uniq)}") | |
| # 按用户问题分组后切分训练/验证，再在各自分区内展开工具调用前缀 | |
| # （mlx_lm --mask-prompt 只训最后一条 assistant，前缀样本让模型学会发出调用； | |
| # 同一问题的会话/前缀必须落在同一分区，否则验证损失虚低） | |
| groups: dict = {} | |
| for d in uniq: | |
| groups.setdefault(d["messages"][1]["content"], []).append(d) | |
| gkeys = list(groups) | |
| rnd.shuffle(gkeys) | |
| valid_sessions, train_sessions, nv = [], [], max(10, int(len(uniq) * a.valid_ratio)) | |
| for k in gkeys: | |
| (valid_sessions if len(valid_sessions) < nv else train_sessions).extend(groups[k]) | |
| def expand(sessions): | |
| out_rows = list(sessions) | |
| for d in sessions: | |
| msgs = d["messages"] | |
| for i, m in enumerate(msgs): | |
| if m["role"] == "assistant" and m.get("tool_calls") and i < len(msgs) - 1: | |
| out_rows.append({**d, "messages": msgs[:i + 1]}) | |
| return out_rows | |
| train_rows, valid_rows = expand(train_sessions), expand(valid_sessions) | |
| rnd.shuffle(train_rows) | |
| with open(out_dir / "valid.jsonl", "w", encoding="utf-8") as f: | |
| for x in valid_rows: | |
| f.write(json.dumps(x, ensure_ascii=False) + "\n") | |
| with open(out_dir / "train.jsonl", "w", encoding="utf-8") as f: | |
| for x in train_rows: | |
| f.write(json.dumps(x, ensure_ascii=False) + "\n") | |
| print(f"完成：train {len(train_rows)} 条（会话 {len(train_sessions)}），valid {len(valid_rows)} 条（会话 {len(valid_sessions)}）→ {out_dir}") | |
| if __name__ == "__main__": | |
| asyncio.run(main()) | |