feat(benchmark): 阶段2 数据采集 — 真实 run 终态自动采集 SwarmMetrics 并落库(+可选导出)
把采集器接进真实运行，让数据自己累积，为后续【经验标定】O/系数/S_gain 攒真实 用户数据（这些量只能由真实运行事后标定，不能先验写死）。本阶段不编任何公式。 - orchestrator/main.py：run 终态(completed/failed)在终态事件后调 capture_run_metrics。 只读 run 状态、try/except 包裹绝不失败 run；默认开，BENCHMARK_CAPTURE=0 可关。 - orchestrator/swarm_runtime.py：record_benchmark_metrics → 落 run.metadata['benchmark']。 - benchmark/collectors/capture.py：collect(run_collector) → 持久化 → 可选导出(to_thread)。 - benchmark/export/：MetricsExporter；默认 NoopExporter(无依赖/无凭据)； BENCHMARK_EXPORT_TARGET=blob 启用 Azure Blob(连接串或 Workload/Managed Identity， 凭据经环境注入，绝不写进代码)，归档 <scenario>/<swarm_id>.json。 诚实：缺项指标 NaN→null，不伪造(规则#9)。 验证：新增 test-benchmark-capture.py + 现有 collector/selfcert + 契约冒烟(runtime/ merge/freeze)全 PASS。 影响范围：agent_swarm。运行时新增**只读**终态钩子(不改派发/执行/契约/计费/审计字段)； 导出默认关，无密钥落地。Client/Manager/Agnet/CodeGW/发布链路不涉及。 依赖：benchmark/ 需在镜像内(quality.py 早已 import benchmark；由 #44 Dockerfile 修复覆盖)。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Fasthei
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
b5cc68c977
commit
94ef894ace
@@ -0,0 +1,53 @@
|
||||
"""阶段2：真实 run 终态 → 自动采集 SwarmMetrics 并落库（+ 可选导出存储账户）。
|
||||
|
||||
把「采集器」变成「会自己攒数据的管道」：每次真实蜂群 run 跑到终态时，用
|
||||
`SwarmRunMetricsCollector` 算出该 run 的真实指标（缺项 NaN，**不编公式、不伪造**），
|
||||
持久化到 `run.metadata['benchmark']`（Redis），并按 `BENCHMARK_EXPORT_TARGET` 可选导出。
|
||||
|
||||
目的：为后续【经验标定】O / BASE_COEFFICIENTS / S_gain 累积**真实用户**数据——这些量
|
||||
只能由真实运行数据事后标定，无法先验写死（见 docs/benchmark/emergence-evaluation.md §6/§7）。
|
||||
|
||||
只读 run 状态，绝不抛进 run 主路径（调用方包 try/except）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
|
||||
from .run_collector import SwarmRunMetricsCollector
|
||||
from ..export import get_exporter
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _num(x):
|
||||
return None if (isinstance(x, float) and math.isnan(x)) else x
|
||||
|
||||
|
||||
def _metrics_dict(metrics) -> dict:
|
||||
return {k: _num(getattr(metrics, k)) for k in metrics.__dataclass_fields__}
|
||||
|
||||
|
||||
async def capture_run_metrics(run) -> dict:
|
||||
"""Collect + persist (+ optionally export) one terminal run's SwarmMetrics. Returns the payload."""
|
||||
from orchestrator.swarm_runtime import swarm_runtime # lazy: avoid import cycle
|
||||
|
||||
collector = SwarmRunMetricsCollector(run.swarm_id)
|
||||
metrics = await collector.collect()
|
||||
payload = {
|
||||
"swarm_id": run.swarm_id,
|
||||
"scenario": (run.metadata or {}).get("scenario"),
|
||||
"status": run.status,
|
||||
"metrics": _metrics_dict(metrics),
|
||||
"coverage": collector.coverage,
|
||||
}
|
||||
await swarm_runtime.record_benchmark_metrics(run, payload)
|
||||
|
||||
# Export is best-effort and OFF by default (NoopExporter). Run in a thread so a blocking
|
||||
# blob upload never stalls the event loop; never fails the run.
|
||||
try:
|
||||
await asyncio.to_thread(get_exporter().export, run.swarm_id, payload)
|
||||
except Exception as exc: # pragma: no cover - export is opt-in
|
||||
logger.warning("benchmark export failed for run %s: %s", run.swarm_id, exc)
|
||||
return payload
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Benchmark metrics 导出器（阶段2 数据采集）。
|
||||
|
||||
把每次真实 run 终态采集到的 `SwarmMetrics` 快照导出到外部留存，供后续【经验标定】
|
||||
（O / BASE_COEFFICIENTS / S_gain）累积数据。**默认 NoopExporter**：不导出、无外部依赖、
|
||||
不需任何凭据——指标仍持久化在 `run.metadata['benchmark']`（Redis）。
|
||||
|
||||
启用 Azure Blob 导出（凭据经环境/连接串注入，**绝不写进代码或日志**）：
|
||||
BENCHMARK_EXPORT_TARGET=blob
|
||||
BENCHMARK_BLOB_CONTAINER=benchmark-selfcert # 默认 benchmark-selfcert
|
||||
AZURE_STORAGE_CONNECTION_STRING=... # 方式一：连接串
|
||||
# 或方式二（无密钥，推荐）：账户 URL + Workload Identity / Managed Identity
|
||||
BENCHMARK_BLOB_ACCOUNT_URL=https://heicode.blob.core.windows.net
|
||||
|
||||
归档路径：`<scenario>/<swarm_id>.json`。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MetricsExporter:
|
||||
"""导出一次 run 的 benchmark 指标快照。子类实现 export()。"""
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class NoopExporter(MetricsExporter):
|
||||
"""默认：不导出（指标已落 run.metadata）。无外部依赖、无凭据。"""
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
logger.debug("benchmark export disabled (BENCHMARK_EXPORT_TARGET=none); "
|
||||
"run %s metrics persisted to run.metadata only", swarm_id)
|
||||
|
||||
|
||||
class BlobExporter(MetricsExporter):
|
||||
"""Azure Blob 导出（懒加载 SDK；凭据来自连接串或 Managed/Workload Identity）。"""
|
||||
|
||||
def __init__(self, container: str, *, connection_string: Optional[str] = None,
|
||||
account_url: Optional[str] = None):
|
||||
if not (connection_string or account_url):
|
||||
raise RuntimeError("BlobExporter 需 AZURE_STORAGE_CONNECTION_STRING 或 BENCHMARK_BLOB_ACCOUNT_URL")
|
||||
self.container = container
|
||||
self.connection_string = connection_string
|
||||
self.account_url = account_url
|
||||
|
||||
def _client(self):
|
||||
from azure.storage.blob import BlobServiceClient # lazy: keep core import-light & key-free
|
||||
if self.connection_string:
|
||||
return BlobServiceClient.from_connection_string(self.connection_string)
|
||||
from azure.identity import DefaultAzureCredential # Workload/Managed Identity, no secret
|
||||
return BlobServiceClient(account_url=self.account_url, credential=DefaultAzureCredential())
|
||||
|
||||
def export(self, swarm_id: str, payload: dict) -> None:
|
||||
scenario = (payload.get("scenario") or "unknown")
|
||||
blob_name = f"{scenario}/{swarm_id}.json"
|
||||
client = self._client().get_blob_client(container=self.container, blob=blob_name)
|
||||
client.upload_blob(json.dumps(payload, ensure_ascii=False), overwrite=True)
|
||||
logger.info("benchmark metrics exported → %s/%s", self.container, blob_name)
|
||||
|
||||
|
||||
def get_exporter() -> MetricsExporter:
|
||||
"""按 BENCHMARK_EXPORT_TARGET 选择导出器（默认 none → NoopExporter）。"""
|
||||
target = (os.getenv("BENCHMARK_EXPORT_TARGET") or "none").strip().lower()
|
||||
if target in ("none", "off", ""):
|
||||
return NoopExporter()
|
||||
if target == "blob":
|
||||
return BlobExporter(
|
||||
container=os.getenv("BENCHMARK_BLOB_CONTAINER", "benchmark-selfcert"),
|
||||
connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING"),
|
||||
account_url=os.getenv("BENCHMARK_BLOB_ACCOUNT_URL"),
|
||||
)
|
||||
logger.warning("unknown BENCHMARK_EXPORT_TARGET=%s; falling back to noop", target)
|
||||
return NoopExporter()
|
||||
@@ -669,6 +669,18 @@ async def refresh_swarm_run_status(run):
|
||||
await swarm_runtime.emit_event(run, "timeline.updated", payload=timeline_payload)
|
||||
await maybe_emit_budget_alert(run)
|
||||
|
||||
# Benchmark data capture (阶段2): on a terminal run, compute + persist the run's SwarmMetrics
|
||||
# so real-user data accrues for later EMPIRICAL calibration (O / coefficients / S_gain — these
|
||||
# cannot be defined a priori, only fitted from real runs). Read-only over run state; never
|
||||
# fails the run. Default-on; set BENCHMARK_CAPTURE=0 to disable. Export is separately gated
|
||||
# (BENCHMARK_EXPORT_TARGET, off by default) — see benchmark/export.
|
||||
if (os.getenv("BENCHMARK_CAPTURE", "on").strip().lower() not in ("0", "off", "false", "no")):
|
||||
try:
|
||||
from benchmark.collectors.capture import capture_run_metrics
|
||||
await capture_run_metrics(run)
|
||||
except Exception as exc:
|
||||
logger.warning("benchmark capture failed for run %s: %s", run.swarm_id, exc)
|
||||
|
||||
|
||||
# Lifespan context manager
|
||||
@asynccontextmanager
|
||||
|
||||
@@ -420,6 +420,15 @@ class SwarmRuntime:
|
||||
run.quality = quality or {}
|
||||
await self.save_run(run)
|
||||
|
||||
async def record_benchmark_metrics(self, run: SwarmRun, payload: Dict[str, Any]) -> None:
|
||||
"""Persist a terminal run's benchmark SwarmMetrics snapshot (阶段2 数据采集).
|
||||
|
||||
Stored on run.metadata['benchmark'] so real-user runs accrue data for later empirical
|
||||
calibration (O / coefficients). Read-only over run state; never fabricates (NaN → null).
|
||||
"""
|
||||
run.metadata["benchmark"] = payload or {}
|
||||
await self.save_run(run)
|
||||
|
||||
async def record_decision(self, run: SwarmRun, decision: Dict[str, Any]) -> None:
|
||||
"""Append one ACO assignment decision (Group A telemetry). Bounded list."""
|
||||
run.decisions = (run.decisions or [])[-999:] + [decision]
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Test 阶段2 benchmark data capture: terminal run → SwarmMetrics persisted to run.metadata.
|
||||
|
||||
Hermetic (REDIS_FAKE), default NoopExporter (no storage dependency). Verifies:
|
||||
- capture_run_metrics persists run.metadata['benchmark'] with metrics + coverage.
|
||||
- real metrics are numbers; uncollected ones are null (NaN→null, never fabricated).
|
||||
- default export is a no-op and does not fail the capture.
|
||||
|
||||
Run: python scripts/test-benchmark-capture.py
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
os.environ["REDIS_FAKE"] = "1"
|
||||
os.environ.pop("BENCHMARK_EXPORT_TARGET", None) # ensure default Noop
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from orchestrator.redis_client import redis_client
|
||||
from orchestrator import swarm_runtime as sr_mod
|
||||
from orchestrator.swarm_runtime import swarm_runtime
|
||||
from orchestrator.task_queue import task_queue, TaskStatus
|
||||
from benchmark.collectors.capture import capture_run_metrics
|
||||
|
||||
failures = []
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(("PASS" if cond else "FAIL"), "-", name)
|
||||
if not cond:
|
||||
failures.append(name)
|
||||
|
||||
|
||||
async def _noop(self, *a, **k):
|
||||
return None
|
||||
|
||||
|
||||
async def add_task(run, task_id, *, status, agent, cost=0.0, retry=0, depends_on=None):
|
||||
t = await task_queue.create_task(task_id=task_id, description=task_id,
|
||||
agent_role=task_id.split("-")[-1], depends_on=depends_on or [],
|
||||
enqueue=False)
|
||||
t.status = status
|
||||
t.assigned_agent_id = agent
|
||||
t.retry_count = retry
|
||||
t.result = json.dumps({"usage": {"model_cost_usd": cost}})
|
||||
await task_queue._save_task(t)
|
||||
await swarm_runtime.attach_task(run, t.task_id)
|
||||
return t
|
||||
|
||||
|
||||
async def main():
|
||||
await redis_client.connect()
|
||||
sr_mod.SwarmRuntime._post_callback = _noop
|
||||
|
||||
body = {
|
||||
"mode": "swarm",
|
||||
"requirement": {"objective": "capture test"},
|
||||
"orchestration_plan": {"budget": {"max_cost_usd": 10}},
|
||||
"callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
|
||||
"metadata": {"manager_deployment_id": "m-cap", "scenario": "coding"},
|
||||
}
|
||||
run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="c")
|
||||
await add_task(run, "t-implementation", status=TaskStatus.COMPLETED, agent="A", cost=2.0)
|
||||
await add_task(run, "t-testing", status=TaskStatus.COMPLETED, agent="B", cost=3.0)
|
||||
|
||||
payload = await capture_run_metrics(run)
|
||||
|
||||
# persisted onto the run
|
||||
reloaded = await swarm_runtime.get_run(run.swarm_id)
|
||||
bench = (reloaded.metadata or {}).get("benchmark")
|
||||
check("run.metadata['benchmark'] persisted", isinstance(bench, dict))
|
||||
check("payload scenario carried", bench.get("scenario") == "coding")
|
||||
check("metrics dict present", isinstance(bench.get("metrics"), dict))
|
||||
check("coverage dict present", isinstance(bench.get("coverage"), dict))
|
||||
# s_completion real (2/2*100 = 100), serialized as a number
|
||||
check("s_completion real (100)", bench["metrics"].get("s_completion") == 100.0
|
||||
and bench["coverage"].get("s_completion") is True)
|
||||
# uncollected metric (gain) → null, coverage False (no fabrication)
|
||||
check("s_gain null + coverage False", bench["metrics"].get("s_gain") is None
|
||||
and bench["coverage"].get("s_gain") is False)
|
||||
# JSON-serializable (would raise on NaN-as-float being non-serializable only with allow_nan=False;
|
||||
# we converted NaN→None, so strict dumps must succeed)
|
||||
try:
|
||||
json.dumps(payload, allow_nan=False)
|
||||
check("payload strictly JSON-serializable (no NaN)", True)
|
||||
except ValueError:
|
||||
check("payload strictly JSON-serializable (no NaN)", False)
|
||||
|
||||
print()
|
||||
if failures:
|
||||
print(f"{len(failures)} capture check(s) FAILED: {failures}")
|
||||
sys.exit(1)
|
||||
print("all benchmark capture checks passed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user