1 章节标题
1.1 向量数据库概述
01.基础功能
a.核心特性
a.功能说明
向量数据库的定义和特点。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 向量数据库概述示例代码
print("向量数据库概述功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：向量数据库的定义和特点")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["向量数据库概述最佳实践1", "向量数据库概述最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
1.2 发展历程
01.基础功能
a.核心特性
a.功能说明
向量数据库的发展历史。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 发展历程示例代码
print("发展历程功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：向量数据库的发展历史")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["发展历程最佳实践1", "发展历程最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
1.3 应用场景
01.基础功能
a.核心特性
a.功能说明
向量数据库的典型应用。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 应用场景示例代码
print("应用场景功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：向量数据库的典型应用")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["应用场景最佳实践1", "应用场景最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
2 章节标题
2.1 向量表示
01.基础功能
a.核心特性
a.功能说明
数据的向量化表示方法。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 向量表示示例代码
print("向量表示功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：数据的向量化表示方法")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["向量表示最佳实践1", "向量表示最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
2.2 距离度量
01.基础功能
a.核心特性
a.功能说明
向量间距离度量方式。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 距离度量示例代码
print("距离度量功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：向量间距离度量方式")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["距离度量最佳实践1", "距离度量最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
2.3 相似度计算
01.基础功能
a.核心特性
a.功能说明
相似度计算方法。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 相似度计算示例代码
print("相似度计算功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：相似度计算方法")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["相似度计算最佳实践1", "相似度计算最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
2.4 关系数据库扩展
01.pgvector
a.特点介绍
a.功能说明
pgvector是PostgreSQL的向量扩展。SQL和向量结合。数据一致性强。适合小中规模。
b.代码示例
---
import psycopg2
from pgvector.psycopg2 import register_vector
conn = psycopg2.connect("...")
cur = conn.cursor()
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
cur.execute("CREATE TABLE items (id SERIAL, embedding VECTOR(128))")
cur.execute("CREATE INDEX ON items USING hnsw (embedding vector_l2_ops)")
print("✓ pgvector完成")
---
b.使用场景
a.功能说明
已有PostgreSQL、小规模应用、需要事务保证的场景。
b.代码示例
---
# pgvector SQL查询
cur.execute("SELECT * FROM items ORDER BY embedding <-> %s LIMIT 10", (query_vector,))
print("✓ pgvector查询完成")
---
02.Elasticsearch向量
a.特点介绍
a.功能说明
ES 8.0+支持向量检索。结合全文和向量搜索。
b.代码示例
---
from elasticsearch import Elasticsearch
es = Elasticsearch(["http://localhost:9200"])
# k-NN搜索
query = {"knn": {"field": "embedding", "query_vector": [0.1]*128, "k": 10}}
results = es.search(index="docs", body=query)
print("✓ ES向量检索完成")
---
b.混合检索
a.功能说明
关键词+向量混合排序。提升检索准确性。
b.代码示例
---
# 混合检索
print("✓ 混合检索完成")
---
3 章节标题
3.1 产品对比
01.基础功能
a.核心特性
a.功能说明
主流向量数据库产品对比。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 产品对比示例代码
print("产品对比功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：主流向量数据库产品对比")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["产品对比最佳实践1", "产品对比最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
3.2 性能评测
01.基础功能
a.核心特性
a.功能说明
向量数据库性能基准测试。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 性能评测示例代码
print("性能评测功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：向量数据库性能基准测试")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["性能评测最佳实践1", "性能评测最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
3.3 选型建议
01.基础功能
a.核心特性
a.功能说明
如何选择合适的向量数据库。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 选型建议示例代码
print("选型建议功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：如何选择合适的向量数据库")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["选型建议最佳实践1", "选型建议最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
3.4 扩展性
01.分布式能力
a.横向扩展
a.功能说明
Milvus和Pinecone支持分布式。水平扩展QPS和容量。
b.代码示例
---
# Milvus分布式
# QueryNode: 多实例扩展QPS
# DataNode: 多实例扩展写入
print("✓ 分布式扩展完成")
---
b.纵向扩展
a.功能说明
升级硬件提升单机性能。CPU、内存、GPU。
b.代码示例
---
# GPU加速
"""
gpu_index = faiss.index_cpu_to_gpu(res, 0, cpu_index)
# 性能提升10倍
"""
print("✓ 纵向扩展完成")
---
02.高可用
a.副本机制
a.功能说明
多副本保证可用性。主从复制和故障转移。
b.代码示例
---
# Milvus副本
collection.load(replica_number=3)
print("✓ 副本机制完成")
---
b.故障恢复
a.功能说明
自动故障检测和切换。最小化服务中断。
b.代码示例
---
# 故障转移
if not health_check(primary):
switch_to_backup()
print("✓ 故障恢复完成")
---
4 章节标题
4.1 文本Embedding
01.基础功能
a.核心特性
a.功能说明
文本向量化Embedding模型。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 文本Embedding示例代码
print("文本Embedding功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：文本向量化Embedding模型")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["文本Embedding最佳实践1", "文本Embedding最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
4.2 图像Embedding
01.基础功能
a.核心特性
a.功能说明
图像向量化Embedding模型。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 图像Embedding示例代码
print("图像Embedding功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：图像向量化Embedding模型")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["图像Embedding最佳实践1", "图像Embedding最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
4.3 多模态Embedding
01.基础功能
a.核心特性
a.功能说明
多模态数据Embedding。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 多模态Embedding示例代码
print("多模态Embedding功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：多模态数据Embedding")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["多模态Embedding最佳实践1", "多模态Embedding最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
4.4 资源消耗
01.内存占用
a.对比分析
a.功能说明
HNSW内存占用最大（1.8-2.3倍）。IVF_PQ最节省（0.2倍）。
b.代码示例
---
# 内存估算
"""
100万向量，128维:
- Flat: 512MB
- HNSW: 900MB
- IVF_PQ: 100MB
"""
print("✓ 内存对比完成")
---
b.内存优化
a.功能说明
使用PQ压缩、Float16量化、mmap节省内存。
b.代码示例
---
# PQ压缩
index_pq = faiss.IndexIVFPQ(quantizer, dim, nlist, m=8, nbits=8)
# 压缩比5:1
print("✓ 内存优化完成")
---
02.CPU消耗
a.负载分析
a.功能说明
单查询CPU 5-10%，高并发可达100%。索引构建CPU密集。
b.代码示例
---
import psutil
cpu_percent = psutil.cpu_percent(interval=1)
print(f"CPU: {cpu_percent}%")
---
b.优化策略
a.功能说明
GPU卸载计算、批量查询、连接池降低CPU负载。
b.代码示例
---
# GPU加速降低CPU
gpu_index = faiss.index_cpu_to_gpu(res, 0, index)
# CPU负载降低80%
print("✓ CPU优化完成")
---
5 章节标题
5.1 RAG架构
01.基础功能
a.核心特性
a.功能说明
检索增强生成架构设计。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# RAG架构示例代码
print("RAG架构功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：检索增强生成架构设计")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["RAG架构最佳实践1", "RAG架构最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
5.2 语义搜索
01.基础功能
a.核心特性
a.功能说明
基于向量的语义搜索。提供完整的功能支持。适合生产环境使用。
b.代码示例
---
# 语义搜索示例代码
print("语义搜索功能实现")
---
b.使用场景
a.功能说明
适用于各种应用场景。灵活配置。高性能。
b.代码示例
---
# 使用场景示例
print("适用场景：基于向量的语义搜索")
---
02.高级应用
a.配置优化
a.功能说明
根据实际需求优化配置。提升性能和稳定性。
b.代码示例
---
# 配置优化示例
config = {"enabled": True, "optimized": True}
print(f"配置: {config}")
---
b.最佳实践
a.功能说明
遵循最佳实践。确保系统稳定可靠。
b.代码示例
---
# 最佳实践示例
best_practices = ["语义搜索最佳实践1", "语义搜索最佳实践2"]
for practice in best_practices:
print(f"- {practice}")
---
5.3 云平台支持
01.公有云支持
a.AWS集成
a.功能说明
AWS提供多种向量数据库服务。Amazon OpenSearch Service支持向量检索。RDS PostgreSQL可安装pgvector。托管服务简化运维。
b.代码示例
---
# AWS OpenSearch向量检索
from opensearchpy import OpenSearch
client = OpenSearch(
hosts=[{'host': 'domain.region.es.amazonaws.com', 'port': 443}],
http_auth=('user', 'pass'), use_ssl=True
)
# 创建向量索引
index_body = {
"settings": {"index": {"knn": True}},
"mappings": {"properties": {"embedding": {"type": "knn_vector", "dimension": 128}}}
}
client.indices.create(index="vectors", body=index_body)
# k-NN搜索
query = {"size": 10, "query": {"knn": {"embedding": {"vector": [0.1]*128, "k": 10}}}}
results = client.search(index="vectors", body=query)
print("✓ AWS OpenSearch完成")
---
b.阿里云/腾讯云
a.功能说明
国内公有云支持。阿里云DashVector、腾讯云向量数据库。国内网络环境优化。
b.代码示例
---
# 国内云平台示例（伪代码）
# 阿里云DashVector / 腾讯云向量数据库API调用示例
print("✓ 国内云平台支持完成")
---
02.混合云部署
a.跨云方案
a.功能说明
混合云和多云部署策略。数据同步和灾备。跨地域部署优化延迟。
b.代码示例
---
# 混合云数据同步
from pymilvus import Collection, connections
connections.connect(alias="local", host="on-premise", port="19530")
local_coll = Collection("demo", using="local")
def sync_to_cloud():
results = local_coll.query(expr="timestamp > last_sync", output_fields=["id", "embedding"])
# 同步到云端
print(f"已同步 {len(results)} 条")
print("✓ 混合云同步完成")
---
b.灾备策略
a.功能说明
故障转移和灾备恢复。主备切换机制。
b.代码示例
---
# 故障转移
class FailoverManager:
def __init__(self, primary, backup):
self.primary = primary
self.backup = backup
def get_active_host(self):
if self.health_check(self.primary):
return self.primary
return self.backup
print("✓ 故障转移完成")
---
6 章节标题
6.1 按数据规模选型
01.小规模数据
a.10万以内
a.功能说明
小规模数据选型建议。Chroma或FAISS即可满足。开发简单成本低。
b.代码示例
---
# 小规模向量数据库选择
"""
数据量: <10万向量
推荐方案:
1. Chroma: 开箱即用，LangChain集成好
2. FAISS: 内存索引，速度最快
3. HNSWLib: 轻量级，易集成
不推荐: Milvus, Pinecone（杀鸡用牛刀）
"""
import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("demo")
collection.add(ids=["1"], embeddings=[[0.1]*128], documents=["text"])
results = collection.query(query_embeddings=[[0.1]*128], n_results=10)
print("✓ 小规模完成")
---
b.100万级别
a.功能说明
中等规模选型。Qdrant或单机Milvus。需要持久化和过滤。
b.代码示例
---
# 100万级别推荐Qdrant
from qdrant_client import QdrantClient
client = QdrantClient(path="./qdrant_storage")
print("✓ 中等规模完成")
---
02.大规模数据
a.千万级别
a.功能说明
千万级数据需要分布式。Milvus最佳选择。支持横向扩展。
b.代码示例
---
# 千万级用Milvus
from pymilvus import connections, Collection
connections.connect(host="localhost", port="19530")
collection = Collection("large_dataset")
collection.load(replica_number=3)
print("✓ 千万级完成")
---
b.亿级以上
a.功能说明
亿级数据必须分布式+优化。Milvus集群或Pinecone。DiskANN突破内存限制。
b.代码示例
---
# 亿级用Milvus DiskANN
"""
collection.create_index(
field_name="embedding",
index_params={"index_type": "DISKANN", "metric_type": "L2"}
)
"""
print("✓ 亿级以上完成")
---
6.2 按性能要求选型
01.QPS要求
a.低QPS场景
a.功能说明
QPS<100场景。任何向量库都可满足。选择简单易用的。
b.代码示例
---
# 低QPS场景：选Chroma/pgvector
print("低QPS(<100)：Chroma、pgvector都可")
---
b.高QPS场景
a.功能说明
QPS>1000场景。需要FAISS、Milvus或Qdrant。GPU加速更佳。
b.代码示例
---
# 高QPS场景：Milvus+GPU
"""
gpu_index_params = {"index_type": "GPU_IVF_FLAT", "metric_type": "L2"}
collection.create_index("embedding", gpu_index_params)
# GPU可提升10倍QPS
"""
print("高QPS(>1000)：Milvus GPU / FAISS")
---
02.延迟要求
a.低延迟场景
a.功能说明
P99<20ms要求。FAISS或HNSWLib最佳。纯内存索引。
b.代码示例
---
# 低延迟：FAISS
import faiss
index = faiss.IndexHNSWFlat(128, 32)
# P99延迟 <15ms
print("低延迟(<20ms)：FAISS、HNSWLib")
---
b.延迟容忍场景
a.功能说明
可接受50-100ms延迟。云服务Pinecone、Zilliz Cloud。
b.代码示例
---
# 云服务可接受更高延迟
print("延迟容忍(50-100ms)：Pinecone、Zilliz Cloud")
---
6.3 按部署环境选型
01.本地部署
a.开发测试
a.功能说明
本地开发测试环境。Chroma最简单。Docker一键部署Qdrant。
b.代码示例
---
# 本地开发：Chroma
import chromadb
client = chromadb.PersistentClient(path="./local_db")
print("本地开发：Chroma（零配置）")
---
b.生产环境
a.功能说明
本地生产环境。Milvus自建集群。完全可控。
b.代码示例
---
# 生产环境：Milvus集群
"""
docker-compose部署Milvus
- QueryNode: 多实例
- DataNode: 多实例
- MinIO: 对象存储
"""
print("本地生产：Milvus集群")
---
02.云端部署
a.公有云
a.功能说明
公有云托管服务。Pinecone、Zilliz Cloud。零运维。
b.代码示例
---
# 云服务：Pinecone
import pinecone
pinecone.init(api_key="...", environment="...")
index = pinecone.Index("demo")
print("公有云：Pinecone（托管）")
---
b.信创环境
a.功能说明
信创环境适配。Milvus+达梦+Ollama。国产化全栈。
b.代码示例
---
# 信创：Milvus+达梦+Ollama
"""
- Milvus: 开源向量库
- 达梦DM8: 元数据存储
- Ollama: 本地LLM
- 昇腾NPU: 加速推理
"""
print("信创环境：Milvus+达梦+Ollama")
---
6.4 按预算成本选型
01.预算有限
a.开源方案
a.功能说明
预算有限选开源。Chroma、Qdrant、Milvus都免费。自建节省成本。
b.代码示例
---
# 零成本方案
"""
开源免费方案:
1. Chroma: 完全免费，适合小规模
2. Qdrant: 开源，Docker部署
3. Milvus: 开源，适合大规模
4. FAISS: Meta开源，纯内存
成本: 仅服务器成本（$50-200/月）
"""
print("开源方案：Chroma/Qdrant/Milvus")
---
b.低成本云服务
a.功能说明
低预算云服务。Pinecone免费层或Zilliz Serverless。按使用付费。
b.代码示例
---
# Pinecone免费层
"""
Pinecone Starter:
- 免费100K向量
- 适合原型验证
- 超出按需付费
"""
print("低成本云：Pinecone免费层")
---
02.预算充足
a.企业级服务
a.功能说明
预算充足选企业级。Pinecone、Zilliz Cloud付费版。SLA保证和技术支持。
b.代码示例
---
# 企业级服务
"""
Pinecone Standard:
- $70+/月
- 99.9% SLA
- 专属支持
Zilliz Cloud Dedicated:
- 定制价格
- 私有集群
- 7x24支持
"""
print("企业级：Pinecone/Zilliz付费版")
---
b.混合方案
a.功能说明
预算平衡混合方案。核心数据自建，扩展用云。成本优化。
b.代码示例
---
# 混合方案节省成本
"""
成本对比（100万向量）:
- 纯自建: $500/月
- 纯云服务: $300/月
- 混合方案: $350/月
混合策略:
- 热数据：本地Milvus（快）
- 冷数据：云端Pinecone（便宜）
"""
print("混合方案：自建+云服务")
---
7 章节标题
7.1 RAG问答系统
01.方案选择
a.小规模RAG
a.功能说明
文档量<1万，Chroma最佳。LangChain集成简单。适合MVP。
b.代码示例
---
from langchain.vectorstores import Chroma
vectorstore = Chroma.from_documents(documents, embeddings, persist_directory="./db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
print("小规模RAG：Chroma")
---
b.生产级RAG
a.功能说明
文档量>10万，Milvus或Qdrant。支持复杂过滤和高并发。
b.代码示例
---
from langchain.vectorstores import Milvus
vectorstore = Milvus.from_documents(documents, embeddings)
print("生产级RAG：Milvus")
---
02.优化策略
a.检索优化
a.功能说明
混合检索提升准确率。向量+关键词。重排序refine结果。
b.代码示例
---
docs = vectorstore.max_marginal_relevance_search(query, k=5)
print("检索优化：MMR")
---
b.缓存加速
a.功能说明
常见问题缓存。减少向量检索次数。
b.代码示例
---
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_search(query):
return vectorstore.similarity_search(query)
print("缓存完成")
---
7.2 推荐系统
01.协同过滤
a.向量化推荐
a.功能说明
用户和物品向量化。向量检索实现推荐。
b.代码示例
---
user_vector = get_user_embedding(user_id)
items = vectorstore.similarity_search_by_vector(user_vector, k=10)
print("推荐完成")
---
b.实时推荐
a.功能说明
实时更新用户行为。毫秒级推荐响应。
b.代码示例
---
print("实时推荐完成")
---
02.冷启动
a.内容推荐
a.功能说明
新用户基于内容推荐。解决冷启动问题。
b.代码示例
---
item_features = extract_features(item)
similar = vectorstore.similarity_search_by_vector(item_features, k=10)
print("内容推荐完成")
---
b.混合策略
a.功能说明
协同过滤+内容推荐。
b.代码示例
---
if user_history < 5:
results = content_based()
else:
results = collaborative()
print("混合完成")
---
7.3 图像检索
01.以图搜图
a.特征提取
a.功能说明
使用CLIP提取图像特征。支持语义检索。
b.代码示例
---
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('clip-ViT-B-32')
vector = model.encode(image)
results = vectorstore.similarity_search_by_vector(vector, k=10)
print("以图搜图完成")
---
b.跨模态检索
a.功能说明
文本搜图片、图片搜文本。
b.代码示例
---
text_vector = model.encode("红色汽车")
images = vectorstore.similarity_search_by_vector(text_vector, k=5)
print("跨模态完成")
---
02.去重检测
a.图像去重
a.功能说明
识别重复图片。优化存储。
b.代码示例
---
duplicates = vectorstore.similarity_search_by_vector(img_vector, k=5)
print("去重完成")
---
b.版权检测
a.功能说明
检测侵权图片。相似度匹配。
b.代码示例
---
matches = copyright_db.similarity_search_by_vector(suspect_vector, k=1)
print("版权检测完成")
---
7.4 语义搜索
01.企业搜索
a.知识库检索
a.功能说明
企业文档向量化。语义搜索突破关键词限制。
b.代码示例
---
docs = load_company_docs()
vectorstore = Milvus.from_documents(docs, embeddings)
results = vectorstore.similarity_search("如何申请年假", k=3)
print("企业搜索完成")
---
b.权限控制
a.功能说明
基于角色的检索结果过滤。
b.代码示例
---
results = vectorstore.similarity_search(query, filter={"access_level": user_role})
print("权限控制完成")
---
02.电商搜索
a.商品检索
a.功能说明
商品向量化。支持模糊搜索和同义词。
b.代码示例
---
products = vectorstore.similarity_search("手机充电宝", k=20)
print("商品检索完成")
---
b.个性化排序
a.功能说明
结合用户画像重排序。
b.代码示例
---
base_results = vectorstore.similarity_search(query, k=50)
final = rerank(base_results, user_vector)
print("个性化排序完成")
---
8 章节标题
8.1 国产化支持
01.数据库适配
a.达梦数据库
a.功能说明
达梦DM8适配。BLOB存储向量。自实现相似度计算。
b.代码示例
---
import dmPython
conn = dmPython.connect(host="localhost", user="SYSDBA")
# 使用BLOB存储向量
print("达梦适配完成")
---
b.人大金仓
a.功能说明
人大金仓KingBase。基于PostgreSQL。可安装pgvector扩展。
b.代码示例
---
# KingBase兼容pgvector
print("人大金仓适配完成")
---
02.硬件适配
a.昇腾NPU
a.功能说明
华为昇腾NPU加速。CANN算子库。替代CUDA。
b.代码示例
---
# 昇腾NPU推理
print("昇腾NPU适配完成")
---
b.飞腾/鲲鹏CPU
a.功能说明
ARM架构CPU适配。Docker镜像编译。性能优化。
b.代码示例
---
# ARM架构部署Milvus
print("飞腾/鲲鹏适配完成")
---
8.2 昇腾NPU适配
01.推理加速
a.CANN集成
a.功能说明
CANN（Compute Architecture for Neural Networks）集成。替代CUDA。
b.代码示例
---
# 昇腾CANN推理
print("CANN集成完成")
---
b.性能对比
a.功能说明
昇腾910 vs NVIDIA V100性能对比。推理速度接近。
b.代码示例
---
# 性能测试
print("昇腾性能对比完成")
---
02.部署方案
a.容器部署
a.功能说明
Docker镜像适配昇腾。K8S调度NPU资源。
b.代码示例
---
# Dockerfile for Ascend
print("昇腾容器部署完成")
---
b.优化建议
a.功能说明
算子优化、批处理、FP16量化。
b.代码示例
---
# 昇腾优化
print("优化完成")
---
8.3 达梦数据库集成
01.存储方案
a.BLOB存储
a.功能说明
向量序列化为BLOB。达梦表存储。
b.代码示例
---
import dmPython
import pickle
vector_blob = pickle.dumps(vector)
cursor.execute("INSERT INTO vectors (id, data) VALUES (?, ?)", (1, vector_blob))
print("BLOB存储完成")
---
b.检索实现
a.功能说明
应用层实现相似度计算。性能有限，适合小规模。
b.代码示例
---
def cosine_similarity(v1, v2):
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
print("检索实现完成")
---
02.混合方案
a.达梦+Milvus
a.功能说明
达梦存业务数据，Milvus存向量。混合架构。
b.代码示例
---
# 业务数据：达梦
# 向量数据：Milvus
print("混合方案完成")
---
b.性能优化
a.功能说明
索引优化、批量查询、缓存策略。
b.代码示例
---
# 达梦索引优化
print("性能优化完成")
---
9 章节标题
9.1 数据迁移
01.迁移策略
a.全量迁移
a.功能说明
一次性迁移所有数据。适合小规模或停机窗口。
b.代码示例
---
# 全量迁移
source_data = source_db.query_all()
target_db.insert_batch(source_data)
print("全量迁移完成")
---
b.增量迁移
a.功能说明
先迁移存量，再同步增量。最小化停机时间。
b.代码示例
---
# 增量迁移
while True:
new_data = source_db.query_since(last_timestamp)
target_db.insert_batch(new_data)
print("增量迁移完成")
---
02.数据验证
a.一致性检查
a.功能说明
对比源和目标数据。确保完整性。
b.代码示例
---
source_count = source_db.count()
target_count = target_db.count()
assert source_count == target_count
print("一致性检查完成")
---
b.回滚方案
a.功能说明
迁移失败回滚机制。备份原数据。
b.代码示例
---
# 备份和回滚
backup = source_db.export()
# 迁移失败时恢复
print("回滚方案完成")
---
9.2 代码适配
01.API迁移
a.接口映射
a.功能说明
不同向量库API映射。统一抽象层。
b.代码示例
---
class VectorDBAdapter:
def search(self, vector, k):
if self.db_type == "milvus":
return self.milvus_search(vector, k)
elif self.db_type == "qdrant":
return self.qdrant_search(vector, k)
print("API映射完成")
---
b.功能对齐
a.功能说明
目标库不支持的功能替代方案。
b.代码示例
---
# 功能适配
print("功能对齐完成")
---
02.测试验证
a.功能测试
a.功能说明
迁移后功能验证。回归测试。
b.代码示例
---
def test_search():
results = vectordb.search(query_vector, k=10)
assert len(results) == 10
print("功能测试完成")
---
b.性能测试
a.功能说明
对比迁移前后性能。QPS、延迟、准确率。
b.代码示例
---
# 性能对比
print("性能测试完成")
---
9.3 性能对比测试
01.基准测试
a.QPS测试
a.功能说明
迁移前后QPS对比。确保性能不降级。
b.代码示例
---
import time
start = time.time()
for i in range(1000):
vectordb.search(query, k=10)
qps = 1000 / (time.time() - start)
print(f"QPS: {qps}")
---
b.延迟测试
a.功能说明
P50、P99延迟对比。
b.代码示例
---
latencies = []
for query in queries:
start = time.time()
vectordb.search(query, k=10)
latencies.append(time.time() - start)
print(f"P99: {np.percentile(latencies, 99)}")
---
02.准确率验证
a.Recall测试
a.功能说明
迁移后Recall@K验证。确保准确率一致。
b.代码示例
---
# Recall测试
print("Recall测试完成")
---
b.结果对比
a.功能说明
抽样对比迁移前后检索结果。
b.代码示例
---
# 结果对比
print("结果对比完成")
---
10 章节标题
10.1 技术选型决策树
01.决策流程
a.需求评估
a.功能说明
数据规模、性能要求、预算、团队技能评估。
b.代码示例
---
"""
决策树:
数据量 < 10万?
Yes → Chroma/FAISS
No → 继续
需要分布式?
Yes → Milvus/Pinecone
No → Qdrant
预算有限?
Yes → 开源方案
No → 云服务
"""
print("决策流程完成")
---
b.场景匹配
a.功能说明
RAG、推荐、图像检索等场景推荐方案。
b.代码示例
---
# 场景推荐
print("场景匹配完成")
---
02.评估标准
a.技术指标
a.功能说明
QPS、延迟、Recall、成本、运维复杂度。
b.代码示例
---
# 评估矩阵
print("技术指标评估完成")
---
b.POC验证
a.功能说明
小规模验证候选方案。实测性能和易用性。
b.代码示例
---
# POC测试
print("POC验证完成")
---
10.2 混合使用策略
01.热冷分离
a.数据分层
a.功能说明
热数据本地高性能库，冷数据云端归档。
b.代码示例
---
# 热数据：本地Milvus
# 冷数据：云端Pinecone
if data_age < 7:
results = hot_db.search(query)
else:
results = cold_db.search(query)
print("数据分层完成")
---
b.成本优化
a.功能说明
混合方案降低总体成本。平衡性能和预算。
b.代码示例
---
# 成本优化
print("成本优化完成")
---
02.多库协同
a.并行检索
a.功能说明
多个向量库并行检索，合并结果。
b.代码示例
---
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=3) as executor:
results1 = executor.submit(db1.search, query)
results2 = executor.submit(db2.search, query)
merged = merge_results([results1.result(), results2.result()])
print("并行检索完成")
---
b.智能路由
a.功能说明
根据查询特征路由到最优数据库。
b.代码示例
---
if query_type == "image":
results = image_db.search(query)
else:
results = text_db.search(query)
print("智能路由完成")
---
10.3 成本优化
01.资源优化
a.索引选择
a.功能说明
根据场景选择合适索引。PQ压缩节省内存。
b.代码示例
---
# IVF_PQ高压缩比
index_params = {"index_type": "IVF_PQ", "params": {"m": 8, "nbits": 8}}
# 内存节省80%
print("索引优化完成")
---
b.硬件配置
a.功能说明
合理配置CPU、内存、GPU。避免资源浪费。
b.代码示例
---
# 硬件配置建议
"""
小规模: 4核8G
中规模: 8核16G
大规模: 16核32G + GPU
"""
print("硬件配置完成")
---
02.运维优化
a.监控告警
a.功能说明
QPS、延迟、错误率监控。Prometheus+Grafana。
b.代码示例
---
# Prometheus监控
print("监控告警完成")
---
b.自动扩缩容
a.功能说明
基于负载自动扩缩容。降低闲时成本。
b.代码示例
---
# K8S HPA自动扩容
print("自动扩缩容完成")
---