131 lines
4.5 KiB
Python
131 lines
4.5 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')}")
|
|||
|
|
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())
|