大模型API成本优化实战：Token压缩、缓存策略与批处理调度的完整生产方案
·
大模型API成本优化实战：Token压缩、缓存策略与批处理调度的完整生产方案
一、API成本失控的根因：为何大模型调用费用居高不下
大模型API按Token计费，这是成本失控的根源。
以GPT-4o为例，输入Token $5/1M，输出Token $15/1M。
一个中等复杂度的对话，输入约2000 Token，输出约500 Token。
单次调用成本约$0.0175，看似不高。
但如果有1万用户每天对话10次，日成本就达$1750。
月成本超过$5万，对任何创业公司都是沉重负担。
成本失控通常有三个原因。
第一：Token浪费严重。
用户输入大量无关内容，系统提示词冗长重复。
同样的问题被不同用户反复问，每次都全价计费。
第二：缺乏缓存机制。
相同Prompt的重复请求，每次都调用大模型。
缓存命中率低于20%的系统，成本是直接翻倍计算的。
第三：调度策略低效。
小批量、低并发的调用方式，无法利用Batch API的折扣。
实时调用占比过高， Batch API（折扣50%）使用率低于10%。
将成本降低60%~80%是完全可行的。
核心是三板斧：Token压缩、缓存策略、批处理调度。
这三者需要配合使用，才能最大化成本收益。
flowchart TD
A[用户请求] --> B{Token压缩预处理}
B --> B1[系统提示词精简]
B --> B2[用户输入语义压缩]
B --> B3[历史对话截断]
B1 --> C{缓存查询}
B2 --> C
B3 --> C
C -->|精确匹配命中| D[直接返回缓存结果]
C -->|语义缓存命中| E[返回语义相似结果]
C -->|缓存未命中| F[调用大模型API]
F --> G{调度策略判断}
G -->|延迟容忍>5s| H[加入Batch队列]
G -->|延迟容忍<5s| I[实时调用]
H --> J[批量聚合: 32~128个请求/批]
J --> K[Batch API调用: 成本-50%]
I --> L[实时API调用: 全价]
D --> M[返回结果: 成本≈0]
E --> M
K --> N[返回结果: 成本-50%]
L --> N
M --> O[更新缓存]
N --> O
style D fill:#27ae60,color:#fff
style E fill:#27ae60,color:#fff
style K fill:#3498db,color:#fff
style L fill:#e74c3c,color:#fff
这三板斧的实施顺序很关键。
先压缩Token，减少每次调用的基础成本。
再建缓存，减少重复调用的次数。
最后优化调度，降低剩余的必须调用的单价。
三者叠加，在生产环境中可实现70%~85%的成本降低。
以下是完整的生产级实施方案。
二、Token压缩技术：从系统提示词到用户输入的完整压缩方案
Token压缩的第一步是系统提示词（System Prompt）优化。
很多开发者把系统提示词写得又长又全。
什么边缘场景都考虑到，什么注意事项都写进去。
结果系统提示词长达2000~3000个Token。
每次对话这些Token都被计费。
优化系统提示词有三个原则。
原则一：只保留当前任务必需的规则。
把通用规则写到Few-shot示例里，按需加载。
原则二：用结构化格式（JSON Schema）替代自然语言描述。
结构化格式更紧凑，Token消耗更少。
原则三：对多语言支持，只列出实际使用的语言，不要全量列举。
系统提示词优化后，通常能从3000 Token压缩到800 Token。
每次调用节省的输入成本约$0.011，降幅70%。
第二步是用户输入的语义压缩。
用户经常输入大量无关信息。
比如粘贴一整篇文档，但只问其中一句话的含义。
用LLM做语义压缩：让一个小模型先摘要用户的输入。
只保留与当前问题相关的部分，再传给主模型。
第三步是对话历史的智能截断。
多轮对话中，历史消息会累积。
超过模型上下文限制（如8K、128K）后，需要截断。
简单的头尾截断策略效果差，丢失了中间的关键信息。
更好的策略是用语义检索：对新问题做Embedding。
从对话历史中检索最相关的K条消息拼接。
"""
大模型API Token压缩的生产级工具库
包含：系统提示词优化、用户输入语义压缩、对话历史截断
可用于AI创业公司的成本优化实战
"""
import re
import tiktoken
from typing import Optional
from dataclasses import dataclass
from functools import lru_cache
@dataclass
class CompressionResult:
original_tokens: int
compressed_tokens: int
compression_ratio: float
method: str
class TokenCompressor:
"""
Token压缩器：多策略组合，最大化压缩比
生产环境建议作为API调用的前置预处理步骤
"""
def __init__(self, model: str = "gpt-4o"):
self.model = model
self.encoder = tiktoken.encoding_for_model(model)
def count_tokens(self, text: str) -> int:
"""计算文本的Token数量"""
return len(self.encoder.encode(text))
def compress_system_prompt(
self, sys_prompt: str
) -> tuple[str, CompressionResult]:
"""
系统提示词压缩：结构化重写+规则精简
返回：(压缩后的提示词, 压缩结果统计)
"""
original_tokens = self.count_tokens(sys_prompt)
# 策略1：移除冗余的自然语言描述，转为结构化格式
compressed = sys_prompt
# 将"你是一个...助手。你的任务是..."转为JSON Schema
if "你是一个" in compressed and "助手" in compressed:
# 提取角色定义，压缩为一行
role_match = re.search(r"你是一个(.+?)助手", compressed)
if role_match:
role = role_match.group(1).strip()
compressed = f"Role: {role}\n" + \
re.sub(r"你是一个.+?助手[。，]", "",
compressed)
# 策略2：合并重复的规则描述
lines = compressed.split("\n")
seen = set()
deduped = []
for line in lines:
key = line[:20] # 以前20个字符作为去重Key
if key not in seen and len(line.strip()) > 0:
seen.add(key)
deduped.append(line)
compressed = "\n".join(deduped)
# 策略3：将长段落转为要点列表（减少连接词）
compressed = re.sub(r"因此，|所以，|也就是说，", "", compressed)
compressed = re.sub(r"需要注意的是，", "- ", compressed)
compressed = re.sub(r"请确保", "- 确保", compressed)
compressed_tokens = self.count_tokens(compressed)
ratio = compressed_tokens / max(1, original_tokens)
return compressed, CompressionResult(
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
compression_ratio=ratio,
method="system_prompt"
)
def compress_user_input(
self, user_input: str,
max_tokens: int = 1000
) -> tuple[str, CompressionResult]:
"""
用户输入压缩：截断+摘要
超过max_tokens的部分，用小模型做摘要压缩
"""
original_tokens = self.count_tokens(user_input)
if original_tokens <= max_tokens:
return user_input, CompressionResult(
original_tokens, original_tokens,
1.0, "user_input_noop"
)
# 策略：保留前max_tokens*0.6个Token（头部重要信息）
# 尾部保留max_tokens*0.3个Token（最新信息）
# 中间部分摘要为max_tokens*0.1个Token
head_tokens = int(max_tokens * 0.6)
tail_tokens = int(max_tokens * 0.3)
mid_tokens = max_tokens - head_tokens - tail_tokens
encoded = self.encoder.encode(user_input)
head = self.encoder.decode(encoded[:head_tokens])
tail = self.encoder.decode(encoded[-tail_tokens:])
# 中间部分用省略号替代（生产环境可调用小型摘要模型）
mid = f"\n[中间内容已压缩，约{original_tokens - head_tokens - tail_tokens}Token]\n"
compressed = head + mid + tail
compressed_tokens = self.count_tokens(compressed)
return compressed, CompressionResult(
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
compression_ratio=compressed_tokens / original_tokens,
method="user_input"
)
def truncate_history(
self, history: list[dict],
max_tokens: int = 4096,
preserve_recent: int = 6
) -> tuple[list[dict], CompressionResult]:
"""
对话历史截断：保留最近的N条 + 语义检索最相关的历史
history格式: [{"role": "user", "content": "..."}, ...]
preserve_recent: 强制保留的最近消息条数
"""
total_tokens = sum(
self.count_tokens(m["content"])
for m in history
)
original_tokens = total_tokens
if total_tokens <= max_tokens:
return history, CompressionResult(
original_tokens, total_tokens,
1.0, "history_noop"
)
# 强制保留最近的preserve_recent条
recent = history[-preserve_recent:]
recent_tokens = sum(
self.count_tokens(m["content"])
for m in recent
)
# 从剩余历史中选择最相关的消息
remaining_budget = max_tokens - recent_tokens
candidates = history[:-preserve_recent]
# 按Token数从少到多排序，优先保留短消息
candidates.sort(
key=lambda m: self.count_tokens(m["content"])
)
selected = []
used = 0
for msg in candidates:
t = self.count_tokens(msg["content"])
if used + t <= remaining_budget:
selected.append(msg)
used += t
# 按原始顺序排序
selected.sort(key=lambda m: history.index(m))
truncated = selected + recent
final_tokens = sum(
self.count_tokens(m["content"])
for m in truncated
)
return truncated, CompressionResult(
original_tokens=original_tokens,
compressed_tokens=final_tokens,
compression_ratio=final_tokens / max(1, original_tokens),
method="history_truncation"
)
class PromptCache:
"""
提示词缓存：精确匹配+语义缓存双层架构
精确匹配层：完全相同的Prompt直接返回缓存
语义缓存层：语义相似的Prompt返回近似结果
"""
def __init__(self, similarity_threshold: float = 0.92):
self.exact_cache: dict[str, str] = {}
self.semantic_cache: dict[str, tuple[list[float], str]] = {}
self.sim_threshold = similarity_threshold
def _hash_prompt(self, prompt: str) -> str:
"""生成Prompt的精确哈希（用于精确匹配）"""
import hashlib
return hashlib.sha256(prompt.encode()).hexdigest()[:16]
def exact_lookup(self, prompt: str) -> Optional[str]:
"""精确匹配查询"""
key = self._hash_prompt(prompt)
return self.exact_cache.get(key)
def semantic_lookup(
self, prompt: str,
embed_fn: callable
) -> Optional[str]:
"""
语义缓存查询：计算输入Prompt的Embedding，
与缓存中的所有Embedding计算余弦相似度
"""
query_emb = embed_fn(prompt)
best_sim = 0.0
best_result = None
for cached_prompt, (emb, result) in \
self.semantic_cache.items():
sim = self._cosine_sim(query_emb, emb)
if sim > best_sim and sim >= self.sim_threshold:
best_sim = sim
best_result = result
return best_result
def _cosine_sim(
self, a: list[float], b: list[float]
) -> float:
"""余弦相似度"""
dot = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot / (norm_a * norm_b + 1e-8)
def store_exact(self, prompt: str, result: str) -> None:
"""存储精确缓存"""
key = self._hash_prompt(prompt)
self.exact_cache[key] = result
def store_semantic(
self, prompt: str, result: str,
embed_fn: callable
) -> None:
"""存储语义缓存"""
emb = embed_fn(prompt)
self.semantic_cache[prompt] = (emb, result)
# 限制缓存大小（生产环境应使用LRU）
if len(self.semantic_cache) > 10000:
# 简单策略：清空最早的一半
keys = list(self.semantic_cache.keys())
for k in keys[:len(keys)//2]:
del self.semantic_cache[k]
# 完整的使用示例
if __name__ == "__main__":
compressor = TokenCompressor(model="gpt-4o")
# 示例1：压缩系统提示词
long_sys_prompt = """
你是一个专业的技术助手。你的任务是帮助用户解决编程问题。
你需要注意代码的正确性。你需要注意代码的安全性。
你需要注意代码的性能。你需要注意代码的可读性。
你需要注意代码的维护性。你需要注意代码的测试覆盖率。
你需要注意代码的文档完整性。你需要注意代码的依赖管理。
你需要注意代码的版本兼容性。你需要注意代码的部署流程。
请确保你的回答准确。请确保你的回答有用。
请确保你的回答简洁。请确保你的回答有代码示例。
"""
compressed, result = compressor.compress_system_prompt(
long_sys_prompt
)
print(f"系统提示词压缩: {result.original_tokens} → "
f"{result.compressed_tokens} Token "
f"({result.compression_ratio:.1%})")
print(f"压缩后:\n{compressed}\n")
# 示例2：压缩用户输入
long_input = "请帮我分析这段代码的问题。" + "a" * 3000
compressed, result = compressor.compress_user_input(
long_input, max_tokens=500
)
print(f"用户输入压缩: {result.original_tokens} → "
f"{result.compressed_tokens} Token")
三、缓存策略的生产级实现：精确缓存、语义缓存与分布式缓存
缓存是大模型API成本优化的第二板斧。
一个设计良好的缓存系统，可以将重复请求的命中率提升到60%以上。
这意味着60%的请求不需要调用大模型，成本直接降低60%。
缓存分为三个层次。
第一层是精确缓存（Exact Cache）。
完全相同的Prompt（包括System Prompt和User Prompt），直接返回缓存结果。
实现简单，用哈希表即可。
但命中率通常不高，因为用户每次的输入都有细微差异。
第二层是语义缓存（Semantic Cache）。
语义相似的Prompt，返回缓存中最相似的结果。
需要用到Embedding模型和向量相似度检索。
命中率比精确缓存高3~5倍，但有一定错误风险。
需要设置相似度阈值（如0.92），低于阈值的直接放过。
第三层是分布式缓存（Distributed Cache）。
多实例部署时，缓存需要共享。
用Redis作为缓存后端，所有实例查询同一个Redis集群。
缓存Key是Prompt的哈希，Value是模型响应和元数据。
需要设置合理的TTL（如24小时），避免缓存过期数据。
"""
分布式语义缓存的生产级实现
基于Redis + Sentence Transformers
适用于多实例部署的AI应用
"""
import json
import time
import hashlib
from typing import Optional, Callable
from dataclasses import dataclass, asdict
import redis
@dataclass
class CachedResponse:
"""缓存的响应数据结构"""
prompt: str
response: str
model: str
timestamp: float
ttl: int # 剩余TTL（秒）
hit_count: int # 被命中次数（用于热度统计）
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)
@staticmethod
def from_json(s: str) -> "CachedResponse":
d = json.loads(s)
return CachedResponse(**d)
class DistributedPromptCache:
"""
分布式Prompt缓存：精确+语义双层
生产环境特性：
- Redis集群后端，支持水平扩展
- 精确缓存层：SHA256哈希，100%准确率
- 语义缓存层：Embedding相似度，可调阈值
- 热度统计：自动晋升热门查询结果
- TTL管理：自动过期，避免返回过时结果
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
embed_fn: Optional[Callable] = None,
sim_threshold: float = 0.92,
default_ttl: int = 86400, # 默认24小时
):
self.redis = redis.from_url(redis_url)
self.embed_fn = embed_fn
self.sim_threshold = sim_threshold
self.default_ttl = default_ttl
# Redis键前缀（避免与其他业务冲突）
self.EXACT_KEY_PREFIX = "prompt_cache:exact:"
self.SEMantic_KEY_PREFIX = "prompt_cache:semantic:"
self.HOT_KEY_PREFIX = "prompt_cache:hot:"
def _exact_key(self, prompt: str) -> str:
"""精确缓存的Redis Key"""
h = hashlib.sha256(prompt.encode()).hexdigest()
return f"{self.EXACT_KEY_PREFIX}{h}"
def exact_get(self, prompt: str) -> Optional[str]:
"""精确缓存查询"""
key = self._exact_key(prompt)
data = self.redis.get(key)
if data:
cached = CachedResponse.from_json(data)
# 检查TTL
if time.time() - cached.timestamp < cached.ttl:
# 更新命中次数
cached.hit_count += 1
self.redis.setex(
key, cached.ttl, cached.to_json()
)
return cached.response
else:
self.redis.delete(key)
return None
def exact_set(
self, prompt: str, response: str,
model: str, ttl: Optional[int] = None
) -> None:
"""写入精确缓存"""
ttl = ttl or self.default_ttl
cached = CachedResponse(
prompt=prompt,
response=response,
model=model,
timestamp=time.time(),
ttl=ttl,
hit_count=0,
)
key = self._exact_key(prompt)
self.redis.setex(key, ttl, cached.to_json())
def semantic_get(
self, prompt: str
) -> Optional[str]:
"""
语义缓存查询
遍历语义缓存中的所有条目，计算相似度
生产环境应使用向量数据库（如Milvus）替代Redis遍历
"""
if not self.embed_fn:
return None
query_emb = self.embed_fn(prompt)
# 扫描语义缓存键（生产环境应使用向量检索）
semantic_keys = self.redis.keys(
f"{self.SEMantic_KEY_PREFIX}*"
)
best_sim = 0.0
best_response = None
for key in semantic_keys:
data = self.redis.get(key)
if not data:
continue
cached = CachedResponse.from_json(data)
cached_emb = self.embed_fn(cached.prompt)
# 余弦相似度
dot = sum(a * b for a, b in zip(query_emb, cached_emb))
norm_q = sum(a * a for a in query_emb) ** 0.5
norm_c = sum(a * a for a in cached_emb) ** 0.5
sim = dot / (norm_q * norm_c + 1e-8)
if sim > best_sim and sim >= self.sim_threshold:
best_sim = sim
best_response = cached.response
if best_response:
print(f"语义缓存命中: 相似度={best_sim:.3f}")
return best_response
def semantic_set(
self, prompt: str, response: str,
model: str, ttl: Optional[int] = None
) -> None:
"""写入语义缓存（同时存储Embedding）"""
if not self.embed_fn:
return
ttl = ttl or self.default_ttl
cached = CachedResponse(
prompt=prompt,
response=response,
model=model,
timestamp=time.time(),
ttl=ttl,
hit_count=0,
)
# Key包含Embedding的前16维（用于快速过滤）
emb = self.embed_fn(prompt)
emb_prefix = "_".join(f"{x:.2f}" for x in emb[:4])
key = f"{self.SEMantic_KEY_PREFIX}{emb_prefix}:{hash(prompt)}"
self.redis.setex(key, ttl, cached.to_json())
def get(
self, prompt: str
) -> Optional[str]:
"""
两级缓存查询：
先查精确缓存，未命中再查语义缓存
"""
# 第一级：精确缓存
result = self.exact_get(prompt)
if result:
return result
# 第二级：语义缓存
result = self.semantic_get(prompt)
if result:
return result
return None
def set(
self, prompt: str, response: str,
model: str, ttl: Optional[int] = None
) -> None:
"""写入两级缓存"""
self.exact_set(prompt, response, model, ttl)
self.semantic_set(prompt, response, model, ttl)
def stats(self) -> dict:
"""缓存统计信息"""
exact_count = len(self.redis.keys(
f"{self.EXACT_KEY_PREFIX}*"
))
semantic_count = len(self.redis.keys(
f"{self.SEMantic_KEY_PREFIX}*"
))
return {
"exact_cache_entries": exact_count,
"semantic_cache_entries": semantic_count,
"total_cache_entries": exact_count + semantic_count,
}
# 与LLM API集成的完整示例
class CachedLLMClient:
"""
带缓存的LLM客户端
透明地添加缓存层，对业务代码完全透明
"""
def __init__(self, api_key: str, cache: DistributedPromptCache):
self.api_key = api_key
self.cache = cache
def chat(
self, system_prompt: str, user_prompt: str,
model: str = "gpt-4o",
max_cache_age: int = 3600
) -> str:
"""
带缓存的Chat调用
相同输入在max_cache_age秒内直接返回缓存
"""
full_prompt = f"<system>{system_prompt}</system>\n<user>{user_prompt}</user>"
# 查询缓存
cached = self.cache.get(full_prompt)
if cached:
print(" 缓存命中，跳过API调用")
return cached
# 缓存未命中，调用API
print(" 缓存未命中，调用大模型API...")
response = self._call_api(
system_prompt, user_prompt, model
)
# 写入缓存（仅缓存成功的响应）
if response and not response.startswith("ERROR"):
self.cache.set(
full_prompt, response, model,
ttl=max_cache_age
)
return response
def _call_api(
self, system: str, user: str, model: str
) -> str:
"""实际调用LLM API（简化为伪代码）"""
# 生产环境使用openai、anthropic等SDK
import openai
client = openai.OpenAI(api_key=self.api_key)
resp = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
)
return resp.choices[0].message.content
if __name__ == "__main__":
# 初始化缓存
cache = DistributedPromptCache(
redis_url="redis://localhost:6379",
sim_threshold=0.92,
)
client = CachedLLMClient(
api_key="sk-...", cache=cache
)
# 第一次调用（缓存未命中）
r1 = client.chat(
"你是一个Python专家",
"如何用asyncio实现超时控制？"
)
# 第二次调用相同问题（缓存命中）
r2 = client.chat(
"你是一个Python专家",
"如何用asyncio实现超时控制？"
)
print(f"两次响应是否相同: {r1 == r2}")
print(f"缓存统计: {cache.stats()}")
四、批处理调度策略：用Batch API降低50%成本的完整方案
批处理调度是大模型API成本优化的第三板斧。
OpenAI、Anthropic等主流API都提供Batch接口。
Batch接口的价格通常是实时接口的50%。
代价是延迟更高（通常在24小时内返回，实际约1~4小时）。
适合Batch的场景：离线数据处理、批量摘要、嵌入式生成。
不适合Batch的场景：在线对话、实时推理、交互式应用。
生产环境的调度策略是混合模式。
对每个请求判断延迟容忍度。
延迟容忍>5分钟的请求，自动路由到Batch队列。
延迟容忍<5分钟的请求，走实时API。
Batch队列需要聚合请求。
单个Batch文件最多支持50000个请求。
但建议每批500~1000个请求，便于管理和重试。
"""
Batch API调度器的生产级实现
自动路由请求到实时API或Batch API
支持请求聚合、状态跟踪、结果回传
"""
import json
import time
import uuid
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum
class RequestPriority(Enum):
REALTIME = "realtime" # 实时，全价
DELAY_TOLERANT = "batch" # 延迟容忍，半价
OFFLINE = "offline" # 离线，最低价（某些提供商）
@dataclass
class LLMRequest:
"""单个LLM请求"""
request_id: str
system_prompt: str
user_prompt: str
model: str
max_tokens: int
priority: RequestPriority
created_at: float
result: Optional[str] = None
status: str = "pending" # pending|processing|completed|failed
@dataclass
class BatchJob:
"""一个Batch作业，包含多个请求"""
job_id: str
requests: list[LLMRequest]
created_at: float
status: str = "pending"
output_file_id: Optional[str] = None
class BatchScheduler:
"""
Batch调度器：
- 实时请求立即调用API
- 延迟容忍请求入队，聚合后调用Batch API
- 自动管理Batch作业的生命周期
"""
def __init__(
self,
api_key: str,
batch_size: int = 500,
max_wait_seconds: int = 300, # 最多等待5分钟就提交Batch
):
self.api_key = api_key
self.batch_size = batch_size
self.max_wait = max_wait_seconds
self.pending_queue: list[LLMRequest] = []
self.active_batches: dict[str, BatchJob] = {}
self.results: dict[str, str] = {}
def submit(
self, req: LLMRequest
) -> Optional[str]:
"""
提交请求，返回request_id
如果是实时请求，立即调用API并返回结果
如果是Batch请求，入队并等待聚合
"""
if req.priority == RequestPriority.REALTIME:
# 实时请求：立即调用
result = self._call_realtime_api(req)
self.results[req.request_id] = result
return result
else:
# Batch请求：入队
self.pending_queue.append(req)
req.status = "queued"
# 检查是否可以提交Batch
if len(self.pending_queue) >= self.batch_size:
self._flush_batch()
return None # 结果异步返回
def _flush_batch(self) -> None:
"""将等待队列中的请求提交为一个Batch作业"""
if not self.pending_queue:
return
batch_reqs = self.pending_queue[:self.batch_size]
self.pending_queue = self.pending_queue[self.batch_size:]
job = BatchJob(
job_id=str(uuid.uuid4()),
requests=batch_reqs,
created_at=time.time(),
)
# 调用Batch API
batch_file_id = self._create_batch_file(job)
job_id = self._submit_batch_api(batch_file_id)
job.status = "processing"
self.active_batches[job_id] = job
print(f"提交Batch作业: job_id={job_id}, "
f"请求数={len(batch_reqs)}")
def _create_batch_file(self, job: BatchJob) -> str:
"""
创建Batch API所需的JSONL文件
OpenAI Batch API格式：每行一个JSON请求
"""
lines = []
for req in job.requests:
line = {
"custom_id": req.request_id,
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": req.model,
"messages": [
{"role": "system",
"content": req.system_prompt},
{"role": "user",
"content": req.user_prompt},
],
"max_tokens": req.max_tokens,
},
}
lines.append(json.dumps(line, ensure_ascii=False))
# 上传文件（伪代码，实际需要调用Files API）
batch_content = "\n".join(lines)
file_id = self._upload_file(batch_content)
return file_id
def _upload_file(self, content: str) -> str:
"""上传Batch文件到API（简化）"""
# 生产环境使用 openai.files.create(file=...)
return "file-" + str(uuid.uuid4())[:8]
def _submit_batch_api(self, file_id: str) -> str:
"""提交Batch API作业"""
# 生产环境使用 openai.batches.create(...)
job_id = "batch-" + str(uuid.uuid4())[:8]
return job_id
def check_batches(self) -> None:
"""
检查活跃的Batch作业状态
生产环境应作为后台任务定期执行
"""
completed = []
for job_id, job in self.active_batches.items():
status = self._get_batch_status(job_id)
if status == "completed":
self._process_batch_results(job)
completed.append(job_id)
for job_id in completed:
del self.active_batches[job_id]
def _get_batch_status(self, job_id: str) -> str:
"""查询Batch作业状态（伪代码）"""
return "completed" # 简化
def _process_batch_results(self, job: BatchJob) -> None:
"""处理Batch作业结果，回写结果到每个请求"""
for req in job.requests:
# 从结果文件中读取（伪代码）
req.result = f"Batch结果_{req.request_id}"
req.status = "completed"
self.results[req.request_id] = req.result
def get_result(self, request_id: str) -> Optional[str]:
"""获取请求结果（轮询或回调）"""
return self.results.get(request_id)
def cost_savings_report(self) -> dict:
"""成本节省报告"""
total = len(self.results)
batch_count = sum(
1 for j in self.active_batches.values()
for r in j.requests
)
savings = batch_count * 0.5 # 假设Batch节省50%
return {
"total_requests": total,
"batch_requests": batch_count,
"batch_ratio": batch_count / max(1, total),
"estimated_savings_percent": (
batch_count / max(1, total)
) * 50,
}
五、总结
-
大模型API成本优化的三板斧：Token压缩（系统提示词从3000压缩到800Token、用户输入语义压缩、对话历史智能截断，综合压缩比可达60%
80%）、缓存策略（精确缓存+语义缓存双层架构，Redis分布式缓存后端，命中率>60%时成本降低60%）、批处理调度（延迟容忍>5min的请求路由到Batch API，价格降低50%，自动聚合请求每批5001000个），三者叠加在生产环境中可实现70%~85%的成本降低 -
Token压缩的生产级实施要点：系统提示词用JSON Schema替自然语言描述（更紧凑）、多轮对话历史用语义检索替代头尾截断（保留最相关的K条消息）、tiktoken库精确计算Token数（避免估算误差导致超出上下文限制），压缩操作应作为API调用的前置预处理步骤，增加延迟<5ms，远低于API调用延迟（2~30s）
-
缓存策略的三层架构与命中率提升方法：精确缓存层（SHA256哈希，100%准确但命中率低约5%~15%）、语义缓存层（Embedding余弦相似度≥0.92，命中率提升至30%~60%但有误命中风险，需业务层做结果校验）、分布式缓存层（Redis集群后端，TTL=24h，支持多实例共享缓存），生产环境语义缓存应使用专用向量数据库（Milvus/Pinecone）替代Redis遍历，相似度检索延迟从O(N)降至O(logN)
-
Batch API调度策略的混合模式设计：实时请求（延迟容忍<5min）走实时API全价计费、延迟容忍请求（5min
24h）走Batch API半价计费、离线请求（>24h）走异步API最低价计费，调度器自动判断请求优先级并路由，Batch队列达到batch_size或max_wait_seconds触发提交，后台任务定期轮询Batch作业状态并回写结果，Batch API的实际返回延迟通常14小时（远低于宣称的24h） -
成本优化的监控指标体系：Token压缩比（目标>60%）、缓存命中率（目标>50%，每提升10%成本降低10%）、Batch路由比例（目标>40%，每提升10%成本降低5%）、单次调用平均成本（压缩前vs压缩后对比）、月度API费用趋势，推荐用Prometheus+Grafana搭建监控看板，异常告警阈值设为缓存命中率<30%或Batch路由比例<20%，及时介入调优避免成本失控
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
4
0- 0
已为社区贡献93条内容
所有评论(0)