LangChain Redis 包：利用先进向量存储和更快速缓存构建更智能的 AI 应用

import os

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
print(f"Connecting to Redis at: {REDIS_URL}")

from langchain_redis import RedisVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain.docstore.document import Document

# Initialize RedisVectorStore
embeddings = OpenAIEmbeddings()
vector_store = RedisVectorStore(embeddings, redis_url=REDIS_URL, index_name="my_docs")

# Add documents
docs = [
    Document(page_content="Redis is a powerful in-memory data structure store", metadata={"source": "redis_intro"}),
    Document(page_content="LangChain is a framework for developing applications powered by language models", metadata={"source": "langchain_intro"}),
]
vector_store.add_documents(docs)

# Perform a similarity search
query = "What is Redis?"
results = vector_store.similarity_search(query)
print(f"Search results for '{query}':")
for doc in results:
    print(f"- {doc.page_content} (Source: {doc.metadata['source']})")

from langchain_redis import RedisCache
from langchain_openai import OpenAI
from langchain.globals import set_llm_cache
import time

# Initialize RedisCache
redis_cache = RedisCache(redis_url=REDIS_URL)
set_llm_cache(redis_cache)

# Initialize the language model
llm = OpenAI(temperature=0)

# Function to measure execution time
def timed_completion(prompt):
    start_time = time.time()
    result = llm.invoke(prompt)
    end_time = time.time()
    return result, end_time - start_time

# First call (not cached)
prompt = "Explain the concept of caching in three sentences."
result1, time1 = timed_completion(prompt)
print(f"First call (not cached):\nResult: {result1}\nTime: {time1:.2f} seconds\n")

# Second call (should be cached)
result2, time2 = timed_completion(prompt)
print(f"Second call (cached):\nResult: {result2}\nTime: {time2:.2f} seconds\n")

print(f"Speed improvement: {time1 / time2:.2f}x faster")

from langchain_redis import RedisSemanticCache
from langchain_openai import OpenAIEmbeddings

# Initialize RedisSemanticCache
embeddings = OpenAIEmbeddings()
semantic_cache = RedisSemanticCache(redis_url=REDIS_URL, embeddings=embeddings, distance_threshold=0.2)
set_llm_cache(semantic_cache)

# Test semantic caching
original_prompt = "What is the capital of France?"
result1, time1 = timed_completion(original_prompt)
print(f"Original query:\nPrompt: {original_prompt}\nResult: {result1}\nTime: {time1:.2f} seconds\n")

similar_prompt = "Can you tell me the capital city of France?"
result2, time2 = timed_completion(similar_prompt)
print(f"Similar query:\nPrompt: {similar_prompt}\nResult: {result2}\nTime: {time2:.2f} seconds\n")

print(f"Speed improvement: {time1 / time2:.2f}x faster")

from langchain_redis import RedisChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

# Initialize RedisChatMessageHistory
message_history = RedisChatMessageHistory("user_123", redis_url=REDIS_URL)

# Create a conversational chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful AI assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])
chain = prompt | llm

# Function to get or create a RedisChatMessageHistory instance
def get_redis_history(session_id: str):
    return RedisChatMessageHistory(session_id, redis_url=REDIS_URL)

# Create a runnable with message history
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_redis_history,
    input_messages_key="input",
    history_messages_key="history"
)

# Use the chain in a conversation
response1 = chain_with_history.invoke(
    {"input": "Hi, my name is Alice."},
    config={"configurable": {"session_id": "alice_123"}}
)
print("AI Response 1:", response1.content)

response2 = chain_with_history.invoke(
    {"input": "What's my name?"},
    config={"configurable": {"session_id": "alice_123"}}
)
print("AI Response 2:", response2.content)