AI Agent成本控制指南：Token花得肉疼怎么办？
一、开场：账单来了，我的心凉了
大家好，我是老金。
上个月底，收到OpenAI的账单，差点心梗。
一个小小的客服Agent，一个月烧了3000美元。平均每天100刀，比我日均咖啡预算高了50倍。
老板问我：”这钱花得值吗？”
我支支吾吾答不上来——因为我自己都不知道这些Token都花哪儿了。
痛定思痛，我花了一周时间研究成本控制，现在每天成本降到了30美元，降了97%。
今天把这些经验分享出来，希望能救救你的钱包。
二、Token都花在哪儿了？
成本拆解
首先，你得知道钱花哪儿了：
import tiktoken
from collections import defaultdict
class TokenAnalyzer:
"""Token使用分析器"""
def __init__(self, model="gpt-4"):
self.encoding = tiktoken.encoding_for_model(model)
self.usage_log = []
def analyze_trace(self, trace):
"""分析一次对话的Token使用"""
analysis = {
"user_input": self.count_tokens(trace.user_input),
"system_prompt": self.count_tokens(trace.system_prompt),
"tool_definitions": 0,
"conversation_history": 0,
"agent_responses": 0,
"tool_results": 0
}
# 工具定义
for tool in trace.tools_used:
analysis["tool_definitions"] += self.count_tokens(
json.dumps(tool.definition)
)
# 对话历史
for msg in trace.history:
analysis["conversation_history"] += self.count_tokens(
msg.content
)
# Agent响应
for response in trace.agent_responses:
analysis["agent_responses"] += self.count_tokens(response)
# 工具返回结果
for result in trace.tool_results:
analysis["tool_results"] += self.count_tokens(
json.dumps(result)
)
self.usage_log.append(analysis)
return analysis
def get_cost_breakdown(self, pricing):
"""获取成本拆解"""
total_by_category = defaultdict(int)
for log in self.usage_log:
for category, tokens in log.items():
total_by_category[category] += tokens
# 计算成本
input_cost = sum(v for k, v in total_by_category.items()
if k != "agent_responses")
output_cost = total_by_category["agent_responses"]
return {
"input_tokens": input_cost,
"output_tokens": output_cost,
"input_cost": input_cost * pricing["input"],
"output_cost": output_cost * pricing["output"],
"categories": dict(total_by_category)
}
def count_tokens(self, text):
"""计算Token数"""
return len(self.encoding.encode(text))我的成本分布
分析后发现：
| 成本来源 | Token占比 | 成本占比 | 问题 |
|---|---|---|---|
| 系统提示词 | 15% | 10% | 太长了 |
| 工具定义 | 20% | 15% | 描述太详细 |
| 对话历史 | 35% | 25% | 全量保留 |
| 工具返回结果 | 20% | 15% | 返回数据太多 |
| Agent响应 | 10% | 35% | 输出token贵 |
发现问题：输入token占了65%的成本，这是优化重点！
三、系统提示词优化
问题：提示词像写论文
看看我之前的系统提示词：
你是一个专业的客服AI助手，负责处理用户的订单查询、退换货请求、
商品咨询等问题。你需要具备以下能力：
1. 订单查询能力：
- 能够查询订单状态
- 能够查询物流信息
- 能够查询订单详情
- 支持按订单号、手机号查询
2. 退换货处理能力：
- 了解退换货政策
- 能够判断是否符合退换货条件
- 能够生成退换货申请
...（还有500多字）这个提示词有800多个token，每次对话都要发送。
优化方案：精简+模块化
# 优化后的提示词模块化设计
SYSTEM_PROMPT_CORE = """你是客服助手。处理订单查询、退换货、商品咨询。
回复简洁，语气友好。"""
TOOL_PROMPTS = {
"order": "查订单用query_order，需要订单号或手机号。",
"return": "退换货先check_return_policy，再create_return。",
"product": "商品咨询用search_product。"
}
def get_system_prompt(context):
"""动态生成系统提示词"""
# 根据上下文选择需要的模块
relevant_tools = infer_relevant_tools(context.user_input)
prompt = SYSTEM_PROMPT_CORE + "n"
for tool in relevant_tools:
prompt += TOOL_PROMPTS.get(tool, "") + "n"
return prompt效果：系统提示词从800 token降到150 token，降了81%
提示词优化清单
| 优化项 | 原始 | 优化后 | 节省 |
|---|---|---|---|
| 移除冗余描述 | 300字 | 100字 | 200 token |
| 合并相似规则 | 15条 | 8条 | 100 token |
| 删除示例对话 | 5个 | 0个 | 300 token |
| 精简工具描述 | 详细版 | 简洁版 | 150 token |
四、对话历史管理
问题：照单全收
每次对话把所有历史都发给模型：
# ❌ 错误做法
def build_messages(history, user_input):
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(history) # 全部历史
messages.append({"role": "user", "content": user_input})
return messages对话越长，成本越高。一个20轮的对话，历史可能消耗5000+ token。
方案1：滑动窗口
def sliding_window(history, max_turns=5):
"""滑动窗口：只保留最近N轮对话"""
return history[-max_turns * 2:] # 每轮包含user和assistant问题：丢失早期上下文
方案2：摘要压缩
async def summarize_history(history, llm):
"""将早期对话压缩为摘要"""
if len(history) 2:
self.recent_turns.pop(0)
def get_relevant_history(self, current_input):
"""检索相关历史"""
# 检索最相关的3条
query_embedding = self.embedder.embed(current_input)
relevant = self.vector_store.search(query_embedding, k=3)
# 组合：最近历史 + 相关历史
context = []
for turn in self.recent_turns:
context.append({"role": "user", "content": turn[0]})
context.append({"role": "assistant", "content": turn[1]})
if relevant:
context.insert(0, {
"role": "system",
"content": f"[相关历史]n{chr(10).join(relevant)}"
})
return context五、工具返回优化
问题：工具返回数据太多
# ❌ 返回完整数据
def query_order(order_id):
order = db.get_order(order_id)
return order # 返回整个订单对象，可能几百个字段Agent收到的可能是：
{
"order_id": "12345",
"user_id": "67890",
"items": [...50个商品...],
"shipping_address": {...},
"billing_address": {...},
"payment_info": {...},
"tracking": {...},
"timestamps": {...},
"metadata": {...},
// ... 总共5000+ token
}优化：按需返回
# ✅ 只返回必要信息
def query_order(order_id, fields=None):
order = db.get_order(order_id)
# 默认只返回关键字段
default_fields = ["order_id", "status", "items.name", "total", "tracking"]
fields = fields or default_fields
return extract_fields(order, fields)
# 或者让Agent指定需要的字段
def query_order(order_id, fields):
return db.get_order(order_id, projection=fields)效果：工具返回从5000 token降到500 token
工具返回优化策略
| 策略 | 说明 | 节省比例 |
|---|---|---|
| 字段裁剪 | 只返回必要字段 | 60-80% |
| 分页返回 | 大数据分页获取 | 90%+ |
| 摘要返回 | 返回摘要而非全文 | 70% |
| 懒加载 | 按需调用额外接口 | 50% |
六、模型选择策略
不同任务用不同模型
不是所有任务都需要GPT-4：
class ModelSelector:
"""模型选择器"""
def __init__(self):
self.model_costs = {
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"claude-3-haiku": {"input": 0.00025, "output": 0.00125},
}
def select_model(self, task_type, complexity):
"""根据任务选择模型"""
if task_type == "simple_qa":
return "gpt-3.5-turbo" # 简单问答
elif task_type == "complex_reasoning":
return "gpt-4-turbo" # 复杂推理
elif task_type == "tool_calling":
if complexity == "high":
return "gpt-4-turbo"
else:
return "gpt-3.5-turbo"
else:
return "gpt-3.5-turbo" # 默认级联调用策略
先用便宜模型，不行再换贵的：
async def cascading_call(user_input, fallback_model="gpt-4"):
"""级联调用：先用便宜模型"""
# 第一轮：用便宜模型
cheap_response = await call_model("gpt-3.5-turbo", user_input)
# 评估响应质量
if is_good_response(cheap_response):
return cheap_response
# 质量不够，换贵模型
expensive_response = await call_model(fallback_model, user_input)
return expensive_response
def is_good_response(response):
"""简单评估响应质量"""
# 检查是否包含"我不知道"等不确定表达
uncertain_patterns = ["我不确定", "可能", "应该"]
for pattern in uncertain_patterns:
if pattern in response:
return False
return True成本对比
| 模型 | 输入价格 | 输出价格 | 适用场景 |
|---|---|---|---|
| GPT-4 Turbo | $0.01/1K | $0.03/1K | 复杂推理、重要决策 |
| GPT-3.5 Turbo | $0.0005/1K | $0.0015/1K | 简单问答、工具调用 |
| Claude 3 Haiku | $0.00025/1K | $0.00125/1K | 高并发场景 |
| DeepSeek | ¥0.001/1K | ¥0.002/1K | 成本敏感场景 |
选择GPT-3.5处理80%的请求，能节省90%的成本！
七、缓存策略
相同问题缓存答案
import hashlib
from functools import lru_cache
class ResponseCache:
"""响应缓存"""
def __init__(self, ttl=3600):
self.cache = {}
self.ttl = ttl
def get_cache_key(self, messages, tools):
"""生成缓存键"""
content = json.dumps(messages + [str(tools)], sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()
async def get_or_generate(self, messages, tools, generator):
"""获取缓存或生成新响应"""
key = self.get_cache_key(messages, tools)
if key in self.cache:
cached = self.cache[key]
if not self._is_expired(cached):
return cached["response"]
# 生成新响应
response = await generator(messages, tools)
self.cache[key] = {
"response": response,
"timestamp": time.time()
}
return response
def _is_expired(self, cached):
return time.time() - cached["timestamp"] > self.ttl缓存命中场景
| 场景 | 缓存命中率 | 节省成本 |
|---|---|---|
| FAQ问答 | 60%+ | 高 |
| 商品咨询 | 40% | 中 |
| 订单查询 | 10% | 低 |
| 个性化推荐 | self.alert_threshold: |
self.send_alert()
def send_alert(self):
"""发送告警"""
message = f"""
⚠️ 成本告警
今日累计成本：${self.daily_cost:.2f}
各Agent消耗：
{self.format_cost_breakdown()}
"""
send_slack_alert(message)
def format_cost_breakdown(self):
return "n".join(
f"- {agent}: ${cost:.2f}"
for agent, cost in self.cost_by_agent.items()
)
### 成本优化效果追踪
```python
# 优化前
before = {
"daily_cost": 100,
"avg_tokens_per_request": 5000,
"model_distribution": {"gpt-4": 1.0}
}
# 优化后
after = {
"daily_cost": 30,
"avg_tokens_per_request": 1500,
"model_distribution": {"gpt-4": 0.2, "gpt-3.5": 0.8}
}
# 效果
improvement = {
"cost_reduction": "70%",
"token_reduction": "70%",
"cost_per_successful_task": "从$0.15降到$0.04"
}九、总结与行动清单
成本优化公式
总成本 = (输入Token × 输入单价 + 输出Token × 输出单价) × 请求次数优化每一项都能降低成本！
行动清单
| 优先级 | 行动项 | 预期效果 | 时间 |
|---|---|---|---|
| P0 | 精简系统提示词 | -80%系统提示token | 1小时 |
| P0 | 限制对话历史长度 | -60%历史token | 30分钟 |
| P0 | 工具返回裁剪 | -70%工具返回token | 2小时 |
| P1 | 简单任务用便宜模型 | -50%模型成本 | 1小时 |
| P1 | 实现响应缓存 | 30%命中率可省30% | 2小时 |
| P2 | 级联调用策略 | -20%混合成本 | 3小时 |
| P2 | 成本监控告警 | 及时发现问题 | 2小时 |
我的优化成果
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 日均成本 | $100 | $30 | -70% |
| 单次请求token | 5000 | 1500 | -70% |
| GPT-4使用比例 | 100% | 20% | -80% |
| 缓存命中率 | 0% | 35% | +35% |
下期预告
明天聊聊AI Agent知识库构建——RAG系统怎么搭建才不踩坑？
往期回顾
正文完
我们为何放弃了CrewAI：一个关于AI框架选型的深度复盘 - 技术老金 […] AI写不出“干净架构”：从代码生成到软件匠艺的进阶之路 […]