From e4a120b715c2f967cc37397e3ac257b9e1a4d5d0 Mon Sep 17 00:00:00 2001 From: Songhaoz666 Date: Thu, 11 Jun 2026 14:36:07 +0800 Subject: [PATCH] =?UTF-8?q?/metrics=20=E6=9A=B4=E9=9C=B2=20run=20=E7=BA=A7?= =?UTF-8?q?=20cost=5Fby=5Fphase{initial,review=5Fretry}=EF=BC=88Refs=20#16?= =?UTF-8?q?=20#37=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 回应 @Mem0ried 在 PR #37 的对接问题：除逐条 budget.alert 事件的 cost_phase 外，详情 聚合也需能拆 initial/review_retry。build_runtime_metrics 新增 cost_by_phase 滚动汇总 （按 swarm_id，从各任务 usage.model_cost_usd/model_tokens 按 rework_attributions 归类）， GET …/{id}/metrics 返回 {initial,review_retry}:{cost_usd,model_tokens}。HM 详情聚合 / 客户端 08 用量抽屉可直接展示拆分，无需从事件推导。 测试 test-usage-cost-phase.py 增 /metrics 断言（reworked→review_retry、非 reworked→initial）； usage-billing §5 记录该字段。 Refs #16 Refs #37 Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/integration/usage-billing-schema.md | 1 + orchestrator/main.py | 13 +++++++++++++ scripts/test-usage-cost-phase.py | 12 ++++++++++++ 3 files changed, 26 insertions(+) diff --git a/docs/integration/usage-billing-schema.md b/docs/integration/usage-billing-schema.md index 2fc6707..4e32902 100644 --- a/docs/integration/usage-billing-schema.md +++ b/docs/integration/usage-billing-schema.md @@ -87,6 +87,7 @@ PayPal 说明 §5 给出的运行时用量回传目标： - 一次 swarm 请求拆成多 task / 多 agent / 多评审轮；**每个 task 的 `usage` 可按 `swarm_id` 聚合**得到运行级合计（观测用途）。 - **评审重做成本（#16，已打标）**：每轮重做重新执行任务并累计 `usage`，且其用量事件带 `cost_phase="review_retry"`（cross-review 已把该任务记入 `rework_attributions` 之后的再执行）；初次执行为 `cost_phase="initial"`。计费/采集侧按 `swarm_id` 聚合时，可用 `cost_phase` 拆分 **initial vs review_retry 成本**——「review retry 成本可归属」由此满足，无需从 `retry_count` 反推。 +- **详情聚合（#37）**：除逐条事件的 `cost_phase` 外，`GET …/{id}/metrics` 另返回 run 级滚动汇总 `cost_by_phase: { initial: {cost_usd, model_tokens}, review_retry: {cost_usd, model_tokens} }`，供 HM 详情用量聚合 / 客户端用量抽屉直接展示「初次 vs 返工」拆分，无需从事件流推导。 ## 6. 字段覆盖与缺口 diff --git a/orchestrator/main.py b/orchestrator/main.py index 53dc711..0c1e1d6 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -1886,6 +1886,18 @@ async def build_runtime_metrics(run, window: str, step: str) -> Dict[str, Any]: budget = ((run.request_body.get("orchestration_plan") or {}).get("budget") or {}) duration_budget = budget.get("duration_seconds") or budget.get("max_duration_seconds") budget_ratio = duration_seconds / float(duration_budget) if duration_budget else None + # Run-level cost split by phase (#16/#37): lets HM's detail usage aggregation show + # initial vs review_retry without deriving from the event stream. Same attribution as + # emit_usage_event — a task counts as review_retry once cross-review recorded it in + # run.metadata["rework_attributions"]. + rework_targets = {a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])} + cost_by_phase = {"initial": {"cost_usd": 0.0, "model_tokens": 0}, + "review_retry": {"cost_usd": 0.0, "model_tokens": 0}} + for task in tasks: + usage = (parse_task_result(task) or {}).get("usage") or {} + phase = "review_retry" if task.task_id in rework_targets else "initial" + cost_by_phase[phase]["cost_usd"] += float(usage.get("model_cost_usd") or 0.0) + cost_by_phase[phase]["model_tokens"] += int(usage.get("model_tokens") or 0) return { "deployment_id": run.deployment_id, "swarm_id": run.swarm_id, @@ -1903,6 +1915,7 @@ async def build_runtime_metrics(run, window: str, step: str) -> Dict[str, Any]: sum(completed_durations) / len(completed_durations) if completed_durations else 0 ), + "cost_by_phase": cost_by_phase, # {initial,review_retry}:{cost_usd,model_tokens} (#16/#37) "budget": { "duration_seconds": duration_budget, "duration_ratio": budget_ratio, diff --git a/scripts/test-usage-cost-phase.py b/scripts/test-usage-cost-phase.py index 00c3861..12716d6 100644 --- a/scripts/test-usage-cost-phase.py +++ b/scripts/test-usage-cost-phase.py @@ -78,6 +78,18 @@ async def main(): t2_ev = [e for e in evs if e["payload"].get("task_id") == t2.task_id][-1] check("non-reworked task stays cost_phase=initial", t2_ev["payload"].get("cost_phase") == "initial") + # --- /metrics run-level cost_by_phase rollup (#37) --- + # Persist task results with usage so build_runtime_metrics can split them; t1 is a rework target. + await task_queue.complete_task(t1.task_id, result=json.dumps({"usage": {"model_cost_usd": 0.05, "model_tokens": 500}})) + await task_queue.complete_task(t2.task_id, result=json.dumps({"usage": {"model_cost_usd": 0.02, "model_tokens": 200}})) + refreshed = await swarm_runtime.get_run(run.swarm_id) + metrics = await orch.build_runtime_metrics(refreshed, "15m", "60s") + cbp = metrics.get("cost_by_phase") or {} + check("/metrics exposes cost_by_phase", set(cbp.keys()) == {"initial", "review_retry"}) + check("/metrics review_retry cost = reworked task (t1=0.05)", abs(cbp["review_retry"]["cost_usd"] - 0.05) < 1e-9) + check("/metrics initial cost = non-reworked task (t2=0.02)", abs(cbp["initial"]["cost_usd"] - 0.02) < 1e-9) + check("/metrics review_retry tokens = 500", cbp["review_retry"]["model_tokens"] == 500) + print() if failures: print(f"{len(failures)} usage cost-phase check(s) FAILED: {failures}")