135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试「知你客服6号」Agent：登录 -> 创建执行 -> 轮询直到结束。
|
||
|
||
用法:
|
||
python test_zhini_kefu_6.py
|
||
python test_zhini_kefu_6.py --base-url http://127.0.0.1:8037
|
||
set PLATFORM_BASE_URL=... && python test_zhini_kefu_6.py
|
||
|
||
依赖: requests（与项目其他测试脚本一致）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
import requests
|
||
|
||
# 知你客服6号（本地平台 Agent 管理中的名称对应 ID，若你环境不同请改此处或传 --agent-id）
|
||
DEFAULT_AGENT_ID = "2acc84d5-814b-4d61-9703-94a4b117375f"
|
||
DEFAULT_MESSAGE = "你好"
|
||
|
||
|
||
def main() -> int:
|
||
if sys.platform == "win32" and hasattr(sys.stdout, "reconfigure"):
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
sys.stderr.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
parser = argparse.ArgumentParser(description="测试知你客服6号，发送一条用户消息并打印结果")
|
||
parser.add_argument(
|
||
"--base-url",
|
||
default=os.getenv("PLATFORM_BASE_URL", "http://127.0.0.1:8037"),
|
||
help="平台 API 根地址（默认 http://127.0.0.1:8037）",
|
||
)
|
||
parser.add_argument("--username", default=os.getenv("PLATFORM_USERNAME", "admin"))
|
||
parser.add_argument("--password", default=os.getenv("PLATFORM_PASSWORD", "123456"))
|
||
parser.add_argument("--agent-id", default=os.getenv("ZHINI_6_AGENT_ID", DEFAULT_AGENT_ID))
|
||
parser.add_argument("--message", "-m", default=DEFAULT_MESSAGE, help="用户消息，默认：你好")
|
||
parser.add_argument(
|
||
"--user-id",
|
||
default="script_test_zhini6",
|
||
help="多轮记忆隔离用 user_id（可选）",
|
||
)
|
||
parser.add_argument("--timeout", type=int, default=180, help="轮询最长秒数")
|
||
parser.add_argument("--poll", type=float, default=0.8, help="轮询间隔秒")
|
||
args = parser.parse_args()
|
||
|
||
base = args.base_url.rstrip("/")
|
||
|
||
print("=" * 60)
|
||
print("知你客服6号 执行测试")
|
||
print(" base_url :", base)
|
||
print(" agent_id :", args.agent_id)
|
||
print(" message :", args.message)
|
||
print("=" * 60)
|
||
|
||
# 1. 登录
|
||
r = requests.post(
|
||
f"{base}/api/v1/auth/login",
|
||
data={"username": args.username, "password": args.password},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
timeout=15,
|
||
)
|
||
if r.status_code != 200:
|
||
print("登录失败:", r.status_code, r.text[:500], file=sys.stderr)
|
||
return 1
|
||
token = r.json().get("access_token")
|
||
if not token:
|
||
print("登录响应无 access_token", file=sys.stderr)
|
||
return 1
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
msg = args.message
|
||
body = {
|
||
"agent_id": args.agent_id,
|
||
"input_data": {
|
||
"query": msg,
|
||
"USER_INPUT": msg,
|
||
"user_id": args.user_id,
|
||
},
|
||
}
|
||
|
||
# 2. 创建执行
|
||
r2 = requests.post(f"{base}/api/v1/executions", headers=headers, json=body, timeout=30)
|
||
if r2.status_code != 201:
|
||
print("创建执行失败:", r2.status_code, r2.text[:1000], file=sys.stderr)
|
||
return 1
|
||
|
||
ex = r2.json()
|
||
eid = ex["id"]
|
||
print("已创建执行:", eid)
|
||
print("初始状态:", ex.get("status"))
|
||
|
||
# 3. 轮询
|
||
deadline = time.time() + args.timeout
|
||
status = ex.get("status", "pending")
|
||
while status in ("pending", "running") and time.time() < deadline:
|
||
time.sleep(args.poll)
|
||
rs = requests.get(f"{base}/api/v1/executions/{eid}", headers=headers, timeout=60)
|
||
if rs.status_code != 200:
|
||
print("查询执行失败:", rs.status_code, rs.text[:500], file=sys.stderr)
|
||
return 1
|
||
detail = rs.json()
|
||
status = detail.get("status", status)
|
||
print(" ...", status)
|
||
|
||
final = requests.get(f"{base}/api/v1/executions/{eid}", headers=headers, timeout=60).json()
|
||
status = final.get("status")
|
||
print()
|
||
print("最终状态:", status)
|
||
if final.get("error_message"):
|
||
print("错误信息:", final["error_message"])
|
||
od = final.get("output_data")
|
||
if od is not None:
|
||
print("output_data:")
|
||
print(json.dumps(od, ensure_ascii=False, indent=2))
|
||
else:
|
||
print("output_data: null")
|
||
|
||
return 0 if status == "completed" else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|