🧠 图解记忆：预算控制要先可归因，再用分层阈值告警，并为异常增长准备降级与止损动作；点击图片可查看原图。
💡 答案要点
Agent 成本构成：
总成本 = Token成本 + API调用次数成本 + 计算资源成本
Token成本 = Σ(输入Token数 × 单价) + Σ(输出Token数 × 单价)预算控制四层架构：
┌─────────────────────────────────────────────────────────┐
│ Agent 成本控制四层架构 │
├─────────────────────────────────────────────────────────┤
│ Layer 1: 每日预算硬限制（最后防线） │
│ ├── 日预算 $200，超 $200 自动熔断 │
│ ├── 按 Agent 拆分预算（Orchestrator: $80, Worker: $120）│
│ └── 余额不足时拒绝新任务，排队等待 │
├─────────────────────────────────────────────────────────┤
│ Layer 2: 实时告警（提前发现问题） │
│ ├── 预计日成本 > 80% 阈值 → warning │
│ ├── 预计日成本 > 100% 阈值 → critical │
│ └── Token消耗速率异常（突增 > 3x）→ 立即告警 │
├─────────────────────────────────────────────────────────┤
│ Layer 3: 成本优化（主动降本） │
│ ├── 模型降级：GPT-4 → DeepSeek V4-Flash（简单任务） │
│ ├── 缓存命中：相同问题直接返回（省 100%） │
│ ├── 步数限制：> 10步强制结束（防死循环） │
│ └── Prompt压缩：LLMLingua 减少 30% token │
├─────────────────────────────────────────────────────────┤
│ Layer 4: 成本归因（搞清楚钱花在哪） │
│ ├── 按 Agent 类型归因（SearchAgent 消耗最多） │
│ ├── 按用户归因（Top 10 用户占 60% 成本） │
│ └── 按任务类型归因（多跳问答比简单问答贵 5x） │
└─────────────────────────────────────────────────────────┘实现代码：
展开 Python 代码示例（75 行）
python
from datetime import datetime, timedelta
from collections import defaultdict
import asyncio
class AgentBudgetController:
def __init__(
self,
daily_budget_usd: float = 200.0,
warning_threshold: float = 0.8
):
self.daily_budget = daily_budget_usd
self.warning_threshold = warning_threshold
self.spent_today = 0.0
self.agent_budgets = {
"orchestrator": daily_budget_usd * 0.4,
"search": daily_budget_usd * 0.3,
"executor": daily_budget_usd * 0.3
}
self.alerted_agents = set()
async def check_budget(self, agent_name: str, task_cost: float):
"""每次 Agent 调用前检查预算"""
remaining = self.agent_budgets.get(agent_name, 0) - self.spent_today
# Layer 1: 硬限制
if self.spent_today + task_cost > self.daily_budget:
raise BudgetExceededError(
f"日预算已超，当前${self.spent_today:.2f}，剩余${self.daily_budget - self.spent_today:.2f}"
)
# Layer 2: 实时告警
cost_rate = self.spent_today / (datetime.now().hour + 1)
projected_daily = cost_rate * 24
if projected_daily > self.daily_budget * self.warning_threshold:
if agent_name not in self.alerted_agents:
await self.send_alert(
agent_name=agent_name,
severity="warning",
message=f"预计日成本${projected_daily:.2f}，超过{self.warning_threshold*100}%阈值"
)
self.alerted_agents.add(agent_name)
async def record_cost(
self,
agent_name: str,
input_tokens: int,
output_tokens: int,
model: str
):
"""记录成本并更新预算"""
cost = self.calculate_cost(input_tokens, output_tokens, model)
self.spent_today += cost
cost_gauge.labels(
agent=agent_name,
model=model,
date=datetime.now().strftime("%Y-%m-%d")
).inc(cost)
def calculate_cost(self, input_tok: int, output_tok: int, model: str) -> float:
pricing = {
"gpt-4o": (0.005, 0.015),
"deepseek-v4-flash": (0.0005, 0.0015),
"claude-3-opus": (0.015, 0.075)
}
in_price, out_price = pricing.get(model, (0.0, 0.0))
return (input_tok / 1000 * in_price) + (output_tok / 1000 * out_price)
async def send_alert(self, agent_name: str, severity: str, message: str):
await pagerduty.create_incident(
title=f"Agent成本告警: {agent_name}",
severity=severity,
body=message
)成本归因 Dashboard：
yaml
# Grafana Panel: Agent成本归因
- title: 成本按Agent分布
expr: |
sum(increase(agent_cost_total{date="2026-05-15"}[1d])) by (agent_name)
type: pie
- title: 成本按用户分布（Top 10）
expr: |
topk(10, sum(increase(agent_cost_total[1d])) by (user_id))
type: table
- title: 成本按任务类型分布
expr: |
sum(increase(agent_cost_total[1d])) by (task_type)
/ sum(increase(agent_cost_total[1d])) * 100
legend: "{{task_type}}: {{value}}%"面试话术：
"成本控制包括预算/配额、预测告警、按租户和步骤归因，以及缓存、路由、上下文和步数优化。硬限制要设计关键业务的降级与豁免；缓存键还要包含权限、版本和时效，不能为省钱返回越权或过期结果。异常增长要沿 trace 查请求量、token、重试、模型和缓存命中变化。"
📚 参考：Langfuse：预算与成本告警