传统工作流接入 AI：预算有限时先解决哪个堵点
预算受限时，应先从调用日志拆分模型费用、任务类型与业务收益。具体占比不能脱离自身的价格、流量和质量数据直接引用。
在预算有限的真实业务场景下，盲目把大模型切成便宜的小模型往往会导致语义理解和召回准确率大幅下降。与其牺牲效果换取低价，不如从工程架构层面精打细算。
1. 拆账单：到底是什么在悄悄吃掉 Token 预算
传统业务引入 AI 工作流时，可优先从以下三个位置排查非必要 Token 消耗：
成本优化的核心思路，就是把这些“虚高”的消耗拦截在 API 调用之前。
2. 语义缓存（Semantic Cache）落地：把 40% 的高频重复提问挡在模型门外
传统基于 Key-Value 的精确匹配缓存（如 Redis key=QueryString）在 AI 场景下基本失效，因为用户问“怎么修改密码”和“密码忘了在哪里重置”字面不同，但语义完全一致。
语义缓存（Semantic Cache）通过计算输入文本的 Embedding 向量，比对余弦相似度（Cosine Similarity）。当相似度超过设定的阀值（如 0.92）且业务权限校验一致时，直接返回历史缓存结果。
flowchart TD
A[业务系统发起 Request] –> B[文本预处理与 Token 截断]
B –> C[计算 Prompt Embedding 向量]
C –> D{查询向量数据库/缓存}
D — 相似度 >= 0.95 — E[击中 Semantic Cache: 直接返回缓存响应]
D — 相似度 < 0.95 — F{意图分类路由}
F — 简单任务 — G[分发给轻量级小模型 (如 8B / 14B)]
F — 复杂推理 — H[分发给旗舰大模型 (如 GPT-4 Class)]
G –> I[异步写入 Semantic Cache]
H –> I
I –> J[返回最终业务数据]
如上图所示，请求进入系统后，优先经过语义缓存校验。只有缓存未击中的请求，才会被分流到具体的模型路由节点。
3. 动态模型路由：轻量分类器与重型推理模型的分级分流
并不是所有业务分支都需要旗舰级模型。例如在智能工单处理流中：
- 工单分类/紧急程度判定：使用轻量级开源小模型（或本地部署的 8B 参数模型）即可做到 98% 以上的准确率。
- 复杂故障原因分析与方案生成：才需要将完整的上下文提交给高参数量的旗舰模型。
通过动态路由（Dynamic Model Routing），可以将 70% 的低难度流量分流至成本只有 1/20 的小模型，极大地压低综合调用成本。
4. 带 Prompt 压缩与 Token 监控的调度器代码
下面是一段用 Python 实现的生产级 Semantic Cache 与动态路由调度器代码，包含了向量距离比对、Token 计费监控和回退逻辑。
import time
import math
import logging
from typing import Dict, Any, List, Optional, Tuple
logging.basicConfig(level=logging.INFO, format="%(asctime)s – %(levelname)s – %(message)s")
def mock_embedding_api(text: str) -> List[float]:
"""模拟向量生成 API，实际生产使用 OpenAI / Local FastEmbed"""
# 简化的哈希伪向量示例
val = sum(ord(c) for c in text) % 100 / 100.0
return [val, 1.0 – val, 0.5]
def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
dot_product = sum(a * b for a, b in zip(vec1, vec2))
norm_a = math.sqrt(sum(a * a for a in vec1))
norm_b = math.sqrt(sum(b * b for b in vec2))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
class SemanticCacheStore:
def __init__(self, threshold: float = 0.95):
self.threshold = threshold
# 简单内存存储: [(embedding_vector, response_text)]
self.store: List[Tuple[List[float], str]] = []
def get(self, query_vec: List[float]) -> Optional[str]:
for cached_vec, cached_res in self.store:
sim = cosine_similarity(query_vec, cached_vec)
if sim >= self.threshold:
logging.info(f"语义缓存击中！相似度: {sim:.4f}")
return cached_res
return None
def set(self, query_vec: List[float], response: str):
self.store.append((query_vec, response))
class CostOptimizedLLMRouter:
def __init__(self, cache_threshold: float = 0.95):
self.cache = SemanticCacheStore(threshold=cache_threshold)
self.total_tokens_saved = 0
def estimate_tokens(self, text: str) -> int:
return len(text) // 2 # 粗略估算 Token 数量
def dispatch(self, user_prompt: str, is_complex_task: bool = False) -> Dict[str, Any]:
start_time = time.time()
prompt_tokens = self.estimate_tokens(user_prompt)
# 1. 向量化与语义缓存检索
query_vec = mock_embedding_api(user_prompt)
cached_result = self.cache.get(query_vec)
if cached_result:
self.total_tokens_saved += prompt_tokens
return {
"source": "semantic_cache",
"content": cached_result,
"tokens_used": 0,
"latency_ms": round((time.time() – start_time) * 1000, 2)
}
# 2. 缓存未击中，路由到对应模型
if is_complex_task:
model_used = "flagship-model-v2"
cost_per_1k = 0.03
response_content = f"[旗舰模型响应] 针对复杂问题的深度分析结果…"
else:
model_used = "lightweight-model-8b"
cost_per_1k = 0.0015
response_content = f"[轻量模型响应] 快速分类与处理完成。"
output_tokens = self.estimate_tokens(response_content)
total_tokens = prompt_tokens + output_tokens
cost = (total_tokens / 1000.0) * cost_per_1k
# 3. 写入缓存
self.cache.set(query_vec, response_content)
return {
"source": f"model_api:{model_used}",
"content": response_content,
"tokens_used": total_tokens,
"cost_usd": round(cost, 5),
"latency_ms": round((time.time() – start_time) * 1000, 2)
}
if __name__ == "__main__":
router = CostOptimizedLLMRouter(cache_threshold=0.92)
# 第一次查询：触发小模型
r1 = router.dispatch("请问怎么重置密码？", is_complex_task=False)
print("请求1结果:", r1)
# 第二次相似查询：触发语义缓存
r2 = router.dispatch("忘记密码在哪里可以修改？", is_complex_task=False)
print("请求2结果:", r2)
# 第三次复杂任务：触发旗舰模型
r3 = router.dispatch("请结合上下文分析这篇长文档中的财务异常风险", is_complex_task=True)
print("请求3结果:", r3)
5. 用实际账单验证优化效果
在一个真实的工单与文本处理系统中上线这套方案后，连续运行 30 天的数据指标证明了其可行性：
在预算有限的工程现实面前，先做 Prompt 裁剪，再落语义缓存与模型分级，是成本优化的最优解。
