各位同仁、技术爱好者们，大家下午好！
今天，我们将深入探讨一个在大型语言模型（LLM）应用开发中日益关键且充满挑战的议题——提示缓存（Prompt Caching）的深度集成。尤其，我们将聚焦于如何针对不同的LLM厂商，例如OpenAI和Anthropic Claude，定制化我们的缓存命中策略，从而最大限度地提升效率、降低成本并优化用户体验。
在LLM技术飞速发展的今天，我们享受着AI带来的巨大便利，但同时也面临着API调用成本高昂、响应延迟不稳定以及API速率限制等实际问题。提示缓存正是解决这些痛点的一把利器。它并非简单的“存储键值对”，而是一项需要深思熟虑、精巧设计的系统工程，尤其当我们要应对不同厂商API的微妙差异时。
第一章：提示缓存（Prompt Caching）的核心概念与必要性
1.1 什么是提示缓存？
提示缓存，顾名思义，是存储LLM API请求的输入（即“提示”或“Prompt”）及其对应的输出（即LLM生成的响应）的一种机制。当系统再次收到与已缓存请求“足够相似”的请求时，它会直接返回先前存储的响应，而非再次调用LLM API。
1.2 为什么我们需要提示缓存？
提示缓存的价值体现在多个维度：
- 成本节约：LLM API调用通常按输入/输出的Token数量计费。缓存可以显著减少重复调用，从而降低运营成本。
- 响应延迟优化：从缓存中获取响应通常比通过网络调用远程LLM API快得多，尤其对于低延迟要求的应用至关重要。
- API速率限制缓解：减少API调用次数有助于我们更好地遵守厂商的速率限制，避免因频繁调用而被限流。
- 用户体验提升：更快的响应速度直接转化为更流畅、更愉悦的用户体验。
- 可扩展性增强：通过减少对外部API的依赖，系统能够处理更大规模的并发请求。
1.3 提示缓存的基本工作原理
一个典型的提示缓存系统通常包含以下步骤：
- 请求拦截：在实际调用LLM API之前，拦截所有出站请求。
- 缓存键生成：根据请求的各种参数（提示内容、模型名称、温度等），生成一个唯一的“缓存键”（Cache Key）。这是整个缓存机制的核心。
- 缓存查找：使用生成的缓存键在缓存存储中查找是否有对应的响应。
- 缓存命中：如果找到匹配项（缓存命中），则直接返回缓存的响应，并记录命中事件。
- 缓存未命中：如果未找到匹配项（缓存未命中），则继续调用实际的LLM API。
- 响应处理与存储：接收到LLM API的响应后，将其与之前生成的缓存键一同存储到缓存中，以备后用。
- 响应返回：将LLM API的响应返回给原始请求方。
下图是一个简化的流程图（此处不使用图片，但可以想象为一个流程图）。
[用户应用] --- 请求 ---> [缓存代理层]
|
|--- (生成缓存键) ---> [缓存存储]
| |
| |--- (查找)
| |
| <--- (命中? / 未命中?)
|
|--- 命中 ---> [返回缓存响应]
|
|--- 未命中 ---> [LLM API] --- 请求 ---> [LLM 服务商]
| |
<------------------- (响应)
|
|--- (存储响应到缓存)
|
<--- [返回LLM响应]第二章：缓存命中策略的复杂性与挑战
简单地对整个Prompt字符串进行哈希处理，在很多情况下是不足够的。LLM API请求的复杂性远超我们的想象，这给缓存键的设计带来了巨大挑战。
2.1 简单的哈希匹配的局限性
- 语义等价而非文本等价：两个在字面上略有不同但语义完全相同的提示，可能需要得到相同的响应。例如，“你好世界”和“Hello World”在某些场景下可能是等价的。简单的字符串哈希无法识别这种语义等价。
- 参数差异：除了Prompt本身，模型名称、温度（temperature）、top_p、stop tokens、系统消息、Function/Tool定义等参数都会影响LLM的响应。即使Prompt完全相同，这些参数的微小差异也可能导致截然不同的输出。
- 模型与版本差异：不同的模型（例如GPT-3.5 vs GPT-4）或同一模型的不同版本（例如
gpt-3.5-turbo-0613vsgpt-3.5-turbo-1106）即使收到相同的Prompt和参数，也可能产生不同的响应。 - 输出非确定性：LLM的生成过程本身具有一定的随机性（尤其当
temperature > 0时）。这意味着即使是完全相同的请求，也可能在每次调用时产生略有不同的响应。
2.2 LLM API请求的构成分析
为了设计鲁棒的缓存键，我们首先需要深入理解LLM API请求的构成。以OpenAI和Claude为例，虽然它们都遵循“消息”格式，但具体字段和语义有所不同。
通用请求参数类别：
- 核心提示内容 (Prompt Content):
messages数组：包含多个消息对象，每个对象通常有role(system, user, assistant, tool) 和content。- 特定于厂商的字段：例如Claude的
system消息可以独立于messages数组。
- 模型选择 (Model Selection):
model：指定使用的LLM模型，如gpt-4-turbo-preview,claude-3-opus-20240229。
- 生成参数 (Generation Parameters):
temperature：控制输出的随机性。top_p：控制核采样的多样性。max_tokens：限制最大输出长度。stop：指定停止生成的序列。seed(OpenAI)：用于确定性采样，固定随机种子。
- 功能调用/工具使用 (Function Calling / Tool Use):
tools/functions(legacy)：定义可供LLM调用的工具或函数。tool_choice/function_call：指定是否强制调用某个工具。
- 其他参数：
response_format(OpenAI)：指定输出格式（如JSON）。user(OpenAI)：用于监控和防止滥用。stream：是否以流式方式返回响应。
2.3 缓存键（Cache Key）的设计原则
一个理想的缓存键应该具备以下特性：
- 唯一性：对于任何可能导致不同响应的请求，都应生成不同的缓存键。
- 一致性：对于任何可能导致相同响应的请求，都应生成相同的缓存键。
- 简洁性：缓存键不应过长，以避免不必要的存储和计算开销。
- 稳定性：缓存键的生成逻辑应该稳定，不受外部环境或非关键因素的影响。
为了实现这些原则，我们需要对请求参数进行标准化和哈希处理。
第三章：针对OpenAI的定制化缓存策略
OpenAI的Chat Completions API是目前最常用的接口之一。其请求结构相对复杂，特别是messages数组和tools定义。
3.1 OpenAI API请求特性分析
一个典型的OpenAI Chat Completions API请求负载如下：
{
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "system", "content": "你是一个严谨的编程助手。"},
{"role": "user", "content": "请解释一下Prompt Caching。"},
{"role": "assistant", "content": "Prompt Caching是一种优化LLM应用的技术..."},
{"role": "user", "content": "那它的主要优势是什么？"}
],
"temperature": 0.7,
"top_p": 1,
"max_tokens": 150,
"stop": ["nn"],
"seed": 42,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
}
],
"tool_choice": "auto",
"response_format": {"type": "json_object"}
}关键参数分析与缓存策略：
model: 必须精确匹配。gpt-3.5-turbo和gpt-4的输出通常截然不同。即使是版本号差异（如gpt-3.5-turbo-0613vsgpt-3.5-turbo-1106），也可能导致行为差异，建议精确匹配。messages: 这是最复杂的组件。- 消息顺序：消息的顺序至关重要，改变顺序通常会改变LLM的行为。
- 角色 (role)：
system,user,assistant,tool。 - 内容 (content)：文本内容。
- 工具调用 (tool_calls)：
assistant角色可能包含tool_calls。 - 工具响应 (tool_message)：
tool角色包含tool_call_id和content。 - 标准化：
- 对
messages数组中的每个消息对象进行标准化处理，例如去除content中的多余空白字符。 - 对消息对象内部的键进行排序（如
role和content），确保{"content": "...", "role": "user"}和{"role": "user", "content": "..."}生成相同的哈希。 - 处理
tool_calls和tool_message的结构。
- 对
temperature/top_p:- 严格模式：精确匹配。这是最安全的策略，因为即使是微小的差异也可能影响输出的随机性。
- 宽松模式：可以考虑在一个可接受的阈值范围内进行匹配（例如，
temperature在0.6到0.8之间都视为0.7）。但这会增加缓存不确定性，需要谨慎使用。对于高度确定性的应用，应避免此策略。
max_tokens: 必须精确匹配。它直接限制了响应的长度。stop: 必须精确匹配。停止序列会影响生成内容的结束点。seed: 必须精确匹配。seed参数的引入使得输出可以具有确定性。如果设置了seed，则必须精确匹配才能保证缓存的有效性。tools/tool_choice:tools是一个工具定义数组，需要对其进行标准化和哈希。这通常涉及对工具对象内部的字段（如name,description,parameters）进行排序和哈希。tool_choice也需要精确匹配。
response_format: 必须精确匹配。JSON模式和文本模式的输出格式截然不同。
3.2 缓存键生成示例 (OpenAI)
为了构建一个稳定且唯一的缓存键，我们将使用一个哈希函数（如SHA256）对请求的关键参数进行处理。
import hashlib
import json
from typing import Dict, Any, List, Optional
def _standardize_message(message: Dict[str, Any]) -> Dict[str, Any]:
"""标准化OpenAI消息对象，确保键序一致且内容规范化。"""
standardized = {}
for key in sorted(message.keys()): # 确保键的顺序一致
value = message[key]
if key == "content" and isinstance(value, str):
standardized[key] = value.strip() # 移除内容首尾空白
elif key == "tool_calls" and isinstance(value, list):
# 对tool_calls列表中的每个tool_call对象进行标准化
standardized[key] = sorted(
[_standardize_tool_call(tc) for tc in value],
key=lambda x: json.dumps(x, sort_keys=True)
)
elif isinstance(value, dict):
standardized[key] = _standardize_message(value) # 递归处理嵌套字典
elif isinstance(value, list):
# 简单列表，如果元素是dict，则递归标准化并排序
standardized[key] = sorted(
[_standardize_message(item) if isinstance(item, dict) else item for item in value],
key=lambda x: json.dumps(x, sort_keys=True) if isinstance(x, dict) else str(x)
)
else:
standardized[key] = value
return standardized
def _standardize_tool_call(tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""标准化OpenAI tool_call对象。"""
standardized = {}
for key in sorted(tool_call.keys()):
value = tool_call[key]
if key == "function" and isinstance(value, dict):
# 对function对象内部的arguments进行json解析和重排序
func_obj = {}
for f_key in sorted(value.keys()):
f_val = value[f_key]
if f_key == "arguments" and isinstance(f_val, str):
try:
# 尝试解析arguments为JSON并标准化
args = json.loads(f_val)
func_obj[f_key] = json.dumps(args, sort_keys=True, ensure_ascii=False)
except json.JSONDecodeError:
func_obj[f_key] = f_val # 解析失败则保留原字符串
else:
func_obj[f_key] = f_val
standardized[key] = func_obj
else:
standardized[key] = value
return standardized
def _standardize_tools_definition(tools: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]:
"""标准化OpenAI工具定义，确保顺序一致。"""
if not tools:
return None
standardized_tools = []
for tool in tools:
standardized_tool = {}
for key in sorted(tool.keys()):
value = tool[key]
if key == "function" and isinstance(value, dict):
# 对function定义内部的parameters进行标准化
func_def = {}
for f_key in sorted(value.keys()):
f_val = value[f_key]
if f_key == "parameters" and isinstance(f_val, dict):
func_def[f_key] = json.dumps(f_val, sort_keys=True, ensure_ascii=False)
else:
func_def[f_key] = f_val
standardized_tool[key] = func_def
else:
standardized_tool[key] = value
standardized_tools.append(standardized_tool)
# 对整个工具列表进行排序，以确保列表顺序不影响哈希
return sorted(standardized_tools, key=lambda x: json.dumps(x, sort_keys=True, ensure_ascii=False))
def generate_openai_cache_key(request_payload: Dict[str, Any]) -> str:
"""
为OpenAI Chat Completions API请求生成一个唯一的缓存键。
"""
cache_components = {}
# 1. 模型 (Model) - 必须精确匹配
cache_components["model"] = request_payload.get("model")
# 2. 消息 (Messages) - 最复杂的部分，需要深度标准化
messages = request_payload.get("messages", [])
standardized_messages = [
_standardize_message(msg) for msg in messages
]
# messages数组的顺序很重要，无需额外排序整个列表，因为_standardize_message已经处理了内部排序
cache_components["messages"] = standardized_messages
# 3. 生成参数 (Generation Parameters)
# temperature, top_p, max_tokens, stop, seed, response_format
# 这些参数通常需要精确匹配
cache_components["temperature"] = request_payload.get("temperature", 1.0) # 默认值
cache_components["top_p"] = request_payload.get("top_p", 1.0) # 默认值
cache_components["max_tokens"] = request_payload.get("max_tokens")
cache_components["stop"] = sorted(request_payload.get("stop", [])) if request_payload.get("stop") else None # 停止词列表排序
cache_components["seed"] = request_payload.get("seed")
cache_components["response_format"] = request_payload.get("response_format")
# 4. 工具定义 (Tools)
tools = request_payload.get("tools")
cache_components["tools"] = _standardize_tools_definition(tools)
cache_components["tool_choice"] = request_payload.get("tool_choice")
# 5. 其他可忽略或次要参数 (如user, stream, logprobs等，通常不影响核心语义)
# 对于缓存键，我们通常选择忽略这些参数，除非业务场景有特殊要求。
# 例如，user字段用于区分用户，但不会改变LLM的生成内容。
# stream=True/False 影响的是传输方式，而不是最终内容。
# 将标准化后的组件字典转换为一个规范的JSON字符串
# sort_keys=True 确保字典键的顺序一致，保证哈希结果稳定
# ensure_ascii=False 允许非ASCII字符，避免编码问题
canonical_json_string = json.dumps(cache_components, sort_keys=True, ensure_ascii=False)
# 使用SHA256哈希生成最终缓存键
return hashlib.sha256(canonical_json_string.encode('utf-8')).hexdigest()
# 示例使用
if __name__ == "__main__":
req1 = {
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "system", "content": "你是一个严谨的编程助手。"},
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.7,
"seed": 42,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
}
]
}
req2 = { # 仅仅改变了content中的空白符，应该命中
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "system", "content": "你是一个严谨的编程助手。 "}, # 多了空白
{"role": "user", "content": " 请解释一下Prompt Caching。"} # 多了空白
],
"temperature": 0.7,
"seed": 42,
"tools": [
{
"type": "function",
"function": {
"description": "获取指定城市的天气信息", # 键顺序不同
"name": "get_weather",
"parameters": {
"required": ["city"],
"type": "object",
"properties": {
"city": {"description": "城市名称", "type": "string"}
}
}
}
}
]
}
req3 = { # 改变了temperature，不应该命中
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "system", "content": "你是一个严谨的编程助手。"},
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.8, # 改变了
"seed": 42,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
}
]
}
req4 = { # 改变了工具定义，不应该命中
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "system", "content": "你是一个严谨的编程助手。"},
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.7,
"seed": 42,
"tools": [
{
"type": "function",
"function": {
"name": "get_weather_forecast", # 改变了工具名称
"description": "获取指定城市的天气预报",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
}
]
}
req5 = { # 改变了messages顺序，不应该命中
"model": "gpt-4-turbo-preview",
"messages": [
{"role": "user", "content": "请解释一下Prompt Caching。"}, # 顺序改变
{"role": "system", "content": "你是一个严谨的编程助手。"}
],
"temperature": 0.7,
"seed": 42,
"tools": []
}
key1 = generate_openai_cache_key(req1)
key2 = generate_openai_cache_key(req2)
key3 = generate_openai_cache_key(req3)
key4 = generate_openai_cache_key(req4)
key5 = generate_openai_cache_key(req5)
print(f"Key for req1: {key1}")
print(f"Key for req2: {key2}")
print(f"Key for req3: {key3}")
print(f"Key for req4: {key4}")
print(f"Key for req5: {key5}")
print(f"req1 == req2: {key1 == key2}") # 应该为 True
print(f"req1 == req3: {key1 == key3}") # 应该为 False
print(f"req1 == req4: {key1 == key4}") # 应该为 False
print(f"req1 == req5: {key1 == key5}") # 应该为 False代码解释：
_standardize_message: 递归地标准化消息对象。它通过sorted(message.keys())确保字典键的顺序一致，并对content进行strip()操作以忽略首尾空白。tool_calls列表及其内部对象也会被标准化。_standardize_tool_call: 专门处理tool_calls内部的function.arguments，因为arguments本身是JSON字符串，需要解析并重新序列化以确保其内部键的顺序一致。_standardize_tools_definition: 负责标准化tools数组。它会对每个工具定义内部的function.parameters进行JSON序列化和排序，并对整个工具列表进行排序，以确保工具定义顺序不影响哈希。generate_openai_cache_key: 组合所有标准化后的关键参数。它明确列出并处理了model,messages,temperature,top_p,max_tokens,stop,seed,response_format,tools,tool_choice等参数。- 最终，通过
json.dumps(..., sort_keys=True, ensure_ascii=False)将标准化后的字典转换为一个规范的JSON字符串，再进行SHA256哈希。sort_keys=True是确保稳定哈希的关键。 ensure_ascii=False可以避免非ASCII字符被转义成uXXXX形式，这在某些情况下可能导致不同编码的字符串产生不同的哈希。
第四章：针对Claude的定制化缓存策略
Anthropic Claude API的请求结构与OpenAI有相似之处，但也有其独特之处，尤其是system消息的处理。
4.1 Claude API请求特性分析
一个典型的Anthropic Claude Messages API请求负载如下：
{
"model": "claude-3-opus-20240229",
"system": "你是一个严谨的编程助手。请用中文回答。",
"messages": [
{"role": "user", "content": "请解释一下Prompt Caching。"},
{"role": "assistant", "content": "Prompt Caching是一种优化LLM应用的技术..."},
{"role": "user", "content": "那它的主要优势是什么？"}
],
"temperature": 0.7,
"top_p": 1,
"max_tokens": 150,
"stop_sequences": ["nn"],
"tools": [
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"input_schema": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
]
}关键参数分析与缓存策略：
model: 必须精确匹配。与OpenAI相同。system: 这是Claude独有的顶级参数。它通常用于设置模型角色或提供全局指令。必须精确匹配，且需要进行内容标准化（如strip()）。messages: 与OpenAI类似，但具体内部结构略有不同。- 消息顺序：同样重要，需要保留。
- 角色 (role)：
user,assistant,tool. (注意Claudesystem是单独字段)。 - 内容 (content)：文本内容。
- 工具使用 (tool_use)：
assistant角色可能包含tool_use对象。 - 工具结果 (tool_result)：
tool角色包含tool_use_id和content。 - 标准化：与OpenAI类似，对每个消息对象进行标准化，包括键排序、内容去除空白等。
temperature/top_p: 策略与OpenAI相同，建议精确匹配。max_tokens: 必须精确匹配。stop_sequences: 必须精确匹配。注意字段名为stop_sequences。tools:- Claude的工具定义结构略有不同，使用
input_schema而非parameters。 - 需要对其进行标准化和哈希，包括对
input_schema内部的字段进行排序和哈希。
- Claude的工具定义结构略有不同，使用
4.2 缓存键生成示例 (Claude)
import hashlib
import json
from typing import Dict, Any, List, Optional
def _standardize_claude_message(message: Dict[str, Any]) -> Dict[str, Any]:
"""标准化Claude消息对象。"""
standardized = {}
for key in sorted(message.keys()):
value = message[key]
if key == "content" and isinstance(value, str):
standardized[key] = value.strip()
elif key == "tool_use" and isinstance(value, dict):
# 对tool_use对象内部的input进行json解析和重排序
tool_use_obj = {}
for tu_key in sorted(value.keys()):
tu_val = value[tu_key]
if tu_key == "input" and isinstance(tu_val, dict): # Claude input是dict
tool_use_obj[tu_key] = json.dumps(tu_val, sort_keys=True, ensure_ascii=False)
else:
tool_use_obj[tu_key] = tu_val
standardized[key] = tool_use_obj
elif isinstance(value, dict):
standardized[key] = _standardize_claude_message(value)
elif isinstance(value, list):
standardized[key] = sorted(
[_standardize_claude_message(item) if isinstance(item, dict) else item for item in value],
key=lambda x: json.dumps(x, sort_keys=True) if isinstance(x, dict) else str(x)
)
else:
standardized[key] = value
return standardized
def _standardize_claude_tools_definition(tools: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]:
"""标准化Claude工具定义。"""
if not tools:
return None
standardized_tools = []
for tool in tools:
standardized_tool = {}
for key in sorted(tool.keys()):
value = tool[key]
if key == "input_schema" and isinstance(value, dict):
# 对input_schema进行json序列化和排序
standardized_tool[key] = json.dumps(value, sort_keys=True, ensure_ascii=False)
else:
standardized_tool[key] = value
standardized_tools.append(standardized_tool)
return sorted(standardized_tools, key=lambda x: json.dumps(x, sort_keys=True, ensure_ascii=False))
def generate_claude_cache_key(request_payload: Dict[str, Any]) -> str:
"""
为Anthropic Claude Messages API请求生成一个唯一的缓存键。
"""
cache_components = {}
# 1. 模型 (Model)
cache_components["model"] = request_payload.get("model")
# 2. 系统消息 (System Prompt) - Claude独有
system_message = request_payload.get("system")
if system_message is not None:
cache_components["system"] = system_message.strip() # 标准化内容
# 3. 消息 (Messages)
messages = request_payload.get("messages", [])
standardized_messages = [
_standardize_claude_message(msg) for msg in messages
]
cache_components["messages"] = standardized_messages
# 4. 生成参数 (Generation Parameters)
cache_components["temperature"] = request_payload.get("temperature", 1.0)
cache_components["top_p"] = request_payload.get("top_p", 1.0)
cache_components["max_tokens"] = request_payload.get("max_tokens") # 注意max_tokens是必须参数
cache_components["stop_sequences"] = sorted(request_payload.get("stop_sequences", [])) if request_payload.get("stop_sequences") else None
# 5. 工具定义 (Tools)
tools = request_payload.get("tools")
cache_components["tools"] = _standardize_claude_tools_definition(tools)
canonical_json_string = json.dumps(cache_components, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(canonical_json_string.encode('utf-8')).hexdigest()
# 示例使用
if __name__ == "__main__":
claude_req1 = {
"model": "claude-3-opus-20240229",
"system": "你是一个严谨的编程助手。请用中文回答。",
"messages": [
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.7,
"max_tokens": 150,
"tools": [
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"input_schema": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
]
}
claude_req2 = { # 仅仅改变了content中的空白符，应该命中
"model": "claude-3-opus-20240229",
"system": "你是一个严谨的编程助手。请用中文回答。 ", # 多了空白
"messages": [
{"role": "user", "content": " 请解释一下Prompt Caching。"} # 多了空白
],
"temperature": 0.7,
"max_tokens": 150,
"tools": [
{
"description": "获取指定城市的天气信息", # 键顺序不同
"name": "get_weather",
"input_schema": {
"required": ["city"],
"type": "object",
"properties": {
"city": {"description": "城市名称", "type": "string"}
}
}
}
]
}
claude_req3 = { # 改变了system消息，不应该命中
"model": "claude-3-opus-20240229",
"system": "你是一个严谨的编程助手。请用英文回答。", # 改变
"messages": [
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.7,
"max_tokens": 150,
"tools": []
}
claude_req4 = { # 改变了max_tokens，不应该命中
"model": "claude-3-opus-20240229",
"system": "你是一个严谨的编程助手。请用中文回答。",
"messages": [
{"role": "user", "content": "请解释一下Prompt Caching。"}
],
"temperature": 0.7,
"max_tokens": 200, # 改变
"tools": []
}
key_c1 = generate_claude_cache_key(claude_req1)
key_c2 = generate_claude_cache_key(claude_req2)
key_c3 = generate_claude_cache_key(claude_req3)
key_c4 = generate_claude_cache_key(claude_req4)
print(f"Key for claude_req1: {key_c1}")
print(f"Key for claude_req2: {key_c2}")
print(f"Key for claude_req3: {key_c3}")
print(f"Key for claude_req4: {key_c4}")
print(f"claude_req1 == claude_req2: {key_c1 == key_c2}") # 应该为 True
print(f"claude_req1 == claude_req3: {key_c1 == key_c3}") # 应该为 False
print(f"claude_req1 == claude_req4: {key_c1 == key_c4}") # 应该为 False代码解释：
_standardize_claude_message: 类似OpenAI的消息标准化，但特别处理Claude的tool_use结构，其input字段直接是字典，需要json.dumps处理以确保内部排序。_standardize_claude_tools_definition: 处理Claude工具定义中的input_schema字段，同样通过json.dumps进行标准化。generate_claude_cache_key: 核心差异在于将system消息作为一个独立的组件加入缓存键，并对stop_sequences（而非stop）进行处理。其他参数处理逻辑与OpenAI类似。
第五章：深度集成与架构考量
仅仅有缓存键生成策略是不够的，我们还需要将其融入到整个系统架构中。
5.1 缓存存储选择
- 内存缓存 (In-memory Cache):
- 优点：速度极快，无需网络开销。
- 缺点：容量有限，进程重启数据丢失，不适合分布式系统。
- 适用场景：单体应用，短期、小规模缓存，如使用
functools.lru_cache。
- Redis：
- 优点：高性能，支持多种数据结构，分布式，可持久化。
- 缺点：需要额外部署和维护Redis服务。
- 适用场景：大多数需要高性能、分布式缓存的生产环境。
- 数据库 (Database):
- 优点：持久化，数据结构灵活（如JSONB），易于查询和管理。
- 缺点：相对Redis和内存缓存，读写延迟较高。
- 适用场景：对缓存数据一致性要求较高，或需要复杂查询和分析缓存数据时。例如，PostgreSQL的
jsonb字段非常适合存储LLM请求/响应。
5.2 缓存失效策略
- TTL (Time-To-Live)：设置缓存项的生存时间。过期后自动失效。
- 优点：简单易行，适用于数据时效性要求不高的场景。
- 缺点：无法保证数据在TTL内仍是最新。
- LRU (Least Recently Used) / LFU (Least Frequently Used)：根据使用频率或最近使用情况淘汰缓存。
- 优点：自动管理缓存大小，保留最“有用”的数据。
- 缺点：需要额外的管理开销。
- 主动失效 (Proactive Invalidation)：当源数据发生变化时，主动通知缓存系统失效相关缓存项。
- 优点：保证数据强一致性。
- 缺点：实现复杂，需要有机制来检测源数据变化并触发失效。
考虑到LLM的响应不随时间变化而变化（除非模型更新），TTL通常用于控制缓存的生命周期以适应模型更新或数据新鲜度要求。对于生产环境，我们通常会结合使用TTL和LRU/LFU。
5.3 幂等性与并发控制
当缓存未命中时，多个并发请求可能会尝试同时调用LLM API生成相同的响应。这不仅浪费资源，还可能导致重复计费或触及API限制。我们需要确保：
- 对于一个给定的缓存键，只有一个请求能够触发LLM API调用。
- 其他并发请求应等待该调用完成，然后从缓存中获取结果。
这可以通过分布式锁（例如Redis锁）来实现。
import threading
import time
from concurrent.futures import ThreadPoolExecutor
# 模拟一个分布式锁服务
_locks = {}
_lock_lock = threading.Lock()
class DistributedLock:
def __init__(self, key):
self.key = key
def __enter__(self):
with _lock_lock:
if self.key not in _locks:
_locks[self.key] = threading.Lock()
self.local_lock = _locks[self.key]
self.local_lock.acquire()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.local_lock.release()
with _lock_lock:
# 清理，防止_locks无限增长，但在实际分布式系统中由Redis等管理
# if self.key in _locks and not _locks[self.key].locked():
# del _locks[self.key]
pass
# 模拟LLM API调用
def mock_llm_api_call(prompt_key: str):
print(f"Calling LLM API for key: {prompt_key}...")
time.sleep(2) # 模拟网络延迟和计算时间
response = f"Response for {prompt_key} at {time.time()}"
print(f"LLM API finished for key: {prompt_key}")
return response
# 模拟缓存存储
_cache = {}
def get_or_set_cached_response(prompt_key: str):
# 1. 尝试从缓存获取
if prompt_key in _cache:
print(f"Cache HIT for key: {prompt_key}")
return _cache[prompt_key]
# 2. 缓存未命中，获取分布式锁
print(f"Cache MISS for key: {prompt_key}. Acquiring lock...")
with DistributedLock(prompt_key):
# 再次检查缓存，因为在等待锁的过程中，其他线程可能已经填充了缓存
if prompt_key in _cache:
print(f"Cache HIT (after lock) for key: {prompt_key}")
return _cache[prompt_key]
# 3. 仍未命中，调用LLM API
response = mock_llm_api_call(prompt_key)
# 4. 存储到缓存
_cache[prompt_key] = response
print(f"Cache SET for key: {prompt_key}")
return response
# 模拟并发请求
def simulate_request(key: str, request_id: int):
print(f"[Request {request_id}] Starting for key: {key}")
response = get_or_set_cached_response(key)
print(f"[Request {request_id}] Received response for key {key}: {response[:30]}...")
if __name__ == "__main__":
print("--- First set of requests for 'key_A' ---")
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(simulate_request, "key_A", i) for i in range(1, 6)]
for future in futures:
future.result() # 等待所有请求完成
print("n--- Second set of requests for 'key_A' (should all be cache hits) ---")
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(simulate_request, "key_A", i) for i in range(6, 11)]
for future in futures:
future.result()
print("n--- Requests for 'key_B' (new key) ---")
with ThreadPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(simulate_request, "key_B", i) for i in range(11, 14)]
for future in futures:
future.result()上述代码演示了通过DistributedLock模拟的幂等性处理。当多个请求同时首次请求同一个prompt_key时，只有一个请求会实际调用mock_llm_api_call，其他请求会等待锁释放后从缓存中获取结果。
5.4 可观察性 (Observability)
一个健壮的缓存系统需要良好的监控。关键指标包括：
- 缓存命中率 (Cache Hit Ratio)：
命中次数 / (命中次数 + 未命中次数)。这是衡量缓存效率最重要的指标。 - 缓存未命中率 (Cache Miss Ratio)：1 – 命中率。
- 平均缓存响应时间：从缓存获取响应的平均时间。
- 平均LLM API响应时间：实际调用LLM API的平均时间。
- 缓存存储大小/数量：缓存占用的空间和存储的条目数。
- 缓存驱逐率：因LRU/LFU或TTL过期而被移除的缓存项数量。
- 错误率：缓存系统自身的错误率。
这些指标可以帮助我们评估缓存策略的效果，并进行优化。
5.5 代理层集成
提示缓存通常以代理层（Proxy Layer）的形式集成到应用和LLM API之间。这个代理层可以是：
- 应用内部的SDK/库：直接在应用代码中集成缓存逻辑。
- 独立的服务代理：一个独立运行的服务，所有LLM请求都通过它转发。
- API Gateway插件：在API网关层实现缓存。
独立的服务代理是最灵活和可扩展的方式，它允许集中管理缓存逻辑、监控和策略调整，而无需修改每个应用。
# 伪代码：代理层结构
class LLMCachingProxy:
def __init__(self, cache_store, openai_client, claude_client):
self.cache_store = cache_store # RedisClient, PostgresAdapter等
self.openai_client = openai_client
self.claude_client = claude_client
self.cache_locks = {} # 进程内锁，生产环境用分布式锁
def chat_completion(self, vendor: str, request_payload: Dict[str, Any]):
if vendor == "openai":
cache_key = generate_openai_cache_key(request_payload)
llm_client = self.openai_client
elif vendor == "claude":
cache_key = generate_claude_cache_key(request_payload)
llm_client = self.claude_client
else:
raise ValueError("Unsupported LLM vendor")
# 1. 尝试从缓存读取
cached_response = self.cache_store.get(cache_key)
if cached_response:
print(f"Cache HIT for {vendor} key: {cache_key}")
# 记录缓存命中指标
return json.loads(cached_response)
# 2. 缓存未命中，使用分布式锁确保单次LLM调用
# 实际生产环境应使用Redis或其他分布式锁
with self._get_or_create_lock(cache_key):
# 再次检查缓存，避免在等待锁期间重复调用
cached_response = self.cache_store.get(cache_key)
if cached_response:
print(f"Cache HIT (after lock) for {vendor} key: {cache_key}")
return json.loads(cached_response)
print(f"Cache MISS for {vendor} key: {cache_key}. Calling LLM API...")
# 3. 调用实际的LLM API
try:
if vendor == "openai":
response_obj = llm_client.chat.completions.create(**request_payload).model_dump_json()
elif vendor == "claude":
# Claude API调用方式略有不同
response_obj = llm_client.messages.create(**request_payload).model_dump_json()
# 4. 存储响应到缓存，设置TTL (例如24小时)
self.cache_store.set(cache_key, response_obj, ex=86400)
print(f"Cache SET for {vendor} key: {cache_key}")
return json.loads(response_obj)
except Exception as e:
print(f"Error calling LLM API for {vendor}: {e}")
raise
def _get_or_create_lock(self, key):
"""简单的进程内锁管理，生产环境应替换为分布式锁。"""
if key not in self.cache_locks:
self.cache_locks[key] = threading.Lock()
return self.cache_locks[key]
# 示例：假设您有OpenAI和Claude的客户端实例
# from openai import OpenAI
# from anthropic import Anthropic
# openai_client = OpenAI(api_key="your_openai_key")
# claude_client = Anthropic(api_key="your_anthropic_key")
# 模拟一个简单的Redis缓存接口
class MockRedisCache:
def __init__(self):
self._cache = {}
self._ttl = {}
def get(self, key):
if key in self._cache:
if self._ttl.get(key, 0) == 0 or self._ttl[key] > time.time():
return self._cache[key]
else:
del self._cache[key]
del self._ttl[key]
return None
def set(self, key, value, ex=0):
self._cache[key] = value
if ex > 0:
self._ttl[key] = time.time() + ex
else:
self._ttl[key] = 0
# if __name__ == "__main__":
# mock_redis = MockRedisCache()
# # 假设openai_client和claude_client已初始化
# # proxy = LLMCachingProxy(mock_redis, openai_client, claude_client)
# # proxy.chat_completion("openai", openai_req1)
# # proxy.chat_completion("claude", claude_req1)第六章：缓存命中策略的动态调整与展望
6.1 基于成本和延迟的动态策略
我们不必对所有请求都采用最严格的缓存策略。可以根据业务场景对成本和延迟的敏感度来动态调整：
- 高成本/高延迟请求：采用更宽松的匹配策略（例如，
temperature在小范围内匹配），以提高命中率。 - 低成本/低延迟且要求精确的请求：采用最严格的精确匹配策略。
- 实时性要求高的请求：可以缩短缓存TTL，甚至跳过缓存。
这可以通过在请求负载中添加自定义的缓存控制参数来实现，例如：
{
"model": "...",
"messages": [...],
"temperature": 0.7,
"x_cache_control": {
"strategy": "loose_temperature",
"temperature_tolerance": 0.1,
"ttl_seconds": 3600
}
}代理层在生成缓存键时，可以根据x_cache_control中的策略来决定是否对某些参数进行模糊匹配或忽略。
6.2 语义缓存 (Semantic Caching) 的引入
尽管我们对请求参数进行了标准化，但仍然无法解决“语义等价但文本不同”的问题。语义缓存是解决这一问题的高级方法：
- 工作原理：它不直接哈希原始文本，而是将Prompt通过嵌入模型（Embedding Model）转换为高维向量。
- 缓存查找：当新请求到来时，也将其Prompt转换为向量，然后使用向量相似度搜索（例如余弦相似度）在缓存中查找“语义上最相似”的已缓存Prompt。
- 挑战：需要额外的嵌入模型和向量数据库，计算开销大，且相似度阈值难以确定。
这是一个更先进但复杂度更高的方向，适合对缓存命中率有极高要求且能接受更高开销的场景。
6.3 多模态输入缓存的挑战
随着LLM支持多模态输入（如图像、音频），缓存策略将面临新的挑战：
- 图像哈希：如何对图像进行标准化和哈希？简单的文件哈希可能无法应对图像的微小修改或不同编码。感知哈希（Perceptual Hashing）可能是方向之一，但其“相似性”定义仍需商榷。
- 多模态语义：如何判断一个图像-文本组合与另一个图像-文本组合是“等价”的？这比纯文本语义等价复杂得多。
- 存储开销：多模态内容通常较大，对缓存存储的容量和性能提出更高要求。
6.4 未来的发展方向
- AI驱动的缓存策略：利用机器学习模型分析历史请求和响应，自动学习最佳的缓存键生成策略和失效规则。
- 跨厂商的通用缓存抽象：构建一个更高级的抽象层，能够以更统一的方式处理不同LLM厂商的请求，减少定制化工作。
- 更细粒度的缓存：不仅缓存整个API响应，甚至可以缓存单个消息的生成结果，或特定工具调用的结果，以实现更细粒度的复用。
通过今天的探讨，我们深入剖析了提示缓存的核心概念、其在LLM应用中的重要性，以及如何针对OpenAI和Claude等主流LLM厂商，精心设计定制化的缓存命中策略。我们强调了标准化请求参数、构建鲁棒缓存键的重要性，并通过具体的代码示例展示了实现细节。同时，我们也探讨了深度集成中的架构考量、幂等性、可观察性，并展望了语义缓存、多模态缓存及动态策略等未来发展方向。
提示缓存不仅仅是一种技术优化，更是构建高效、经济、可靠LLM应用的关键组成部分。它要求我们不仅理解LLM API的表面接口，更要洞察其内部机制和行为模式，才能真正发挥其潜力。希望今天的分享能为大家在LLM应用开发中带来启发，共同推动这一领域的技术进步。感谢大家！