9.5 成本控制
在上一节中，我们讨论了如何通过降级和容错让系统在故障中存活。但有一个问题贯穿始终，也是所有 Agent 产品负责人最关心的：这个系统每天烧多少钱？
在传统软件开发中，服务的主要成本来自服务器、带宽和人力，边际成本很低——多一个用户不过多消耗一点带宽。但 Agent 系统完全不同：每次用户交互都在消耗 Token，而 Token 直接对应金钱。 用户越多，成本越高，而且是线性甚至超线性增长的。
9.5.1 为什么成本控制是 Agent 系统的核心挑战？
让我们算一笔账。
假设一个智能客服 Agent，每天处理 10,000 次对话，每次对话平均 3 轮交互，每轮消耗约 2,000 input tokens + 500 output tokens：
使用 GPT-4o（$2.50/1M input, $10/1M output）：
- 日均成本 = 10,000 × 3 × (2000 × 2.50/1M + 500 × 10/1M) = $300/天
- 月均成本 ≈ $9,000/月
如果 30% 的请求可以通过缓存或简单模型处理：
- 月均成本降至 ≈ $3,000~$4,000/月
生活类比：传统软件的成本就像开一辆电车——充电费很便宜，开多开少差别不大。Agent 系统的成本就像开出租车——每跑一公里都在烧油，而且油价还不便宜。你得精打细算每一趟行程。
这就是成本控制的巨大价值。对于一个拥有百万用户的 Agent 产品，成本可以从每月数十万美元降至数万美元。
9.5.2 Token 用量监控
Token 计算基础
不同模型使用不同的 Tokenizer，但大多数情况下可以用 tiktoken 库进行估算：
import tiktoken # 导入 OpenAI 的 Token 计数库
encoding = tiktoken.encoding_for_model("gpt-4o") # 获取 GPT-4o 的编码器
text = "你好，这是一个测试。Hello, this is a test."
tokens = len(encoding.encode(text)) # 编码并计算 Token 数
print(f"文本: '{text}'") # 打印原始文本
print(f"Token 数: {tokens}") # 打印 Token 数
# 中文通常 1 个字符 ≈ 1.5~2 个 tokens
# 英文通常 1 个单词 ≈ 1.3 个 tokens成本追踪中间件
import time # 导入时间模块
import json # 导入 JSON 处理
from typing import Dict, List, Optional # 导入类型注解
from dataclasses import dataclass, field # 导入数据类
from datetime import datetime, timedelta # 导入日期时间
import redis.asyncio as aioredis # 导入异步 Redis
from fastapi import FastAPI, Request, Response # 导入 FastAPI
from pydantic import BaseModel # 导入数据模型
@dataclass
class TokenUsage:
"""Token 用量记录"""
model: str # 使用的模型
prompt_tokens: int # 输入 Token 数
completion_tokens: int # 输出 Token 数
total_tokens: int # 总 Token 数
cost: float # 本次调用成本（美元）
timestamp: float = field(default_factory=time.time) # 时间戳
user_id: str = "anonymous" # 用户 ID
endpoint: str = "" # 调用端点
class CostTracker:
"""成本追踪器"""
# 各模型价格（每百万 tokens，美元）
MODEL_PRICES = {
"gpt-4o": {"input": 2.50, "output": 10.00}, # GPT-4o
"gpt-4o-mini": {"input": 0.15, "output": 0.60}, # GPT-4o-mini
"gpt-4-turbo": {"input": 10.00, "output": 30.00}, # GPT-4-Turbo
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00}, # Claude
"claude-3-haiku": {"input": 0.25, "output": 1.25}, # Claude Haiku
"deepseek-chat": {"input": 0.14, "output": 0.28}, # DeepSeek
"deepseek-reasoner": {"input": 0.55, "output": 2.19}, # DeepSeek-R1
}
def __init__(self, redis_url: str = "redis://localhost"):***@staticmethod
def compress_context(messages: List[dict], max_tokens: int = 2000) -> List[dict]:
"""压缩上下文：从最近的消息开始保留"""
compressed = [] # 压缩后的消息列表
current_tokens = 0 # 当前 Token 计数
# 从最近的消息开始保留（最近的最重要）
for msg in reversed(messages): # 逆序遍历
msg_tokens = len(tiktoken.encoding_for_model("gpt-4o").encode(msg["content"]))
if current_tokens + msg_tokens > max_tokens: # 超出预算
# 如果放不下，尝试截断用户消息
if msg["role"] == "user":
content = msg["content"][:int(len(msg["content"]) * 0.5)] # 保留前半
compressed.append({"role": msg["role"], "content": content + "..."})
break # 停止处理更早的消息
compressed.append(msg) # 完整保留
current_tokens += msg_tokens
return list(reversed(compressed)) # 恢复正序
@staticmethod
def optimize_system_prompt(prompt: str) -> str:
"""优化 System Prompt：去除多余空格和重复说明"""
prompt = " ".join(prompt.split()) # 压缩多余空白
lines = prompt.split("。") # 按句号分割
seen = set() # 已见句集合
unique_lines = []
for line in lines: # 去重
if line.strip() and line.strip() not in seen:
seen.add(line.strip())
unique_lines.append(line.strip())
return "。".join(unique_lines) # 重新拼接
@staticmethod
def truncate_tool_results(results: str, max_chars: int = 500) -> str:
"""截断工具调用结果"""
if len(results) <= max_chars: # 未超限
return results
return results[:max_chars] + f"\n...(截断，原始长度 {len(results)} 字符)"9.5.6 成本优化全景图
┌─────────────────────────────────────────────────────────────┐
│ 成本优化策略全景 │
├───────────────┬──────────────┬───────────────┬───────────────┤
│ 策略 │ 预期节省 │ 实现难度 │ 适用场景 │
├───────────────┼──────────────┼───────────────┼───────────────┤
│ Prefix Cache │ 30-50% │ 零（自动） │ 长 System Prompt│
│ Response Cache│ 10-30% │ 低 │ FAQ/重复问题 │
│ Semantic Cache│ 20-40% │ 高 │ 通用问答 │
│ 模型分层 │ 40-70% │ 中 │ 所有场景 │
│ Prompt 压缩 │ 15-30% │ 低 │ 长上下文 │
│ 限制输出长度 │ 10-20% │ 低 │ 所有场景 │
│ 批量处理 │ 5-15% │ 中 │ 后台任务 │
└───────────────┴──────────────┴───────────────┴───────────────┘常见误区
"成本控制就是用便宜模型"——便宜模型的能力也弱，简单任务可以用，复杂任务用了反而需要更多轮交互，最终成本更高。模型分层是关键：按任务复杂度选模型。
"Token 计数不精确没关系"——小误差累积起来可能偏差很大。特别是长上下文场景，1,000 个请求每个偏差 100 Token，一天就差 10 万 Token，对应几美元的误差。
"预算告警设了就够了"——告警只是第一步，更重要的是自动响应。当成本达到 80% 时应该自动切换到便宜模型，达到 100% 时应该熔断非核心请求。
"Prompt 长一点无所谓"——每 1,000 个 Token 的输入成本约 $0.0025（GPT-4o）。如果 System Prompt 有 5,000 Token，每次调用多花 $0.0125。一天 10,000 次调用就是 $125——仅 System Prompt 就占了月成本的 $3,750。
"缓存命中率不影响成本"——缓存命中率每提高 10%，成本就降低约 10%。监控缓存命中率并持续优化是成本控制的关键手段。
"不需要按用户追踪成本"——不同用户的成本差异可能很大。一个高频用户可能消耗 80% 的成本。按用户追踪成本有助于设计差异化定价策略或设置配额。
本节小结
| 要点 | 说明 |
|---|---|
| Token 计费 | 了解各模型的 input/output 价格，这是成本控制的基础 |
| 实时监控 | 按小时/天/用户维度追踪 Token 用量和成本 |
| 模型分层 | 简单任务用便宜模型，复杂任务用贵模型，能节省 40~70% |
| 缓存复用 | Prefix Cache + Semantic Cache 是成本节省的利器 |
| Prompt 优化 | 压缩上下文、截断工具结果、优化 System Prompt |
| 预算告警 | 设置每日预算上限，80% 警告，100% 熔断 |
成本控制让系统的运行变得可持续。但还有一个问题影响用户体验：用户在等待 LLM 响应的几秒甚至几十秒里，看到的是什么？ 如果只是一个旋转的加载图标，再好的系统也会让人觉得"慢"。下一节我们讨论流式输出——如何让用户在等待中就能看到内容逐渐生成，体验"实时打字"的效果。
下一节：9.6 流式输出 Streaming——SSE 与 WebSocket，让用户在等待中看见进展。