使用 Upstash 构建无服务器向量数据库与缓存系统的最佳实践

技术背景介绍

在构建现代智能应用时，开发者经常需要处理大量数据，包括向量嵌入、缓存管理和消息存储。然而传统数据库和内存存储的部署和运维往往带来复杂性。为了简化这些挑战，Upstash 提供了基于 HTTP 的无服务器数据库和消息队列，其中包含两大关键产品：

1. Upstash Vector —— 用于管理向量嵌入的无服务器数据库。
2. Upstash Redis —— 用作缓存与内存存储。

这些数据库支持 HTTP 协议，使其能够在无服务器平台（如 AWS Lambda、边缘计算环境等）中运行，而无需依赖传统的 TCP 连接。

本文将详细讲解如何通过 Upstash 开发强大的 LangChain 应用，包括向量存储与查询、缓存使用等场景。

核心原理解析

Upstash Vector

Upstash Vector 是一个支持向量嵌入存储与查询的无服务器数据库。它允许开发者存储高维向量，并快速进行相似性搜索。其核心功能包括：

1. 向量插入与批量存储。
2. 基于文本或向量的相似性搜索。
3. 命名空间（namespace）用于数据分区。
4. 支持基于 HTTP 的嵌入操作。

Upstash Redis

Upstash Redis 是 Redis 的无服务器实现，适用于以下场景：

1. 低延迟缓存（如缓存 LLM 的输入/输出）。
2. 聊天消息历史的存储，支持对话记忆。
3. 高并发访问的全球分布式数据库。

代码实现演示

环境配置

在开始之前，请在 Upstash 控制台创建账户，并获取相应的 REST URL 和令牌。在代码中设置这些环境变量以确保安全。

import os

os.environ["UPSTASH_VECTOR_REST_URL"] = "https://example.upstash.io"
os.environ["UPSTASH_VECTOR_REST_TOKEN"] = "your-upstash-token"
os.environ["UPSTASH_REDIS_REST_URL"] = "https://redis-example.upstash.io"
os.environ["UPSTASH_REDIS_REST_TOKEN"] = "your-redis-token"

Upstash Vector 的实践

1. 安装 SDK

首先安装必要的 Python SDK：

pip install upstash-vector langchain

2. 创建向量存储对象

以下代码展示如何初始化一个 Upstash Vector 存储对象，并使用 OpenAI 嵌入模型生成向量。

from langchain_community.vectorstores.upstash import UpstashVectorStore
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings

# 加载文本数据
loader = TextLoader("state_of_the_union.txt")
documents = loader.load()

# 分割文档为小段落
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

# 初始化 OpenAI 嵌入
embeddings = OpenAIEmbeddings()

# 创建 Upstash Vector 存储对象
store = UpstashVectorStore(embedding=embeddings)

# 插入文档的嵌入向量
store.add_documents(docs)

3. 查询向量

对向量的查询可以基于文本或向量：

# 基于文本的相似性搜索
result = store.similarity_search(
    "The United States of America",
    k=5  # 返回前 5 个匹配
)
print(result)

# 基于向量的相似性搜索
query_vector = embeddings.embed_query("Global economics")
result = store.similarity_search_by_vector(query_vector, k=5)
print(result)

4. 命名空间管理

使用命名空间将数据进行逻辑分区，适合大规模数据管理：

store_with_namespace = UpstashVectorStore(
    embedding=embeddings,
    namespace="economic_reports"
)

Upstash Redis 的实践

1. 安装 SDK

安装 Redis SDK：

pip install upstash-redis

2. 配置缓存

将 Upstash Redis 集成到 LangChain 缓存中，用于存储 LLM 请求与响应的结果：

from langchain.cache import UpstashRedisCache
from upstash_redis import Redis

redis_client = Redis(
    url=os.getenv("UPSTASH_REDIS_REST_URL"),
    token=os.getenv("UPSTASH_REDIS_REST_TOKEN")
)

# 设置 LLM 缓存
from langchain.llm_cache import UpstashRedisCache
langchain.llm_cache = UpstashRedisCache(redis_=redis_client)

3. 对话记忆管理

使用 Upstash Redis 存储聊天消息历史：

from langchain_community.chat_message_histories import UpstashRedisChatMessageHistory

# 初始化聊天消息历史存储
message_history = UpstashRedisChatMessageHistory(
    redis_=redis_client,
    session_id="user-session-1"
)

# 保存对话历史
message_history.add_message({"role": "user", "content": "Hello AI!"})
message_history.add_message({"role": "assistant", "content": "Hello User!"})

# 获取对话历史
print(message_history.messages)

应用场景分析

1. AI 搜索引擎：结合 Upstash Vector 构建支持语义查询的搜索引擎。
2. 实时对话系统：通过 Upstash Redis 存储对话历史，实现上下文记忆。
3. 边缘推理应用：基于 Upstash 的 HTTP 支持，可以轻松部署在 AWS Lambda 等无服务器平台。

实践建议

1. 优化批量插入：在插入大规模向量时，使用 batch_size 参数控制请求批量大小，以提升性能。
2. 使用命名空间：为不同类别的数据设置单独的命名空间，便于分区管理和加速查询。
3. 缓存 LLM 结果：使用 Upstash Redis 减少重复计算，降低延迟与成本。

通过本文，你可以轻松在 LangChain 中集成 Upstash 提供的服务，构建高效、可扩展的 AI 应用。如果遇到问题欢迎在评论区交流。

—END—