0Pricing
AI Prompt Engineering · 课时

提示词缓存策略

语义缓存、精确匹配缓存和 Anthropic 提示词缓存。

提示词缓存策略 是 CoddyKit 上的免费 AI Prompt Engineering 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Prompt Engineering 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Prompt Engineering 课程共包含 4 节课。

为什么要缓存提示词结果

LLM 接口调用成本高且速度慢。许多生产应用会反复发送相同（或非常相似）的提示词。缓存会为重复查询返回已存储的结果，从而消除冗余接口调用，大幅降低成本和延迟。

使用哈希键进行精确匹配缓存

最简单的缓存方式是：对完全相同的提示词字符串进行哈希处理并存储结果。如果再次出现相同的提示词字符串，就直接返回缓存结果，而不调用接口。

import hashlib
import json
from functools import lru_cache

class ExactMatchCache:
    def __init__(self, backend=None):
        # backend: a dict (in-memory) or Redis client
        self.store = backend or {}

    def _key(self, messages, model, max_tokens):
        content = json.dumps({'messages': messages, 'model': model,
                               'max_tokens': max_tokens}, sort_keys=True)
        return 'llm:' + hashlib.sha256(content.encode()).hexdigest()

    def get(self, messages, model, max_tokens):
        key = self._key(messages, model, max_tokens)
        return self.store.get(key)

    def set(self, messages, model, max_tokens, result, ttl_seconds=3600):
        key = self._key(messages, model, max_tokens)
        self.store[key] = result
        # In Redis: self.store.setex(key, ttl_seconds, json.dumps(result))

cache = ExactMatchCache()

# Usage
messages = [{'role': 'user', 'content': 'What is the capital of France?'}]
cached = cache.get(messages, 'gpt-4o-mini', 100)
if cached:
    print('Cache HIT:', cached[:50])
else:
    print('Cache MISS — calling API...')

带缓存封装的 LLM 客户端

使用缓存装饰器封装 LLM 接口调用，这样所有调用方都能透明地获得缓存功能，而无需修改自身代码。

import openai
from typing import Optional

client = openai.OpenAI(api_key='YOUR_API_KEY')
cache = ExactMatchCache()

def cached_completion(messages, model='gpt-4o-mini', max_tokens=500,
                       temperature=0.0, use_cache=True) -> str:
    if use_cache and temperature == 0.0:
        # Only cache deterministic requests (temperature=0)
        cached = cache.get(messages, model, max_tokens)
        if cached:
            return cached

    response = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens,
        temperature=temperature
    )
    result = response.choices[0].message.content

    if use_cache and temperature == 0.0:
        cache.set(messages, model, max_tokens, result)

    return result

# Important: only cache temperature=0 responses
# Non-deterministic responses (temp>0) may return stale results
print('Cache wrapping: only deterministic (temp=0) calls are cached.')

使用嵌入向量进行语义缓存

语义缓存会为含义相似的查询返回缓存结果，而不仅仅是匹配完全相同的字符串。它使用嵌入向量和余弦相似度来查找近似重复的查询。

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticCache:
    def __init__(self, similarity_threshold=0.95):
        self.entries = []  # [(embedding, query, result)]
        self.threshold = similarity_threshold

    def embed(self, text):
        '''Get embedding for text using OpenAI embeddings API.'''
        response = client.embeddings.create(
            model='text-embedding-3-small',
            input=text
        )
        return np.array(response.data[0].embedding)

    def get(self, query):
        if not self.entries:
            return None
        query_emb = self.embed(query)
        for emb, stored_query, result in self.entries:
            sim = cosine_similarity([query_emb], [emb])[0][0]
            if sim >= self.threshold:
                print(f'Semantic cache HIT (similarity={sim:.3f}): {stored_query[:40]}...')
                return result
        return None

    def set(self, query, result):
        emb = self.embed(query)
        self.entries.append((emb, query, result))

sem_cache = SemanticCache(similarity_threshold=0.95)
print('Semantic cache ready. Threshold: 0.95 cosine similarity.')

GPT 缓存库

GPT 缓存库是一个开源语义缓存库，支持多种嵌入模型、相似度后端（FAISS、内存数据库）和驱逐策略。它可以直接与 OpenAI 和 LangChain 客户端集成。

# pip install gptcache
# GPTCache integration example

# from gptcache import cache
# from gptcache.adapter import openai
# from gptcache.embedding import Onnx
# from gptcache.manager import CacheBase, VectorBase, get_data_manager
# from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

# Initialize GPTCache
# onnx = Onnx()
# data_manager = get_data_manager(
#     CacheBase('sqlite'),
#     VectorBase('faiss', dimension=onnx.dimension)
# )
# cache.init(
#     embedding_func=onnx.to_embeddings,
#     data_manager=data_manager,
#     similarity_evaluation=SearchDistanceEvaluation(),
# )

# After init, use openai from gptcache.adapter instead of standard openai
# response = openai.ChatCompletion.create(
#     model='gpt-4o-mini',
#     messages=[{'role': 'user', 'content': 'What is Python?'}]
# )
# Same API, but cache is checked first

print('GPTCache: drop-in semantic cache for OpenAI API calls.')
print('Supports: FAISS, Redis, SQLite, Milvus as vector backends.')

Anthropic 提示词缓存（原生功能）

Anthropic 提供原生的提示词缓存功能，会在其服务器上缓存系统提示词的处理结果。命中缓存后，您只需支付正常输入令牌价格的 10%。这与应用层响应缓存是分开的。

import anthropic

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

LONG_SYSTEM_PROMPT = '''You are an expert financial analyst with 20 years of experience.
''' + 'Domain knowledge: ' + 'analysis context...' * 500  # large system prompt

# Enable prompt caching with cache_control
response = client.messages.create(
    model='claude-opus-4-5',
    max_tokens=1024,
    system=[
        {
            'type': 'text',
            'text': LONG_SYSTEM_PROMPT,
            'cache_control': {'type': 'ephemeral'}  # cache this prefix
        }
    ],
    messages=[{'role': 'user', 'content': 'Analyze Q3 2024 earnings.'}]
)

print('Cache write tokens:', response.usage.cache_creation_input_tokens)
print('Cache read tokens: ', response.usage.cache_read_input_tokens)
print('Regular input tokens:', response.usage.input_tokens)
# On cache HIT: cache_read_input_tokens shows the cached tokens
# Cost: cached tokens charged at 10% of normal rate

缓存 TTL 与驱逐策略

当底层知识发生变化或模型更新时，缓存结果会变得过时。TTL（生存时间）和驱逐策略用于管理新鲜度。

import time
from collections import OrderedDict

class TTLCache:
    def __init__(self, max_size=1000, default_ttl=3600):
        self.store = OrderedDict()  # key: (value, expire_at)
        self.max_size = max_size
        self.default_ttl = default_ttl

    def set(self, key, value, ttl=None):
        ttl = ttl or self.default_ttl
        expire_at = time.time() + ttl
        if key in self.store:
            del self.store[key]
        self.store[key] = (value, expire_at)
        # LRU eviction: remove oldest if over capacity
        if len(self.store) > self.max_size:
            self.store.popitem(last=False)

    def get(self, key):
        if key not in self.store:
            return None
        value, expire_at = self.store[key]
        if time.time() > expire_at:
            del self.store[key]
            return None  # expired
        # Move to end (LRU update)
        self.store.move_to_end(key)
        return value

# TTL strategy guidelines
ttl_guidelines = {
    'Static knowledge': 86400,  # 24h (facts, definitions)
    'Semi-static': 3600,        # 1h (product info, FAQs)
    'Dynamic content': 300,     # 5min (news, prices)
    'Personalized': 0           # no cache (user-specific)
}
for k, v in ttl_guidelines.items():
    print(f'{k}: {v}s TTL')

缓存失效模式

缓存失效——确定何时清除过时数据——是计算领域最困难的问题之一。对于 LLM 缓存，以下模式可以处理最常见的失效需求。

class InvalidationAwareCache(TTLCache):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.tags = {}  # key: set of tags
        self.tag_index = {}  # tag: set of keys

    def set_with_tags(self, key, value, tags, ttl=None):
        self.set(key, value, ttl)
        self.tags[key] = set(tags)
        for tag in tags:
            self.tag_index.setdefault(tag, set()).add(key)

    def invalidate_by_tag(self, tag):
        keys_to_delete = self.tag_index.pop(tag, set())
        for key in keys_to_delete:
            self.store.pop(key, None)
            self.tags.pop(key, None)
        print(f'Invalidated {len(keys_to_delete)} entries with tag={tag}')

# Usage: tag cache entries by data source
cache = InvalidationAwareCache()
cache.set_with_tags('product_faq_123', 'Product FAQs...', tags=['product:123', 'faqs'])
cache.set_with_tags('product_spec_123', 'Spec sheet...', tags=['product:123', 'specs'])

# When product 123 is updated, invalidate all its cache entries
cache.invalidate_by_tag('product:123')  # Invalidated 2 entries

衡量缓存性能

请跟踪缓存性能指标，以了解缓存对成本和延迟的影响。经过良好调优的缓存，在大多数生产用例中应达到 >50% 的命中率。

class CacheMetrics:
    def __init__(self):
        self.hits = 0
        self.misses = 0
        self.total_latency_saved_ms = 0
        self.total_cost_saved_usd = 0
        self.avg_api_latency_ms = 1500  # typical LLM call latency
        self.avg_api_cost_usd = 0.002   # typical cost per call

    def record_hit(self):
        self.hits += 1
        self.total_latency_saved_ms += self.avg_api_latency_ms
        self.total_cost_saved_usd += self.avg_api_cost_usd

    def record_miss(self):
        self.misses += 1

    def report(self):
        total = self.hits + self.misses
        hit_rate = self.hits / total if total else 0
        return {
            'hit_rate': f'{hit_rate:.1%}',
            'total_requests': total,
            'cache_hits': self.hits,
            'latency_saved_sec': round(self.total_latency_saved_ms / 1000, 1),
            'cost_saved_usd': round(self.total_cost_saved_usd, 2)
        }

metrics = CacheMetrics()
for i in range(100):
    if i % 3 == 0:  # simulate 33% hit rate
        metrics.record_hit()
    else:
        metrics.record_miss()
print(metrics.report())

基于 Redis 的生产环境缓存

内存缓存会在重启后丢失，也无法在多个服务器实例之间共享。Redis 提供了持久化的共享缓存，可在生产环境部署中的多个 API 服务器之间使用。

import redis
import json
import hashlib

class RedisLLMCache:
    def __init__(self, host='localhost', port=6379, db=0, default_ttl=3600):
        self.client = redis.Redis(host=host, port=port, db=db,
                                   decode_responses=True)
        self.default_ttl = default_ttl

    def _key(self, messages, model):
        content = json.dumps({'messages': messages, 'model': model},
                              sort_keys=True)
        return 'llmcache:' + hashlib.sha256(content.encode()).hexdigest()

    def get(self, messages, model):
        key = self._key(messages, model)
        value = self.client.get(key)
        if value:
            self.client.expire(key, self.default_ttl)  # refresh TTL on hit
            return json.loads(value)
        return None

    def set(self, messages, model, result, ttl=None):
        key = self._key(messages, model)
        self.client.setex(key, ttl or self.default_ttl, json.dumps(result))

    def stats(self):
        keys = self.client.keys('llmcache:*')
        return {'cached_entries': len(keys),
                'memory_bytes': self.client.memory_usage('llmcache:') or 0}

# Usage: drop-in replacement for in-memory cache
# cache = RedisLLMCache(host='redis.internal', port=6379)
print('RedisLLMCache: shared across all server instances, survives restarts.')

不应缓存的情况

并非所有 LLM 调用都适合缓存。了解何时跳过缓存，可以避免提供过时或错误的结果。

DONT_CACHE_WHEN = {
    'High temperature': (
        'temperature > 0 produces different outputs for the same input. '
        'Caching would always return the first generation, defeating the purpose.'
    ),
    'Real-time data required': (
        'Queries about current prices, live news, or real-time status '
        'must always hit the API and live data source.'
    ),
    'Personalized responses': (
        'Responses that depend on user_id, session context, or personal data '
        'should not be shared across users.'
    ),
    'Safety-critical': (
        'Medical, legal, or financial responses where staleness could cause harm '
        'require fresh responses with the most current model version.'
    ),
    'Non-deterministic tools': (
        'If the prompt includes a current timestamp or random seed, '
        'the response is by design non-repeatable.'
    )
}

for condition, reason in DONT_CACHE_WHEN.items():
    print(f'Skip cache: {condition}')
    print(f'  Reason: {reason[:60]}...')
    print()

快速检查

LLM 响应的精确匹配缓存与语义缓存之间的关键区别是什么？

缓存策略总结

有效的提示词缓存需要结合多种策略：

  • 精确匹配：基于哈希，命中时零额外开销，但不同措辞会导致命中率较低
  • 语义缓存：通过嵌入向量相似度查找改写后的匹配项，命中率更高
  • GPTCache：结合两种策略并支持 FAISS/Redis 后端的开源库
  • Anthropic 原生缓存：在服务器端缓存系统提示词，令牌成本为原来的 10%
  • TTL + LRU 淘汰：通过时间保证新鲜度，并管理缓存容量
  • 基于标签的失效处理：源数据发生变化时，使相关条目失效
  • 不应缓存的情况：非零温度、实时数据、个性化内容以及安全攸关的请求

常见问题解答

「提示词缓存策略」课时是免费的吗？

是的 — 「提示词缓存策略」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Prompt Engineering 课程的其余内容，请升级到 CoddyKit PRO。 AI Prompt Engineering 课程共包含 4 节课。

「提示词缓存策略」这节课中我会学到什么？

语义缓存、精确匹配缓存和 Anthropic 提示词缓存。 你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Prompt Engineering 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Prompt Engineering 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。

「提示词缓存策略」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Prompt Engineering 课中编写并运行代码吗？

能。每节 AI Prompt Engineering 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 提示词缓存策略
  2. 批处理与异步执行
  3. 跨模型负载均衡
  4. 提示词流程的监控与告警
← 返回 AI Prompt Engineering