Quick Start: 5分でRedisVLを動かす
LLMを1回呼び出すたびに2〜3秒と一定量のトークンを消費します。これが1日に数千リクエストあれば、無視できないコストになります。RedisVLはRedisをvector storeとsemantic cacheに変えるPythonライブラリです。そして最大のポイントは、言い回しが違っても意味が同じ2つの質問に対して、LLMを再度呼び出すのではなくキャッシュされた結果を返すことです。
Redis Stackのインストールと起動（RediSearchモジュールが必要）：
pip install redisvl
# DockerでRedis Stackを起動
docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latestsemantic cacheをすぐにテストする：
from redisvl.extensions.llmcache import SemanticCache
cache = SemanticCache(
name="llm_cache",
redis_url="redis://localhost:6379",
distance_threshold=0.15 # cosine distance、低いほど類似度が高い
)
# 初回：cache miss、結果を保存
response = cache.check(prompt="Pythonとは何ですか？")
if not response:
answer = "Pythonは高水準で読みやすく汎用的なプログラミング言語です..."
cache.store(prompt="Pythonとは何ですか？", response=answer)
print("Cache miss — 保存しました")
else:
print(f"Cache hit: {response[0]['response']}")
# 次回「Pythonについて教えて」と聞いても → cache hitになる
# "Pythonとは何ですか？"とvector embeddingが近いためそれだけです。次回ユーザーが「Pythonについて教えて」や「Pythonはどんな言語？」と聞いても、言い回しが違っても、RedisはAPIを呼ぶのではなく数ミリ秒で以前の結果を返します。
RedisVLが解決する問題
なぜexact-match cacheでは不十分なのか？
通常のキャッシュは質問のハッシュをキーとして使います。「Pythonとは何ですか？」と「Pythonはどんな言語？」は答えが同じでも全く異なるキーになります。結果として、実際のcache hit rateは非常に低く、5%を下回ることもあります。
Semantic cachingはハッシュの代わりにvector embeddingを使います。「Pythonとは何ですか？」と「Pythonはどんな言語？」はembeddingが非常に近く、threshold 0.15で同じcache entryにhitできます。実際のhit rateは35〜45%に達することもあり、exact-matchよりもはるかに高くなります。
RedisがVector Searchに対応しているか確認する
redis-cli MODULE LIST | grep search
# outputに"search"が表示されるはず
# 表示されない場合 → Redis StackまたはSearch有効なRedis Cloudが必要詳細解説：カスタムVector Search Indexの構築
RAG向けknowledge baseの構築
RedisVLはcachingだけにとどまりません。Redis上に直接vector search indexを構築でき、RAGやknowledge base searchに最適です。実際に社内ドキュメント（約3,000エントリ）のindexに使ったところ、そのデータセットでクエリのlatencyは15ms以下で安定していました：
from redisvl.index import SearchIndex
from redisvl.schema import IndexSchema
import numpy as np
from openai import OpenAI
client = OpenAI()
# schemaを定義
schema = IndexSchema.from_dict({
"index": {
"name": "docs_index",
"prefix": "doc:",
"storage_type": "hash"
},
"fields": [
{"name": "content", "type": "text"},
{"name": "category", "type": "tag"},
{"name": "embedding", "type": "vector", "attrs": {
"dims": 1536,
"distance_metric": "cosine",
"algorithm": "hnsw",
"datatype": "float32"
}}
]
})
index = SearchIndex(schema, redis_url="redis://localhost:6379")
index.create(overwrite=True)データのロードとクエリ
from redisvl.query import VectorQuery
def get_embedding(text):
res = client.embeddings.create(input=text, model="text-embedding-3-small")
return res.data[0].embedding
docs = [
{"content": "Dockerはアプリケーションのコンテナ化プラットフォームです", "category": "devops"},
{"content": "Kubernetes orchestrates container deployments", "category": "devops"},
{"content": "Redisは高速なin-memory databaseです", "category": "database"},
]
data = []
for i, doc in enumerate(docs):
emb = get_embedding(doc["content"])
data.append({
"content": doc["content"],
"category": doc["category"],
"embedding": np.array(emb, dtype=np.float32).tobytes()
})
index.load(data)
# semantic検索
query_emb = get_embedding("Container orchestration tool")
results = index.query(
VectorQuery(
vector=query_emb,
vector_field_name="embedding",
return_fields=["content", "category"],
num_results=3
)
)
for r in results:
print(f"Score: {r['vector_distance']:.3f} | {r['content']}")応用：Metadata filter、TTL、Async
ユーザーコンテキスト別のキャッシュ
cache = SemanticCache(
name="app_cache",
redis_url="redis://localhost:6379",
distance_threshold=0.15,
ttl=3600 # キャッシュの有効期限は1時間
)
# metadataと共に保存
cache.store(
prompt="ニューラルネットワークについて説明して",
response="ニューラルネットワークは機械学習モデルの一種です...",
metadata={"language": "ja", "user_tier": "premium"}
)
# filterつきでチェック — 同じ言語のキャッシュのみhit
results = cache.check(
prompt="ニューラルネットワークとは何ですか？",
filter_expression="@language:{ja}"
)
if results:
print(f"Cache hit: {results[0]['response']}")高トラフィックなproduction向けAsync
from redisvl.extensions.llmcache import AsyncSemanticCache
async def handle_request(prompt: str, llm_func):
cache = AsyncSemanticCache(
name="async_cache",
redis_url="redis://localhost:6379",
distance_threshold=0.15
)
cached = await cache.acheck(prompt=prompt)
if cached:
return cached[0]["response"]
response = await llm_func(prompt)
await cache.astore(prompt=prompt, response=response)
return response実践的なTips
適切なdistance_thresholdの選び方
10,000以上のリクエストで約1週間productionを運用した結果、各thresholdの範囲で観察できたことを以下にまとめます：
- 0.05–0.10：非常にstrict。Cache hit rateは低め（〜15%）だが、responseの精度は高い。
- 0.10–0.20：バランスが良い。Hit rate〜35〜45%で、ほとんどのuse caseで十分な精度。
- 0.20+：Aggressive。Hit rateは高いが、正確な結果ではなく「近似的な」結果を返す可能性がある。
0.15から始めて、1〜2週間モニタリングしてから調整しましょう。
日本語対応のembedding modelを使う
RedisVLはデフォルトでtext-embedding-ada-002を使います。コストをかけずローカルで実行し、日本語により良く対応したい場合：
from redisvl.utils.vectorize import HFTextVectorizer
vectorizer = HFTextVectorizer(
model="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
# 日本語を含む50以上の言語をサポート、完全ローカルで実行可能
)
cache = SemanticCache(
name="ja_cache",
vectorizer=vectorizer,
redis_url="redis://localhost:6379",
distance_threshold=0.15
)LangChainとの統合
LangChainを使っているスタックなら？数行のconfig追加だけで、すべてのLLM callが自動的にsemantic cacheを通るようになります：
from langchain_community.cache import RedisSemanticCache
from langchain_openai import OpenAIEmbeddings
import langchain
langchain.llm_cache = RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=OpenAIEmbeddings(),
score_threshold=0.2
)
# 以降、LangChain内のすべてのLLM callが自動的にsemantic cacheを使用する
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke("Redisとは何ですか？") # 2回目はcache hitになるcache poisoningを防ぐ
Cache poisoningは見落とされがちなリスクです。LLMが最初にhallucinateした場合、誤ったresponseがキャッシュされ、以降の類似する質問すべてで誤った結果が返されます。対策：
- 適切なTTLを設定する（データの種類に応じて3600〜86400秒）
- リアルタイムデータに関する質問はキャッシュしない（株価、天気など）
- metadataを使ってソースをタグ付けし、必要な時にバッチで無効化しやすくする
- モニタリングし、システムエラー検出時にキャッシュをクリアする仕組みを設ける：
# statsを確認
stats = cache.index.info()
print(f"Total cached entries: {stats['num_docs']}")
# 新モデルのrollout時にキャッシュを全クリア
cache.clear()すでにRedisを使っているスタックなら、RedisVLを追加するだけで新しいインフラのデプロイは不要です。LLM呼び出しの1〜3秒に対してcache hitは10ms以下——threshold 0.15でhit rate 35〜45%なら、数週間で実際のトークンコストは大幅に削減されます。専用のvector DBに移行する前に試す価値があります。
