File size: 5,837 Bytes
42da440 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | import logging
from typing import Dict, List, Union, Optional
from tabulate import tabulate
def calculate_pipeline_cost_efficiency(
total_input_tokens: int,
token_counters: List[Dict[str, Union[int, float, str]]],
) -> Dict[str, Union[float, int, Dict]]:
"""
Calculates the cost efficiency metrics for a pipeline based on token usage.
Args:
total_input_tokens: Total number of tokens in the original input
token_counters: List of dictionaries containing token counts, costs, and names
in the format {"name": str, "input_tokens": int, "input_cost": float,
"output_tokens": int, "output_cost": float}
log_level: Logging level to use for reporting metrics
Returns:
Dict containing cost efficiency metrics:
{
"total_input_tokens": int,
"total_model_input_tokens": int,
"total_model_output_tokens": int,
"total_cost": float,
"cost_per_million_original_tokens": float,
"input_to_model_token_ratio": float,
"component_metrics": Dict[str, Dict]
}
"""
# Initialize counters
total_model_input_tokens = 0
total_model_output_tokens = 0
total_input_cost = 0.0
total_output_cost = 0.0
# Dictionary to store metrics for each component
component_metrics = {}
# Aggregate token counts and costs from all counters
for counter in token_counters:
name = counter.get("name", f"Component {len(component_metrics) + 1}")
input_tokens = counter.get("input_tokens", 0)
output_tokens = counter.get("output_tokens", 0)
input_cost = counter.get("input_cost", 0.0)
output_cost = counter.get("output_cost", 0.0)
total_model_input_tokens += input_tokens
total_model_output_tokens += output_tokens
total_input_cost += input_cost
total_output_cost += output_cost
component_total_cost = input_cost + output_cost
component_metrics[name] = {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost": input_cost,
"output_cost": output_cost,
"total_cost": component_total_cost,
"input_tokens_percentage": 0.0, # Will calculate after getting totals
"output_tokens_percentage": 0.0, # Will calculate after getting totals
"cost_percentage": 0.0, # Will calculate after getting totals
}
total_cost = total_input_cost + total_output_cost
# Calculate percentages for each component
for name, metrics in component_metrics.items():
if total_model_input_tokens > 0:
metrics["input_tokens_percentage"] = (
metrics["input_tokens"] / total_model_input_tokens
) * 100
if total_model_output_tokens > 0:
metrics["output_tokens_percentage"] = (
metrics["output_tokens"] / total_model_output_tokens
) * 100
if total_cost > 0:
metrics["cost_percentage"] = (metrics["total_cost"] / total_cost) * 100
# Calculate cost per million original tokens
cost_per_million_original = 0.0
if total_input_tokens > 0:
cost_per_million_original = (total_cost / total_input_tokens) * 1_000_000
# Calculate input to model token ratio (how many times each token is processed)
input_to_model_ratio = 0.0
if total_input_tokens > 0:
input_to_model_ratio = total_model_input_tokens / total_input_tokens
# Prepare results
results = {
"total_input_tokens": total_input_tokens,
"total_model_input_tokens": total_model_input_tokens,
"total_model_output_tokens": total_model_output_tokens,
"total_input_cost": total_input_cost,
"total_output_cost": total_output_cost,
"total_cost": total_cost,
"cost_per_million_original_tokens": cost_per_million_original,
"input_to_model_token_ratio": input_to_model_ratio,
"component_metrics": component_metrics,
}
# Log the results
print("\n=== Pipeline Cost Efficiency Metrics ===")
print(f"Original input tokens: {total_input_tokens:,}")
print(f"Total model input tokens: {total_model_input_tokens:,}")
print(f"Total model output tokens: {total_model_output_tokens:,}")
print(f"Total cost: ${total_cost:.2f}")
print(f"Cost per million original tokens: ${cost_per_million_original:.2f}")
print(f"Input-to-model token ratio: {input_to_model_ratio:.2f}x")
# Create a table for component breakdown
table_data = []
headers = [
"Component",
"Input Tokens",
"Output Tokens",
"Input Cost",
"Output Cost",
"Total Cost",
"% of Total Cost",
]
for name, metrics in component_metrics.items():
table_data.append(
[
name,
f"{metrics['input_tokens']:,} ({metrics['input_tokens_percentage']:.1f}%)",
f"{metrics['output_tokens']:,} ({metrics['output_tokens_percentage']:.1f}%)",
f"${metrics['input_cost']:.2f}",
f"${metrics['output_cost']:.2f}",
f"${metrics['total_cost']:.2f}",
f"{metrics['cost_percentage']:.1f}%",
]
)
# Add totals row
table_data.append(
[
"TOTAL",
f"{total_model_input_tokens:,} (100.0%)",
f"{total_model_output_tokens:,} (100.0%)",
f"${total_input_cost:.2f}",
f"${total_output_cost:.2f}",
f"${total_cost:.2f}",
"100.0%",
]
)
table = tabulate(table_data, headers=headers, tablefmt="grid")
print(f"\n=== Component Breakdown ===\n{table}")
return results
|