南方科技 · 智能客服 本地调试控制台（回答全部来自公司资料库）
可以直接试：基金赎回几天到账 · 季季盈90天起投多少 ·
C1 客户能买什么 · 然后再问 那它风险高吗（试试多轮指代）
"""
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",
)
def build_console(user_id: str) -> FastAPI:
console = FastAPI(title="客服本地控制台")
# 底座以 in-process 方式挂载：不经过网络、不新增任何路由，私钥只在签发时用到
base_app = create_app()
auth = {"Authorization": f"Bearer {token(user_id)}"}
@console.get("/", response_class=HTMLResponse)
async def index() -> HTMLResponse:
return HTMLResponse(PAGE)
@console.post("/api/chat")
async def chat(payload: dict[str, Any]) -> dict[str, Any]:
message = str(payload.get("message") or "").strip()
session_id = str(payload.get("session_id") or f"web-{uuid.uuid4().hex[:8]}")
if not message:
return {"answer": "请先输入内容。", "intent": "", "guided": False}
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=base_app), base_url="http://base", timeout=120
) as client:
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:
body = accepted.text[:200]
return {"answer": f"受理失败（{accepted.status_code}）：{body}",
"intent": "", "guided": False}
run_id = accepted.json()["data"]["run_id"]
# 本控制台自己驱动这一条 run；若常驻 Worker 同时在跑会与它抢队列
await WorkerRuntime().execute(run_id)
detail = (await client.get(
f"/api/v1/agent-runs/{run_id}", headers=auth
)).json()["data"]
result = detail.get("result") or {}
answer = str(result.get("content") or "")
if detail.get("status") != "succeeded":
answer = answer or f"运行未成功：{detail.get('status')} {detail.get('error_code')}"
# intent 在结果里可能是 {"intent": ..., "confidence": ...}，也可能已是字符串
raw_intent = result.get("intent")
intent = raw_intent.get("intent") if isinstance(raw_intent, dict) else raw_intent
return {
"answer": answer,
"intent": intent or "",
"guided": FALLBACK_MARK in answer,
}
return console
def main() -> None:
parser = argparse.ArgumentParser(description="客服本地控制台")
parser.add_argument("--port", type=int, default=8098)
parser.add_argument("--user", default="9001", help="以哪个客户身份对话（默认 9001）")
args = parser.parse_args()
print(f"控制台：http://127.0.0.1:{args.port} 身份={args.user}")
print("提示：运行前请停掉常驻 Worker，否则它会抢走任务、页面会一直转圈。")
uvicorn.run(build_console(args.user), host="127.0.0.1", port=args.port, log_level="warning")
if __name__ == "__main__":
main()