朱子纯 / xlyNewAi

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'}")