diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 0a7ebd5..d7279cc 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -404,13 +404,18 @@ class OcrClient: source_name=source.name, usage_source="measured", ) + # 错误带异常类名前缀(metric ocr-call-success 注册口径: 按类名归组) + if error is None or isinstance(error, str): + error_text = error + else: + error_text = f"{type(error).__name__}: {error}" await self._emitter.emit_attempt( request=request, source=source, call_id=call_id, latency_ms=latency_ms, response=response, - error=None if error is None else str(error), + error=error_text, ) @staticmethod diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index a81caa8..5265f36 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -350,7 +350,7 @@ class TestTelemetry: for row in recorder.rows: assert "" in row["messages"] assert "RAW-IMAGE-BYTES" not in row["messages"] # 图像字节绝不入库 - assert recorder.rows[0]["error"] is not None + assert recorder.rows[0]["error"].startswith("TransientError:") # 类名前缀口径 assert recorder.rows[1]["error"] is None assert recorder.rows[1]["prompt_tokens"] == 0 diff --git a/tests/unit/test_soak_scoreboard.py b/tests/unit/test_soak_scoreboard.py new file mode 100644 index 0000000..c37e2c9 --- /dev/null +++ b/tests/unit/test_soak_scoreboard.py @@ -0,0 +1,54 @@ +"""P7 记分板新增纯函数测试(M3 计划 T8): 成功率/坏源占比/错误分类。""" + +import pytest + +from tools.soak.scoreboard import ( + inv_errors_classified, + inv_fault_share, + inv_success_rate, +) + +_KNOWN = ("TransientError", "AllSourcesExhausted", "cancelled") + + +def _row(source="monkey_1", error=None, cache_hit=0): + return {"source_name": source, "error": error, "cache_hit": cache_hit} + + +class TestSuccessRate: + def test_pass_at_threshold(self): + inv_success_rate(98, 100, min_rate=0.98) + + def test_fail_below(self): + with pytest.raises(AssertionError, match="成功率"): + inv_success_rate(97, 100, min_rate=0.98) + + def test_zero_total_rejected(self): + with pytest.raises(AssertionError): + inv_success_rate(0, 0, min_rate=0.98) + + +class TestFaultShare: + def test_share_within_threshold(self): + rows = [_row()] * 9 + [_row("monkey_3")] + inv_fault_share(rows, ["monkey_3", "monkey_4"], max_share=0.15) + + def test_share_exceeds(self): + rows = [_row()] * 4 + [_row("monkey_3")] * 2 + with pytest.raises(AssertionError, match="坏源尝试占比"): + inv_fault_share(rows, ["monkey_3"], max_share=0.15) + + def test_cache_rows_excluded(self): + rows = [_row(cache_hit=1)] * 100 + [_row("monkey_3")] + [_row()] * 9 + inv_fault_share(rows, ["monkey_3"], max_share=0.15) # 缓存行不入分母 + + +class TestErrorsClassified: + def test_known_prefixes_pass(self): + rows = [_row(error="TransientError: m1 超时"), _row(error="cancelled"), _row()] + inv_errors_classified(rows, _KNOWN) + + def test_unknown_prefix_fails(self): + rows = [_row(error="KeyError: 'oops'")] + with pytest.raises(AssertionError, match="未分类"): + inv_errors_classified(rows, _KNOWN) diff --git a/tools/soak/run_soak.py b/tools/soak/run_soak.py index c9d1d7a..9e83b55 100644 --- a/tools/soak/run_soak.py +++ b/tools/soak/run_soak.py @@ -122,14 +122,18 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None: """单 worker: 消费场景生成器,信号量限并发,双预算任一命中即停。""" from polygateway import GatewayClient from polygateway.telemetry.sqlite import SQLiteRecorder - from tools.soak.scenarios import SCENARIOS, SoakCorpus env = _merged_env() run_id = args.run_id telemetry_path = _ROOT / f"data/soak/telemetry_{run_id}_{worker_idx}.db" recorder = SQLiteRecorder(telemetry_path) - client = GatewayClient.from_env(args.scope, telemetry=recorder, env=env) + if args.scenario == "P7": + from polygateway.ocr import OcrClient + + client = OcrClient.from_env(args.scope, telemetry=recorder, env=env) + else: + client = GatewayClient.from_env(args.scope, telemetry=recorder, env=env) corpus = SoakCorpus( harness_db=_ROOT / "data/soak/harness.db", telemetry_db=_ROOT / "data/soak/generate_questions_telemetry.db", @@ -148,11 +152,16 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None: async def _one(kind: str, kwargs: dict) -> None: try: - # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处) - kwargs.setdefault("cache_namespace", run_id) - resp = await client.chat(**kwargs) + if kind == "ocr_text": + await client.recognize_text(**kwargs) + elif kind == "ocr_layout": + await client.parse_layout(**kwargs) + else: + # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处) + kwargs.setdefault("cache_namespace", run_id) + resp = await client.chat(**kwargs) + stats["tokens"] += resp.prompt_tokens + resp.completion_tokens stats["ok"] += 1 - stats["tokens"] += resp.prompt_tokens + resp.completion_tokens except asyncio.CancelledError: stats["cancelled"] += 1 raise @@ -263,8 +272,11 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None: # 交报告"错误分布"人工核对(findings §4 条 2 的比例校准留待 P5 常态化) if args.scenario in ("P5", "P6"): _check("故障混编生效(P5/P6)", sb.inv_any_errors, rows) - rate = sb.structured_success_rate(rows, session_suffix="-p3") - verdicts.append(("P3 结构化成功率", f"{rate:.3f}(基线首跑建立)")) + if args.scenario == "P7": + _p7_checks(args, env, results, rows, calls, verdicts, _check) + else: + rate = sb.structured_success_rate(rows, session_suffix="-p3") + verdicts.append(("P3 结构化成功率", f"{rate:.3f}(基线首跑建立)")) report = sb.render_report(args.run_id, rows, verdicts) path = sb.write_report(args.run_id, report) print(report) @@ -273,6 +285,78 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None: raise SystemExit("硬不变量被击穿,见报告") +def _p7_checks(args, env, results, rows, calls, verdicts, check) -> None: + """P7 专属裁决(M3 计划 T8): 成功率/坏源压制/故障生效/错误分类/取消泄漏/一致性。""" + from tools.soak import scoreboard as sb + + ok_total = sum(r["stats"]["ok"] for r in results) + check("成功率 ≥98%(P7)", sb.inv_success_rate, ok_total, calls, min_rate=0.98) + fault_names = [ + s.strip() for s in env.get(f"{args.scope}_FAULT_SOURCES", "").split(",") if s.strip() + ] + check("坏源尝试占比 ≤15%(P7)", sb.inv_fault_share, rows, fault_names, max_share=0.15) + check("故障混编生效(P7)", sb.inv_fault_errors_present, rows, fault_source_names=fault_names) + check("错误全部可分类(P7)", sb.inv_errors_classified, rows, _KNOWN_ERROR_PREFIXES) + cancelled = sum(r["stats"]["cancelled"] for r in results) + verdicts.append( + ("零取消泄漏(P7)", "PASS" if cancelled == 0 else f"FAIL — cancelled={cancelled}") + ) + asyncio.run(_ocr_consistency_check(env, args.scope, verdicts)) + + +# P7 已知错误前缀(metric ocr-call-success 口径: 异常类名 + 取消哨兵) +_KNOWN_ERROR_PREFIXES = ( + "TransientError", + "SourceDeadError", + "RequestRejectedError", + "ResultInvalidError", + "AllSourcesExhausted", + "CircuitOpenError", + "GatewayUnavailableError", + "cancelled", +) + + +async def _ocr_consistency_check( + env: dict[str, str], scope: str, verdicts: list[tuple[str, str]], n: int = 20 +) -> None: + """P7 不变量⑥: tables/para_blocks 一致性抽查(跑后活检查,设计 §1.2 护栏)。""" + import io + import random + import zipfile + + import httpx + + from polygateway.transports.monkey_ocr import _parse_middle_json + + base = env[f"{scope}__MONKEY__1__BASE_URL"] + images = sorted((_ROOT / "data/soak/chs_images").glob("chs_*.jpg")) + sample = random.sample(images, min(n, len(images))) + mismatches: list[str] = [] + async with httpx.AsyncClient(base_url=base, trust_env=False, timeout=300) as client: + for path in sample: + resp = await client.post( + "/parse", files={"file": ("image.jpg", path.read_bytes(), "image/jpeg")} + ) + resp.raise_for_status() + zip_resp = await client.get(resp.json()["download_url"]) + zip_resp.raise_for_status() + elements, _ = _parse_middle_json(zip_resp.content) + lib = sorted(e.bbox for e in elements if e.type == "table") + with zipfile.ZipFile(io.BytesIO(zip_resp.content)) as archive: + member = next(m for m in archive.namelist() if m.endswith("_middle.json")) + payload = json.load(archive.open(member)) + raw = sorted( + tuple(float(v) for v in t["bbox"]) + for page in payload["pdf_info"] + for t in page.get("tables", []) + ) + if lib != raw: + mismatches.append(path.name) + verdict = "PASS" if not mismatches else f"FAIL — 不一致: {mismatches[:5]}" + verdicts.append((f"tables/para_blocks 一致性抽查 n={len(sample)}", verdict)) + + async def _redis_clock_offset_s(redis_url: str) -> float: """Redis 服务器钟相对本机钟的偏移(秒);限流窗口 id 以服务器钟为准。""" import redis.asyncio as aioredis @@ -292,7 +376,6 @@ async def _live_checks( from polygateway.backends.redis.breaker import RedisGate from polygateway.backends.redis.limiter import RedisLimiter from polygateway.config import GatewaySettings - from tools.soak import scoreboard as sb settings = GatewaySettings.from_env(args.scope, env=env) @@ -322,7 +405,9 @@ async def _live_checks( def main() -> None: parser = argparse.ArgumentParser(description="PolyGateway 真实数据压测") - parser.add_argument("--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6"]) + parser.add_argument( + "--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6", "P7"] + ) parser.add_argument("--budget-calls", type=int) parser.add_argument("--budget-tokens", type=int) parser.add_argument("--workers", type=int, default=1) diff --git a/tools/soak/scenarios.py b/tools/soak/scenarios.py index 7b32df3..b69c9dc 100644 --- a/tools/soak/scenarios.py +++ b/tools/soak/scenarios.py @@ -1,8 +1,9 @@ -"""P1-P6 场景请求生成器(findings §2 矩阵;async 生成器,产出调用参数)。 +"""P1-P7 场景请求生成器(findings §2 矩阵;async 生成器,产出调用参数)。 -每项产出 `(kind, kwargs)`: kind ∈ {"chat"};kwargs 直接喂 -`GatewayClient.chat(**kwargs)`。回放/组装场景一律掺 `cache_salt=run_id` -破缓存(缓存行为归 P4);P4 子流量特意重复 messages 且不掺 salt。 +每项产出 `(kind, kwargs)`: kind ∈ {"chat", "ocr_text", "ocr_layout"}; +chat 喂 `GatewayClient.chat(**kwargs)`,ocr_* 喂 `OcrClient` 对应方法 +(M3 计划 T8)。回放/组装场景一律掺 `cache_salt=run_id` 破缓存(缓存 +行为归 P4);P4 子流量特意重复 messages 且不掺 salt。 """ from __future__ import annotations @@ -10,10 +11,13 @@ from __future__ import annotations import random from dataclasses import dataclass from pathlib import Path -from typing import Any, AsyncIterator +from typing import TYPE_CHECKING, Any from pydantic import BaseModel +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from tools.soak.corpus import ( assemble_frame_messages, chs_image_messages, @@ -182,6 +186,19 @@ async def p6_mixed_soak( }[key](corpus, f"{run_id}-r{rng()}", rng) +async def p7_ocr( + corpus: SoakCorpus, run_id: str, rng=random.random, *, layout_ratio: float = 0.2 +) -> AsyncIterator[Item]: + """P7 OCR 双端点混合(M3 计划 T8): 真实图像语料循环,text/layout 8:2。 + + 故障性来自 SOAK_OCR 源池配置(黑洞/坏端口),生成器本身不造故障。 + """ + while True: + image = corpus.images[int(rng() * len(corpus.images)) % len(corpus.images)] + kind = "ocr_layout" if rng() < layout_ratio else "ocr_text" + yield (kind, {"image": image.read_bytes(), "session_id": f"{run_id}-p7"}) + + SCENARIOS = { "P1": p1_trace_chains, "P2": p2_multimodal_replay, @@ -189,4 +206,5 @@ SCENARIOS = { "P4": p4_cache_bidirectional, "P5": p5_fault_mixed, "P6": p6_mixed_soak, + "P7": p7_ocr, } diff --git a/tools/soak/scoreboard.py b/tools/soak/scoreboard.py index 299cdc9..a9101e8 100644 --- a/tools/soak/scoreboard.py +++ b/tools/soak/scoreboard.py @@ -9,7 +9,7 @@ from __future__ import annotations import sqlite3 from collections import Counter -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -50,7 +50,7 @@ def _admit_second(row: Row, clock_offset_s: float) -> float: dt = datetime.fromisoformat(str(row["created_at"])) if dt.tzinfo is None: # SQLite datetime('now') 落库为 UTC naive;按本机时区解析会错位整时 - dt = dt.replace(tzinfo=timezone.utc) + dt = dt.replace(tzinfo=UTC) latency_ms = row.get("latency_ms") or 0 return dt.timestamp() - float(latency_ms) / 1000.0 + clock_offset_s @@ -193,13 +193,14 @@ def write_report(run_id: str, content: str, out_dir: Path | str = "tests/outputs # —— 活后端检查与预算帽(2026-07-21 verifier I1/M6/M7 补齐)—— -SCENARIO_CALL_CAPS = { # 设计 §8.1 签字的单场景请求数上限 +SCENARIO_CALL_CAPS = { # 设计 §8.1 签字的单场景请求数上限(P7: M3 计划 T8) "P1": 500, "P2": 450, "P3": 2500, "P4": 2500, "P5": 1000, "P6": 8000, + "P7": 1500, } @@ -239,6 +240,34 @@ def inv_fault_errors_present(rows: list[Row], *, fault_source_names: list[str]) assert fault_errors, f"故障源 {fault_source_names} 零错误行——故障混编未生效" +def inv_success_rate(ok: int, total: int, *, min_rate: float) -> None: + """P7 不变量①: 调用级成功率(worker stats 口径,与 M2.5 验收同源)。""" + assert total > 0, "零调用无法评估成功率" + rate = ok / total + assert rate >= min_rate, f"成功率 {rate:.4f}({ok}/{total})< 阈值 {min_rate}" + + +def inv_fault_share(rows: list[Row], fault_source_names: list[str], *, max_share: float) -> None: + """P7 不变量②: 坏源尝试占比受健康选源压制(排除缓存行)。""" + attempts = [r for r in rows if not r.get("cache_hit")] + assert attempts, "无尝试行" + fault = sum(1 for r in attempts if r["source_name"] in fault_source_names) + share = fault / len(attempts) + assert share <= max_share, ( + f"坏源尝试占比 {share:.3f}({fault}/{len(attempts)})> 阈值 {max_share}" + ) + + +def inv_errors_classified(rows: list[Row], known_prefixes: tuple[str, ...]) -> None: + """P7 不变量⑦/⑧: 一切错误行可归入已知异常类名前缀(零未分类异常)。""" + unknown = Counter( + str(r["error"]).split(":")[0] + for r in rows + if r.get("error") and not str(r["error"]).startswith(known_prefixes) + ) + assert not unknown, f"未分类错误前缀: {dict(unknown)}" + + def inv_any_errors(rows: list[Row]) -> None: """不变量 2c 的兜底形态(P5/P6): 故障源混编池下错误行必然存在。