Function get_usage_stats

experiments/token_cost/token_usage_tracker.py:166–219

获取指定日期范围的 token 使用统计

(
    start_date: str,
    end_date: str,
    group_by: str = "date"
)

def get_usage_stats(
    start_date: str,
    end_date: str,
    group_by: str = "date"
) -> Dict[str, Any]:
    """获取指定日期范围的 token 使用统计"""
    if not TOKEN_LOG_PATH.exists():
        return {"error": "No token log found"}

    stats = defaultdict(lambda: {
        "embedding_input": 0,
        "embedding_output": 0,
        "llm_input": 0,
        "llm_output": 0,
        "cost": 0,
        "calls": 0,
    })

    with open(TOKEN_LOG_PATH, "r", encoding="utf-8") as f:
        for line in f:
            record = json.loads(line)
            record_date = record.get("date", "")

            if not (start_date <= record_date <= end_date):
                continue

            if _is_daily_aggregate_record(record):
                key = record.get(group_by, record_date) if group_by == "date" else "aggregate"
                s = stats[key]
                s["embedding_input"] += int(record.get("embedding_tokens", 0))
                s["llm_input"] += int(record.get("llm_tokens", 0))
                s["calls"] += int(record.get("call_count", 0))
                continue

            key = record.get(group_by, "unknown")
            s = stats[key]

            task_type = record.get("task_type", "")
            input_tokens = record.get("input_tokens", 0)
            output_tokens = record.get("output_tokens", 0)
            model = record.get("model", "")

            s["calls"] += 1

            if "embedding" in task_type.lower():
                s["embedding_input"] += input_tokens
                s["embedding_output"] += output_tokens
            else:
                s["llm_input"] += input_tokens
                s["llm_output"] += output_tokens

            s["cost"] += calculate_cost(task_type, model, input_tokens, output_tokens)

    return dict(stats)


def benchmark_one_day(