import json import time import urllib.request URL = "http://112.82.245.194:41434/v1/chat/completions" MODEL = "qwen3.6-27b-iq3:latest" # (utterance, fine-grained expectation for agent judge) CASES = [ ("有多少个客户？", "查询"), ("必胜客的销售员是谁？", "查询"), ("列出所有供应商", "查询"), ("查一下这个月的送货单", "查询"), ("库存数量最多的前5个物料是什么？", "查询"), ("上个月销售额最高的客户是谁？", "查询"), ("报价纸盒", "新增"), ("给苏州华为报价彩盒1000个", "新增"), ("纸盒5000个多少钱？", "新增"), ("帮我新建一个客户，名字叫上海创远包装", "新增"), ("把必胜客的电话改成13800138000", "修改"), ("修改客户地址", "修改"), ("删除报价单BJD202607001", "删除"), ("作废昨天那张送货单", "删除"), ("审核一下报价单BJD202607082", "审核"), ("你好", "闲聊"), ("谢谢，辛苦了", "闲聊"), ("报价", "不清楚"), ("纸盒", "不清楚"), ("查询报价单有哪些", "查询"), ("上海创远包装的地址是什么？", "查询"), ("这个月一共开了几张销售订单？", "查询"), ("查一下BJD202607001这张报价单", "查询"), ("铜版纸的库存还有多少？", "查询"), ("有哪些客户是上海的？", "查询"), ("最近有没有新的采购订单？", "查询"), ("必胜客上个月下了几单？", "查询"), ("看看今天的生产任务", "查询"), ("应收账款总共多少钱？", "查询"), ("供应商里有没有叫金田纸业的？", "查询"), ("画册1000本什么价格？", "新增"), ("给必胜客做5000个披萨盒报个价", "新增"), ("新建一个供应商：杭州新新材料", "新增"), ("帮我录一张报价单", "新增"), ("添加物料：300g白卡纸", "新增"), ("小册子500本多少钱", "新增"), ("来一单纸箱报价，规格60*40*30", "新增"), ("把上海创远包装的地址改成浦东新区川沙路500号", "修改"), ("必胜客的销售员换成王五", "修改"), ("供应商金田纸业的电话改一下，改成0571-88881234", "修改"), ("帮我改一下那张报价单", "修改"), ("把BJD202607001的数量改成8000", "修改"), ("把BJD202607001这张报价单作废", "删除"), ("那张给必胜客的报价单不要了，作废掉", "删除"), ("删掉刚才建错的那个客户", "删除"), ("把BJD202607082审核通过", "审核"), ("那张报价单审核错了，帮我反审核", "审核"), ("送货单DH202607005需要审核一下", "审核"), ("你是谁？你能做什么？", "闲聊"), ("送货单", "不清楚"), ] # fine label -> 4-class gate expectation GATE_MAP = {"查询": "查询", "新增": "新增", "修改": "操作已有单据", "删除": "操作已有单据", "审核": "操作已有单据", "闲聊": "其他", "不清楚": "其他"} READ_TOOLS = {"findForms", "readFormData", "lookupRecord"} # ---------------- 意图门（生产新版逐字复刻：IntentService.SYSTEM） ---------------- INTENT_SYSTEM = ( "你是印刷/包装 ERP 的意图与实体抽取器。把一句用户话分类为四种意图之一，并抽取句中实体及其业务角色。\n" "意图定义：\n" "- 查询：查看、统计或查找系统里**已有**的数据/单据。\n" "- 新增：要创建一张新单据或一条新记录。行业常识：对定制产品问价格（多少钱/什么价/报个价）" "就是要**新建报价单**——价格由系统核价算出，没有现成价；只有给了单号或明确说查已有报价才算查询。\n" "- 操作已有单据：对系统里已存在的单据/记录做修改、作废/删除、审核/反审核、复原。" "只要动作属于这些，即使还没说具体哪条记录也算本类（缺的信息之后澄清）。\n" "- 其他：闲聊、问候、与 ERP 业务无关，或只说名词、没有任何动作、无法判断想做什么。\n" "实体角色：客户=购买方的公司/单位名；产品=要生产/加工/报价的物品（如纸盒、彩盒、画册）；其余按字面。\n" "danju=该意图针对的单据或实体类型：新增时=要新建的单据类型（如问价即 报价），" "查询/操作时=要查或要操作的单据/实体（如 客户/销售订单）；确实没有才留空。" "missing=完成该意图还缺的关键信息。只输出 JSON，不要解释。" ) INTENT_SCHEMA = { "type": "object", "properties": { "intent": {"type": "string", "enum": ["查询", "新增", "操作已有单据", "其他"]}, "danju": {"type": "string"}, "entities": {"type": "array", "items": { "type": "object", "properties": { "value": {"type": "string"}, "role": {"type": "string", "enum": ["客户", "产品", "物料", "供应商", "数量", "尺寸", "日期", "金额", "其他", "未知"]}, }, "required": ["value", "role"], }}, "missing": {"type": "array", "items": {"type": "string"}}, }, "required": ["intent", "danju", "entities"], } # ---------------- 统一 agent prompt（生产新版逐字复刻：SystemPromptService.render） ---------------- DOMAIN_MAP = """- 生产（生产智能体，585 张单据） 上游←[其他，品质，基础资料，报价，销售] 下游→[其他，发外，品质，库存，报价，采购，销售] - 销售（销售智能体，325 张单据） 上游←[其他，库存，报价，生产] 下游→[其他，发外，品质，库存，报价，生产] - 基础资料（ERP代理人，234 张单据） 下游→[报价，生产] - 其他（ERP代理人，134 张单据） 上游←[库存，生产，采购，销售] 下游→[发外，品质，生产，财务，采购，销售] - 库存（库存智能体，134 张单据） 上游←[品质，生产，财务，采购，销售] 下游→[其他，发外，品质，销售] - 报价（报价智能体，99 张单据） 上游←[基础资料，生产，销售] 下游→[生产，销售] - 品质（品质智能体，63 张单据） 上游←[其他，发外，库存，生产，财务，采购，销售] 下游→[发外，库存，生产，采购] - 采购（采购智能体，59 张单据） 上游←[其他，品质，生产] 下游→[其他，品质，库存，财务] - 发外（发外智能体，51 张单据） 上游←[其他，品质，库存，生产，销售] 下游→[品质] - 财务（财务智能体，51 张单据） 上游←[其他，采购] 下游→[品质，库存] - 设备（ERP代理人，13 张单据） """ HEADER = ("【硬性要求】必须始终用**简体中文**回答；严禁输出英文/泰文等非中文。**只给最终答复**，" "不要复述你在调用哪个工具、不要输出思考过程或过程性旁白。\n\n" "你是「小羚羊」，小羚羊印刷 ERP 的智能助手，服务印刷/包装行业的企业用户，帮他们查询和操作 ERP 业务单据。\n") UNIFIED_PROMPT = HEADER + f""" 【业务域地图（先据此判断问题属于哪个域、涉及哪些单据）】 {DOMAIN_MAP} 【业务常识】 - 对定制产品问价格（多少钱/什么价/报个价）= 要**新建一张报价单**：价格由系统核价算出，系统里没有现成价。只有给了单号、或明确说查已有报价/统计报价额，才是查询。 - 业务单据的「删除/取消」= **作废**（proposeWrite action=invalid，可复原），不要物理删除。 【可用工具（只有这些）】 - findForms(keyword)：按关键词把「某类单据/报表」定位到具体表单，拿到 formId 与 moduleId。 - readFormData(formId, moduleId, keyword?, page?)：读某表单的真实数据（每页若干行+总条数）。问数量/概况时 keyword 留空；找某个名称的记录才填 keyword；用户要看下一页时 page 递增。 - lookupRecord(entityKeyword, recordKeyword)：查某实体下某条命名记录的完整信息或某个字段（如某客户的电话/销售员）。问「某记录的某字段」优先用它。 - collectForm(entityKeyword, knownFieldsJson?)：新增字段较多的单据（尤其**报价**）时，弹一张表单让用户一次填齐；把用户已说的信息作为 knownFieldsJson 预填。**新增一律先用它**，用户提交后你再用 proposeWrite(action=create)。 - proposeWrite(action, entityKeyword, recordKeyword?, fieldChinese?, newValue?, fieldsJson?)：**唯一的写工具**（人在环：只生成待确认提议，用户点【确认】才执行）。action：create=新增（fieldsJson）；update=改字段（recordKeyword+fieldChinese+newValue）；invalid=作废；cancelInvalid=复原；examine=审核；cancelExamine=销审；delete=物理删除（业务单据别用）。它自行定位主表与记录，无需先 findForms。 - askUser(question, options?)：缺关键信息时向用户提**一个**澄清问题（尽量给选项）。 【准则】 1. 查询：findForms 定位 → readFormData/lookupRecord 读 → 如实汇报。同名多张表单优先选检索结果靠前那张。**绝不编造**表单名、单号或数据——答案里的每个数字都必须来自工具结果。 2. 写操作一律人在环：只生成待确认提议、**绝不声称已完成/已写入**。新增先 collectForm；proposeWrite 生成提议后就停下、提示用户点【确认】，不要继续调别的工具。 3. 实体角色不能错：要报价/加工的物品（纸盒/彩盒/画册）是**产品**，购买方公司名才是客户。客户/产品/物料必须是系统里已有的真实记录——找不到就让用户从下拉里选，**绝不新建不存在的客户**。 4. 信息足够就直接做，不要无谓反问；缺关键参数才 askUser 问**一次**并停下等回答，不要反复追问。回答面向业务人员：用记录名称而不是内部 ID。 """ def fn(name, desc, props, required): return {"type": "function", "function": { "name": name, "description": desc, "parameters": {"type": "object", "properties": props, "required": required}}} T_FINDFORMS = fn("findForms", "按关键词检索 ERP 业务表单目录，返回匹配的表单名、底层数据源、所属菜单id(moduleId) 与表单id(formId)。" "当用户提到某类单据或报表、但你不确定具体是哪一张表单时，先用它来定位。最多返回 15 条。", {"keyword": {"type": "string", "description": "表单名或业务关键词，例如：报价 / 客户 / 库存 / 送货 / 应收 / 采购订单"}}, ["keyword"]) T_READFORM = fn("readFormData", "读取指定 ERP 表单的真实业务数据（返回前若干行 + 总条数）。可选 keyword 用于按名称模糊过滤" "（如查某个客户 / 物料）；可选 page 翻页（从 1 开始，用户要看「下一页/更多」时递增）。" "必须先用 findForms 得到目标表单的 formId 与 moduleId。", {"formId": {"type": "string", "description": "表单id（findForms 返回的 formId）"}, "moduleId": {"type": "string", "description": "菜单id（findForms 返回的 moduleId）"}, "keyword": {"type": "string", "description": "可选：仅当要查找某个具体名称的记录时才填；问数量 / 全部 / 概况时必须留空"}, "page": {"type": "integer", "description": "可选：页码，从 1 开始；默认 1"}}, ["formId", "moduleId"]) T_LOOKUP = fn("lookupRecord", "查询某个实体下某条命名记录的**完整信息**（返回该记录的所有可读字段）。" "用于「某客户 / 某物料 的详细资料 / 某个具体字段（如联系电话、销售员）」这类精确查询。", {"entityKeyword": {"type": "string", "description": "实体类型，如 客户 / 物料 / 供应商"}, "recordKeyword": {"type": "string", "description": "记录名称关键词，如某个客户名"}}, ["entityKeyword", "recordKeyword"]) T_COLLECT = fn("collectForm", "在对话里弹出一张 ERP 表单让用户一次性填齐多个字段（而不是逐个追问、更不是自己猜）。用于字段较多的新建/录入场景" "（尤其**新建报价**）。entityKeyword = 单据类型（如 报价 / 客户）；knownFieldsJson = 用户已说的字段(中文名->值)用于预填。" "客户/产品/物料会渲染成下拉让用户从真实数据里选。用户填完提交后，你再用 proposeWrite(action=create) 生成待确认的新增。", {"entityKeyword": {"type": "string", "description": "要新建的实体/单据类型，如 报价 / 客户 / 物料"}, "knownFieldsJson": {"type": "string", "description": "可选：用户已说的字段 JSON（中文名->值），用于预填表单"}}, ["entityKeyword"]) T_PROPOSE = fn("proposeWrite", "提议一次写操作（人在环：只生成待确认提议、绝不立即执行；用户点【确认】才写入，你绝不能声称已完成）。" "用 action 指定动作：create=新增；update=改某字段；invalid=作废（业务单据要“删除/取消”一律用它，可复原）；" "cancelInvalid=复原/取消作废；examine=审核；cancelExamine=销审/反审核；delete=物理删除（明细行/极少用，业务单据别用）。" "按 action 传参：create 用 fieldsJson；update 用 recordKeyword+fieldChinese+newValue；" "invalid/cancelInvalid/examine/cancelExamine/delete 用 recordKeyword 定位单据。本工具自行定位主表，无需先 findForms。", {"action": {"type": "string", "description": "动作：create|update|invalid|cancelInvalid|examine|cancelExamine|delete"}, "entityKeyword": {"type": "string", "description": "实体/单据类型，如 报价 / 客户 / 销售订单"}, "recordKeyword": {"type": "string", "description": "记录名称或单号关键词（create 不需要）"}, "fieldChinese": {"type": "string", "description": "要改的字段中文名（仅 update）"}, "newValue": {"type": "string", "description": "新值（仅 update）"}, "fieldsJson": {"type": "string", "description": "字段 JSON 中文名->值（仅 create）"}}, ["action", "entityKeyword"]) T_ASK = fn("askUser", "向用户提出一个澄清/消歧的小问题并给出可选项。当用户意图不明确、或需要在几个候选中二选一时使用；" "问完即结束本轮、等待用户回答。options 用中文顿号或逗号分隔，可留空表示自由回答。", {"question": {"type": "string", "description": "要问用户的问题"}, "options": {"type": "string", "description": "候选选项，用、或,分隔；没有明确候选时留空"}}, ["question"]) ALL6 = [T_FINDFORMS, T_READFORM, T_LOOKUP, T_COLLECT, T_PROPOSE, T_ASK] def call(body, timeout=200): req = urllib.request.Request( URL, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", "Authorization": "Bearer ollama"}) t0 = time.time() with urllib.request.urlopen(req, timeout=timeout) as r: out = json.loads(r.read()) return out, time.time() - t0 def has_digits(s): return any(c.isdigit() for c in s) def judge(exp, tool, args, text): """返回 'ok' | 'clarify' | 'fail'。clarify=用文字/askUser 反问（宽松口径算对）。""" action = (args or {}).get("action", "") asked = tool == "askUser" or (tool is None and text and ("？" in text or "?" in text)) if exp == "查询": if tool in READ_TOOLS: return "ok" if tool is None and text and not has_digits(text): return "ok" return "fail" if exp == "新增": if tool == "collectForm" or (tool == "proposeWrite" and action == "create"): return "ok" return "clarify" if asked else "fail" if exp == "修改": if (tool == "proposeWrite" and action == "update") or tool in ("askUser", "lookupRecord"): return "ok" return "clarify" if asked else "fail" if exp == "删除": if tool == "proposeWrite" and action in ("invalid", "delete", "cancelInvalid"): return "ok" return "clarify" if asked else "fail" if exp == "审核": if tool == "proposeWrite" and action in ("examine", "cancelExamine"): return "ok" return "clarify" if asked else "fail" if exp == "闲聊": return "ok" if tool is None else "fail" if exp == "不清楚": if tool == "askUser" or (tool is None and text): return "ok" return "fail" return "fail" def run_gate(): print("=" * 20, "意图门 4 类（50题）", "=" * 20) ok = 0 wrong = [] for utt, fine in CASES: exp = GATE_MAP[fine] body = {"model": MODEL, "stream": False, "temperature": 0.1, "top_p": 0.9, "reasoning_effort": "none", "response_format": {"type": "json_schema", "json_schema": {"name": "output", "schema": INTENT_SCHEMA}}, "messages": [{"role": "system", "content": INTENT_SYSTEM}, {"role": "user", "content": utt}]} try: resp, dt = call(body) j = json.loads(resp["choices"][0]["message"]["content"]) got = j.get("intent", "?") hit = got == exp ok += hit if not hit: wrong.append((utt, exp, got, j.get("danju", ""))) print(f'{"✓" if hit else "✗"} [{exp}] {utt} → {got}/{j.get("danju","")} ({dt:.1f}s)') except Exception as ex: wrong.append((utt, exp, "ERROR", str(ex))) print(f"✗ [{exp}] {utt} → ERROR {ex}") print(f"\n意图门: {ok}/{len(CASES)} = {ok/len(CASES)*100:.0f}%") for utt, exp, got, d in wrong: print(f" 错例: 「{utt}」 期望{exp} → 得到{got}({d})") print() return ok def run_agent(): print("=" * 20, "统一prompt+固定6工具（50题）", "=" * 20) strict = 0 lenient = 0 per = {} fails = [] for utt, exp in CASES: body = {"model": MODEL, "stream": False, "temperature": 0.1, "top_p": 0.9, "reasoning_effort": "none", "tools": ALL6, "messages": [{"role": "system", "content": UNIFIED_PROMPT}, {"role": "user", "content": utt}]} try: resp, dt = call(body) msg = resp["choices"][0]["message"] tcs = msg.get("tool_calls") or [] tool, args = None, None if tcs: tool = tcs[0]["function"]["name"] try: args = json.loads(tcs[0]["function"].get("arguments") or "{}") except Exception: args = {} text = (msg.get("content") or "").strip() v = judge(exp, tool, args, text) except Exception as ex: v, tool, args, text, dt = "fail", None, None, f"ERROR {ex}", 0 strict += v == "ok" lenient += v in ("ok", "clarify") s, l, n = per.get(exp, (0, 0, 0)) per[exp] = (s + (v == "ok"), l + (v in ("ok", "clarify")), n + 1) mark = {"ok": "✓", "clarify": "◐", "fail": "✗"}[v] detail = f"tool={tool or '文字'}" + (f" {(args or {}).get('action','')}" if tool == "proposeWrite" else "") if v == "fail" and tool is None: detail += " 「" + text[:50].replace("\n", " ") + "」" if v == "fail": fails.append((utt, exp, detail)) print(f"{mark} [{exp}] {utt} → {detail} ({dt:.1f}s)") n = len(CASES) print(f"\n统一6工具: 严格 {strict}/{n} = {strict/n*100:.0f}% 宽松 {lenient}/{n} = {lenient/n*100:.0f}%") print("按意图分（严格/宽松/总数）:", {k: f"{s}/{l}/{c}" for k, (s, l, c) in per.items()}) for utt, exp, d in fails: print(f" 失败: 「{utt}」[{exp}] → {d}") print() return strict, lenient import sys if __name__ == "__main__": n = len(CASES) if "--gate-only" in sys.argv: g = run_gate() print(f"验收: 意图门100%={'PASS' if g == n else 'FAIL'}") else: g = run_gate() a_s, a_l = run_agent() print("=" * 60) print(f"总结(50题): 意图门 {g}/{n}={g/n*100:.0f}% | agent 严格{a_s}/{n}={a_s/n*100:.0f}% 宽松{a_l}/{n}={a_l/n*100:.0f}%") print(f"验收: 意图门100%={'PASS' if g == n else 'FAIL'} agent严格≥76%={'PASS' if a_s/n >= 0.76 else 'FAIL'}")