65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import ssl
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def _json_request(url: str, method: str = "GET", payload: dict | None = None, timeout: int = 20) -> dict:
|
|
body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
req = urllib.request.Request(url=url, method=method, data=body)
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
ctx = ssl._create_unverified_context() if url.startswith("https://") else None
|
|
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Quick QA for text pipeline and metrics")
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:8080")
|
|
parser.add_argument("--rounds", type=int, default=5)
|
|
args = parser.parse_args()
|
|
|
|
health = _json_request(f"{args.base_url}/health")
|
|
print("health.ok:", bool(health.get("llm")))
|
|
latencies = []
|
|
for i in range(args.rounds):
|
|
t0 = time.perf_counter()
|
|
data = _json_request(
|
|
f"{args.base_url}/chat/text",
|
|
method="POST",
|
|
payload={"text": f"第{i+1}轮压测，请简短回复。"},
|
|
timeout=60,
|
|
)
|
|
dt = int((time.perf_counter() - t0) * 1000)
|
|
latencies.append(dt)
|
|
print(
|
|
f"round={i+1} ok={data.get('ok')} llm={data.get('llm_latency_ms')} "
|
|
f"tts={data.get('tts_latency_ms')} total={dt}"
|
|
)
|
|
|
|
p50 = sorted(latencies)[len(latencies) // 2] if latencies else 0
|
|
print("rounds:", args.rounds, "p50_ms:", p50, "max_ms:", max(latencies) if latencies else 0)
|
|
|
|
health2 = _json_request(f"{args.base_url}/health")
|
|
print(
|
|
"final.health:",
|
|
{
|
|
"pipeline_runs": health2.get("pipeline_runs"),
|
|
"last_latency_ms": health2.get("last_latency_ms"),
|
|
"last_llm_latency_ms": health2.get("last_llm_latency_ms"),
|
|
"last_tts_latency_ms": health2.get("last_tts_latency_ms"),
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except urllib.error.URLError as exc:
|
|
raise SystemExit(f"request failed: {exc}")
|