"""Cost estimation agent."""
from typing import Any
from .base import BaseAgent, Tool
class CloudCostCalculator:
"""Tool to calculate cloud infrastructure costs."""
name = "cloud_cost_calculator"
description = "Calculate monthly cloud costs based on compute, storage, and network usage"
def execute(
self,
provider: str = "aws",
compute_instances: int = 1,
instance_type: str = "medium",
storage_gb: int = 100,
bandwidth_gb: int = 100,
database_type: str = "managed",
**kwargs: Any,
) -> dict[str, Any]:
"""Calculate cloud costs.
Args:
provider: Cloud provider (aws, gcp, azure, railway)
compute_instances: Number of compute instances
instance_type: Instance size (small, medium, large, xlarge)
storage_gb: Storage in GB
bandwidth_gb: Monthly bandwidth in GB
database_type: Database type (managed, self-hosted)
**kwargs: Additional parameters
Returns:
Dictionary with cost breakdown
"""
# Simplified pricing (monthly USD)
pricing = {
"aws": {
"small": 20, # t3.small
"medium": 35, # t3.medium
"large": 70, # t3.large
"xlarge": 140, # t3.xlarge
"storage_per_gb": 0.10, # EBS
"bandwidth_per_gb": 0.09,
"managed_db": 50, # RDS smallest
"load_balancer": 20,
"nat_gateway": 35,
},
"gcp": {
"small": 18,
"medium": 32,
"large": 65,
"xlarge": 130,
"storage_per_gb": 0.09,
"bandwidth_per_gb": 0.08,
"managed_db": 45,
"load_balancer": 18,
"nat_gateway": 30,
},
"azure": {
"small": 21,
"medium": 38,
"large": 75,
"xlarge": 145,
"storage_per_gb": 0.11,
"bandwidth_per_gb": 0.10,
"managed_db": 55,
"load_balancer": 22,
"nat_gateway": 35,
},
"railway": {
"small": 10,
"medium": 20,
"large": 40,
"xlarge": 80,
"storage_per_gb": 0.05,
"bandwidth_per_gb": 0.05,
"managed_db": 10,
"load_balancer": 0, # Included
"nat_gateway": 0, # Included
},
}
if provider not in pricing:
provider = "aws"
prices = pricing[provider]
# Calculate costs
compute_cost = compute_instances * prices[instance_type]
storage_cost = storage_gb * prices["storage_per_gb"]
bandwidth_cost = bandwidth_gb * prices["bandwidth_per_gb"]
db_cost = prices["managed_db"] if database_type == "managed" else 0
lb_cost = prices["load_balancer"] if compute_instances > 1 else 0
nat_cost = prices["nat_gateway"] if compute_instances > 0 else 0
total = compute_cost + storage_cost + bandwidth_cost + db_cost + lb_cost + nat_cost
return {
"provider": provider,
"monthly_total": round(total, 2),
"breakdown": {
"compute": round(compute_cost, 2),
"storage": round(storage_cost, 2),
"bandwidth": round(bandwidth_cost, 2),
"database": round(db_cost, 2),
"load_balancer": round(lb_cost, 2),
"nat_gateway": round(nat_cost, 2),
},
"annual_total": round(total * 12, 2),
"config": {
"instances": compute_instances,
"instance_type": instance_type,
"storage_gb": storage_gb,
"bandwidth_gb": bandwidth_gb,
},
}
class ServiceCostEstimator:
"""Tool to estimate costs for additional services."""
name = "service_cost_estimator"
description = "Estimate costs for CDN, caching, monitoring, and other services"
def execute(
self,
cdn_enabled: bool = False,
cache_enabled: bool = False,
monitoring_tier: str = "basic",
backup_enabled: bool = True,
**kwargs: Any,
) -> dict[str, Any]:
"""Estimate service costs.
Args:
cdn_enabled: Whether CDN is used
cache_enabled: Whether caching layer is used
monitoring_tier: Monitoring level (basic, standard, premium)
backup_enabled: Whether backups are enabled
**kwargs: Additional parameters
Returns:
Dictionary with service cost breakdown
"""
costs = {}
if cdn_enabled:
costs["cdn"] = 20 # CloudFlare/Cloudfront basic
if cache_enabled:
costs["cache"] = 15 # Redis/Memcached small instance
monitoring_costs = {
"basic": 0, # Free tier
"standard": 50, # Datadog/New Relic
"premium": 150,
}
costs["monitoring"] = monitoring_costs.get(monitoring_tier, 0)
if backup_enabled:
costs["backups"] = 10
# Additional common services
costs["logging"] = 10 # CloudWatch/Stackdriver
costs["secrets_management"] = 5
costs["dns"] = 1
total = sum(costs.values())
return {
"monthly_total": round(total, 2),
"breakdown": costs,
"annual_total": round(total * 12, 2),
}
class CostAgent(BaseAgent):
"""Agent specialized in cost estimation and optimization."""
def __init__(self) -> None:
"""Initialize the cost agent."""
tools: list[Tool] = [
CloudCostCalculator(), # type: ignore[list-item]
ServiceCostEstimator(), # type: ignore[list-item]
]
super().__init__(
name="cost",
role="financial analyst specializing in cloud cost estimation and optimization",
tools=tools,
)
async def analyze(self, context: dict[str, Any]) -> dict[str, Any]:
"""Analyze tech stack and estimate costs.
Args:
context: Dictionary with keys like:
- user_query: str
- infrastructure_recommendations: dict (from infrastructure agent)
- database_recommendations: dict (from database agent)
- dau: int
- budget_target: float (optional monthly budget)
Returns:
Dictionary with cost estimates
"""
self.logger.info("cost_analysis_start", context=context)
# Extract context
user_query = context.get("user_query", "")
dau = context.get("dau", 0)
budget_target = context.get("budget_target", 0)
api_key = context.get("api_key")
# Extract from other agents if available
infra = context.get("infrastructure_recommendations", {})
db = context.get("database_recommendations", {})
# Estimate infrastructure requirements based on DAU
if dau < 10_000:
instances = 2
instance_type = "small"
storage_gb = 50
bandwidth_gb = 100
elif dau < 100_000:
instances = 4
instance_type = "medium"
storage_gb = 200
bandwidth_gb = 500
elif dau < 500_000:
instances = 10
instance_type = "large"
storage_gb = 500
bandwidth_gb = 2000
else:
instances = 20
instance_type = "xlarge"
storage_gb = 1000
bandwidth_gb = 5000
# Calculate for different providers
providers = ["aws", "gcp", "railway"]
cost_comparisons = []
for provider in providers:
cloud_costs = self._execute_tool(
"cloud_cost_calculator",
provider=provider,
compute_instances=instances,
instance_type=instance_type,
storage_gb=storage_gb,
bandwidth_gb=bandwidth_gb,
database_type="managed",
)
service_costs = self._execute_tool(
"service_cost_estimator",
cdn_enabled=dau > 10_000,
cache_enabled=dau > 10_000,
monitoring_tier="standard" if dau > 50_000 else "basic",
backup_enabled=True,
)
total_monthly = cloud_costs["monthly_total"] + service_costs["monthly_total"]
cost_comparisons.append({
"provider": provider,
"monthly_cost": total_monthly,
"cloud_costs": cloud_costs,
"service_costs": service_costs,
})
# Sort by cost
cost_comparisons.sort(key=lambda x: x["monthly_cost"])
# Build prompt for LLM
prompt = f"""Analyze these cost estimates and provide EXTREMELY DETAILED recommendations with full explanations:
User Query: {user_query}
Daily Active Users: {dau:,}
Budget Target: ${budget_target}/month (if specified)
Configuration:
- Compute Instances: {instances} x {instance_type}
- Storage: {storage_gb} GB
- Bandwidth: {bandwidth_gb} GB/month
- Managed Database: Yes
Cost Comparison:
{self._format_costs(cost_comparisons)}
Provide a COMPREHENSIVE, DETAILED analysis with:
1. **Recommended Provider** (150+ words):
- Which provider and why (specific reasons)
- Compare all 3 providers with detailed pros/cons for THIS specific use case
- Total cost comparison with breakdown
- Long-term value assessment (not just cheapest)
2. **Detailed Cost Breakdown** (200+ words):
- Explain EACH cost component in detail (compute, storage, database, network, services)
- Why each component costs what it does
- Which components will scale most with growth
- Hidden costs to watch out for
- Cost allocation by feature/function
3. **Cost Optimization Strategies** (250+ words):
- 5-8 SPECIFIC optimization tactics for THIS stack
- Reserved instances / savings plans (exact savings %)
- Right-sizing opportunities
- Auto-scaling strategies
- Spot instances / preemptible VMs where applicable
- Data transfer optimization
- Storage tier optimization
- Database optimization (read replicas, caching)
- Each strategy should include: description, expected savings, implementation difficulty, risks
4. **Scaling Cost Projections** (150+ words):
- Detailed cost at 2x scale: ${cost_comparisons[0]['monthly_cost'] * 2:.2f}/month
- Detailed cost at 5x scale: ${cost_comparisons[0]['monthly_cost'] * 4:.2f}/month
- Detailed cost at 10x scale: ${cost_comparisons[0]['monthly_cost'] * 7:.2f}/month
- Explain which costs scale linearly vs sub-linearly
- Break down where the money goes at each scale
- When to consider architecture changes
5. **Budget Management** (100+ words):
- Set up cost alerts (specific thresholds)
- Budget tracking recommendations
- Cost anomaly detection
- Monthly review checklist
- What to monitor daily/weekly/monthly
6. **Free Tier Opportunities** (100+ words):
- What free tier services apply to this stack
- How long free tiers last
- Migration strategy when free tier expires
- Alternative free/cheap services
7. **Cost vs Performance Trade-offs** (100+ words):
- Where NOT to cut costs (critical services)
- Safe cost cuts that won't hurt performance
- Monitoring to ensure cost cuts don't degrade UX
Respond with extensive, paragraph-form explanations. Be specific and detailed. Use real numbers and percentages.
"""
# Get LLM recommendation
response = self._call_llm(prompt, api_key=api_key)
self.logger.info(
"cost_analysis_complete",
cheapest_provider=cost_comparisons[0]["provider"],
monthly_range=f"${cost_comparisons[0]['monthly_cost']}-${cost_comparisons[-1]['monthly_cost']}",
)
return {
"agent": self.name,
"cost_comparisons": cost_comparisons,
"recommendations": response,
"configuration": {
"instances": instances,
"instance_type": instance_type,
"storage_gb": storage_gb,
"bandwidth_gb": bandwidth_gb,
},
}
def _format_costs(self, comparisons: list[dict[str, Any]]) -> str:
"""Format cost comparisons for prompt."""
lines = []
for comp in comparisons:
lines.append(f"\n{comp['provider'].upper()}:")
lines.append(f" Monthly: ${comp['monthly_cost']:.2f}")
lines.append(f" Annual: ${comp['monthly_cost'] * 12:.2f}")
breakdown = comp['cloud_costs']['breakdown']
lines.append(f" Breakdown: Compute ${breakdown['compute']}, "
f"Storage ${breakdown['storage']}, "
f"DB ${breakdown['database']}, "
f"Network ${breakdown['bandwidth']}")
return "\n".join(lines)