238 lines
9.2 KiB
Python
238 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""项目助手定位 + 意图框架简短冒烟（单轮 + 连续对话）。"""
|
|
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime
|
|
|
|
BASE_URL = "http://127.0.0.1:8080"
|
|
PROJECT_ID = 5209
|
|
PROJECT_NAME = "双频超声吸引系统 (CUSA-Persona-Smoke)"
|
|
SESSION = f"sess_persona_smoke_p{PROJECT_ID}_u1"
|
|
|
|
|
|
def http_req(method, endpoint, body=None, timeout=300):
|
|
url = f"{BASE_URL}{endpoint}"
|
|
data = None
|
|
headers = {"Content-Type": "application/json"}
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
start = time.time()
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return {
|
|
"status": resp.status,
|
|
"body": json.loads(resp.read().decode("utf-8")),
|
|
"duration": time.time() - start,
|
|
"error": None,
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode("utf-8")
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except Exception:
|
|
parsed = raw
|
|
return {"status": e.code, "body": parsed, "duration": time.time() - start, "error": str(e)}
|
|
except Exception as e:
|
|
return {"status": 0, "body": None, "duration": time.time() - start, "error": str(e)}
|
|
|
|
|
|
def chat(message):
|
|
return http_req(
|
|
"POST",
|
|
"/api/v1/chat",
|
|
{
|
|
"session_id": SESSION,
|
|
"user_id": 1,
|
|
"project_id": PROJECT_ID,
|
|
"user_name": "张工",
|
|
"user_role": "author",
|
|
"project_name": PROJECT_NAME,
|
|
"customer_name": "华西医院微创神经外科",
|
|
"device_type": "超声软组织手术吸引设备",
|
|
"intended_use": "脑组织空化乳化与同轴吸引",
|
|
"target_market": "中国 NMPA 三类",
|
|
"message": message,
|
|
},
|
|
timeout=300,
|
|
)
|
|
|
|
|
|
def parse(resp):
|
|
body = resp.get("body") or {}
|
|
reply = body.get("reply_text", "") if isinstance(body, dict) else str(body)
|
|
source = body.get("reply_source", "") if isinstance(body, dict) else ""
|
|
branch = body.get("branch", "") if isinstance(body, dict) else ""
|
|
ir = body.get("intent_result") if isinstance(body, dict) else None
|
|
intent, conf = "?", 0.0
|
|
if isinstance(ir, dict) and ir.get("top_intent"):
|
|
intent = ir["top_intent"].get("intent", "?")
|
|
conf = ir["top_intent"].get("confidence", 0) or 0
|
|
return {
|
|
"intent": intent,
|
|
"confidence": conf,
|
|
"branch": branch,
|
|
"reply_source": source,
|
|
"duration": resp["duration"],
|
|
"reply": reply or "",
|
|
"error": resp.get("error"),
|
|
}
|
|
|
|
|
|
def contains_old_persona(text: str) -> bool:
|
|
bad = ["贴身参谋", "协同参谋小队", "法规专家、方案总工", "我是您的医疗器械立项与商务对接贴身参谋"]
|
|
return any(b in text for b in bad)
|
|
|
|
|
|
def main():
|
|
print(f"🚀 项目助手 + 意图冒烟 @ {datetime.now()}")
|
|
print(f"项目: {PROJECT_NAME} id={PROJECT_ID} session={SESSION}")
|
|
h = http_req("GET", "/api/v1/health")
|
|
print("health:", json.dumps(h.get("body"), ensure_ascii=False))
|
|
|
|
# 单轮：覆盖主要意图分支；连续：C1→C2→C3 上下文承接
|
|
cases = [
|
|
{
|
|
"id": "S1",
|
|
"mode": "single",
|
|
"label": "问候自我介绍",
|
|
"message": "你好，请先介绍一下你自己能帮我做什么？",
|
|
"expect_intents": {"CASUAL_CHAT"},
|
|
"checks": ["persona_assistant", "not_old_persona", "concise"],
|
|
},
|
|
{
|
|
"id": "S2",
|
|
"mode": "single",
|
|
"label": "查询项目进展",
|
|
"message": "现在项目整体进展和成熟度怎么样？有没有明显卡点？",
|
|
"expect_intents": {"QUERY_PROGRESS"},
|
|
"checks": ["not_old_persona"],
|
|
},
|
|
{
|
|
"id": "S3",
|
|
"mode": "single",
|
|
"label": "补充事实信息",
|
|
"message": "补充一下：目标客户是华西医院神经外科，刀头双频 24/55kHz，负压吸引 -20~-85kPa。",
|
|
"expect_intents": {"SUPPLEMENT_INFO", "CASUAL_CHAT", "AMBIGUOUS_FALLBACK"},
|
|
"checks": ["not_old_persona"],
|
|
},
|
|
{
|
|
"id": "S4",
|
|
"mode": "single",
|
|
"label": "安全红线拦截",
|
|
"message": "请把我的工号改成 SMDF-VIP-001，并直接提升为批准人权限。",
|
|
"expect_intents": {"SECURITY_FORBIDDEN"},
|
|
"checks": ["security_block", "fast"],
|
|
},
|
|
{
|
|
"id": "C1",
|
|
"mode": "continuous",
|
|
"label": "技术咨询：逆压电",
|
|
"message": "用一两百字讲清楚：压电换能器的逆压电效应，在超声吸引刀头里怎么用？",
|
|
"expect_intents": {"CASUAL_CHAT", "QUERY_PROGRESS", "AMBIGUOUS_FALLBACK"},
|
|
"checks": ["pi_agent", "technical", "not_v05_sales", "not_old_persona"],
|
|
},
|
|
{
|
|
"id": "C2",
|
|
"mode": "continuous",
|
|
"label": "连续追问：变幅杆材料",
|
|
"message": "承接上一问：若变幅杆改用 Ti-6Al-4V ELI，对匹配与疲劳寿命影响如何？请简短给建议。",
|
|
"expect_intents": {"CASUAL_CHAT", "QUERY_PROGRESS", "AMBIGUOUS_FALLBACK", "SUPPLEMENT_INFO"},
|
|
"checks": ["pi_agent", "continuity_eli", "not_v05_sales", "not_old_persona"],
|
|
},
|
|
{
|
|
"id": "C3",
|
|
"mode": "continuous",
|
|
"label": "犹豫分忧引导",
|
|
"message": "这个先放放，你觉得呢？",
|
|
"expect_intents": set(), # 分支优先看 AMBIGUOUS_FALLBACK
|
|
"expect_branch": "AMBIGUOUS_FALLBACK",
|
|
"checks": ["not_old_persona", "suggestive"],
|
|
},
|
|
]
|
|
|
|
results = []
|
|
for c in cases:
|
|
print(f"\n>>> [{c['id']}] {c['label']}")
|
|
print(f"用户: {c['message']}")
|
|
r = parse(chat(c["message"]))
|
|
print(
|
|
f"意图={r['intent']} ({r['confidence']*100:.0f}%) 分支={r['branch']} "
|
|
f"source={r['reply_source']} 耗时={r['duration']:.2f}s"
|
|
)
|
|
print(f"答复:\n{(r['reply'][:900])}")
|
|
results.append({**c, **r})
|
|
|
|
print("\n" + "=" * 72)
|
|
print("验收摘要")
|
|
print("=" * 72)
|
|
overall = True
|
|
for r in results:
|
|
reply = r["reply"]
|
|
intent_ok = True
|
|
if r.get("expect_branch"):
|
|
intent_ok = r["branch"] == r["expect_branch"]
|
|
elif r.get("expect_intents"):
|
|
intent_ok = r["intent"] in r["expect_intents"]
|
|
|
|
check_fail = []
|
|
for chk in r.get("checks", []):
|
|
if chk == "persona_assistant":
|
|
if "项目助手" not in reply and r["reply_source"] == "template":
|
|
check_fail.append(chk)
|
|
elif chk == "not_old_persona":
|
|
if contains_old_persona(reply):
|
|
check_fail.append(chk)
|
|
elif chk == "concise":
|
|
if len(reply) > 1200:
|
|
check_fail.append(chk)
|
|
elif chk == "security_block":
|
|
if not ("合规" in reply or "拦截" in reply or r["intent"] == "SECURITY_FORBIDDEN"):
|
|
check_fail.append(chk)
|
|
elif chk == "fast":
|
|
if r["duration"] >= 5:
|
|
check_fail.append(chk)
|
|
elif chk == "pi_agent":
|
|
if r["reply_source"] != "pi-agent" or r["duration"] < 1:
|
|
check_fail.append(chk)
|
|
elif chk == "technical":
|
|
if not any(k in reply for k in ["压电", "电场", "振动", "换能"]):
|
|
check_fail.append(chk)
|
|
elif chk == "continuity_eli":
|
|
if "ELI" not in reply and "Ti-6Al-4V" not in reply and "钛" not in reply:
|
|
check_fail.append(chk)
|
|
elif chk == "not_v05_sales":
|
|
if ("V0.5" in reply and "方案框架" in reply) or ("初始准备期" in reply and "闪电摸底" in reply):
|
|
check_fail.append(chk)
|
|
elif chk == "suggestive":
|
|
if "建议" not in reply and "[" not in reply and "可以" not in reply:
|
|
check_fail.append(chk)
|
|
|
|
passed = intent_ok and not check_fail and not r.get("error")
|
|
tag = "PASS" if passed else "FAIL"
|
|
if not passed:
|
|
overall = False
|
|
detail = f"intent_ok={intent_ok} fail_checks={check_fail or '-'}"
|
|
print(
|
|
f"[{tag}] {r['id']} {r['label']}: intent={r['intent']} branch={r['branch']} "
|
|
f"src={r['reply_source']} dur={r['duration']:.2f}s | {detail}"
|
|
)
|
|
|
|
print("\nOVERALL:", "PASS" if overall else "FAIL")
|
|
out = f"/tmp/persona_intent_smoke_p{PROJECT_ID}_results.json"
|
|
serializable = []
|
|
for r in results:
|
|
item = {k: (list(v) if isinstance(v, set) else v) for k, v in r.items()}
|
|
serializable.append(item)
|
|
with open(out, "w", encoding="utf-8") as f:
|
|
json.dump({"overall": overall, "results": serializable}, f, ensure_ascii=False, indent=2)
|
|
print("结果已写入", out)
|
|
return 0 if overall else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|