_{🧠 图解记忆：成本先按租户、任务、模型与步骤归因，再优化缓存、路由、上下文和调用次数；点击图片可查看原图。}

**成本监控架构：**

展开 Python 代码示例（53 行）

import prometheus_client as prom
from prometheus_client import Counter, Histogram, Gauge

# 定义指标
TOKEN_USAGE = Counter(
    'agent_tokens_total',
    'Total tokens consumed',
    ['model', 'agent_type', 'user_tier']
)

TASK_COST = Histogram(
    'agent_task_cost_usd',
    'Cost per task in USD',
    ['agent_type'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5]
)

MONTHLY_BUDGET = Gauge(
    'agent_monthly_budget_remaining_usd',
    'Remaining monthly budget'
)

# 成本追踪装饰器
def track_cost(model: str, price_per_1k_input: float, price_per_1k_output: float):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            start_tokens = get_token_count()
            result = await func(*args, **kwargs)
            
            end_tokens = get_token_count()
            input_tokens = end_tokens['input'] - start_tokens['input']
            output_tokens = end_tokens['output'] - start_tokens['output']
            
            cost = (input_tokens / 1000 * price_per_1k_input + 
                    output_tokens / 1000 * price_per_1k_output)
            
            TOKEN_USAGE.labels(model=model, agent_type=func.__name__).inc(
                input_tokens + output_tokens
            )
            TASK_COST.labels(agent_type=func.__name__).observe(cost)
            
            return result
        return wrapper
    return decorator

# 预算告警
def check_budget_alert():
    monthly_spent = get_monthly_cost()
    monthly_limit = get_monthly_limit()
    MONTHLY_BUDGET.set(monthly_limit - monthly_spent)
    
    if monthly_spent > monthly_limit * 0.8:
        send_alert(f"月度预算已达 80%，剩余 ${monthly_limit - monthly_spent:.2f}")

成本优化策略：

策略	节省比例	实现方式
语义缓存	30-50%	Embedding 相似度 > 0.95 直接返回缓存
模型路由	30-40%	简单任务用 DeepSeek V4-Flash，复杂用 GPT-4
上下文压缩	40-90%	LLMLingua / Recomp 压缩历史
Token 配额	动态	按用户 tier 设置每日上限

展开 Python 代码示例（30 行）

# 语义缓存实现
class SemanticCache:
    def __init__(self, similarity_threshold=0.95):
        self.cache = FAISS.from_texts(CACHE_TEXTS, CACHE_EMBEDDINGS)
        self.similarity_threshold = similarity_threshold
        self.cache_hits = 0
        self.cache_misses = 0
    
    async def get(self, query: str) -> Optional[str]:
        query_emb = get_embedding(query)
        scores, indices = self.cache.search(query_emb, k=1)
        
        if scores[0] > self.similarity_threshold:
            self.cache_hits += 1
            return self.cache_results[indices[0]]
        
        self.cache_misses += 1
        return None
    
    async def set(self, query: str, response: str):
        # 异步写入，避免阻塞
        await asyncio.get_event_loop().run_in_executor(
            None, 
            lambda: self.cache.add_texts([query], [get_embedding(query)])
        )
        self.cache_results.append(response)
    
    def hit_rate(self) -> float:
        total = self.cache_hits + self.cache_misses
        return self.cache_hits / total if total > 0 else 0.0

📚 参考：LiteLLM（Token 消耗追踪与预算）

面试话术：

"成本治理先按租户、任务、模型、步骤和输入/输出 token 做归因，再评估缓存、模型路由、上下文裁剪、批处理和步数限制。每项优化都可能影响新鲜度、质量或延迟，因此必须在同一任务集和 SLO 下比较；面试只使用本人真实账单与实验数据。"

---