Redis在AI向量数据库场景下的高性能实践
·
Redis在AI向量数据库场景下的高性能实践
深入探讨Redis在AI向量搜索和RAG应用中的核心作用，从原理到实践，全面解析高性能向量存储方案。
📋 目录
🚀 引言
随着AI应用的快速发展，向量数据库成为了知识检索、推荐系统、RAG（检索增强生成）等场景的核心基础设施。Redis凭借其卓越的性能和丰富的数据结构，在向量存储和搜索领域展现出了强大的优势。
为什么选择Redis作为向量数据库
- 极致性能：内存存储，毫秒级查询响应
- 成熟生态：丰富的客户端支持和运维工具
- 灵活架构：支持多种部署模式和扩展方案
- AI原生：RedisStack提供了专门的向量搜索能力
🔍 Redis向量数据库基础
Redis向量搜索架构
┌─────────────────────────────────────────────────────────────┐
│ Redis Vector Database │
├─────────────────┬─────────────────┬─────────────────────────┤
│ 客户端层 │ API网关 │ 负载均衡器 │
│ (Spring AI) │ (Spring Cloud)│ (Redis Proxy) │
├─────────────────┼─────────────────┼─────────────────────────┤
│ Redis Stack / Redis Enterprise │
│ ┌─────────────┬─────────────────┬─────────────────────┐ │
│ │ RediSearch │ RedisJSON │ RedisBloom │ │
│ │ (向量索引) │ (结构化数据) │ (布隆过滤器) │ │
│ └─────────────┴─────────────────┴─────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ 存储与持久化层 │
│ ┌─────────────┬─────────────────┬─────────────────────┐ │
│ │ RDB快照 │ AOF日志 │ 集群复制 │ │
│ │ (数据持久化)│ (操作日志) │ (高可用) │ │
│ └─────────────┴─────────────────┴─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
核心数据结构
// 1. 向量文档实体
@Data
@Builder
@Document
public class VectorDocument {
@Id
private String id;
@Indexed
private String title;
@Indexed
private String content;
@Vector(algorithm = VectorAlgorithm.HNSW,
dimensions = 1536,
distanceMetric = DistanceMetric.COSINE)
private float[] embedding;
@Indexed
private String category;
@Indexed
private String source;
@CreatedDate
private Instant createdAt;
@Indexed
private Map<String, String> metadata;
// 构建向量索引配置
public static IndexDefinition createIndexDefinition() {
return new IndexDefinition("vector_idx")
.addField(FieldType.TEXT, "title", 1.0)
.addField(FieldType.TEXT, "content", 0.5)
.addField(FieldType.VECTOR, "embedding",
VectorFieldOptions.builder()
.algorithm(VectorAlgorithm.HNSW)
.dimensions(1536)
.distanceMetric(DistanceMetric.COSINE)
.initialCapacity(10000)
.m(16) // HNSW参数
.efConstruction(200) // 构建时的ef值
.build())
.addField(FieldType.TAG, "category")
.addField(FieldType.TAG, "source");
}
}
// 2. Redis向量存储配置
@Configuration
@EnableConfigurationProperties(RedisVectorProperties.class)
public class RedisVectorConfig {
@Bean
public RedisVectorStore redisVectorStore(
RedisTemplate<String, Object> redisTemplate,
EmbeddingClient embeddingClient,
RedisVectorProperties properties) {
return RedisVectorStore.builder()
.redisTemplate(redisTemplate)
.embeddingClient(embeddingClient)
.indexName(properties.getIndexName())
.keyPrefix(properties.getKeyPrefix())
.vectorDimensions(properties.getVectorDimensions())
.similarityMetric(properties.getSimilarityMetric())
.build();
}
@Bean
public RedisTemplate<String, Object> vectorRedisTemplate(
LettuceConnectionFactory connectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
// 使用JSON序列化器，支持复杂数据结构
template.setDefaultSerializer(new GenericJackson2JsonRedisSerializer());
template.setKeySerializer(new StringRedisSerializer());
template.setHashKeySerializer(new StringRedisSerializer());
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
template.afterPropertiesSet();
return template;
}
}
// 3. 向量存储配置属性
@ConfigurationProperties(prefix = "redis.vector")
@Data
public class RedisVectorProperties {
private String indexName = "vector_search_idx";
private String keyPrefix = "doc:";
private int vectorDimensions = 1536;
private SimilarityMetric similarityMetric = SimilarityMetric.COSINE;
private HnswParameters hnsw = new HnswParameters();
private PerformanceConfig performance = new PerformanceConfig();
@Data
public static class HnswParameters {
private int m = 16; // 每个节点的最大连接数
private int efConstruction = 200; // 构建时的搜索深度
private int efRuntime = 10; // 运行时的搜索深度
private int initialCapacity = 10000; // 初始容量
}
@Data
public static class PerformanceConfig {
private int batchSize = 100; // 批量操作大小
private int maxConnections = 20; // 最大连接数
private Duration connectionTimeout = Duration.ofSeconds(5);
private Duration readTimeout = Duration.ofSeconds(30);
private boolean enablePipelining = true; // 开启管道
}
}
⚡ 向量搜索算法优化
HNSW算法实现与优化
// 1. 高性能向量搜索实现
@Service
@Slf4j
public class RedisVectorSearchService {
private final RedisTemplate<String, Object> redisTemplate;
private final EmbeddingClient embeddingClient;
private final VectorIndexManager indexManager;
private final SearchResultCache searchCache;
public RedisVectorSearchService(RedisTemplate<String, Object> redisTemplate,
EmbeddingClient embeddingClient) {
this.redisTemplate = redisTemplate;
this.embeddingClient = embeddingClient;
this.indexManager = new VectorIndexManager(redisTemplate);
this.searchCache = new SearchResultCache(redisTemplate);
}
public List<VectorSearchResult> searchSimilar(String query,
VectorSearchOptions options) {
try {
// 1. 查询向量化
float[] queryVector = embeddingClient.embed(query);
// 2. 检查缓存
String cacheKey = generateCacheKey(queryVector, options);
List<VectorSearchResult> cachedResults = searchCache.get(cacheKey);
if (cachedResults != null) {
log.debug("Cache hit for query: {}", query);
return cachedResults;
}
// 3. 执行向量搜索
List<VectorSearchResult> results = performVectorSearch(queryVector, options);
// 4. 结果后处理
List<VectorSearchResult> processedResults = postProcessResults(results, options);
// 5. 缓存结果
searchCache.put(cacheKey, processedResults, Duration.ofMinutes(30));
return processedResults;
} catch (Exception e) {
log.error("向量搜索失败: {}", query, e);
throw new VectorSearchException("搜索执行失败", e);
}
}
private List<VectorSearchResult> performVectorSearch(float[] queryVector,
VectorSearchOptions options) {
// 构建Redis Search查询
Query searchQuery = buildVectorQuery(queryVector, options);
// 执行搜索
SearchResult searchResult = redisTemplate.execute((RedisCallback<SearchResult>) connection -> {
return ((RedisSearchCommands) connection).search(
options.getIndexName(),
searchQuery
);
});
// 转换结果
return convertSearchResults(searchResult, options);
}
private Query buildVectorQuery(float[] queryVector, VectorSearchOptions options) {
Query.Builder queryBuilder = Query.builder();
// 向量相似度查询
queryBuilder.vector("embedding", queryVector)
.limit(0, options.getTopK())
.scoreField("vector_score");
// 添加过滤条件
if (options.hasFilters()) {
String filterExpression = buildFilterExpression(options.getFilters());
queryBuilder.filter(filterExpression);
}
// 添加排序
if (options.hasSortBy()) {
queryBuilder.sortBy(options.getSortBy(), options.getSortDirection());
}
// 设置返回字段
if (options.hasReturnFields()) {
queryBuilder.returnFields(options.getReturnFields());
}
return queryBuilder.build();
}
private String buildFilterExpression(Map<String, Object> filters) {
List<String> conditions = new ArrayList<>();
filters.forEach((field, value) -> {
if (value instanceof String) {
conditions.add(String.format("@%s:%s", field, escapeValue((String) value)));
} else if (value instanceof List) {
List<String> values = ((List<?>) value).stream()
.map(v -> escapeValue(v.toString()))
.collect(Collectors.toList());
conditions.add(String.format("@%s:{%s}", field, String.join(" | ", values)));
} else if (value instanceof NumberRange) {
NumberRange range = (NumberRange) value;
conditions.add(String.format("@%s:[%s %s]", field, range.getMin(), range.getMax()));
}
});
return String.join(" ", conditions);
}
// 批量向量搜索优化
public Map<String, List<VectorSearchResult>> batchSearch(List<String> queries,
VectorSearchOptions options) {
// 1. 批量向量化
List<float[]> queryVectors = embeddingClient.embedBatch(queries);
// 2. 并行搜索
Map<String, CompletableFuture<List<VectorSearchResult>>> futures = new HashMap<>();
for (int i = 0; i < queries.size(); i++) {
String query = queries.get(i);
float[] vector = queryVectors.get(i);
CompletableFuture<List<VectorSearchResult>> future = CompletableFuture
.supplyAsync(() -> performVectorSearch(vector, options));
futures.put(query, future);
}
// 3. 收集结果
Map<String, List<VectorSearchResult>> results = new HashMap<>();
futures.forEach((query, future) -> {
try {
results.put(query, future.get(5, TimeUnit.SECONDS));
} catch (Exception e) {
log.warn("批量搜索中查询失败: {}", query, e);
results.put(query, Collections.emptyList());
}
});
return results;
}
}
// 2. 向量索引管理器
@Component
public class VectorIndexManager {
private final RedisTemplate<String, Object> redisTemplate;
private final IndexMetrics indexMetrics;
public void createVectorIndex(IndexDefinition indexDef) {
try {
// 检查索引是否已存在
if (indexExists(indexDef.getName())) {
log.info("索引已存在: {}", indexDef.getName());
return;
}
// 创建索引
redisTemplate.execute((RedisCallback<Void>) connection -> {
((RedisSearchCommands) connection).createIndex(indexDef);
return null;
});
log.info("向量索引创建成功: {}", indexDef.getName());
// 记录索引指标
indexMetrics.recordIndexCreation(indexDef.getName());
} catch (Exception e) {
log.error("创建向量索引失败: {}", indexDef.getName(), e);
throw new VectorIndexException("索引创建失败", e);
}
}
public void optimizeIndex(String indexName) {
try {
// 执行索引优化
redisTemplate.execute((RedisCallback<Void>) connection -> {
// 重建索引以优化性能
((RedisSearchCommands) connection).rebuildIndex(indexName);
return null;
});
log.info("索引优化完成: {}", indexName);
} catch (Exception e) {
log.error("索引优化失败: {}", indexName, e);
}
}
public IndexStatistics getIndexStatistics(String indexName) {
return redisTemplate.execute((RedisCallback<IndexStatistics>) connection -> {
IndexInfo indexInfo = ((RedisSearchCommands) connection).getIndexInfo(indexName);
return IndexStatistics.builder()
.indexName(indexName)
.documentCount(indexInfo.getDocumentCount())
.averageDocumentSize(indexInfo.getAverageDocumentSize())
.memoryUsage(indexInfo.getMemoryUsage())
.indexingProgress(indexInfo.getIndexingProgress())
.build();
});
}
@Scheduled(fixedRate = 300000) // 每5分钟检查一次
public void monitorIndexHealth() {
List<String> indexNames = getAllIndexNames();
for (String indexName : indexNames) {
try {
IndexStatistics stats = getIndexStatistics(indexName);
// 检查索引健康状况
if (stats.getMemoryUsage() > getMemoryThreshold()) {
log.warn("索引内存使用过高: {} - {}MB",
indexName, stats.getMemoryUsage() / 1024 / 1024);
}
if (stats.getIndexingProgress() < 1.0) {
log.info("索引构建进行中: {} - {}%",
indexName, stats.getIndexingProgress() * 100);
}
// 更新监控指标
indexMetrics.updateIndexStats(indexName, stats);
} catch (Exception e) {
log.error("监控索引健康状况失败: {}", indexName, e);
}
}
}
}
// 3. 搜索结果缓存
@Component
public class SearchResultCache {
private final RedisTemplate<String, Object> redisTemplate;
private static final String CACHE_PREFIX = "search_cache:";
public List<VectorSearchResult> get(String cacheKey) {
try {
String key = CACHE_PREFIX + cacheKey;
Object cached = redisTemplate.opsForValue().get(key);
if (cached != null) {
return (List<VectorSearchResult>) cached;
}
} catch (Exception e) {
log.warn("获取搜索缓存失败: {}", cacheKey, e);
}
return null;
}
public void put(String cacheKey, List<VectorSearchResult> results, Duration ttl) {
try {
String key = CACHE_PREFIX + cacheKey;
redisTemplate.opsForValue().set(key, results, ttl);
} catch (Exception e) {
log.warn("保存搜索缓存失败: {}", cacheKey, e);
}
}
public void evict(String pattern) {
try {
Set<String> keys = redisTemplate.keys(CACHE_PREFIX + pattern);
if (!keys.isEmpty()) {
redisTemplate.delete(keys);
}
} catch (Exception e) {
log.warn("清除搜索缓存失败: {}", pattern, e);
}
}
}
🌐 Spring Boot集成实践
完整的向量存储服务
// 1. 向量文档服务
@Service
@Slf4j
public class VectorDocumentService {
private final RedisVectorStore vectorStore;
private final DocumentProcessor documentProcessor;
private final EmbeddingClient embeddingClient;
private final RedisTemplate<String, Object> redisTemplate;
public String addDocument(DocumentAddRequest request) {
try {
// 1. 文档预处理
ProcessedDocument processed = documentProcessor.process(request);
// 2. 生成向量
float[] embedding = embeddingClient.embed(processed.getContent());
// 3. 构建向量文档
VectorDocument document = VectorDocument.builder()
.id(UUID.randomUUID().toString())
.title(processed.getTitle())
.content(processed.getContent())
.embedding(embedding)
.category(request.getCategory())
.source(request.getSource())
.metadata(request.getMetadata())
.createdAt(Instant.now())
.build();
// 4. 存储到Redis
String key = buildDocumentKey(document.getId());
redisTemplate.opsForHash().putAll(key, documentToHash(document));
// 5. 更新索引
updateSearchIndex(document);
log.info("文档添加成功: {}", document.getId());
return document.getId();
} catch (Exception e) {
log.error("添加文档失败", e);
throw new DocumentStorageException("文档存储失败", e);
}
}
public void addDocumentsBatch(List<DocumentAddRequest> requests) {
if (requests.isEmpty()) return;
try {
// 1. 批量预处理
List<ProcessedDocument> processed = requests.parallelStream()
.map(documentProcessor::process)
.collect(Collectors.toList());
// 2. 批量向量化
List<String> contents = processed.stream()
.map(ProcessedDocument::getContent)
.collect(Collectors.toList());
List<float[]> embeddings = embeddingClient.embedBatch(contents);
// 3. 批量构建文档
List<VectorDocument> documents = new ArrayList<>();
for (int i = 0; i < processed.size(); i++) {
ProcessedDocument proc = processed.get(i);
DocumentAddRequest req = requests.get(i);
float[] embedding = embeddings.get(i);
VectorDocument document = VectorDocument.builder()
.id(UUID.randomUUID().toString())
.title(proc.getTitle())
.content(proc.getContent())
.embedding(embedding)
.category(req.getCategory())
.source(req.getSource())
.metadata(req.getMetadata())
.createdAt(Instant.now())
.build();
documents.add(document);
}
// 4. 批量存储
batchStoreDocuments(documents);
log.info("批量添加文档成功: {} 个", documents.size());
} catch (Exception e) {
log.error("批量添加文档失败", e);
throw new DocumentStorageException("批量文档存储失败", e);
}
}
private void batchStoreDocuments(List<VectorDocument> documents) {
// 使用Pipeline提高批量操作性能
redisTemplate.executePipelined(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
for (VectorDocument document : documents) {
String key = buildDocumentKey(document.getId());
Map<String, Object> hash = documentToHash(document);
hash.forEach((field, value) -> {
connection.hSet(key.getBytes(), field.getBytes(),
serializeValue(value));
});
}
return null;
}
});
}
public List<VectorSearchResult> searchDocuments(DocumentSearchRequest request) {
VectorSearchOptions options = VectorSearchOptions.builder()
.indexName("vector_search_idx")
.topK(request.getTopK())
.similarityThreshold(request.getSimilarityThreshold())
.filters(request.getFilters())
.returnFields(Arrays.asList("id", "title", "content", "category", "source"))
.build();
return vectorStore.searchSimilar(request.getQuery(), options);
}
public RecommendationResult getRecommendations(String documentId, int count) {
try {
// 1. 获取源文档
VectorDocument sourceDoc = getDocumentById(documentId);
if (sourceDoc == null) {
throw new DocumentNotFoundException("文档不存在: " + documentId);
}
// 2. 基于向量相似度推荐
VectorSearchOptions options = VectorSearchOptions.builder()
.indexName("vector_search_idx")
.topK(count + 1) // +1 排除自身
.filters(Map.of("category", sourceDoc.getCategory())) // 同类别推荐
.build();
List<VectorSearchResult> results = vectorStore.searchSimilarByVector(
sourceDoc.getEmbedding(), options);
// 3. 过滤掉源文档
List<VectorSearchResult> recommendations = results.stream()
.filter(result -> !result.getId().equals(documentId))
.collect(Collectors.toList());
return RecommendationResult.builder()
.sourceDocumentId(documentId)
.recommendations(recommendations)
.algorithm("vector_similarity")
.generatedAt(Instant.now())
.build();
} catch (Exception e) {
log.error("获取推荐失败: {}", documentId, e);
throw new RecommendationException("推荐生成失败", e);
}
}
private Map<String, Object> documentToHash(VectorDocument document) {
Map<String, Object> hash = new HashMap<>();
hash.put("id", document.getId());
hash.put("title", document.getTitle());
hash.put("content", document.getContent());
hash.put("embedding", serializeEmbedding(document.getEmbedding()));
hash.put("category", document.getCategory());
hash.put("source", document.getSource());
hash.put("created_at", document.getCreatedAt().toString());
hash.put("metadata", serializeMetadata(document.getMetadata()));
return hash;
}
private byte[] serializeEmbedding(float[] embedding) {
ByteBuffer buffer = ByteBuffer.allocate(embedding.length * 4);
for (float value : embedding) {
buffer.putFloat(value);
}
return buffer.array();
}
}
// 2. 文档处理器
@Component
public class DocumentProcessor {
private final TextSplitter textSplitter;
private final ContentExtractor contentExtractor;
public DocumentProcessor() {
this.textSplitter = new RecursiveCharacterTextSplitter(1000, 200);
this.contentExtractor = new TikaContentExtractor();
}
public ProcessedDocument process(DocumentAddRequest request) {
try {
// 1. 内容提取
String rawContent = extractContent(request);
// 2. 文本清理
String cleanedContent = cleanText(rawContent);
// 3. 文本分割（如果需要）
if (cleanedContent.length() > 2000) {
List<String> chunks = textSplitter.split(cleanedContent);
cleanedContent = selectBestChunk(chunks);
}
// 4. 标题提取
String title = extractTitle(request, cleanedContent);
return ProcessedDocument.builder()
.title(title)
.content(cleanedContent)
.originalLength(rawContent.length())
.processedLength(cleanedContent.length())
.build();
} catch (Exception e) {
log.error("文档处理失败", e);
throw new DocumentProcessingException("文档处理失败", e);
}
}
private String extractContent(DocumentAddRequest request) {
if (request.getContent() != null) {
return request.getContent();
} else if (request.getFilePath() != null) {
return contentExtractor.extractFromFile(request.getFilePath());
} else if (request.getUrl() != null) {
return contentExtractor.extractFromUrl(request.getUrl());
} else {
throw new IllegalArgumentException("必须提供内容、文件路径或URL");
}
}
private String cleanText(String text) {
return text
.replaceAll("\\s+", " ") // 合并多个空白字符
.replaceAll("[\\x00-\\x1F\\x7F]", "") // 移除控制字符
.trim();
}
private String selectBestChunk(List<String> chunks) {
// 选择最具代表性的文本块
return chunks.stream()
.max(Comparator.comparingInt(String::length))
.orElse(chunks.get(0));
}
private String extractTitle(DocumentAddRequest request, String content) {
if (request.getTitle() != null && !request.getTitle().isEmpty()) {
return request.getTitle();
}
// 从内容中提取标题（取前50个字符）
return content.length() > 50 ? content.substring(0, 50) + "..." : content;
}
}
🔄 RAG系统架构设计
检索增强生成系统实现
// 1. RAG服务主入口
@Service
@Slf4j
public class RAGService {
private final VectorDocumentService vectorDocumentService;
private final ChatClient chatClient;
private final ContextBuilder contextBuilder;
private final ResponsePostProcessor responsePostProcessor;
public RAGResponse processQuery(RAGRequest request) {
try {
// 1. 查询理解与重写
String enhancedQuery = enhanceQuery(request.getQuery(), request.getContext());
// 2. 向量检索
List<VectorSearchResult> retrievedDocs = retrieveRelevantDocuments(
enhancedQuery, request.getRetrievalOptions());
// 3. 上下文构建
String context = contextBuilder.buildContext(retrievedDocs, request);
// 4. LLM生成
String response = generateResponse(enhancedQuery, context, request);
// 5. 后处理
String finalResponse = responsePostProcessor.process(response, retrievedDocs);
return RAGResponse.builder()
.query(request.getQuery())
.enhancedQuery(enhancedQuery)
.response(finalResponse)
.retrievedDocuments(retrievedDocs)
.contextUsed(context)
.processingTime(calculateProcessingTime())
.confidence(calculateConfidence(response, retrievedDocs))
.build();
} catch (Exception e) {
log.error("RAG处理失败: {}", request.getQuery(), e);
throw new RAGProcessingException("RAG处理失败", e);
}
}
private String enhanceQuery(String originalQuery, Map<String, Object> context) {
// 查询扩展和重写
String prompt = String.format("""
请改进以下查询，使其更适合进行语义搜索：
原始查询：%s
上下文信息：%s
改进要求：
1. 添加相关的同义词和概念
2. 明确查询意图
3. 保持原意不变
只返回改进后的查询，不要其他内容：
""", originalQuery, formatContext(context));
try {
ChatResponse response = chatClient.call(new Prompt(prompt));
String enhanced = response.getResult().getOutput().getContent().trim();
// 如果增强失败，返回原查询
return enhanced.isEmpty() ? originalQuery : enhanced;
} catch (Exception e) {
log.warn("查询增强失败，使用原查询: {}", originalQuery, e);
return originalQuery;
}
}
private List<VectorSearchResult> retrieveRelevantDocuments(String query,
RetrievalOptions options) {
DocumentSearchRequest searchRequest = DocumentSearchRequest.builder()
.query(query)
.topK(options.getTopK())
.similarityThreshold(options.getSimilarityThreshold())
.filters(options.getFilters())
.build();
List<VectorSearchResult> results = vectorDocumentService.searchDocuments(searchRequest);
// 结果重排序
if (options.isEnableReranking()) {
results = rerankResults(query, results);
}
return results;
}
private List<VectorSearchResult> rerankResults(String query, List<VectorSearchResult> results) {
// 使用交叉编码器进行重排序
List<RerankingCandidate> candidates = results.stream()
.map(result -> RerankingCandidate.builder()
.id(result.getId())
.content(result.getContent())
.originalScore(result.getScore())
.build())
.collect(Collectors.toList());
// 计算重排序分数
List<Double> rerankingScores = calculateRerankingScores(query, candidates);
// 更新分数并重新排序
for (int i = 0; i < results.size(); i++) {
VectorSearchResult result = results.get(i);
double newScore = rerankingScores.get(i);
result.setScore(newScore);
result.setRerankingScore(newScore);
}
return results.stream()
.sorted((a, b) -> Double.compare(b.getScore(), a.getScore()))
.collect(Collectors.toList());
}
private String generateResponse(String query, String context, RAGRequest request) {
String systemPrompt = buildSystemPrompt(request.getResponseOptions());
String userPrompt = String.format("""
基于以下上下文信息回答问题：
上下文：
%s
问题：%s
请提供准确、有用的回答。如果上下文中没有相关信息，请明确说明。
""", context, query);
List<Message> messages = Arrays.asList(
new SystemMessage(systemPrompt),
new UserMessage(userPrompt)
);
ChatResponse response = chatClient.call(new Prompt(messages,
OpenAiChatOptions.builder()
.withModel("gpt-4")
.withTemperature(0.3)
.withMaxTokens(1000)
.build()));
return response.getResult().getOutput().getContent();
}
private String buildSystemPrompt(ResponseOptions options) {
StringBuilder systemPrompt = new StringBuilder();
systemPrompt.append("你是一个专业的问答助手。");
if (options.getResponseStyle() != null) {
systemPrompt.append("回答风格：").append(options.getResponseStyle()).append("。");
}
if (options.getMaxLength() > 0) {
systemPrompt.append("回答长度限制在").append(options.getMaxLength()).append("字以内。");
}
if (options.isIncludeSources()) {
systemPrompt.append("请在回答后列出相关的信息来源。");
}
return systemPrompt.toString();
}
}
// 2. 上下文构建器
@Component
public class ContextBuilder {
private static final int MAX_CONTEXT_LENGTH = 8000; // 控制上下文长度
public String buildContext(List<VectorSearchResult> documents, RAGRequest request) {
if (documents.isEmpty()) {
return "没有找到相关信息。";
}
StringBuilder contextBuilder = new StringBuilder();
int currentLength = 0;
for (int i = 0; i < documents.size(); i++) {
VectorSearchResult doc = documents.get(i);
String docSection = formatDocumentSection(doc, i + 1);
// 检查长度限制
if (currentLength + docSection.length() > MAX_CONTEXT_LENGTH) {
break;
}
contextBuilder.append(docSection).append("\n\n");
currentLength += docSection.length();
}
return contextBuilder.toString().trim();
}
private String formatDocumentSection(VectorSearchResult doc, int index) {
return String.format("""
[文档%d] 标题：%s
来源：%s
相关度：%.3f
内容：%s
""",
index,
doc.getTitle(),
doc.getSource(),
doc.getScore(),
truncateContent(doc.getContent(), 500)
);
}
private String truncateContent(String content, int maxLength) {
if (content.length() <= maxLength) {
return content;
}
// 在单词边界截断
int lastSpace = content.lastIndexOf(' ', maxLength);
if (lastSpace > maxLength * 0.8) {
return content.substring(0, lastSpace) + "...";
}
return content.substring(0, maxLength) + "...";
}
}
// 3. 响应后处理器
@Component
public class ResponsePostProcessor {
public String process(String response, List<VectorSearchResult> sources) {
// 1. 添加引用
String responseWithCitations = addCitations(response, sources);
// 2. 格式化输出
String formattedResponse = formatResponse(responseWithCitations);
// 3. 质量检查
validateResponseQuality(formattedResponse, sources);
return formattedResponse;
}
private String addCitations(String response, List<VectorSearchResult> sources) {
if (sources.isEmpty()) {
return response;
}
StringBuilder result = new StringBuilder(response);
result.append("\n\n**参考资料：**\n");
for (int i = 0; i < Math.min(sources.size(), 3); i++) {
VectorSearchResult source = sources.get(i);
result.append(String.format("%d. %s (相关度: %.2f)\n",
i + 1, source.getTitle(), source.getScore()));
}
return result.toString();
}
private String formatResponse(String response) {
// 改进格式和可读性
return response
.replaceAll("(?m)^([^\\n]*?)：", "**$1：**") // 加粗标题
.replaceAll("(?m)^(\\d+\\.)\\s*", "\n$1 ") // 格式化列表
.trim();
}
private void validateResponseQuality(String response, List<VectorSearchResult> sources) {
// 检查响应质量
if (response.length() < 10) {
log.warn("响应内容过短，可能质量不佳");
}
if (sources.isEmpty() && !response.contains("没有找到") && !response.contains("无法找到")) {
log.warn("没有检索到相关文档，但响应没有说明");
}
}
}
📊 性能调优与监控
Redis向量数据库性能优化
// 1. 性能监控和调优
@Component
public class VectorStorePerformanceMonitor {
private final MeterRegistry meterRegistry;
private final RedisTemplate<String, Object> redisTemplate;
// 监控指标
private final Counter searchRequestCounter;
private final Timer searchLatencyTimer;
private final Gauge indexSizeGauge;
private final Counter cacheHitCounter;
private final Counter cacheMissCounter;
public VectorStorePerformanceMonitor(MeterRegistry meterRegistry,
RedisTemplate<String, Object> redisTemplate) {
this.meterRegistry = meterRegistry;
this.redisTemplate = redisTemplate;
// 初始化指标
this.searchRequestCounter = Counter.builder("vector.search.requests")
.description("向量搜索请求总数")
.register(meterRegistry);
this.searchLatencyTimer = Timer.builder("vector.search.latency")
.description("向量搜索延迟")
.register(meterRegistry);
this.indexSizeGauge = Gauge.builder("vector.index.size")
.description("向量索引大小")
.register(meterRegistry, this, VectorStorePerformanceMonitor::getIndexSize);
this.cacheHitCounter = Counter.builder("vector.cache.hits")
.description("缓存命中次数")
.register(meterRegistry);
this.cacheMissCounter = Counter.builder("vector.cache.misses")
.description("缓存未命中次数")
.register(meterRegistry);
}
public void recordSearchRequest(String indexName, Duration latency, boolean cacheHit) {
searchRequestCounter.increment(Tags.of("index", indexName));
searchLatencyTimer.record(latency);
if (cacheHit) {
cacheHitCounter.increment();
} else {
cacheMissCounter.increment();
}
}
private double getIndexSize() {
try {
// 获取索引统计信息
Object info = redisTemplate.execute((RedisCallback<Object>) connection -> {
return connection.execute("FT.INFO", "vector_search_idx".getBytes());
});
// 解析索引大小（简化实现）
return parseIndexSize(info);
} catch (Exception e) {
log.warn("获取索引大小失败", e);
return 0.0;
}
}
@EventListener
public void handleSearchEvent(VectorSearchEvent event) {
recordSearchRequest(
event.getIndexName(),
event.getLatency(),
event.isCacheHit()
);
}
// 性能调优建议
@Scheduled(fixedRate = 300000) // 每5分钟分析一次
public void analyzePerformance() {
PerformanceAnalysis analysis = performPerformanceAnalysis();
if (analysis.hasIssues()) {
generateOptimizationRecommendations(analysis);
}
}
private PerformanceAnalysis performPerformanceAnalysis() {
// 获取性能指标
double avgLatency = searchLatencyTimer.mean(TimeUnit.MILLISECONDS);
double cacheHitRate = calculateCacheHitRate();
double indexMemoryUsage = getIndexMemoryUsage();
return PerformanceAnalysis.builder()
.averageLatency(avgLatency)
.cacheHitRate(cacheHitRate)
.indexMemoryUsage(indexMemoryUsage)
.timestamp(Instant.now())
.build();
}
private void generateOptimizationRecommendations(PerformanceAnalysis analysis) {
List<String> recommendations = new ArrayList<>();
if (analysis.getAverageLatency() > 100) {
recommendations.add("搜索延迟过高，建议优化HNSW参数或增加缓存");
}
if (analysis.getCacheHitRate() < 0.7) {
recommendations.add("缓存命中率偏低，建议调整缓存策略");
}
if (analysis.getIndexMemoryUsage() > 1000) { // MB
recommendations.add("索引内存使用过高，建议优化索引结构");
}
if (!recommendations.isEmpty()) {
log.info("性能优化建议: {}", String.join("; ", recommendations));
publishOptimizationAlert(recommendations);
}
}
}
// 2. 连接池优化
@Configuration
public class RedisConnectionOptimization {
@Bean
public LettuceConnectionFactory lettuceConnectionFactory(RedisVectorProperties properties) {
// 连接池配置
GenericObjectPoolConfig<StatefulRedisConnection<String, String>> poolConfig =
new GenericObjectPoolConfig<>();
poolConfig.setMaxTotal(properties.getPerformance().getMaxConnections());
poolConfig.setMaxIdle(properties.getPerformance().getMaxConnections() / 2);
poolConfig.setMinIdle(5);
poolConfig.setTestOnBorrow(true);
poolConfig.setTestOnReturn(true);
poolConfig.setTestWhileIdle(true);
poolConfig.setTimeBetweenEvictionRunsMillis(30000);
poolConfig.setBlockWhenExhausted(true);
poolConfig.setMaxWaitMillis(3000);
// 客户端配置
ClientOptions clientOptions = ClientOptions.builder()
.autoReconnect(true)
.disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
.timeoutOptions(TimeoutOptions.enabled(
properties.getPerformance().getConnectionTimeout()))
.build();
// 集群配置（如果需要）
RedisClusterConfiguration clusterConfig = new RedisClusterConfiguration();
// 配置集群节点...
LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
.clientOptions(clientOptions)
.commandTimeout(properties.getPerformance().getReadTimeout())
.poolConfig(poolConfig)
.build();
return new LettuceConnectionFactory(clusterConfig, clientConfig);
}
@Bean
public RedisTemplate<String, Object> optimizedRedisTemplate(
LettuceConnectionFactory connectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
// 序列化优化
Jackson2JsonRedisSerializer<Object> serializer =
new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper mapper = new ObjectMapper();
mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
mapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance,
ObjectMapper.DefaultTyping.NON_FINAL);
serializer.setObjectMapper(mapper);
template.setDefaultSerializer(serializer);
template.setKeySerializer(new StringRedisSerializer());
template.setHashKeySerializer(new StringRedisSerializer());
template.setValueSerializer(serializer);
template.setHashValueSerializer(serializer);
// 开启事务支持
template.setEnableTransactionSupport(true);
template.afterPropertiesSet();
return template;
}
}
// 3. 内存优化策略
@Component
public class VectorMemoryOptimizer {
private final RedisTemplate<String, Object> redisTemplate;
@Scheduled(cron = "0 0 3 * * ?") // 每天凌晨3点执行
public void optimizeMemoryUsage() {
try {
// 1. 清理过期向量
cleanupExpiredVectors();
// 2. 压缩索引
compressIndexes();
// 3. 内存碎片整理
defragmentMemory();
log.info("内存优化完成");
} catch (Exception e) {
log.error("内存优化失败", e);
}
}
private void cleanupExpiredVectors() {
// 清理超过保留期的向量文档
String script = """
local cursor = 0
local keys_deleted = 0
repeat
local result = redis.call('SCAN', cursor, 'MATCH', 'doc:*', 'COUNT', 100)
cursor = result[1]
local keys = result[2]
for i, key in ipairs(keys) do
local created_at = redis.call('HGET', key, 'created_at')
if created_at then
local age = os.time() - tonumber(created_at)
if age > 7776000 then -- 90天
redis.call('DEL', key)
keys_deleted = keys_deleted + 1
end
end
end
until cursor == '0'
return keys_deleted
""";
Long deletedCount = redisTemplate.execute((RedisCallback<Long>) connection -> {
return (Long) connection.eval(script.getBytes(), ReturnType.INTEGER, 0);
});
log.info("清理过期向量文档: {} 个", deletedCount);
}
private void compressIndexes() {
// 重建索引以优化内存布局
redisTemplate.execute((RedisCallback<Void>) connection -> {
connection.execute("FT.CONFIG", "SET".getBytes(), "FORK_GC_CLEAN_THRESHOLD".getBytes(), "0".getBytes());
return null;
});
}
private void defragmentMemory() {
// 执行内存碎片整理
redisTemplate.execute((RedisCallback<Void>) connection -> {
connection.execute("MEMORY", "PURGE".getBytes());
return null;
});
}
}
🚀 集群部署与扩展
Redis Cluster向量存储
# Redis Cluster配置
redis:
cluster:
nodes:
- redis-node-1:7000
- redis-node-2:7000
- redis-node-3:7000
- redis-node-4:7000
- redis-node-5:7000
- redis-node-6:7000
max-redirects: 3
password: ${REDIS_PASSWORD}
vector:
index-name: vector_search_idx
key-prefix: "doc:"
vector-dimensions: 1536
similarity-metric: COSINE
hnsw:
m: 16
ef-construction: 200
ef-runtime: 10
initial-capacity: 100000
performance:
batch-size: 200
max-connections: 50
connection-timeout: 5s
read-timeout: 30s
enable-pipelining: true
# 监控配置
management:
endpoints:
web:
exposure:
include: health,info,metrics,prometheus
metrics:
export:
prometheus:
enabled: true
水平扩展策略
// 集群感知的向量存储服务
@Service
public class ClusterAwareVectorStore {
private final RedisClusterConnection clusterConnection;
private final ConsistentHashRing hashRing;
public void addDocumentWithSharding(VectorDocument document) {
// 1. 计算分片
String shardKey = calculateShardKey(document);
// 2. 获取目标节点
RedisNode targetNode = hashRing.getNode(shardKey);
// 3. 存储到指定节点
storeToNode(document, targetNode);
}
public List<VectorSearchResult> searchAcrossCluster(String query,
VectorSearchOptions options) {
// 1. 并行搜索所有分片
List<CompletableFuture<List<VectorSearchResult>>> futures =
clusterConnection.getClusterNodes().stream()
.map(node -> CompletableFuture.supplyAsync(() ->
searchOnNode(query, options, node)))
.collect(Collectors.toList());
// 2. 合并结果
List<VectorSearchResult> allResults = futures.stream()
.map(CompletableFuture::join)
.flatMap(List::stream)
.collect(Collectors.toList());
// 3. 全局排序和截取
return allResults.stream()
.sorted((a, b) -> Double.compare(b.getScore(), a.getScore()))
.limit(options.getTopK())
.collect(Collectors.toList());
}
}
💼 实战案例分析
智能客服知识库
// 智能客服向量知识库实现
@Service
public class CustomerServiceKnowledgeBase {
private final VectorDocumentService vectorService;
private final RAGService ragService;
public void buildKnowledgeBase(List<FAQDocument> faqDocuments) {
// 批量构建FAQ知识库
List<DocumentAddRequest> requests = faqDocuments.stream()
.map(this::convertToDocumentRequest)
.collect(Collectors.toList());
vectorService.addDocumentsBatch(requests);
}
public CustomerServiceResponse handleCustomerQuery(String query, String sessionId) {
// 使用RAG回答客户问题
RAGRequest ragRequest = RAGRequest.builder()
.query(query)
.context(Map.of("sessionId", sessionId, "domain", "customer_service"))
.retrievalOptions(RetrievalOptions.builder()
.topK(5)
.similarityThreshold(0.7)
.enableReranking(true)
.build())
.responseOptions(ResponseOptions.builder()
.responseStyle("友好、专业")
.maxLength(300)
.includeSources(true)
.build())
.build();
RAGResponse response = ragService.processQuery(ragRequest);
return CustomerServiceResponse.builder()
.answer(response.getResponse())
.confidence(response.getConfidence())
.sources(response.getRetrievedDocuments())
.sessionId(sessionId)
.build();
}
}
📈 总结
本文深入探讨了Redis在AI向量数据库场景下的高性能实践，从基础架构到企业级应用，展示了Redis在现代AI应用中的核心价值。
🎯 核心优势
- 极致性能：内存存储 + 优化算法，实现毫秒级向量搜索
- 企业级特性：集群部署、高可用、监控告警完整方案
- AI原生设计：专门优化的向量索引和搜索算法
- 生态完整：丰富的客户端和工具支持
🚀 技术亮点
- HNSW优化：针对高维向量的高效近似搜索算法
- 批量处理：Pipeline和批量操作大幅提升吞吐量
- 智能缓存：多层缓存策略优化查询性能
- 动态扩展：基于一致性哈希的水平扩展方案
🔮 发展趋势
- 多模态支持：图像、音频向量的统一存储和检索
- 混合检索：向量检索与传统搜索的深度融合
- 边缘计算：轻量级向量存储在边缘设备的部署
- 自动调优：基于AI的参数自动优化
Redis作为向量数据库在AI应用中展现出了强大的潜力，通过本文的实践方案，开发者可以构建出高性能、可扩展的AI向量存储系统，为企业AI应用提供坚实的数据基础。
📚 参考资料
作者简介：一名正在实习的Java开发工程师，热爱技术分享，专注于性能优化和系统架构设计。
觉得有用的话可以点点赞 (/ω＼)，支持一下。
如果愿意的话关注一下。会对你有更多的帮助。
每周都会不定时更新哦 >人< 。
版权声明：本文为原创技术文章，转载请注明出处。
更多推荐
13
0- 0
已为社区贡献2条内容
所有评论(0)