LLM 工作流优化平衡术：如何建立延迟与 Token 成本的双维度评估体系

LLM 工作流优化平衡术：如何建立延迟与 Token 成本的双维度评估体系

LLM 工作流的账单和首字延迟往往同时上升：会话变长、上下文重复发送或把简单任务交给高规格模型，都会增加成本和等待时间。具体比例和延迟必须从自身的调用日志中计算。

本文讨论如何建立延迟与成本的双维评估，并据此调整路由、缓存和上下文策略。

盲目降成本（比如直接把模型全部切换到极低参数的开源小模型）会导致输出质量崩塌；而盲目追求性能（把所有任务一律灌给最顶级的旗舰模型）则会在高并发下把云服务预算瞬间拉爆。

解决这个矛盾，关键在于在工程层建立一套延迟与成本双维度量化的评估与路由体系。

1. 拆解账单与耗时：高成本与高延迟到底花在哪了

要进行精准优化，必须先拿到请求链路的颗粒度数据。

我们在 LLM 网关层埋点了可观测性探针，对过去 7 天的 100 万次 Agent 工作流调用做了一次量化统计。分析结果揭示了两个违背直觉的工程事实：

搞清楚了成本与延迟的消耗分布，治理策略就有了明确的杠杆点：分流、缓存与 Prompt 剪枝。

2. 延迟与成本双维治理架构：模型路由与语义缓存

我们在 LLM 交付层引入了分级路由（Model Routing）与语义缓存（Semantic Cache）二元架构。

流量打进来后，优先经过语义缓存匹配；未命中时由轻量路由分类器对 Prompt 复杂度进行评估，精准分发至对应的模型节点处理。

flowchart TD
A[客户端 Agent 请求] –> B{第一级防线: 语义缓存 Semantic Cache}
B — 向量相似度 > 0.95 & 权限校验通过 –> C[立即返回缓存结果 延迟 < 20ms / 零 Token 成本]
B — 未命中缓存 –> D[进入第二级防线: 智能模型路由]
D –> E{评估 Prompt 复杂度与任务类型}
E — 简单分类/抽取/意图识别 –> F[调用轻量高并发模型 / 延迟 100ms / 低成本]
E — 复杂代码生成/多步骤推理 –> G[调用旗舰主推理模型 / 保证极致输出质量]
F & G –> H[写回语义缓存并记录延迟/Token 账单探针]

通过这套架构，原本一刀切发往顶尖模型的请求被合理分流，系统吞吐量与经济性得到了根本性的改善。

3. 基于 Python 3.11 的生产级模型路由与成本计算器实现

下面是完整的 Python 3.11 智能模型路由分发与 Token 成本控制代码。代码中包含了具体的 Prompt 长度估算、路由分发策略、耗时监控以及兜底熔断。

import time
import asyncio
from typing import Dict, Any, Optional, Tuple
from dataclasses import dataclass

@dataclass
class ModelCostConfig:
input_cost_per_1k: float # 每千 Token 输入成本 (美元)
output_cost_per_1k: float # 每千 Token 输出成本 (美元)
avg_ttft_ms: float # 预期首字延迟 (毫秒)

class CostAndLatencyBalancer:
def __init__(self):
# 建立不同模型档位的成本与性能基线表
self.model_registry: Dict[str, ModelCostConfig] = {
"fast-lite-model": ModelCostConfig(
input_cost_per_1k=0.0005,
output_cost_per_1k=0.0015,
avg_ttft_ms=120.0
),
"flagship-reasoning-model": ModelCostConfig(
input_cost_per_1k=0.0100,
output_cost_per_1k=0.0300,
avg_ttft_ms=850.0
)
}
# 简单内存语义缓存模拟
self._semantic_cache: Dict[str, str] = {}

async def dispatch_request(
self,
prompt: str,
task_type: str,
max_cost_budget: float = 0.05
) -> Dict[str, Any]:
"""
带成本与延迟路由控制的 LLM 调度入口
"""
start_time = time.perf_counter()

# 1. 检查语义缓存（极低延迟与零 Token 成本）
cache_key = hash(prompt)
if cache_key in self._semantic_cache:
latency = (time.perf_counter() – start_time) * 1000
return {
"status": "success",
"source": "semantic_cache",
"result": self._semantic_cache[cache_key],
"metrics": {"latency_ms": round(latency, 2), "estimated_cost_usd": 0.0}
}

# 2. 估算输入 Token 规模 (简单词频估算，生产环境可替换为 tiktoken)
estimated_input_tokens = len(prompt.split()) * 1.3

# 3. 智能模型路由选择
selected_model = self._select_optimal_model(
task_type, estimated_input_tokens, max_cost_budget
)

# 4. 发起 API 通信并监控耗时
try:
response_text, output_tokens = await self._call_llm_backend(
selected_model, prompt
)

# 计算本次调用的精确成本
cost_cfg = self.model_registry[selected_model]
total_cost = (
(estimated_input_tokens / 1000.0) * cost_cfg.input_cost_per_1k +
(output_tokens / 1000.0) * cost_cfg.output_cost_per_1k
)

# 写入缓存
self._semantic_cache[cache_key] = response_text
latency = (time.perf_counter() – start_time) * 1000

return {
"status": "success",
"source": selected_model,
"result": response_text,
"metrics": {
"latency_ms": round(latency, 2),
"estimated_cost_usd": round(total_cost, 6),
"input_tokens": int(estimated_input_tokens),
"output_tokens": output_tokens
}
}
except Exception as err:
return {
"status": "error",
"reason": f"模型调用失败: {str(err)}",
"fallback_result": "系统响应超时，已触发降级方案"
}

def _select_optimal_model(
self, task_type: str, input_tokens: float, max_budget: float
) -> str:
"""根据任务类型与预估 Token 预算路由模型"""
# 规则 1: 文本分类、关键词提取等轻量任务，强制走轻量模型
if task_type in ["classification", "extraction", "intent_detect"]:
return "fast-lite-model"

# 规则 2: 超长上下文且预算有限时，降级走轻量模型
cost_flagship_est = (input_tokens / 1000.0) * self.model_registry["flagship-reasoning-model"].input_cost_per_1k
if cost_flagship_est > max_budget:
return "fast-lite-model"

# 默认复杂推理任务走旗舰模型
return "flagship-reasoning-model"

async def _call_llm_backend(self, model_name: str, prompt: str) -> Tuple[str, int]:
"""模拟后端模型 API 通信"""
cfg = self.model_registry[model_name]
# 模拟通信延迟
await asyncio.sleep(cfg.avg_ttft_ms / 1000.0)
return f"[{model_name}] 针对任务的生成结果", 150

4. 评估口径与双维度观察

下表是记录路由与缓存候选方案的格式示例；只有在相同任务集、模型版本和统计窗口下采集的结果才能比较。

路由选择应由任务复杂度、评测结果和回退条件共同决定，不能只按模型规模下结论。

建立起延迟与成本的双维度评估体系，用工程架构把简单的意图分流给轻量模型，用语义缓存过滤重复问题，才能在把用户体验拉满的同时，把 Token 账单牢牢锁在安全线以内。