lzf_0626
a7fd02ec44
批量验收脚本（customer_service_check.py）只覆盖 8 个固定用例，手工试自己的问题时
需要自己签发 JWT、提交 run、驱动执行、解析结果，门槛太高。本工具把这条链路封成一条命令：
python tools/ask_customer_service.py "基金赎回到账要多久" # 单次提问
python tools/ask_customer_service.py # 交互模式，同会话连续追问
python tools/ask_customer_service.py --user 9003 "问题" # 换身份提问
输出包含：运行状态、识别意图与置信度、处置判定（直接回答 / 引导人工客服）、回答正文、
来源引用，并在"给了答案但正文没有依据"时给出人工核对提示。
注意：脚本自己驱动这一条 run 执行（WorkerRuntime().execute），运行前需停掉常驻 Worker，
否则常驻 Worker 会从共享队列抢走任务，脚本将等不到结果。
133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
"""向客服 Agent 提问，打印回答、意图与依据（手工测试用）。
|
||
|
||
用法：
|
||
|
||
```powershell
|
||
# 单次提问
|
||
D:\\conda\\envs\\jr_py313\\python.exe tools\\ask_customer_service.py "基金赎回几天到账"
|
||
|
||
# 交互模式：连续追问，同一会话（可用于试多轮）
|
||
D:\\conda\\envs\\jr_py313\\python.exe tools\\ask_customer_service.py
|
||
```
|
||
|
||
**运行前请先停掉常驻 Worker**：本脚本自己驱动这一条 run 执行
|
||
（`WorkerRuntime().execute`），而常驻 Worker 与脚本共享 `agent_run` 队列，
|
||
它会抢走任务并用自己的执行路径处理，本脚本就只能一直等不到结果。
|
||
|
||
换身份提问用 `--user`（默认 9001 客户）：
|
||
```powershell
|
||
... ask_customer_service.py --user 9003 "帮我推荐一只基金"
|
||
```
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import datetime as dt
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
import jwt
|
||
|
||
from app.core.config import get_settings
|
||
from app.main import create_app
|
||
from app.worker.runtime import WorkerRuntime
|
||
|
||
# GBK 控制台下知识正文可能含 emoji，直接打印会让脚本自身崩掉
|
||
sys.stdout.reconfigure(errors="replace")
|
||
|
||
AGENT_TYPE = "customer_service"
|
||
FALLBACK_MARK = "客服热线"
|
||
|
||
|
||
def token(subject: str) -> str:
|
||
settings = get_settings()
|
||
private_key = Path(settings.jwt_private_key_path).read_text(encoding="utf-8")
|
||
now = dt.datetime.now(dt.UTC)
|
||
return jwt.encode(
|
||
{
|
||
"sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience,
|
||
"exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5),
|
||
"jti": str(uuid.uuid4()),
|
||
},
|
||
private_key,
|
||
algorithm="RS256",
|
||
)
|
||
|
||
|
||
async def ask(
|
||
client: httpx.AsyncClient, auth: dict[str, str], message: str, session_id: str
|
||
) -> None:
|
||
accepted = await client.post(
|
||
"/api/v1/agent-runs",
|
||
json={
|
||
"agent_type": AGENT_TYPE, "message": message,
|
||
"session_id": session_id, "idempotency_key": uuid.uuid4().hex,
|
||
},
|
||
headers=auth,
|
||
)
|
||
if accepted.status_code != 202:
|
||
print(f"受理失败 {accepted.status_code}：{accepted.text[:300]}\n")
|
||
return
|
||
|
||
run_id = accepted.json()["data"]["run_id"]
|
||
print(f"（run={run_id} 执行中…）")
|
||
await WorkerRuntime().execute(run_id)
|
||
body = (await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)).json()["data"]
|
||
result = body.get("result") or {}
|
||
|
||
text = str(result.get("content") or "")
|
||
raw_intent = result.get("intent")
|
||
intent = raw_intent.get("intent") if isinstance(raw_intent, dict) else raw_intent
|
||
confidence = (
|
||
raw_intent.get("confidence") if isinstance(raw_intent, dict)
|
||
else result.get("confidence")
|
||
)
|
||
guided = FALLBACK_MARK in text
|
||
|
||
print(f"运行状态：{body.get('status')}"
|
||
+ (f" 错误码：{body.get('error_code')}" if body.get("error_code") else ""))
|
||
print(f"识别意图：{intent} 意图置信度：{confidence}")
|
||
print(f"处置：{'引导客户致电人工客服' if guided else '直接回答（内容来自公司资料）'}")
|
||
print("─" * 72)
|
||
print(text)
|
||
print("─" * 72)
|
||
references = result.get("source_references") or []
|
||
for reference in references:
|
||
print(f"来源引用：{reference.get('source_type')} / {reference.get('source_id')}")
|
||
if not guided and "依据：" not in text:
|
||
print("提示：这条回答正文里没有「依据：…」，说明知识命中的是通用来源，请人工核对。")
|
||
print()
|
||
|
||
|
||
async def main() -> None:
|
||
parser = argparse.ArgumentParser(description="向客服 Agent 提问")
|
||
parser.add_argument("message", nargs="*", help="要提的问题；省略则进入交互模式")
|
||
parser.add_argument("--user", default="9001", help="以哪个用户身份提问（默认 9001 客户）")
|
||
args = parser.parse_args()
|
||
|
||
app = create_app()
|
||
auth = {"Authorization": f"Bearer {token(args.user)}"}
|
||
async with httpx.AsyncClient(
|
||
transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=120
|
||
) as client:
|
||
if args.message:
|
||
await ask(client, auth, " ".join(args.message), f"ask-{uuid.uuid4().hex[:12]}")
|
||
return
|
||
|
||
session_id = f"ask-{uuid.uuid4().hex[:12]}"
|
||
print(f"交互模式（会话 {session_id}）。直接回车或输入 exit 退出。\n")
|
||
while True:
|
||
try:
|
||
raw = input("请提问> ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
return
|
||
if not raw or raw.lower() in {"exit", "quit", "退出", "q"}:
|
||
return
|
||
await ask(client, auth, raw, session_id)
|
||
|
||
|
||
asyncio.run(main())
|