Agent 成本观测体系：Token、任务粒度、模型路由与预算告警

CREATE TABLE agent_usage_events (
  id BIGSERIAL PRIMARY KEY,
  task_id TEXT NOT NULL,
  project_key TEXT NOT NULL,
  user_id TEXT NOT NULL,
  model TEXT NOT NULL,
  phase TEXT NOT NULL,
  input_tokens INTEGER NOT NULL DEFAULT 0,
  output_tokens INTEGER NOT NULL DEFAULT 0,
  cached_tokens INTEGER NOT NULL DEFAULT 0,
  tool_calls INTEGER NOT NULL DEFAULT 0,
  retry_index INTEGER NOT NULL DEFAULT 0,
  latency_ms INTEGER NOT NULL DEFAULT 0,
  cost_usd NUMERIC(12, 6) NOT NULL DEFAULT 0,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_usage_project_time ON agent_usage_events(project_key, created_at);
CREATE INDEX idx_usage_task ON agent_usage_events(task_id);

model_routing:
  default_model: "gpt-4.1-mini"
  rules:
    - name: "simple_question"
      when:
        task_type: ["explain", "summarize"]
        risk_level: "low"
        estimated_files_lte: 3
      use_model: "gpt-4.1-mini"
      max_budget_usd: 0.05

    - name: "code_patch_medium"
      when:
        task_type: ["bugfix", "feature"]
        estimated_files_lte: 20
      use_model: "claude-sonnet"
      max_budget_usd: 1.50

    - name: "architecture_or_large_refactor"
      when:
        task_type: ["architecture", "large_refactor"]
      use_model: "claude-opus"
      max_budget_usd: 8.00
      requires_approval: true

from decimal import Decimal


class BudgetExceeded(Exception):
    pass


class BudgetGuard:
    def __init__(self, task_budget_usd: str, project_daily_budget_usd: str):
        self.task_budget = Decimal(task_budget_usd)
        self.project_daily_budget = Decimal(project_daily_budget_usd)

    def check_task(self, current_cost: Decimal, next_estimated_cost: Decimal) -> None:
        if current_cost + next_estimated_cost > self.task_budget:
            raise BudgetExceeded(
                f"task budget exceeded: {current_cost} + {next_estimated_cost} > {self.task_budget}"
            )

    def check_project_day(self, daily_cost: Decimal, next_estimated_cost: Decimal) -> None:
        if daily_cost + next_estimated_cost > self.project_daily_budget:
            raise BudgetExceeded(
                f"project daily budget exceeded: {daily_cost} + {next_estimated_cost}"
            )


guard = BudgetGuard(task_budget_usd="1.50", project_daily_budget_usd="50.00")
guard.check_task(Decimal("0.72"), Decimal("0.18"))

alerts:
  - name: "single_task_budget_exceeded"
    condition: "task.cost_usd > task.max_budget_usd"
    severity: "warning"
    notify: ["task_owner"]

  - name: "project_daily_cost_spike"
    condition: "project.daily_cost_usd > project.baseline_daily_cost_usd * 2"
    severity: "critical"
    notify: ["tech_lead", "engineering_manager"]

  - name: "retry_cost_ratio_high"
    condition: "project.retry_cost_ratio > 0.35"
    severity: "warning"
    notify: ["platform_team"]

  - name: "premium_model_without_approval"
    condition: "model.tier == 'premium' and approval.exists == false"
    severity: "critical"
    notify: ["security_team"]

flowchart TD
  A["发现成本异常"] --> B{"异常来自哪里"}
  B -->|上下文过大| C["裁剪上下文包"]
  B -->|模型过强| D["调整路由规则"]
  B -->|重试过多| E["增加停止条件和验收标准"]
  B -->|验证失败| F["补测试和静态检查"]
  C --> G["复测任务成本"]
  D --> G
  E --> G
  F --> G
  G --> H["更新模板和预算策略"]