当前位置：首页 > news >正文

Prompt 成本与延迟怎么评：先固定数据集和调用参数

news 2026/8/16 10:11:06

Prompt 成本与延迟怎么评：先固定数据集和调用参数

Prompt 对成本与延迟的影响可以测，但要先固定模型版本、数据集和生成参数。比较时分别记录输入 Token、输出 Token、TTFT 和完成时间，避免用一次调用下结论。

验证口径与记录方法

很多团队在对接大模型服务时，往往把 Prompt 优化当成单纯的“文案修饰”。但在真正的生产高并发系统里，Prompt 的结构直接决定了模型解码时的 KV Cache 命中率以及 Prompt Token 的计算开销。

如果系统在每次请求中都带上大量重复的 System Prompt、冗余的 Few-shot 样例以及未裁剪的历史对话，不仅会导致首字延迟急剧升高，更会在按 Token 结算的 API 账单上造成严重的资金浪费。

下面的决策树展示了当请求进入 Gateway 时，如何通过上下文剪枝与缓存命中机制来兼顾响应时间与费用控制。

Prompt 瘦身与动态上下文剪枝的具体裁剪策略

为将开销控制在合理的范围内，可以不能依赖大模型自己去理解复杂的长文档，而必须在请求发出前对 Prompt 进行严格的预处理与瘦身。

动态上下文剪枝的核心做法是：拆分固定 Prefix（如系统角色定义、Tool 定义）与动态后缀（用户提问、检索到的 RAG 上下文）。固定 Prefix 保持严格的字符级一致性，方便推理框架（如 vLLM、SGLang 或 OpenAI 的 Prefix Caching）精准命中缓存。

对于历史对话记录，采用“滑动窗口 + 语义摘要”策略。超过 4 轮的对话记录不再保留原文，而是通过小参数模型（如 Qwen2.5-3B）异步生成短摘要替换。

import time import hashlib from typing import List, Dict, Any, Optional class PromptContextTrimmer: def __init__(self, max_prompt_tokens: int = 1500, system_prompt: str = ""): self.max_prompt_tokens = max_prompt_tokens self.system_prompt = system_prompt self.system_prompt_hash = self._compute_hash(system_prompt) def _compute_hash(self, text: str) -> str: return hashlib.sha256(text.encode('utf-8')).hexdigest() def estimate_token_count(self, text: str) -> int: # 生产环境使用 tiktoken 或 transformers tokenizer，此处按粗略比例估计 return int(len(text) * 0.6) def trim_context(self, history: List[Dict[str, str]], rag_docs: List[str], user_query: str) -> Dict[str, Any]: sys_tokens = self.estimate_token_count(self.system_prompt) query_tokens = self.estimate_token_count(user_query) budget = self.max_prompt_tokens - sys_tokens - query_tokens if budget <= 0: raise ValueError("System prompt与User query已超出最大Token预算，无法构建上下文") selected_docs = [] doc_tokens_used = 0 for doc in rag_docs: doc_tok = self.estimate_token_count(doc) if doc_tokens_used + doc_tok <= int(budget * 0.6): selected_docs.append(doc) doc_tokens_used += doc_tok else: break remaining_budget = budget - doc_tokens_used trimmed_history = [] history_tokens_used = 0 # 从最近的对话倒序提取 for msg in reversed(history): msg_tok = self.estimate_token_count(msg.get("content", "")) if history_tokens_used + msg_tok <= remaining_budget: trimmed_history.insert(0, msg) history_tokens_used += msg_tok else: break final_prompt_struct = { "system": self.system_prompt, "prefix_hash": self.system_prompt_hash, "rag_context": selected_docs, "history": trimmed_history, "query": user_query, "estimated_total_tokens": sys_tokens + query_tokens + doc_tokens_used + history_tokens_used } return final_prompt_struct

结合 Prefix Caching 机制重构流式响应流程

在多轮对话以及大模型 Agent 场景中，Prefill（首字填充）阶段往往占用了大半的推理耗时。通过规范化 Prompt 格式，使所有的 Agent 指令与系统设定排在请求的最头部，能够最大限度提升 Prefix Caching 的命中率。

构建流式 Response 处理机制时，需要实时监测首字延迟（TTFT）与每 Token 生成速率（TPOT），以便在后端推理出现挂起或拥堵时及时干预。

import json class LLMCostLatencyMonitor: def __init__(self, input_cost_per_k: float = 0.0015, output_cost_per_k: float = 0.002): self.input_cost_per_k = input_cost_per_k self.output_cost_per_k = output_cost_per_k def process_stream_response(self, response_stream, prompt_tokens: int, cache_hit: bool = False): start_time = time.time() ttft = None output_tokens = 0 chunks = [] # 前缀缓存命中时，Prompt Token 费用打五折 effective_input_cost = self.input_cost_per_k * 0.5 if cache_hit else self.input_cost_per_k for chunk in response_stream: current_time = time.time() if ttft is None: ttft = current_time - start_time output_tokens += 1 chunks.append(chunk) # 实时检测 TPOT 异常 elapsed = current_time - start_time if elapsed > 15.0: # 强制超时防护 break total_latency = time.time() - start_time input_cost = (prompt_tokens / 1000.0) * effective_input_cost output_cost = (output_tokens / 1000.0) * self.output_cost_per_k total_cost = input_cost + output_cost return { "full_text": "".join(chunks), "ttft_seconds": ttft if ttft else 0.0, "total_latency_seconds": total_latency, "prompt_tokens": prompt_tokens, "output_tokens": output_tokens, "cache_hit": cache_hit, "total_cost_usd": round(total_cost, 6) }

成本与延迟的双维度实时监控与兜底熔断器

仅在事后查看账单无法阻止线上突发的费用暴涨。必须在网关层部署熔断控制器，根据实时计算的单次请求预估费用与全局 QPS 进行动态限流。

当某类 Batch 任务的 TTFT 持续越过服务预算时，可将后续请求调度到已验证的小参数模型或本地蒸馏模型。触发条件要包含样本窗口和恢复门槛，避免一次抖动引发频繁切换。

针对大模型 API 调用的监控指标，需要重点关注三个核心物理量：P99 TTFT、单位时间 Token 消耗速率（Tokens/sec）以及 Prefix Cache 命中率。这三个指标构成了模型线上运营的三角约束。

class ModelCircuitBreaker: def __init__(self, max_cost_per_min: float = 5.0, max_allowed_ttft: float = 3.5): self.max_cost_per_min = max_cost_per_min self.max_allowed_ttft = max_allowed_ttft self.current_minute_cost = 0.0 self.last_reset_time = time.time() self.is_degraded = False def check_and_update(self, request_cost: float, last_ttft: float) -> str: now = time.time() if now - self.last_reset_time > 60: self.current_minute_cost = 0.0 self.last_reset_time = now self.is_degraded = False self.current_minute_cost += request_cost if self.current_minute_cost > self.max_cost_per_min: self.is_degraded = True return "DEGRADE_REASON_COST_EXCEEDED" if last_ttft > self.max_allowed_ttft: self.is_degraded = True return "DEGRADE_REASON_LATENCY_TOO_HIGH" return "NORMAL"