Agent 成本周报：如何按项目、成员、任务类型拆账

-- 成本事件表：记录每次 API 调用的成本
CREATE TABLE cost_events (
    id              BIGSERIAL PRIMARY KEY,
    event_time      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    
    -- 任务维度
    task_id         VARCHAR(64) NOT NULL,
    task_type       VARCHAR(32) NOT NULL,       -- bugfix / review / refactor / test / docs
    task_priority   VARCHAR(8),                  -- P0 / P1 / P2 / P3
    
    -- 项目维度
    project_id      VARCHAR(64) NOT NULL,
    project_name    VARCHAR(128),
    repo_name       VARCHAR(256),
    
    -- 人员维度
    user_id         VARCHAR(64) NOT NULL,
    user_name       VARCHAR(128),
    team_id         VARCHAR(64),
    team_name       VARCHAR(128),
    
    -- 模型维度
    model_id        VARCHAR(64) NOT NULL,        -- claude-sonnet-4-6 / claude-opus-4-8 / ...
    model_tier      VARCHAR(16),                 -- haiku / sonnet / opus
    provider        VARCHAR(32),                 -- anthropic / openai
    
    -- Token 维度
    input_tokens    INTEGER NOT NULL,
    output_tokens   INTEGER NOT NULL,
    cache_read_tokens  INTEGER DEFAULT 0,        -- Prompt Cache 命中的 Token
    cache_write_tokens INTEGER DEFAULT 0,        -- Prompt Cache 写入的 Token
    
    -- 成本
    input_cost      DECIMAL(10,6) NOT NULL,
    output_cost     DECIMAL(10,6) NOT NULL,
    cache_discount  DECIMAL(10,6) DEFAULT 0,     -- Cache 节省的成本
    total_cost      DECIMAL(10,6) NOT NULL,
    
    -- 路由信息
    routing_rule    VARCHAR(64),                 -- 匹配的路由规则
    fallback_used   BOOLEAN DEFAULT FALSE,       -- 是否触发了 fallback 升级
    
    -- 质量信息
    success         BOOLEAN DEFAULT TRUE,
    quality_score   DECIMAL(3,2),                -- 0.00 - 1.00
    retry_count     INTEGER DEFAULT 0,
    
    -- 元信息
    metadata        JSONB DEFAULT '{}'
);

-- 索引：按时间范围查询
CREATE INDEX idx_cost_events_time ON cost_events (event_time);
CREATE INDEX idx_cost_events_project ON cost_events (project_id, event_time);
CREATE INDEX idx_cost_events_user ON cost_events (user_id, event_time);
CREATE INDEX idx_cost_events_model ON cost_events (model_tier, event_time);
CREATE INDEX idx_cost_events_task_type ON cost_events (task_type, event_time);

-- 周报查询：按项目、团队、任务类型多维聚合
WITH weekly_summary AS (
    SELECT
        -- 时间范围
        date_trunc('week', event_time) AS week_start,
        
        -- 项目维度
        project_name,
        team_name,
        
        -- 任务维度
        task_type,
        
        -- 模型维度
        model_tier,
        
        -- 成本汇总
        COUNT(*) AS call_count,
        SUM(input_tokens) AS total_input_tokens,
        SUM(output_tokens) AS total_output_tokens,
        SUM(total_cost) AS total_cost,
        SUM(cache_discount) AS total_cache_savings,
        
        -- 质量指标
        AVG(quality_score) AS avg_quality,
        SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) AS failure_count,
        SUM(retry_count) AS total_retries,
        
        -- 重试成本（重试浪费的成本）
        SUM(CASE WHEN retry_count > 0 THEN total_cost * retry_count / (retry_count + 1) ELSE 0 END) AS wasted_cost
        
    FROM cost_events
    WHERE event_time >= NOW() - INTERVAL '7 days'
    GROUP BY 1, 2, 3, 4, 5
)
SELECT * FROM weekly_summary
ORDER BY total_cost DESC;

# app/reports/weekly_cost_report.py
"""
Agent 成本周报生成器。
按项目、团队、任务类型、模型四个维度生成拆账报告。
"""
import json
from datetime import date, timedelta
from dataclasses import dataclass

@dataclass
class WeeklyCostReport:
    week_start: date
    week_end: date
    
    # 总览
    total_cost: float
    total_calls: int
    total_tokens: int
    cost_change_pct: float          # vs 上周
    top_cost_project: str
    top_cost_project_amount: float
    
    # 按项目
    by_project: list[dict]
    
    # 按团队
    by_team: list[dict]
    
    # 按任务类型
    by_task_type: list[dict]
    
    # 按模型
    by_model: list[dict]
    
    # 异常检测
    anomalies: list[dict]
    
    # 优化建议
    recommendations: list[str]

def generate_weekly_report(week_start: date) -> WeeklyCostReport:
    """生成周报"""
    week_end = week_start + timedelta(days=6)
    
    # 1. 总览数据
    overview = db.query("""
        SELECT 
            SUM(total_cost) as total_cost,
            COUNT(*) as total_calls,
            SUM(input_tokens + output_tokens) as total_tokens
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s
    """, (week_start, week_end))
    
    # 2. 按项目拆账
    by_project = db.query("""
        SELECT 
            project_name,
            COUNT(*) as calls,
            SUM(total_cost) as cost,
            AVG(quality_score) as avg_quality,
            team_name
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s
        GROUP BY project_name, team_name
        ORDER BY cost DESC
    """, (week_start, week_end))
    
    # 3. 按任务类型拆账
    by_task_type = db.query("""
        SELECT 
            task_type,
            COUNT(*) as calls,
            SUM(total_cost) as cost,
            model_tier as primary_model,
            AVG(quality_score) as avg_quality
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s
        GROUP BY task_type, model_tier
        ORDER BY cost DESC
    """, (week_start, week_end))
    
    # 4. 按模型拆账
    by_model = db.query("""
        SELECT 
            model_tier,
            COUNT(*) as calls,
            SUM(total_cost) as cost,
            SUM(cache_discount) as cache_savings,
            AVG(total_cost) as avg_cost_per_call
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s
        GROUP BY model_tier
        ORDER BY cost DESC
    """, (week_start, week_end))
    
    # 5. 异常检测
    anomalies = detect_anomalies(week_start, week_end)
    
    # 6. 生成优化建议
    recommendations = generate_recommendations(by_project, by_task_type, by_model)
    
    return WeeklyCostReport(
        week_start=week_start,
        week_end=week_end,
        total_cost=overview["total_cost"],
        total_calls=overview["total_calls"],
        total_tokens=overview["total_tokens"],
        cost_change_pct=calc_change_pct(overview["total_cost"], week_start),
        top_cost_project=by_project[0]["project_name"] if by_project else "",
        top_cost_project_amount=by_project[0]["cost"] if by_project else 0,
        by_project=by_project,
        by_team=aggregate_by_team(by_project),
        by_task_type=by_task_type,
        by_model=by_model,
        anomalies=anomalies,
        recommendations=recommendations,
    )

def detect_anomalies(week_start: date, week_end: date) -> list:
    """异常检测"""
    anomalies = []
    
    # 1. 单任务成本异常（超过均值 5 倍）
    avg_cost = db.query("SELECT AVG(total_cost) FROM cost_events WHERE event_time BETWEEN %s AND %s",
                         (week_start, week_end))
    high_cost_tasks = db.query("""
        SELECT task_id, task_type, total_cost, user_name, project_name
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s AND total_cost > %s * 5
        ORDER BY total_cost DESC
        LIMIT 10
    """, (week_start, week_end, avg_cost))
    
    for task in high_cost_tasks:
        anomalies.append({
            "type": "high_cost_task",
            "severity": "HIGH" if task["total_cost"] > avg_cost * 10 else "MEDIUM",
            "message": f"任务 {task['task_id']} 成本 ${task['total_cost']:.2f}（均值 ${avg_cost:.2f} 的 {task['total_cost']/avg_cost:.1f} 倍）",
            "task_id": task["task_id"],
            "user": task["user_name"],
            "project": task["project_name"],
        })
    
    # 2. 重试率异常（>30% 的任务触发了重试）
    retry_rate = db.query("""
        SELECT task_type, 
               COUNT(*) FILTER (WHERE retry_count > 0)::float / COUNT(*) as retry_rate
        FROM cost_events
        WHERE event_time BETWEEN %s AND %s
        GROUP BY task_type
        HAVING COUNT(*) FILTER (WHERE retry_count > 0)::float / COUNT(*) > 0.3
    """, (week_start, week_end))
    
    for r in retry_rate:
        anomalies.append({
            "type": "high_retry_rate",
            "severity": "MEDIUM",
            "message": f"{r['task_type']} 类型任务重试率 {r['retry_rate']:.0%}",
            "task_type": r["task_type"],
        })
    
    # 3. 成本环比异常（周环比增长 > 50%）
    prev_week_cost = get_weekly_total(week_start - timedelta(days=7))
    if prev_week_cost > 0:
        change_pct = (overview_total - prev_week_cost) / prev_week_cost
        if change_pct > 0.5:
            anomalies.append({
                "type": "cost_spike",
                "severity": "HIGH",
                "message": f"本周成本环比增长 {change_pct:.0%}",
            })
    
    return anomalies

{
  "report_period": "2024-06-10 ~ 2024-06-16",
  "summary": {
    "total_cost": "$1,234.56",
    "total_calls": 12450,
    "total_tokens": "89.2M",
    "cost_change": "+12% vs 上周",
    "top_project": "电商订单系统 ($456.78, 37%)"
  },
  "by_project": [
    {"project": "电商订单系统", "team": "交易组", "cost": "$456.78", "calls": 3200, "avg_quality": 0.82},
    {"project": "SaaS CRM", "team": "平台组", "cost": "$345.67", "calls": 4100, "avg_quality": 0.79},
    {"project": "内部工具", "team": "效能组", "cost": "$234.56", "calls": 2800, "avg_quality": 0.85},
    {"project": "文档站", "team": "前端组", "cost": "$197.55", "calls": 2350, "avg_quality": 0.76}
  ],
  "by_task_type": [
    {"type": "bugfix", "cost": "$456.78", "calls": 2100, "model": "Sonnet 60% / Opus 40%", "quality": 0.81},
    {"type": "code_review", "cost": "$345.67", "calls": 4200, "model": "Sonnet 95%", "quality": 0.78},
    {"type": "test_generation", "cost": "$234.56", "calls": 3100, "model": "Sonnet 80% / Haiku 20%", "quality": 0.75},
    {"type": "refactor", "cost": "$197.55", "calls": 1050, "model": "Opus 70%", "quality": 0.83}
  ],
  "anomalies": [
    {"severity": "HIGH", "message": "任务 TASK-5678 成本 $12.34（均值的 15 倍）— 原因：上下文包过大（80K tokens）"},
    {"severity": "MEDIUM", "message": "refactor 类型任务重试率 42% — 建议：大型重构默认使用 Opus"}
  ],
  "recommendations": [
    "电商订单系统的 bugfix 任务中 30% 可以降级到 Haiku（预估节省 $45/周）",
    "SaaS CRM 的 code_review 任务可以启用 Prompt Cache（预估节省 $30/周）",
    "内部工具项目建议设置每日预算上限 $50（当前无限制）"
  ]
}