xiaohei/Agentswarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

用量按 review-retry 成本归属：usage 事件标 cost_phase（Refs #16）

Browse Source 
emit_usage_event 给每条 budget.alert 用量事件加 cost_phase：任务进入
run.metadata["rework_attributions"]（cross-review 已 reopen）后的再执行标 review_retry，
否则 initial；并带 attempt(retry_count)。初次执行先于任何 attribution → initial，redo 在
run_cross_review 记录 attribution 之后 → review_retry，时序天然正确。计费/采集侧按 swarm_id
聚合时可用 cost_phase 拆分 initial vs review_retry 成本，满足「review retry 成本可归属」。

usage-billing-schema §4.2/§5/§6 同步：payload 加 cost_phase/attempt，§5 改为已打标，§6 缺口表
review_retry 标 ✅。测试 scripts/test-usage-cost-phase.py 接入 CI。

影响范围：仅 agent_swarm（usage 事件加字段 + 文档 + 测试）；向后兼容；不改鉴权/契约/审批链/密钥。

Refs #16

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Songhaoz666
2026-06-11 09:44:36 +08:00
co-authored by Claude Opus 4.8
parent 5138d9a370
commit 56e080bfa1
4 changed files with 108 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+4
View File
@@ -63,6 +63,10 @@ jobs:
env: { REDIS_FAKE: "1" }
run: python scripts/test-contract-events.py
- name: Usage review-retry cost attribution (#16)
env: { REDIS_FAKE: "1" }
run: python scripts/test-usage-cost-phase.py
- name: Audit / lineage trace (replayable) (#17)
env: { REDIS_FAKE: "1" }
run: python scripts/test-audit-trace.py
docs/integration/usage-billing-schema.md
+4 -2
View File
@@ -62,6 +62,8 @@ Swarm 在运行中按时长/成本比例发 `budget.alert`（默认 80% 阈值
{
"model_id": "...", "model_tokens": 0, "prompt_tokens": 0, "completion_tokens": 0,
"model_cost_usd": 0.0, "runtime_seconds": 0.0, "billing_source": "...",
"cost_phase": "initial | review_retry", // #16：返工再执行的用量标 review_retry，供成本归属
"attempt": 0, // 该任务的重做次数（retry_count）
"manager_deployment_id": "...", "swarm_id": "...", "task_id": "...",
"agent_role": "...", "correlation_id": "...",
"budget": { "max_tokens": null, "max_cost_usd": null, "consumed_usd": 0.0, "remaining_usd": null }
@@ -84,7 +86,7 @@ PayPal 说明 §5 给出的运行时用量回传目标：
## 5. 多 Agent / 评审重做 成本聚合
- 一次 swarm 请求拆成多 task / 多 agent / 多评审轮；**每个 task 的 `usage` 可按 `swarm_id` 聚合**得到运行级合计（观测用途）。
- **评审重做成本**：每轮重做都会重新执行任务并累计 `usage`，因此**已隐含计入**运行级合计；但当前**未单独打标**「review_retry 成本」。
- **评审重做成本（#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` 反推。
## 6. 字段覆盖与缺口
@@ -96,7 +98,7 @@ PayPal 说明 §5 给出的运行时用量回传目标：
| `reasoning_tokens` / `cache_tokens` | 🔴 未采集（取决于 provider usage 返回） |
| `tool_cost`（工具调用成本） | 🔴 未采集（无 SK 工具计量） |
| `cpu_core_seconds` / `memory_mb_seconds`（基础设施） | 🔴 未采集（K8s 指标未接入账本） |
| `review_retry` 成本单独打标 | 🟡 隐含累计，未单独标注 |
| `review_retry` 成本单独打标 | ✅ 已打标（usage 事件 `cost_phase=review_retry`，#16） |
| provider 成本拆分 | 🟡 由 NewAPI/账本侧负责，非本仓 |
| tenant 归因 | ❌ 按标准不使用（见 §1） |
orchestrator/main.py
+11
View File
@@ -1701,6 +1701,15 @@ async def emit_usage_event(run, task, agent_id: str, result: Any):
plan = run.request_body.get("orchestration_plan") or {}
budget = plan.get("budget") or {}
# Review-retry cost attribution (#16): a task's usage is `review_retry` once cross-review has
# reopened it for rework (it appears in run.metadata["rework_attributions"]); otherwise
# `initial`. The first execution emits before any attribution exists → "initial"; each redo
# emits after run_cross_review recorded the attribution → "review_retry". So billing can split
# initial vs review-retry cost by summing usage events per cost_phase (usage-billing §5).
rework_targets = {
a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])
}
cost_phase = "review_retry" if task.task_id in rework_targets else "initial"
payload = {
"model_id": usage.get("model_id") or task.context.get("model_id"),
"model_tokens": usage.get("model_tokens", 0),
@@ -1709,6 +1718,8 @@ async def emit_usage_event(run, task, agent_id: str, result: Any):
"model_cost_usd": usage.get("model_cost_usd", 0),
"runtime_seconds": usage.get("runtime_seconds", 0),
"billing_source": usage.get("billing_source", "unknown"),
"cost_phase": cost_phase, # "initial" | "review_retry" (#16 attribution)
"attempt": task.retry_count, # redo count for this task
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"task_id": task.task_id,
scripts/test-usage-cost-phase.py
+89
View File
@@ -0,0 +1,89 @@
"""Usage review-retry cost attribution test (issue #16).
Verifies emit_usage_event tags each usage event with `cost_phase`:
* "initial" before cross-review has reopened the task for rework;
* "review_retry" once the task appears in run.metadata["rework_attributions"]
(i.e. its re-execution cost is attributable as review-retry cost).
Hermetic: REDIS_FAKE, no model key, no callback url.
Run from agent_swarm_v6 (install deps first):
pip install -r orchestrator/requirements.txt
REDIS_FAKE=1 python scripts/test-usage-cost-phase.py
"""
import asyncio
import json
import os
import sys
from pathlib import Path
os.environ["REDIS_FAKE"] = "1"
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
failures = []
def check(name, cond):
print(("PASS" if cond else "FAIL"), "-", name)
if not cond:
failures.append(name)
async def budget_events(swarm_runtime, swarm_id):
raw = await swarm_runtime.list_events(swarm_id, limit=500)
return [e for e in raw["events"] if e["event_type"] == "budget.alert"]
async def main():
from orchestrator.redis_client import redis_client
from orchestrator.swarm_runtime import swarm_runtime
from orchestrator.task_queue import task_queue
from orchestrator import main as orch
await redis_client.connect()
body = {"mode": "swarm", "orchestration_plan": {"objective": "cost phase"},
"callback": {"url": "", "subscribed_events": []},
"metadata": {"manager_deployment_id": "m-cost"}}
run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="c-cost")
t1 = await task_queue.create_task(task_id=f"{run.swarm_id}-t1", description="impl",
agent_role="impl", required_capabilities=["python"], enqueue=True)
await swarm_runtime.attach_task(run, t1.task_id)
usage_result = {"usage": {"model_id": "gpt-x", "model_tokens": 100, "model_cost_usd": 0.01,
"billing_source": "newapi"}}
# 1) Initial execution: no rework attribution yet → cost_phase=initial.
await orch.emit_usage_event(run, t1, "agent-1", usage_result)
evs = await budget_events(swarm_runtime, run.swarm_id)
check("initial usage tagged cost_phase=initial", evs and evs[-1]["payload"].get("cost_phase") == "initial")
check("usage payload carries attempt", "attempt" in evs[-1]["payload"])
# 2) Cross-review reopened t1 for rework → recorded in rework_attributions.
run.metadata["rework_attributions"] = [{"target_task_id": t1.task_id, "root_cause": "test"}]
await swarm_runtime.save_run(run)
t1.retry_count = 1
await orch.emit_usage_event(run, t1, "agent-1", usage_result)
evs = await budget_events(swarm_runtime, run.swarm_id)
check("redo usage tagged cost_phase=review_retry", evs[-1]["payload"].get("cost_phase") == "review_retry")
check("review_retry usage attributable by swarm_id + cost_phase",
any(e["payload"].get("cost_phase") == "review_retry" and e["payload"].get("swarm_id") == run.swarm_id
for e in evs))
# A non-reworked task stays initial even alongside the reworked one.
t2 = await task_queue.create_task(task_id=f"{run.swarm_id}-t2", description="doc",
agent_role="documentation", required_capabilities=["technical-writing"], enqueue=True)
await swarm_runtime.attach_task(run, t2.task_id)
await orch.emit_usage_event(run, t2, "agent-2", usage_result)
evs = await budget_events(swarm_runtime, run.swarm_id)
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")
print()
if failures:
print(f"{len(failures)} usage cost-phase check(s) FAILED: {failures}")
sys.exit(1)
print("all usage cost-phase checks passed")
if __name__ == "__main__":
asyncio.run(main())