企业级 AI Agent 部署成本与 TCO 全解析：从 API 调用到自托管的完整经济账

"""GPU 选型成本计算器 — 帮你算清每百万 token 的真实成本"""
from dataclasses import dataclass

@dataclass
class GPU:
    name: str
    monthly_cost: float        # 月租费用（美元）
    tokens_per_sec: float      # 推理吞吐量（tokens/sec）
    memory_gb: int             # GPU 显存（GB）

def calculate_cost_per_million_tokens(
    gpu: GPU,
    hours_per_day: float = 24,
    days_per_month: float = 30,
    utilization: float = 0.6   # GPU 利用率（不是 24 小时满载）
) -> float:
    """计算每百万 token 的成本"""
    total_seconds = hours_per_day * 3600 * days_per_month
    effective_seconds = total_seconds * utilization
    total_tokens = gpu.tokens_per_sec * effective_seconds
    cost_per_million = (gpu.monthly_cost / total_tokens) * 1_000_000
    return cost_per_million

# 2026 年主流 GPU 对比
gpus = [
    GPU("NVIDIA H200 141GB", 4500, 45, 141),
    GPU("NVIDIA B200 192GB", 7500, 72, 192),
    GPU("双 H100 80GB (TP)", 6000, 55, 160),
    GPU("H100 80GB (INT4 量化)", 3000, 120, 80),
    GPU("AMD MI325X 256GB", 3800, 50, 256),
]

print(f"{'GPU 型号':<25} {'月租($)':>8} {'吞吐(tok/s)':>12} {'$/M tokens':>11}")
print("-" * 60)
for gpu in gpus:
    cost = calculate_cost_per_million_tokens(gpu)
    print(f"{gpu.name:<25} ${gpu.monthly_cost:>7,} {gpu.tokens_per_sec:>12.0f} ${cost:>10.2f}")

# 输出：
# GPU 型号                    月租($)   吞吐(tok/s)  $/M tokens
# ------------------------------------------------------------
# NVIDIA H200 141GB       $  4,500           45 $      0.87
# NVIDIA B200 192GB       $  7,500           72 $      0.72
# 双 H100 80GB (TP)       $  6,000           55 $      0.76
# H100 80GB (INT4 量化)    $  3,000          120 $      0.35
# AMD MI325X 256GB        $  3,800           50 $      0.63

"""Agent 架构成本模拟器 — 不同架构下的 token 消耗对比"""
import random
from dataclasses import dataclass
from typing import List

@dataclass
class Request:
    complexity: str  # "simple", "medium", "complex"
    needs_tool: bool

@dataclass
class ArchitectureCost:
    name: str
    simple_tokens: int
    medium_tokens: int
    complex_tokens: int
    
    def estimate(self, requests: List[Request]) -> int:
        total = 0
        for r in requests:
            if r.complexity == "simple":
                total += self.simple_tokens
            elif r.complexity == "medium":
                total += self.medium_tokens
            else:
                total += self.complex_tokens
        return total

# 五种架构的 token 消耗模型
architectures = {
    "Router": ArchitectureCost("Router", 2000, 5000, 12000),
    "Plan-and-Execute": ArchitectureCost("P&E", 4000, 8000, 15000),
    "ReAct": ArchitectureCost("ReAct", 5000, 12000, 25000),
    "ReAct + Reflection": ArchitectureCost("ReAct+Ref", 8000, 20000, 45000),
    "Multi-Agent": ArchitectureCost("Multi", 10000, 30000, 80000),
}

# 模拟 1000 个请求的分布（70% 简单 / 20% 中等 / 10% 复杂）
random.seed(42)
requests = [
    Request(
        complexity=random.choices(
            ["simple", "medium", "complex"],
            weights=[0.7, 0.2, 0.1]
        )[0],
        needs_tool=random.random() > 0.4
    )
    for _ in range(1000)
]

print(f"{'架构':<20} {'总 Token':>12} {'成本倍数':>8}")
print("-" * 42)
baseline = None
for name, arch in architectures.items():
    total = arch.estimate(requests)
    if baseline is None:
        baseline = total
    ratio = total / baseline
    print(f"{name:<20} {total:>12,} {ratio:>7.1f}x")

# 输出：
# 架构                     总 Token     成本倍数
# ------------------------------------------
# Router                  3,580,000      1.0x
# P&E                     6,480,000      1.8x
# ReAct                   9,780,000      2.7x
# ReAct+Ref              15,280,000      4.3x
# Multi                  23,400,000      6.5x

"""AI Agent TCO 完整计算器 — 输入业务参数，输出三种模式的月成本"""
from dataclasses import dataclass

@dataclass
class AgentTCOCalculator:
    # 业务参数
    monthly_requests: int          # 月请求量
    avg_tokens_per_request: int    # 平均每个请求的总 token（含多步）
    api_input_price: float = 3.0   # API 输入价格 ($/M tokens)
    api_output_price: float = 12.0 # API 输出价格 ($/M tokens)
    input_output_ratio: float = 0.7 # 输入 token 占比（70% input, 30% output）
    
    # 优化参数
    cache_hit_rate: float = 0.0    # 缓存命中率 (0-1)
    router_savings: float = 0.0    # 模型路由节省比例 (0-1)
    security_overhead: float = 0.15 # 安全开销比例 (0.15 = 15%)
    
    # 自托管参数
    gpu_monthly_cost: float = 3000  # GPU 月租
    self_host_tokens_per_sec: float = 100  # 自托管吞吐
    gpu_utilization: float = 0.5    # GPU 利用率
    
    def _effective_tokens(self) -> int:
        """计算优化后的有效 token 消耗"""
        base = self.monthly_requests * self.avg_tokens_per_request
        # 缓存命中直接跳过 LLM
        cached = base * self.cache_hit_rate
        effective = base - cached
        # 模型路由进一步节省
        effective *= (1 - self.router_savings)
        # 安全开销
        effective *= (1 + self.security_overhead)
        return int(effective)
    
    def api_mode_cost(self) -> float:
        """纯 API 模式月成本"""
        tokens = self._effective_tokens()
        input_tokens = tokens * self.input_output_ratio
        output_tokens = tokens * (1 - self.input_output_ratio)
        input_cost = (input_tokens / 1_000_000) * self.api_input_price
        output_cost = (output_tokens / 1_000_000) * self.api_output_price
        return input_cost + output_cost
    
    def self_hosted_cost(self) -> float:
        """纯自托管模式月成本（不含人力）"""
        tokens = self._effective_tokens()
        total_seconds = 24 * 3600 * 30 * self.gpu_utilization
        max_tokens = self.self_host_tokens_per_sec * total_seconds
        if tokens <= max_tokens:
            return self.gpu_monthly_cost
        # 需要多卡
        cards_needed = tokens / max_tokens
        import math
        return self.gpu_monthly_cost * math.ceil(cards_needed)
    
    def hybrid_mode_cost(self) -> float:
        """混合模式：70% 走 API（核心推理），30% 走自托管（工具调用/检索）"""
        tokens = self._effective_tokens()
        api_tokens = int(tokens * 0.7)
        self_tokens = int(tokens * 0.3)
        
        # API 部分
        input_t = api_tokens * self.input_output_ratio
        output_t = api_tokens * (1 - self.input_output_ratio)
        api_cost = (input_t / 1e6) * self.api_input_price + (output_t / 1e6) * self.api_output_price
        
        # 自托管部分
        total_sec = 24 * 3600 * 30 * self.gpu_utilization
        max_t = self.self_host_tokens_per_sec * total_sec
        self_cost = self.gpu_monthly_cost if self_tokens <= max_t else self.gpu_monthly_cost * 2
        
        return api_cost + self_cost
    
    def run(self):
        """运行计算并输出对比"""
        api = self.api_mode_cost()
        self_hosted = self.self_hosted_cost()
        hybrid = self.hybrid_mode_cost()
        
        print(f"=== AI Agent TCO 月度成本分析 ===")
        print(f"月请求量: {self.monthly_requests:,}")
        print(f"平均每请求 token: {self.avg_tokens_per_request:,}")
        print(f"缓存命中率: {self.cache_hit_rate:.0%}")
        print(f"模型路由节省: {self.router_savings:.0%}")
        print(f"安全开销: {self.security_overhead:.0%}")
        print(f"有效 token 消耗: {self._effective_tokens():,}")
        print(f"")
        print(f"{'模式':<12} {'月成本($)':>12} {'年成本($)':>14}")
        print(f"{'─' * 40}")
        print(f"{'纯 API':<12} ${api:>10,.0f} ${api * 12:>12,.0f}")
        print(f"{'纯自托管':<10} ${self_hosted:>10,.0f} ${self_hosted * 12:>12,.0f}")
        print(f"{'混合部署':<10} ${hybrid:>10,.0f} ${hybrid * 12:>12,.0f}")
        print(f"")
        best = min(api, self_hosted, hybrid)
        best_name = "纯 API" if best == api else "纯自托管" if best == self_hosted else "混合部署"
        print(f"💡 推荐方案: {best_name}（月省 ${max(api, self_hosted, hybrid) - best:,.0f}）")

# === 示例 1：中型客服 Agent（100 万次/月）===
print("【场景一：中型客服 Agent】")
AgentTCOCalculator(
    monthly_requests=1_000_000,
    avg_tokens_per_request=12000,
    cache_hit_rate=0.25,
    router_savings=0.30,
).run()

# === 示例 2：企业内部知识助手（50 万次/月）===
print("\n【场景二：企业内部知识助手】")
AgentTCOCalculator(
    monthly_requests=500_000,
    avg_tokens_per_request=20000,
    cache_hit_rate=0.35,
    router_savings=0.40,
    gpu_monthly_cost=4500,
    self_host_tokens_per_sec=80,
).run()

"""AI Agent 预算管理与告警系统"""
from datetime import datetime
from dataclasses import dataclass

@dataclass
class BudgetManager:
    monthly_budget: float          # 月预算上限（美元）
    current_spend: float = 0.0     # 当前累计花费
    alert_thresholds: list = None  # 告警阈值列表
    
    def __post_init__(self):
        if self.alert_thresholds is None:
            self.alert_thresholds = [0.5, 0.8, 0.95, 1.0]
    
    def record_usage(self, tokens: int, cost: float):
        """记录一次 API 调用"""
        self.current_spend += cost
        self._check_alerts()
    
    def _check_alerts(self):
        """检查是否触发告警"""
        usage_ratio = self.current_spend / self.monthly_budget
        for threshold in self.alert_thresholds:
            if usage_ratio >= threshold:
                day = datetime.now().day
                days_left = 30 - day
                remaining = self.monthly_budget - self.current_spend
                daily_rate = self.current_spend / day
                projected = daily_rate * 30
                
                print(f"⚠️ 预算告警：已使用 {usage_ratio:.0%} "
                      f"(${self.current_spend:,.0f}/${self.monthly_budget:,.0f})")
                print(f"   本月剩余 {days_left} 天，预计超支 ${projected - self.monthly_budget:,.0f}")
                print(f"   建议日均支出：${remaining / days_left:,.0f}")
                
                if usage_ratio >= 1.0:
                    print("🚨 预算已耗尽！建议：降低模型等级 / 提高缓存命中率 / 限制非核心功能")
                elif usage_ratio >= 0.95:
                    print("🔴 即将耗尽：建议立即启用成本优化策略")
                elif usage_ratio >= 0.8:
                    print("🟡 预算紧张：建议审查高消耗场景")
                break  # 只触发最高阈值告警
    
    def predict_monthly_cost(self, current_day: int) -> float:
        """基于当前支出预测全月成本"""
        if current_day == 0:
            return 0
        daily_avg = self.current_spend / current_day
        return daily_avg * 30

# 使用示例
manager = BudgetManager(monthly_budget=5000)
# 模拟前 15 天的支出
for day in range(1, 16):
    daily_cost = 200 + day * 10  # 逐步增长的支出模式
    manager.record_usage(tokens=5_000_000, cost=daily_cost)