lzf_0626
1fa5fc7d03
业务方要求客户侧只看到一句固定话术（不构成投资建议），因此从回答正文移除： 1. 中置信的「（以上信息可能不完整，具体以产品说明书与公司制度为准）」提示； 2. 「（依据：…）」出处行——原先它显示的是知识块标题，FAQ 的标题就是问题本身， 展示为「（依据：公司什么时候成立的?）」并没有可读价值。 可追溯性不受影响：本次命中哪个知识块仍由审计（agent.tool_executed 的工具调用记录） 与消息表留痕，source_references（tool 类型）也照常返回，只是不再面向客户展示。 取舍已在代码注释中标明：中置信回答此后不再向客户标注不确定性。 若将来要把出处展示给客户，应当走 source_references 的 knowledge 类型 （前提是让 ToolExecutor 把工具返回的 doc_id 登记为本次可引用来源）， 而不是继续往正文里拼字符串。 同时移除因此不再使用的 INCOMPLETE_NOTICE 常量、_source_note 方法， 以及提问工具里那句"正文没有依据行"的提示。 验证：ruff 通过、mypy 107 文件无错；customer_service_check 9/9 通过； ask_customer_service 实测回答正文为「答案 + 免责声明」两行。
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())
|