假设你有一个需要每日向大模型发送数百次查询的系统，每次调用都产生Token消耗和API费用。随着查询规模增长，成本会线性上升。本文从工程实践角度，介绍一套成本控制方案，涵盖Token成本估算、缓存策略、模型选择和费用监控。方案基于阿里云百炼和DashScope实现，适合每日监测数百个关键词的场景。读者需要具备Python开发基础，并已开通阿里云百炼服务、创建API Key。本文不涉及监测系统的业务逻辑和数据处理细节。
整体方案
flowchart LR
A[查询请求] --> B[缓存检查]
B -- 命中 --> C[返回缓存结果]
B -- 未命中 --> D[模型调用]
D --> E[结果解析与存储]
E --> F[Token与费用记录]
F --> G[费用告警与预算控制]
查询请求先经过缓存层，命中则直接返回，未命中才调用模型。每次调用后记录Token消耗和费用，用于监控和预算控制。
环境与账号准备
操作系统：Linux / macOS / Windows
Python 3.9+
依赖：dashscope、redis、prometheus_client
阿里云百炼服务已开通，并创建API Key
建议使用RAM用户，仅授予百炼调用权限
环境变量配置：
export DASHSCOPE_API_KEY=""
export REDIS_HOST=""
export REDIS_PORT=6379
export REDIS_PASSWORD=""
Token成本估算
在调用模型前，先估算每次查询的Token消耗。不同模型的Token单价不同，以通义千问-Plus为例，输入0.004元/千Token，输出0.012元/千Token。
估算方法：使用tiktoken库计算提示词的Token数。需要注意的是，tiktoken的编码方式与通义千问模型可能存在差异，估算结果仅供预算参考，实际消耗以API返回为准。
import tiktoken
def estimate_input_tokens(prompt: str, model: str = "qwen-plus") -> int:
# 通义千问使用cl100k_base编码
enc = tiktoken.get_encoding("cl100k_base")
return len(enc.encode(prompt))
根据估算结果，可以提前判断单次调用费用，避免超预算。
缓存策略
缓存是成本控制最有效的手段。对于相同问题，短时间内多次查询应直接返回缓存结果。
缓存键设计
缓存键由模型名称、提示词哈希和温度参数组成：
import hashlib
def build_cache_key(model: str, prompt: str, temperature: float = 0.0) -> str:
content = f"{model}:{prompt}:{temperature}"
return hashlib.sha256(content.encode()).hexdigest()
缓存存储与过期
使用Redis存储，设置过期时间（如24小时）：
import redis
r = redis.Redis(
host=REDIS_HOST,
port=REDIS_PORT,
password=REDIS_PASSWORD,
decode_responses=True
)
def get_cached_result(cache_key: str) -> str | None:
return r.get(cache_key)
def set_cached_result(cache_key: str, result: str, ttl: int = 86400):
r.setex(cache_key, ttl, result)
缓存命中率监控
通过Prometheus指标记录缓存命中次数和总请求次数：
from prometheus_client import Counter
cache_hits = Counter("cache_hits_total", "Total cache hits")
cache_misses = Counter("cache_misses_total", "Total cache misses")
def query_with_cache(prompt: str, model: str) -> str:
cache_key = build_cache_key(model, prompt)
cached = get_cached_result(cache_key)
if cached:
cache_hits.inc()
return cached
cache_misses.inc()
result = call_model(prompt, model) # 实际调用
set_cached_result(cache_key, result)
return result
模型选择
不同模型的价格和效果差异很大。对于可见度监测，不需要高创造性，选择性价比高的模型即可。
模型 输入价格(元/千Token) 输出价格(元/千Token) 适用场景
通义千问-Plus 0.004 0.012 一般监测，要求不高
通义千问-Max 0.02 0.06 需要更准确分析
通义千问-Turbo 0.002 0.006 高频简单查询
建议默认使用通义千问-Turbo，因为它在测试中表现足够且成本最低。当监测结果置信度不足时（例如模型输出与预期偏差较大），降级到Plus或Max。置信度不足的判断标准可以根据实际业务定义，比如输出内容与预设答案的相似度低于某个阈值。
MODEL_TIER = {
"high": "qwen-max",
"medium": "qwen-plus",
"low": "qwen-turbo"
}
def select_model(query_type: str) -> str:
# query_type: "high" 重要关键词, "medium" 一般词, "low" 长尾词
return MODEL_TIER.get(query_type, "qwen-turbo")
费用监控与告警
每次调用后记录Token消耗和费用，写入日志或Prometheus：
from prometheus_client import Gauge
total_cost = Gauge("total_cost_cny", "Total cost in CNY")
def record_cost(input_tokens: int, output_tokens: int, model: str):
# 根据模型单价计算费用
price_map = {
"qwen-turbo": (0.002, 0.006),
"qwen-plus": (0.004, 0.012),
"qwen-max": (0.02, 0.06)
}
input_price, output_price = price_map.get(model, (0.004, 0.012))
cost = (input_tokens * input_price + output_tokens * output_price) / 1000
total_cost.inc(cost)
设置预算告警，当日费用超过阈值时触发通知：
BUDGET_DAILY = 50.0 # 每日预算50元
def check_budget():
current = total_cost._value.get()
if current > BUDGET_DAILY:
# 发送告警，如邮件、钉钉
pass
验证结果
正常情况下，部署后应观察到：
缓存命中率逐步提升，预期首次监测后重复查询命中率会逐步提升，具体数值需根据实际查询分布测试验证
日均Token消耗和费用在预算范围内
费用监控指标正常上报，告警可触发
可通过Prometheus查询缓存命中率和费用数据。
费用与资源回收
阿里云百炼提供免费额度，但仅限试用，正式使用需关注计费
测试结束后，释放Redis实例和ECS资源，避免持续计费
API Key应定期轮换，并存储在密钥管理服务中
常见问题
缓存未命中导致费用激增
原因：缓存键设计不合理，如忽略了温度参数。
解决方法：确保缓存键包含所有影响输出的参数。
模型返回结果不稳定
原因：使用低温度但模型仍有随机性。
解决方法：设置temperature=0.0，并多次查询取多数结果。
总结
本文从Token成本估算、缓存策略、模型选择和费用监控四个方面，介绍了大模型API调用中的成本控制工程实践。关键经验是：缓存设计要包含所有影响输出的参数，模型选择需在效果和成本之间权衡，费用监控要实时且可告警。实际部署后，建议根据查询分布和缓存命中率调整策略，以达到成本与效果的平衡。
