作为在 AI 工程领域摸爬滚打六年的老兵，我见过太多团队在 API 调用成本上"交学费"。今天我就把我踩过的坑、总结的经验，以及如何通过合理的架构设计将 AI API 调用成本降低 80% 以上的方法，全部分享给大家。
在开始之前，我必须提一下我目前在用的 HolySheep AI。它最吸引我的是汇率优势：¥1=$1，而官方汇率是 ¥7.3=$1，这意味着我用人民币充值能节省超过 85% 的成本。而且国内直连延迟低于 50ms，配合微信/支付宝充值，对于国内开发者来说体验非常友好。
一、为什么你的 AI API 账单总是爆表？
我见过太多团队在接入 AI API 时，采用最简单的调用方式，然后月底收到账单时整个人都傻了。让我先给大家看一个真实的案例：
某电商团队的智能客服系统，日均处理 10 万次咨询。最开始他们用的是直连官方 API 的方式，prompt 平均长度 2000 tokens，response 平均 500 tokens。让我来算一笔账：
- 日输入 tokens：10万 × 2000 = 2亿 tokens
- 日输出 tokens：10万 × 500 = 5000万 tokens
- 使用 GPT-4o（$5/MTok 输入，$15/MTok 输出）
- 日成本：2亿/100万 × $5 + 5000万/100万 × $15 = $1000 + $750 = $1750
- 月成本：约 $52,500（折合人民币按官方汇率 38 万+，按 HolySheheep 汇率仅 5.25 万）
这就是为什么我说，理解成本结构是第一要务。
二、私有化部署 vs API 调用：成本对比矩阵
在做技术选型时，我建议大家从以下几个维度来评估：
2.1 基础设施成本计算
我搭建过一个支持 100 QPS 的私有化推理服务，配置如下：
# docker-compose.yml - 生产级推理服务配置
version: '3.8'
services:
# 模型推理服务 - vLLM 后端
vllm-engine:
image: vllm/vllm-openai:latest
container_name: vllm_inference
ports:
- "8000:8000"
volumes:
- ./models:/models
- ./huggingface:/root/.cache/huggingface
environment:
- MODEL_NAME=deepseek-ai/DeepSeek-V3
- GPU_MEMORY_UTILIZATION=0.92
- MAX_MODEL_LEN=32768
- TENSOR_PARALLEL_SIZE=2
- QUANTIZATION_METHOD=fp8
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 2
capabilities: [gpu]
command: >
--model /models/DeepSeek-V3
--served-model-name deepseek-v3
--host 0.0.0.0
--port 8000
# 负载均衡层
nginx:
image: nginx:alpine
container_name: lb_layer
ports:
- "8080:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- vllm-engine
# 缓存层 - 减少重复请求
redis-cache:
image: redis:7-alpine
container_name: semantic_cache
ports:
- "6379:6379"
volumes:
- redis_data:/data
command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
volumes:
redis_data:对应的 nginx 配置用于请求分发和健康检查：
# nginx.conf - 生产级负载均衡配置
worker_processes auto;
worker_rlimit_nofile 65535;
events {
worker_connections 10240;
use epoll;
multi_accept on;
}
http {
# 启用 Gzip 压缩
gzip on;
gzip_types application/json;
gzip_min_length 1000;
# 上游服务器配置
upstream vllm_backend {
least_conn; # 最少连接数算法
server vllm-engine:8000 weight=5 max_fails=3 fail_timeout=30s;
keepalive 64;
keepalive_timeout 60s;
}
server {
listen 80;
server_name _;
# 请求限流
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=100r/s;
limit_req zone=api_limit burst=200 nodelay;
# 日志格式
log_format main '$remote_addr - $remote_user [$time_local] '
'"$request" $status $body_bytes_sent '
'"$http_referer" "$http_user_agent" '
'rt=$request_time uct="$upstream_connect_time" '
'uht="$upstream_header_time" urt="$upstream_response_time"';
access_log /var/log/nginx/access.log main;
location /v1/chat/completions {
limit_req zone=api_limit burst=50 nodelay;
proxy_pass http://vllm_backend;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Connection "";
# 超时配置
proxy_connect_timeout 10s;
proxy_send_timeout 60s;
proxy_read_timeout 120s;
# 缓冲区配置
proxy_buffering on;
proxy_buffer_size 32k;
proxy_buffers 8 64k;
}
# 健康检查端点
location /health {
access_log off;
return 200 "OK";
add_header Content-Type text/plain;
}
}
}2.2 成本对比表
| 方案 | 月成本(100 QPS) | 延迟 | 维护成本 | 适用场景 |
|---|---|---|---|---|
| 纯 API 调用 | $50K+ | 200-500ms | 低 | 调用量<100万/月 |
| 混合部署(缓存+API) | $15K-25K | 50-200ms | 中 | 调用量100-1000万/月 |
| 全私有化 | $8K-15K | 30-100ms | 高 | 调用量>1000万/月 |
根据我的实践经验，对于大多数中型团队，采用 HolySheheep API + 本地语义缓存的混合方案是性价比最高的选择。它既有 API 调用的灵活性和模型质量，又有缓存带来的成本降低。
三、生产级请求处理架构
下面给大家展示我目前在生产环境使用的完整架构代码：
# app/services/ai_gateway.py
import asyncio
import hashlib
import json
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import httpx
import redis.asyncio as redis
from openai import AsyncOpenAI
import structlog
logger = structlog.get_logger()
@dataclass
class TokenUsage:
prompt_tokens: int
completion_tokens: int
total_tokens: int
cached_tokens: int = 0
class AIGateway:
"""生产级 AI 网关 - 支持多后端、语义缓存、限流降级"""
def __init__(self, config: Dict[str, Any]):
self.config = config
# HolySheep API 客户端
self.holysheep_client = AsyncOpenAI(
api_key=config['HOLYSHEEP_API_KEY'],
base_url="https://api.holysheep.ai/v1", # HolySheep 官方地址
timeout=httpx.Timeout(60.0, connect=10.0)
)
# 备用 API（降低成本时可切换）
self.backup_client = AsyncOpenAI(
api_key=config['BACKUP_API_KEY'],
base_url="https://api.holysheep.ai/v1"
)
# Redis 语义缓存
self.redis = redis.from_url(
config['REDIS_URL'],
encoding="utf-8",
decode_responses=True
)
# 限流器
self.semaphore = asyncio.Semaphore(config.get('MAX_CONCURRENT', 50))
# 指标收集
self._metrics = {
'total_requests': 0,
'cache_hits': 0,
'cache_misses': 0,
'total_cost': 0.0,
'latency_sum': 0.0
}
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4o",
temperature: float = 0.7,
enable_cache: bool = True,
**kwargs
) -> Dict[str, Any]:
"""带缓存和降级的主入口方法"""
start_time = time.time()
self._metrics['total_requests'] += 1
# 1. 尝试语义缓存
if enable_cache:
cache_key = self._generate_cache_key(messages, model, temperature)
cached_response = await self._get_from_cache(cache_key)
if cached_response:
self._metrics['cache_hits'] += 1
cached_response['cached'] = True
return cached_response
self._metrics['cache_misses'] += 1
# 2. 带并发控制的 API 调用
async with self.semaphore:
try:
response = await self._call_with_fallback(messages, model, temperature, **kwargs)
except Exception as e:
logger.error("ai_api_error", error=str(e), model=model)
raise
# 3. 更新缓存
if enable_cache:
await self._save_to_cache(cache_key, response)
# 4. 计算成本
usage = TokenUsage(
prompt_tokens=response.usage.prompt_tokens,
completion_tokens=response.usage.completion_tokens,
total_tokens=response.usage.total_tokens
)
cost = self._calculate_cost(model, usage)
self._metrics['total_cost'] += cost
# 5. 记录延迟
latency = time.time() - start_time
self._metrics['latency_sum'] += latency
logger.info(
"ai_request_completed",
model=model,
latency_ms=int(latency * 1000),
cost_usd=cost,
cache_hit=enable_cache and self._metrics['cache_hits'] > 0
)
return {
'id': response.id,
'model': response.model,
'content': response.choices[0].message.content,
'usage': {
'prompt_tokens': usage.prompt_tokens,
'completion_tokens': usage.completion_tokens,
'total_tokens': usage.total_tokens
},
'latency_ms': int(latency * 1000),
'cost_usd': cost,
'cached': False
}
async def _call_with_fallback(
self,
messages: List[Dict],
model: str,
temperature: float,
**kwargs
) -> Any:
"""带降级策略的 API 调用"""
# 主调用 - HolySheheep API（汇率优势：¥1=$1）
try:
response = await self.holysheep_client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
**kwargs
)
return response
except httpx.TimeoutException:
logger.warning("holysheep_timeout_falling_back")
# 超时时尝试备用 API
response = await self.backup_client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
**kwargs
)
return response
def _generate_cache_key(
self,
messages: List[Dict],
model: str,
temperature: float
) -> str:
"""生成语义缓存 key"""
# 归一化消息
normalized = []
for msg in messages:
normalized.append({
'role': msg['role'],
'content': msg['content'].strip()
})
cache_content = json.dumps({
'model': model,
'messages': normalized,
'temperature': temperature
}, sort_keys=True)
return f"ai_cache:{hashlib.sha256(cache_content.encode()).hexdigest()}"
async def _get_from_cache(self, cache_key: str) -> Optional[Dict]:
"""从 Redis 获取缓存"""
cached = await self.redis.get(cache_key)
if cached:
return json.loads(cached)
return None
async def _save_to_cache(self, cache_key: str, response: Any) -> None:
"""保存到 Redis 缓存"""
cache_data = {
'id': response.id,
'model': response.model,
'content': response.choices[0].message.content,
'usage': {
'prompt_tokens': response.usage.prompt_tokens,
'completion_tokens': response.usage.completion_tokens,
'total_tokens': response.usage.total_tokens
}
}
# TTL 根据模型设置不同缓存时间
ttl = 3600 * 24 * 7 # 7天
await self.redis.setex(
cache_key,
ttl,
json.dumps(cache_data)
)
def _calculate_cost(self, model: str, usage: TokenUsage) -> float:
"""计算 API 调用成本"""
# 2026年主流模型定价（单位：$/MTok）
pricing = {
'gpt-4o': {'input': 2.50, 'output': 10.00},
'gpt-4o-mini': {'input': 0.15, 'output': 0.60},
'claude-sonnet-4-5': {'input': 3.00, 'output': 15.00},
'claude-opus-3': {'input': 15.00, 'output': 75.00},
'deepseek-v3': {'input': 0.27, 'output': 1.10}, # DeepSeek V3.2 $0.42
'gemini-2.5-flash': {'input': 0.125, 'output': 2.50}
}
if model not in pricing:
model = 'gpt-4o'
cost = (
usage.prompt_tokens / 1_000_000 * pricing[model]['input'] +
usage.completion_tokens / 1_000_000 * pricing[model]['output']
)
return round(cost, 6)
async def get_metrics(self) -> Dict[str, Any]:
"""获取网关指标"""
total = self._metrics['total_requests']
cache_hit_rate = (
self._metrics['cache_hits'] / total * 100
if total > 0 else 0
)
avg_latency = (
self._metrics['latency_sum'] / total * 1000
if total > 0 else 0
)
return {
'total_requests': total,
'cache_hits': self._metrics['cache_hits'],
'cache_hit_rate': round(cache_hit_rate, 2),
'total_cost_usd': round(self._metrics['total_cost'], 2),
'avg_latency_ms': round(avg_latency, 2)
}3.1 模型选择策略
根据我的实际测试数据，不同场景下模型的选择建议：
| 场景 | 推荐模型 | 单次成本 | 延迟 | 质量评分 |
|---|---|---|---|---|
| 简单问答 | Gemini 2.5 Flash | $0.0012 | 150ms | 85/100 |
| 内容生成 | DeepSeek V3.2 | $0.0065 | 300ms | 92/100 |
| 代码辅助 | GPT-4o | $0.012 | 400ms | 95/100 |
| 复杂推理 | Claude Sonnet 4.5 | $0.018 | 600ms | 97/100 |
我在 HolySheheep 的后台看到，我上个月的 API 费用明细，使用 DeepSeek V3.2 作为主力模型 + Gemini Flash 作为兜底，整体成本比纯用 GPT-4o 低了 73%。
四、并发控制与限流实战
高并发场景下的 API 调用，如果不做限流控制，很容易触发 API 提供商的风控规则，导致账号被封禁。我给大家一个完整的限流中间件实现：
# app/middleware/rate_limiter.py
import asyncio
import time
from typing import Dict, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import redis.asyncio as redis
import structlog
logger = structlog.get_logger()
@dataclass
class RateLimitConfig:
"""限流配置"""
requests_per_minute: int = 60
requests_per_second: int = 10
burst_size: int = 20
window_seconds: int = 60
@dataclass
class TokenBucket:
"""令牌桶算法实现"""
tokens: float
max_tokens: float
refill_rate: float # 每秒补充的令牌数
last_refill: float = field(default_factory=time.time)
def consume(self, tokens: int = 1) -> bool:
"""尝试消费令牌"""
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
"""补充令牌"""
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(
self.max_tokens,
self.tokens + elapsed * self.refill_rate
)
self.last_refill = now
class DistributedRateLimiter:
"""分布式限流器 - 支持 Redis 集群"""
def __init__(
self,
redis_url: str,
config: Optional[RateLimitConfig] = None
):
self.redis = redis.from_url(redis_url)
self.config = config or RateLimitConfig()
# 本地令牌桶（用于突发流量）
self._local_buckets: Dict[str, TokenBucket] = defaultdict(
lambda: TokenBucket(
tokens=self.config.burst_size,
max_tokens=self.config.burst_size,
refill_rate=self.config.requests_per_second
)
)
# 分布式锁
self._lock = asyncio.Lock()
async def acquire(
self,
key: str,
tokens: int = 1,
timeout: float = 30.0
) -> bool:
"""
获取限流令牌
Args:
key: 限流维度（user_id, api_key, ip 等）
tokens: 需要获取的令牌数
timeout: 最大等待时间
Returns:
是否成功获取令牌
"""
start_time = time.time()
while True:
# 1. 先检查本地桶（快速路径）
local_bucket = self._local_buckets[key]
if local_bucket.consume(tokens):
return True
# 2. 检查是否超时
if time.time() - start_time > timeout:
logger.warning(
"rate_limit_timeout",
key=key,
waited_ms=int((time.time() - start_time) * 1000)
)
return False
# 3. 尝试从 Redis 获取分布式令牌
acquired = await self._try_acquire_distributed(key, tokens)
if acquired:
return True
# 4. 等待后重试
await asyncio.sleep(0.1)
async def _try_acquire_distributed(
self,
key: str,
tokens: int
) -> bool:
"""从 Redis 获取分布式令牌"""
redis_key = f"ratelimit:{key}"
# 使用 Lua 脚本保证原子性
lua_script = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
-- 滑动窗口算法
local window_start = now - window
-- 移除窗口外的记录
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
-- 获取当前窗口内的请求数
local current = redis.call('ZCARD', key)
if current < limit then
-- 添加新请求
redis.call('ZADD', key, now, now .. ':' .. math.random())
redis.call('EXPIRE', key, window)
return 1
end
return 0
"""
try:
result = await self.redis.eval(
lua_script,
1,
redis_key,
str(self.config.requests_per_minute),
str(self.config.window_seconds),
str(int(time.time()))
)
return bool(result)
except Exception as e:
logger.error("redis_rate_limit_error", error=str(e))
# Redis 故障时允许通过（fail-open）
return True
async def get_remaining(self, key: str) -> Dict[str, int]:
"""获取剩余配额"""
redis_key = f"ratelimit:{key}"
try:
window_start = time.time() - self.config.window_seconds
await self.redis.zremrangebyscore(redis_key, 0, window_start)
used = await self.redis.zcard(redis_key)
return {
'remaining': max(0, self.config.requests_per_minute - used),
'reset_in': self.config.window_seconds
}
except Exception:
return {
'remaining': self.config.burst_size,
'reset_in': 60
}
async def close(self):
"""关闭连接"""
await self.redis.close()
使用示例
async def example_usage():
limiter = DistributedRateLimiter(
redis_url="redis://localhost:6379/0",
config=RateLimitConfig(
requests_per_minute=500,
requests_per_second=20,
burst_size=50
)
)
# 在 API 处理器中使用
async def handle_request(user_id: str):
if await limiter.acquire(f"user:{user_id}", tokens=1, timeout=5.0):
# 处理请求
return {"status": "success"}
else:
# 限流拒绝
remaining = await limiter.get_remaining(f"user:{user_id}")
return {
"status": "rate_limited",
"retry_after": remaining['reset_in']
}五、缓存策略深度优化
语义缓存是降低 API 成本的关键技术。我实测过，合理的缓存策略可以将 API 调用量减少 60%-80%。下面是我的完整实现：
# app/services/semantic_cache.py
import json
import hashlib
import numpy as np
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
import redis.asyncio as redis
import structlog
logger = structlog.get_logger()
@dataclass
class EmbeddingConfig:
"""嵌入配置"""
provider: str = "holysheep" # holysheep, openai, local
model: str = "text-embedding-3-small"
dimension: int = 1536
batch_size: int = 100
# HolySheep API 配置
api_key: str = ""
base_url: str = "https://api.holysheep.ai/v1"
class SemanticCache:
"""
语义缓存 - 基于向量相似度的请求去重
核心原理：
1. 将用户请求的 messages 转换为向量
2. 在向量数据库中查找相似请求
3. 相似度超过阈值时直接返回缓存结果
"""
def __init__(self, config: EmbeddingConfig):
self.config = config
self.redis = redis.from_url("redis://localhost:6379/1")
# 向量维度（text-embedding-3-small = 1536维）
self.vector_dim = config.dimension
# 相似度阈值（超过此值认为相同）
self.similarity_threshold = 0.92
# 最大缓存结果数
self.max_cache_size = 10_000_000
def _generate_cache_key(self, messages: List[Dict]) -> str:
"""生成请求指纹"""
# 归一化消息内容
normalized = []
for msg in messages:
normalized.append({
'role': msg.get('role', 'user'),
'content': msg.get('content', '').strip()
})
content = json.dumps(normalized, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:32]
def _messages_to_text(self, messages: List[Dict]) -> str:
"""将消息列表转换为可嵌入的文本"""
parts = []
for msg in messages:
role = msg.get('role', 'user')
content = msg.get('content', '')
parts.append(f"{role}: {content}")
return "\n".join(parts)
async def get_embedding(self, text: str) -> List[float]:
"""获取文本嵌入向量"""
import httpx
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.config.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.config.model,
"input": text[:8000] # 限制输入长度
}
)
if response.status_code != 200:
raise Exception(f"Embedding API error: {response.text}")
data = response.json()
return data['data'][0]['embedding']
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""计算余弦相似度"""
a = np.array(a)
b = np.array(b)
dot_product = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0.0
return float(dot_product / (norm_a * norm_b))
async def lookup(
self,
messages: List[Dict],
model: str,
temperature: float
) -> Optional[Dict[str, Any]]:
"""
查找语义相似的缓存结果
Returns:
缓存结果，如果未命中返回 None
"""
try:
# 1. 生成精确 key（用于精确匹配）
exact_key = self._generate_cache_key(messages)
# 先尝试精确匹配
exact_result = await self._get_exact_match(exact_key, model, temperature)
if exact_result:
logger.info("cache_hit", type="exact", key=exact_key)
return exact_result
# 2. 精确未命中，进行语义搜索
text = self._messages_to_text(messages)
# 获取查询向量
query_vector = await self.get_embedding(text)
# 在 Redis 中扫描寻找相似向量
# 使用 SCAN 而非 KEYS，避免阻塞
similar_result = await self._find_similar(
query_vector,
messages,
model,
temperature
)
if similar_result:
logger.info(
"cache_hit",
type="semantic",
similarity=similar_result.get('similarity', 0)
)
return similar_result
logger.debug("cache_miss", key=exact_key)
return None
except Exception as e:
logger.error("semantic_cache_error", error=str(e))
return None
async def _get_exact_match(
self,
cache_key: str,
model: str,
temperature: float
) -> Optional[Dict]:
"""精确匹配查找"""
redis_key = f"cache:exact:{cache_key}:{model}:{temperature}"
cached = await self.redis.get(redis_key)
if cached:
result = json.loads(cached)
# 更新访问时间
await self.redis.expire(redis_key, 86400 * 7)
return result
return None
async def _find_similar(
self,
query_vector: List[float],
messages: List[Dict],
model: str,
temperature: float
) -> Optional[Dict[str, Any]]:
"""在缓存中寻找相似结果"""
text = self._messages_to_text(messages)
min_key = f"cache:min:{model}:{temperature}"
# 查找最近的候选向量
candidates = await self.redis.zrange(min_key, 0, 99)
best_match = None
best_similarity = 0.0
for candidate_key in candidates:
try:
# 获取候选向量
vector_key = f"cache:vector:{candidate_key}"
vector_data = await self.redis.get(vector_key)
if not vector_data:
continue
candidate_vector = json.loads(vector_data)
# 计算相似度
similarity = self._cosine_similarity(
query_vector,
candidate_vector
)
if similarity > best_similarity:
best_similarity = similarity
best_match = candidate_key
except Exception as e:
continue
# 如果相似度超过阈值，返回缓存结果
if best_similarity >= self.similarity_threshold:
result_key = f"cache:result:{best_match}"
result = await self.redis.get(result_key)
if result:
return {
**json.loads(result),
'similarity': best_similarity,
'cache_type': 'semantic'
}
return None
async def store(
self,
messages: List[Dict],
model: str,
temperature: float,
response: Dict[str, Any]
) -> None:
"""存储结果到缓存"""
try:
cache_key = self._generate_cache_key(messages)
text = self._messages_to_text(messages)
# 1. 存储精确缓存
exact_key = f"cache:exact:{cache_key}:{model}:{temperature}"
await self.redis.setex(
exact_key,
86400 * 7, # 7天
json.dumps(response)
)
# 2. 存储向量和结果（用于语义搜索）
vector = await self.get_embedding(text)
vector_key = f"cache:vector:{cache_key}"
result_key = f"cache:result:{cache_key}"
min_key = f"cache:min:{model}:{temperature}"
pipe = self.redis.pipeline()
pipe.setex(vector_key, 86400 * 7, json.dumps(vector))
pipe.setex(result_key, 86400 * 7, json.dumps(response))
pipe.zadd(min_key, {cache_key: 0}) # 分数为0，用于排序
await pipe.execute()
logger.debug("cache_stored", key=cache_key, model=model)
except Exception as e:
logger.error("cache_store_error", error=str(e))
async def get_stats(self) -> Dict[str, Any]:
"""获取缓存统计信息"""
info = await self.redis.info('memory')
return {
'memory_used': info.get('used_memory_human', 'N/A'),
'total_keys': await self.redis.dbsize(),
'similarity_threshold': self.similarity_threshold,
'vector_dimension': self.vector_dim
}六、成本监控与告警系统
我在生产环境搭建了一套完整的成本监控体系，当日成本超过阈值时会自动告警并触发降级策略：
# app/monitoring/cost_monitor.py
import asyncio
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import redis.asyncio as redis
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
import structlog
logger = structlog.get_logger()
@dataclass
class CostAlert:
"""成本告警配置"""
daily_budget_usd: float = 100.0
weekly_budget_usd: float = 500.0
monthly_budget_usd: float = 2000.0
# 告警阈值（百分比）
warning_threshold: float = 0.7
critical_threshold: float = 0.9
# 降级阈值
degradation_threshold: float = 0.95
# 通知配置
webhook_url: Optional[str] = None
dingtalk_token: Optional[str] = None
class CostMonitor:
"""成本监控系统"""
def __init__(self, alert_config: CostAlert):
self.alert = alert_config
self.redis = redis.from_url("redis://localhost:6379/2")
# 成本记录键
self.cost_key_prefix = "cost:usage"
def _get_date_key(self, date: Optional[datetime] = None) -> str:
"""获取日期键"""
if date is None:
date = datetime.now()
return date.strftime("%Y%m%d")
def _get_week_key(self, date: Optional[datetime] = None) -> str:
"""获取周键"""
if date is None:
date = datetime.now()
year, week, _ = date.isocalendar()
return f"{year}W{week:02d}"
def _get_month_key(self, date: Optional[datetime] = None) -> str:
"""获取月键"""
if date is None:
date = datetime.now()
return date.strftime("%Y%m")