当前位置： 首页 > news >正文

《AI大模型应知应会100篇》第60篇：Pinecone 与 Milvus，向量数据库在大模型应用中的作用

news 来源：原创 2025/5/13 9:14:32

第60篇：Pinecone与Milvus，向量数据库在大模型应用中的作用

摘要

本文将系统比较Pinecone与Milvus两大主流向量数据库的技术特点、性能表现和应用场景，提供详细的接入代码和最佳实践，帮助开发者为大模型应用选择并优化向量存储解决方案。

在这里插入图片描述

引言：为什么需要向量数据库？

随着AI大模型的发展，向量数据（如文本、图像的嵌入表示）成为处理非结构化数据的核心方式。然而，传统的数据库无法高效地处理高维向量的相似性搜索问题。因此，向量数据库应运而生。

什么是向量数据库？

向量数据库是一种专门用于存储和检索高维向量数据的数据库。它支持快速执行近似最近邻搜索（ANN），即找出与目标向量最接近的一组向量。

向量数据库的应用场景

  • 推荐系统：基于用户行为向量进行个性化推荐。
  • 语义搜索：通过文本嵌入实现文档或网页内容的语义匹配。
  • 图像检索：利用图像特征向量进行相似图片查找。
  • RAG（Retrieval-Augmented Generation）：结合知识库与大模型生成更准确的回答。

核心概念与知识点

1. 向量数据库基础

向量嵌入基础

向量嵌入是将非结构化数据（如文本、图像）转换为固定维度的数值向量的过程。例如：

  • 使用 text-embedding-ada-002 将文本转换为 1536 维向量。
  • 使用 CLIP 模型将图像转换为 512 维向量。
from openai import OpenAIclient = OpenAI()
response = client.embeddings.create(input="这是一个示例文本",model="text-embedding-ada-002"
)
embedding = response.data[0].embedding
print(len(embedding))  # 输出: 1536
相似性搜索原理

常见的相似性度量方法包括：

  • 余弦相似度（Cosine Similarity）：衡量两个向量之间的夹角。
  • 欧氏距离（Euclidean Distance）：衡量两个向量之间的直线距离。

通常使用余弦相似度来判断两个向量是否“语义上”相似。

索引类型对比
索引类型特点适用场景
HNSW (Hierarchical Navigable Small World)高效且支持动态更新实时推荐系统
IVF (Inverted File Index)快速但不支持频繁更新批量处理场景
FLAT精确但慢小规模数据
向量数据库与传统数据库的区别
对比项传统数据库向量数据库
数据类型结构化数据（如整数、字符串）高维向量
查询方式精确匹配（如 SQL WHERE 条件）近似最近邻搜索（ANN）
性能插入/查询速度快支持大规模向量数据的高效检索

Pinecone实战应用

账户设置与索引创建

Pinecone 是一个云托管的向量数据库服务，适合不想自己部署服务器的开发者。

安装依赖
pip install pinecone-client openai
初始化 Pinecone 并创建索引
import pinecone
from openai import OpenAI# 初始化 Pinecone
pinecone.init(api_key="your-api-key", environment="your-environment")# 创建索引（如果不存在）
if "knowledge-base" not in pinecone.list_indexes():pinecone.create_index(name="knowledge-base",dimension=1536,  # OpenAI ada-002 的输出维度metric="cosine")# 连接到索引
index = pinecone.Index("knowledge-base")
插入向量
client = OpenAI()# 生成向量嵌入
embeddings = client.embeddings.create(input="需要向量化的文本内容",model="text-embedding-ada-002"
).data[0].embedding# 插入向量
index.upsert(vectors=[{"id": "doc1","values": embeddings,"metadata": {"source": "article", "author": "Zhang San"}}]
)
相似性搜索
results = index.query(vector=embeddings,top_k=3,include_metadata=True
)for result in results['matches']:print(f"ID: {result['id']}, Score: {result['score']}, Metadata: {result['metadata']}")
输出示例
ID: doc1, Score: 0.987, Metadata: {'source': 'article', 'author': 'Zhang San'}
ID: doc2, Score: 0.876, Metadata: {'source': 'blog', 'author': 'Li Si'}
ID: doc3, Score: 0.765, Metadata: {'source': 'wiki', 'author': 'Wang Wu'}

Milvus实战应用

部署与集群配置

Milvus 是一个开源的向量数据库，支持本地部署和分布式架构，适合对数据隐私要求较高的企业。

安装 Milvus

使用 Docker 快速启动：

docker run -d --name milvusdb -p 19530:19530 milvusdb/milvus:v2.4.3
Python 客户端连接
pip install pymilvus numpy
连接 Milvus 并插入数据
from pymilvus import connections, Collection, utility
import numpy as np# 连接 Milvus
connections.connect(host='localhost', port='19530')# 创建集合
collection_name = "document_store"if utility.has_collection(collection_name):collection = Collection(collection_name)
else:from pymilvus import FieldType, CollectionSchema, DataTypefields = [{"name": "id", "type": DataType.INT64, "is_primary": True},{"name": "embedding", "type": DataType.FLOAT_VECTOR, "dim": 1536},{"name": "source", "type": DataType.VARCHAR, "max_length": 256},{"name": "author", "type": DataType.VARCHAR, "max_length": 256}]schema = CollectionSchema(fields, description="Document Store")collection = Collection(collection_name, schema=schema)# 插入数据
import randomnum_entities = 1000
ids = [i for i in range(num_entities)]
embeddings = [[random.random() for _ in range(1536)] for _ in range(num_entities)]
sources = ["article"] * num_entities
authors = ["Zhang San"] * num_entitiescollection.insert([ids, embeddings, sources, authors])
collection.flush()
向量搜索
search_params = {"metric_type": "COSINE", "params": {"nprobe": 10}}
results = collection.search(data=[embeddings[0]],anns_field="embedding",param=search_params,limit=3,expr="source == 'article'",output_fields=["source", "author"]
)for hit in results[0]:print(f"ID: {hit.id}, Distance: {hit.distance}, Source: {hit.entity.get('source')}")
输出示例
ID: 0, Distance: 0.0, Source: article
ID: 123, Distance: 0.123, Source: article
ID: 456, Distance: 0.135, Source: article

应用架构与集成

与 LangChain 集成

LangChain 是一个强大的框架，可以轻松集成向量数据库。

安装依赖
pip install langchain pinecone-client pymilvus openai
Pinecone 集成
from langchain.vectorstores import Pinecone as LangchainPinecone
from langchain.embeddings.openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings()
vectorstore = LangchainPinecone.from_existing_index(index_name="knowledge-base",embedding=embeddings,text_key="content"
)
Milvus 集成
from langchain.vectorstores import Milvus as LangchainMilvusmilvus_store = LangchainMilvus(embedding_function=embeddings,collection_name="document_store",connection_args={"host": "localhost", "port": "19530"}
)

性能与选型对比

对比项PineconeMilvus
部署方式云托管自托管/云托管
易用性高（开箱即用）中等（需部署）
成本按 API 调用计费开源免费
扩展性自动扩展支持分布式部署
社区活跃度商业支持开源社区活跃
功能丰富度丰富更加灵活可定制

实战案例

案例一：企业搜索引擎

  • 需求：亿级文档的向量检索系统。
  • 方案：
    • 使用 Milvus 构建本地向量数据库。
    • 使用 Elasticsearch 做关键词搜索。
    • 使用 LangChain 整合两者实现混合搜索。

案例二：个性化推荐

  • 需求：根据用户历史行为推荐相关内容。
  • 方案：
    • 用户行为向量化。
    • 使用 Pinecone 进行实时相似性搜索。
    • 返回相似用户的偏好内容作为推荐结果。

实战案例：完整代码与部署指南

案例一：企业搜索引擎（Milvus + Elasticsearch + LangChain）

1. 系统架构设计

[用户查询] ↓
[LangChain混合搜索]├─→ [Elasticsearch关键词搜索]└─→ [Milvus向量相似性搜索]↓
[结果融合排序]↓
[最终返回结果]

2. 技术栈要求

pip install pymilvus elasticsearch langchain openai sentence-transformers fastapi uvicorn
docker pull milvusdb/milvus:v2.4.3
docker pull docker.elastic.co/elasticsearch/elasticsearch:8.11.3

3. Milvus向量数据库初始化

from pymilvus import connections, Collection, utility, FieldSchema, CollectionSchema, DataType# 连接Milvus
connections.connect(host='localhost', port='19530')# 创建集合
collection_name = "document_vectors"if not utility.has_collection(collection_name):fields = [FieldSchema(name="doc_id", dtype=DataType.INT64, is_primary=True),FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535)]schema = CollectionSchema(fields, description="文档向量存储")collection = Collection(collection_name, schema=schema)# 创建索引index_params = {"index_type": "IVF_FLAT","params": {"nlist": 1024},"metric_type": "COSINE"}collection.create_index(field_name="embedding", index_params=index_params)
else:collection = Collection(collection_name)collection.load()

4. Elasticsearch关键词索引创建

from elasticsearch import Elasticsearch
from elasticsearch_dsl import analyzer# 连接Elasticsearch
es_client = Elasticsearch(hosts=["http://localhost:9200"])# 创建索引
index_name = "document_search"if not es_client.indices.exists(index=index_name):es_client.indices.create(index=index_name,body={"settings": {"analysis": {"analyzer": {"custom_analyzer": {"type": "custom","tokenizer": "whitespace","filter": ["lowercase", "stop", "stemmer"]}}}},"mappings": {"properties": {"title": {"type": "text", "analyzer": "custom_analyzer"},"content": {"type": "text", "analyzer": "custom_analyzer"}}}})

5. 向量化处理服务

from sentence_transformers import SentenceTransformerclass VectorService:def __init__(self):self.model = SentenceTransformer('all-MiniLM-L6-v2')def text_to_vector(self, text: str) -> List[float]:return self.model.encode(text).tolist()vector_service = VectorService()

6. LangChain混合搜索实现

from langchain.vectorstores import Milvus
from langchain.schema import Document
from langchain.embeddings import HuggingFaceEmbeddings# 初始化嵌入模型
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")# 连接Milvus向量数据库
milvus_store = Milvus(embedding_function=embeddings,collection_name=collection_name,connection_args={"host": "localhost", "port": "19530"}
)def hybrid_search(query: str, top_k: int = 5):# 向量搜索vector_results = milvus_store.similarity_search(query, k=top_k)# 关键词搜索es_results = es_client.search(index=index_name,body={"query": {"multi_match": {"query": query, "fields": ["title^2", "content"]}}})# 结果融合combined_results = []for hit in es_results['hits']['hits']:combined_results.append({"source": "elasticsearch","score": hit['_score'],"title": hit['_source']['title'],"content": hit['_source']['content']})for doc in vector_results:combined_results.append({"source": "milvus","score": doc.metadata.get("distance", 0),"title": doc.metadata.get("title", ""),"content": doc.page_content})# 按分数排序combined_results.sort(key=lambda x: x["score"], reverse=True)return combined_results[:top_k]

7. 部署指南

# 启动Milvus
docker run -d --name milvusdb -p 19530:19530 milvusdb/milvus:v2.4.3# 启动Elasticsearch
docker run -d --name es-node -p 9200:9200 \-e "discovery.type=single-node" \-e "ES_JAVA_OPTS=-Xms1g -Xmx1g" \docker.elastic.co/elasticsearch/elasticsearch:8.11.3# 安装依赖
pip install -r requirements.txt# 启动API服务
uvicorn search_api:app --host 0.0.0.0 --port 8000

案例二：个性化推荐系统（Pinecone + 用户行为分析）

1. 系统架构设计

[用户行为数据]↓
[特征工程处理]↓
[Pinecone向量存储]↓
[相似用户查找]↓
[内容推荐生成]

2. 技术栈要求

pip install pinecone-client openai pandas numpy scikit-learn fastapi uvicorn

3. Pinecone初始化

import pinecone
from openai import OpenAI# 初始化Pinecone
pinecone.init(api_key="your-api-key", environment="northamerica-northeast1-gcp")# 创建索引
index_name = "user-behavior"if index_name not in pinecone.list_indexes():pinecone.create_index(name=index_name,dimension=1536,  # OpenAI嵌入维度metric="cosine")# 连接索引
index = pinecone.Index(index_name)
client = OpenAI()

4. 用户行为向量化

import pandas as pd
from sklearn.preprocessing import StandardScalerclass UserBehaviorVectorizer:def __init__(self):self.scaler = StandardScaler()def _extract_features(self, user_data: dict) -> pd.DataFrame:# 特征提取示例features = {"avg_session_time": user_data.get("session_duration", 0),"click_rate": len(user_data.get("clicked_items", [])) / max(1, user_data.get("impressions", 1)),"purchase_frequency": len(user_data.get("purchases", [])) / max(1, user_data.get("days_active", 1)),"category_preference": user_data.get("preferred_category", "other"),"device_usage": user_data.get("device_type", "desktop"),"location": user_data.get("location", "unknown")}return pd.DataFrame([features])def _normalize_features(self, df: pd.DataFrame) -> np.ndarray:# 数值特征标准化numerical = self.scaler.fit_transform(df[["avg_session_time", "click_rate", "purchase_frequency"]])# 类别特征one-hot编码categorical = pd.get_dummies(df[["category_preference", "device_usage", "location"]])# 合并特征return np.hstack([numerical, categorical.values])def vectorize(self, user_data: dict) -> List[float]:df = self._extract_features(user_data)features = self._normalize_features(df)# 使用OpenAI嵌入进行降维response = client.embeddings.create(input=features.tolist()[0],model="text-embedding-ada-002")return response.data[0].embedding

5. 推荐引擎实现

class RecommendationEngine:def __init__(self):self.user_vectorizer = UserBehaviorVectorizer()def add_user(self, user_id: str, user_data: dict):# 向量化用户行为vector = self.user_vectorizer.vectorize(user_data)# 存储到Pineconeindex.upsert(vectors=[{"id": user_id,"values": vector,"metadata": {"preferences": user_data.get("preferred_categories", []),"recent_purchases": user_data.get("recent_purchases", []),"location": user_data.get("location", "")}}])def get_recommendations(self, user_id: str, top_k: int = 5):# 获取用户向量result = index.fetch(ids=[user_id])if not result.vectors:return []user_vector = result.vectors[user_id].values# 查找相似用户similar_users = index.query(vector=user_vector,top_k=top_k+1,  # 排除自己include_metadata=True)# 过滤出相似用户的偏好recommendations = []for match in similar_users.matches:if match.id == user_id:continuemetadata = match.metadatarecommendations.extend(metadata.get("recent_purchases", []))# 去重并返回return list(set(recommendations))[:top_k]

6. API接口实现

from fastapi import FastAPI, HTTPException
from pydantic import BaseModelapp = FastAPI()class UserData(BaseModel):session_duration: floatclicked_items: List[str]impressions: intpurchases: List[str]days_active: intpreferred_category: strdevice_type: strlocation: strrecommendation_engine = RecommendationEngine()@app.post("/users/{user_id}/update")
async def update_user_profile(user_id: str, user_data: UserData):try:recommendation_engine.add_user(user_id, user_data.dict())return {"status": "success"}except Exception as e:raise HTTPException(status_code=500, detail=str(e))@app.get("/users/{user_id}/recommendations")
async def get_recommendations(user_id: str, top_k: int = 5):try:results = recommendation_engine.get_recommendations(user_id, top_k)return {"recommendations": results}except Exception as e:raise HTTPException(status_code=500, detail=str(e))

7. 部署指南

# 设置环境变量
export PINECONE_API_KEY="your-pinecone-api-key"
export OPENAI_API_KEY="your-openai-api-key"# 安装依赖
pip install -r requirements.txt# 启动服务
uvicorn recommendation_api:app --host 0.0.0.0 --port 8000

扩展建议

1. 性能优化

  • 对于企业搜索引擎：
    • 使用Milvus的分片功能支持亿级数据
    • 在Elasticsearch中使用倒排索引优化关键词搜索
    • 引入缓存层（如Redis）存储高频查询结果

2. 安全加固

  • 添加身份验证（JWT/OAuth）
  • 对敏感数据进行加密存储
  • 实现速率限制防止滥用

3. 监控方案

  • Prometheus + Grafana监控系统指标
  • ELK日志分析体系
  • 设置异常报警规则

4. 可扩展性设计

  • 将计算密集型任务移至Celery异步任务队列
  • 使用Kubernetes进行容器编排
  • 为不同模块设计独立的微服务架构

以上两个实战案例提供了完整的代码实现和部署指南，可根据具体需求进一步扩展和优化。

优化与故障排除

性能调优清单

  • 索引类型选择：根据数据量和查询频率选择合适的索引。
  • 批量插入：避免单条插入，使用批量插入提高效率。
  • 内存优化：适当调整索引参数以减少内存占用。
  • 缓存机制：对高频查询结果进行缓存。

常见问题与解决方案

问题解决方案
向量维度不匹配检查模型输出维度是否一致
查询超时增加超时时间或优化索引
内存溢出减少索引分片数或升级硬件
插入失败检查主键唯一性约束

总结与扩展思考

未来趋势

  • 多模态检索：支持文本、图像、音频等多种数据类型的统一检索。
  • 混合搜索：结合关键词搜索与向量搜索，提升搜索准确性。
  • 边缘计算：向量数据库将在边缘设备中得到更多应用。

学习资源推荐

  • Pinecone 官方文档
  • Milvus 官方文档
  • LangChain 官方文档
  • OpenAI Embeddings API 文档

如果你喜欢这篇文章，请点赞、收藏，并分享给你的朋友！欢迎关注我的专栏《AI大模型应知应会100篇》，获取更多实用技术干货！

相关文章：

  • Java大师成长计划之第20天：Spring Framework基础
  • java-----------＞代理模式
  • 专业课复习笔记 8
  • 【SSM-SSM整合】将Spring、SpringMVC、Mybatis三者进行整合；本文阐述了几个核心原理知识点，附带对应的源码以及描述解析
  • pdf 不是扫描件，但却无法搜索关键词【问题尝试解决未果记录】
  • Java中堆栈
  • 一个极简单的 VUE3 + Element-Plus 查询表单展开收起功能组件
  • 基于HTTP头部字段的SQL注入：SQLi-labs第17-20关
  • spring中的@PropertySource注解详解
  • 记录裁员后的半年前端求职经历
  • 【氮化镓】GaN在不同电子能量损失的SHI辐射下的损伤
  • 歌曲《忘尘谷》基于C语言的歌曲调性检测技术解析
  • Linux常用命令详解（下）：打包压缩、文本编辑与查找命令
  • Codeforces Round 1024 (Div. 2)(A-D)
  • 五、Hive表类型、分区及数据加载
  • [Java][Leetcode simple] 189. 轮转数组
  • 中国黄土高原中部XF剖面磁化率和粒度数据
  • 信息系统项目管理师-软考高级（软考高项）​​​​​​​​​​​2025最新（十八）
  • 数据库分区与分表详解
  • Java实现MCP server，配合DeepSeek和达梦数据库，实现基于企业数据库的智能问答
  • 外交部：正确认识和对待历史是检验日本能否恪守和平发展承诺的重要标准
  • 食用城市｜食饭识人
  • 盖茨说对中国技术封锁起到反作用
  • 科创板年内第3家！健信超导IPO获受理，拟募资8.65亿
  • 《AI×SCIENCE十大前沿观察》9：合成数据和数据基础设施
  • 中医的千年传承：网络科学描绘其演化之路｜PNAS速递