Pinecone 向量数据库

Pinecone 是一个全托管的向量数据库服务，专为机器学习应用设计。它提供简单的 API、自动扩展和高可用性，是快速构建向量搜索应用的首选方案。

概述

为什么选择 Pinecone

特性	说明
全托管服务	无需运维，自动扩缩容
简单 API	RESTful API 和 Python SDK，易于集成
实时更新	支持实时插入、更新和删除
混合搜索	支持向量相似度 + 元数据过滤
高可用	多副本、自动故障恢复
企业级安全	SOC2、GDPR 合规，支持私有部署

适用场景

快速原型：无需基础设施，几分钟启动
生产应用：自动扩展，处理任意规模流量
RAG 应用：与 OpenAI、LangChain 深度集成
推荐系统：实时个性化推荐
语义搜索：文档、图片、视频搜索

快速开始

1. 注册和获取 API Key

访问 Pinecone 官网注册账号
在控制台创建 API Key
选择区域（us-east-1、eu-west-1 等）

2. 安装 SDK

pip install pinecone-client

3. 第一个示例

from pinecone import Pinecone, ServerlessSpec
import os

# 初始化 Pinecone
pc = Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))

# 创建索引
index_name = "my-first-index"

# 检查并删除已存在的索引
if index_name in pc.list_indexes().names():
    pc.delete_index(index_name)

# 创建新索引
pc.create_index(
    name=index_name,
    dimension=1536,  # OpenAI embedding-3-small 维度
    metric="cosine",  # 相似度度量: cosine, euclidean, dotproduct
    spec=ServerlessSpec(
        cloud="aws",      # aws 或 gcp
        region="us-east-1"
    )
)

# 连接索引
index = pc.Index(index_name)

# 准备数据
vectors = [
    {
        "id": "vec1",
        "values": [0.1] * 1536,  # 1536 维向量
        "metadata": {"category": "tech", "source": "doc1"}
    },
    {
        "id": "vec2", 
        "values": [0.2] * 1536,
        "metadata": {"category": "life", "source": "doc2"}
    },
    {
        "id": "vec3",
        "values": [0.3] * 1536,
        "metadata": {"category": "tech", "source": "doc3"}
    }
]

# 插入数据
index.upsert(vectors=vectors)

# 查询数据
query_vector = [0.1] * 1536
results = index.query(
    vector=query_vector,
    top_k=3,
    include_metadata=True
)

print("查询结果:")
for match in results["matches"]:
    print(f"ID: {match['id']}, Score: {match['score']:.4f}")
    print(f"Metadata: {match['metadata']}")

核心概念

Index（索引）

索引是 Pinecone 中存储向量的容器。

from pinecone import Pinecone, ServerlessSpec, PodSpec

pc = Pinecone(api_key="your-api-key")

# Serverless 索引（推荐，自动扩展）
pc.create_index(
    name="serverless-index",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

# Pod 索引（传统模式，固定资源）
pc.create_index(
    name="pod-index",
    dimension=1536,
    metric="cosine",
    spec=PodSpec(
        environment="us-east1-gcp",  # 环境
        pod_type="p1.x1",           # Pod 类型
        pods=1                       # Pod 数量
    )
)

# 支持的度量类型
# - "cosine": 余弦相似度 (最常用，推荐)
# - "euclidean": 欧几里得距离
# - "dotproduct": 点积

# 查看索引列表
print(pc.list_indexes())

# 查看索引统计
index = pc.Index("my-index")
print(index.describe_index_stats())

# 删除索引
pc.delete_index("my-index")

Namespace（命名空间）

命名空间用于隔离不同数据集，类似于数据库的表。

# 在不同命名空间操作数据
index = pc.Index("my-index")

# 向命名空间插入数据
index.upsert(
    vectors=[{"id": "vec1", "values": [0.1] * 1536}],
    namespace="user-123"
)

index.upsert(
    vectors=[{"id": "vec1", "values": [0.2] * 1536}],
    namespace="user-456"
)

# 在特定命名空间查询
results = index.query(
    vector=[0.1] * 1536,
    top_k=5,
    namespace="user-123"
)

# 删除命名空间
index.delete(delete_all=True, namespace="user-123")

Metadata（元数据）

元数据用于存储与向量相关的附加信息，支持查询过滤。

# 插入带元数据的向量
vectors = [
    {
        "id": "doc1",
        "values": [0.1] * 1536,
        "metadata": {
            "title": "Python 教程",
            "category": "programming",
            "author": "张三",
            "views": 1000,
            "published": True,
            "tags": ["python", "tutorial"]
        }
    }
]

index.upsert(vectors=vectors)

# 元数据限制
# - 最大 40KB 每向量
# - 支持类型: string, number, boolean, list of strings
# - 不支持嵌套对象

数据操作

插入和更新

# 单条插入
index.upsert(
    vectors=[{
        "id": "vec1",
        "values": [0.1] * 1536,
        "metadata": {"key": "value"}
    }]
)

# 批量插入（推荐，每批最多 100 条）
import random

batch_size = 100
vectors = []

for i in range(1000):
    vector = {
        "id": f"vec_{i}",
        "values": [random.random() for _ in range(1536)],
        "metadata": {"index": i}
    }
    vectors.append(vector)
    
    if len(vectors) >= batch_size:
        index.upsert(vectors=vectors)
        vectors = []
        print(f"已插入 {i+1} 条")

# 处理剩余数据
if vectors:
    index.upsert(vectors=vectors)

查询

基本向量搜索

# 基本查询
results = index.query(
    vector=[0.1] * 1536,  # 查询向量
    top_k=10,              # 返回结果数
    include_metadata=True, # 包含元数据
    include_values=False   # 不包含向量值
)

# 处理结果
for match in results["matches"]:
    print(f"ID: {match['id']}")
    print(f"Score: {match['score']:.4f}")
    print(f"Metadata: {match.get('metadata', {})}")

带过滤条件的搜索

from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")
index = pc.Index("my-index")

# 等于过滤
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={"category": {"$eq": "tech"}}
)

# 不等于过滤
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={"status": {"$ne": "deleted"}}
)

# 范围过滤 (数字)
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={"views": {"$gte": 100, "$lte": 1000}}
)

# 包含在列表中
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={"category": {"$in": ["tech", "programming"]}}
)

# 数组包含
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={"tags": {"$in": ["python"]}}
)

# AND 条件
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={
        "$and": [
            {"category": {"$eq": "tech"}},
            {"views": {"$gte": 100}}
        ]
    }
)

# OR 条件
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={
        "$or": [
            {"category": {"$eq": "tech"}},
            {"category": {"$eq": "programming"}}
        ]
    }
)

# 复杂组合条件
results = index.query(
    vector=[0.1] * 1536,
    top_k=10,
    filter={
        "$and": [
            {"published": {"$eq": True}},
            {"$or": [
                {"category": {"$eq": "tech"}},
                {"category": {"$eq": "ai"}}
            ]},
            {"views": {"$gte": 100}}
        ]
    }
)

批量查询

# 一次查询多个向量
results = index.query(
    vectors=[[0.1] * 1536, [0.2] * 1536, [0.3] * 1536],
    top_k=5
)

# results 包含每个查询的结果
for i, query_results in enumerate(results["results"]):
    print(f"查询 {i} 的结果:")
    for match in query_results["matches"]:
        print(f"  {match['id']}: {match['score']:.4f}")

获取和删除数据

# 根据 ID 获取向量
results = index.fetch(ids=["vec1", "vec2", "vec3"])

for vec_id, vector_data in results["vectors"].items():
    print(f"ID: {vec_id}")
    print(f"Values: {vector_data['values'][:5]}...")  # 前5个值
    print(f"Metadata: {vector_data.get('metadata', {})}")

# 删除向量
index.delete(ids=["vec1", "vec2"])

# 按过滤条件删除
index.delete(
    filter={"category": {"$eq": "obsolete"}}
)

# 删除命名空间所有数据
index.delete(delete_all=True, namespace="user-123")

# 删除索引所有数据
index.delete(delete_all=True)

更新元数据

# Pinecone 不支持直接更新元数据，需要重新插入
# 1. 获取现有向量
result = index.fetch(ids=["vec1"])
vector_data = result["vectors"]["vec1"]

# 2. 更新元数据
new_metadata = vector_data.get("metadata", {})
new_metadata.update({"views": 2000, "updated": True})

# 3. 重新插入
index.upsert(vectors=[{
    "id": "vec1",
    "values": vector_data["values"],
    "metadata": new_metadata
}])

与嵌入模型集成

OpenAI Embedding

from pinecone import Pinecone
from openai import OpenAI
import os

pc = Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))
openai = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

index = pc.Index("my-index")

def get_embeddings(texts, model="text-embedding-3-small"):
    """批量获取文本嵌入"""
    response = openai.embeddings.create(
        input=texts,
        model=model
    )
    return [item.embedding for item in response.data]

# 准备文档
documents = [
    {"id": "doc1", "text": "Pinecone 是向量数据库", "category": "database"},
    {"id": "doc2", "text": "Python 是编程语言", "category": "programming"},
    {"id": "doc3", "text": "机器学习是 AI 的分支", "category": "ai"}
]

# 生成嵌入并插入
texts = [doc["text"] for doc in documents]
embeddings = get_embeddings(texts)

vectors = [
    {
        "id": doc["id"],
        "values": embedding,
        "metadata": {"text": doc["text"], "category": doc["category"]}
    }
    for doc, embedding in zip(documents, embeddings)
]

index.upsert(vectors=vectors)

# 语义搜索
query = "什么是向量数据库？"
query_embedding = get_embeddings([query])[0]

results = index.query(
    vector=query_embedding,
    top_k=3,
    include_metadata=True
)

for match in results["matches"]:
    print(f"{match['score']:.4f}: {match['metadata']['text']}")

LangChain 集成

from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader

# 加载文档
loader = TextLoader("document.txt")
documents = loader.load()

# 切分文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 创建 Pinecone 向量存储
embeddings = OpenAIEmbeddings()
vectorstore = PineconeVectorStore.from_documents(
    texts,
    embeddings,
    index_name="my-index"
)

# 相似度搜索
docs = vectorstore.similarity_search("查询内容", k=3)

# 带过滤的搜索
docs = vectorstore.similarity_search(
    "查询内容",
    k=3,
    filter={"category": {"$eq": "tech"}}
)

RAG 应用完整示例

from pinecone import Pinecone
from openai import OpenAI
import os

class PineconeRAG:
    def __init__(self, index_name="rag-knowledge-base"):
        # 初始化 Pinecone
        self.pc = Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))
        self.openai = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
        
        # 获取或创建索引
        self.index_name = index_name
        if index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=index_name,
                dimension=1536,
                metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )
        
        self.index = self.pc.Index(index_name)
    
    def add_documents(self, documents, namespace="default"):
        """
        添加文档到知识库
        documents: [{"id": "...", "content": "...", "metadata": {...}}, ...]
        """
        # 提取内容
        contents = [doc["content"] for doc in documents]
        
        # 生成嵌入
        embeddings = self._get_embeddings(contents)
        
        # 准备向量
        vectors = []
        for doc, embedding in zip(documents, embeddings):
            metadata = doc.get("metadata", {})
            metadata["content"] = doc["content"]  # 存储原文
            
            vectors.append({
                "id": doc["id"],
                "values": embedding,
                "metadata": metadata
            })
        
        # 批量插入
        batch_size = 100
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i+batch_size]
            self.index.upsert(vectors=batch, namespace=namespace)
            print(f"已插入 {min(i+batch_size, len(vectors))}/{len(vectors)}")
    
    def _get_embeddings(self, texts):
        """获取文本嵌入"""
        response = self.openai.embeddings.create(
            input=texts,
            model="text-embedding-3-small"
        )
        return [item.embedding for item in response.data]
    
    def search(self, query, top_k=5, namespace="default", filters=None):
        """搜索相关文档"""
        # 生成查询向量
        query_embedding = self._get_embeddings([query])[0]
        
        # 查询参数
        query_params = {
            "vector": query_embedding,
            "top_k": top_k,
            "namespace": namespace,
            "include_metadata": True
        }
        
        if filters:
            query_params["filter"] = filters
        
        results = self.index.query(**query_params)
        
        return [
            {
                "id": match["id"],
                "content": match["metadata"].get("content", ""),
                "score": match["score"],
                "metadata": {k: v for k, v in match["metadata"].items() if k != "content"}
            }
            for match in results["matches"]
        ]
    
    def answer(self, question, top_k=3, namespace="default"):
        """生成回答"""
        # 检索相关文档
        docs = self.search(question, top_k, namespace)
        
        # 构建上下文
        context = "\n\n".join([
            f"[文档 {i+1}] {doc['content']}"
            for i, doc in enumerate(docs)
        ])
        
        # 调用 LLM
        prompt = f"""基于以下文档回答问题。如果文档中没有相关信息，请说明。

文档：
{context}

问题：{question}

回答："""
        
        response = self.openai.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "你是一个基于文档回答问题的助手。"},
                {"role": "user", "content": prompt}
            ]
        )
        
        return {
            "answer": response.choices[0].message.content,
            "sources": docs
        }

# 使用示例
rag = PineconeRAG()

docs = [
    {
        "id": "doc1",
        "content": "Pinecone 是一个全托管的向量数据库服务，专为机器学习应用设计。",
        "metadata": {"source": "docs", "category": "database"}
    },
    {
        "id": "doc2",
        "content": "向量数据库用于存储和检索高维向量数据，支持语义搜索。",
        "metadata": {"source": "blog", "category": "ai"}
    }
]

rag.add_documents(docs)

result = rag.answer("什么是 Pinecone？")
print(f"回答：{result['answer']}")
print(f"来源：{result['sources']}")

性能优化

批量操作

# 批量插入优化
batch_size = 100  # Pinecone 推荐每批 100 条

for i in range(0, len(vectors), batch_size):
    batch = vectors[i:i+batch_size]
    index.upsert(vectors=batch)

命名空间策略

# 按用户隔离数据
index.upsert(vectors=user_vectors, namespace=f"user-{user_id}")

# 按时间分区
index.upsert(vectors=recent_vectors, namespace="2024-01")

# 查询时指定命名空间，减少搜索范围
results = index.query(
    vector=query_vector,
    top_k=10,
    namespace=f"user-{user_id}"
)

元数据过滤优化

# 先过滤再向量搜索，提高效率
results = index.query(
    vector=query_vector,
    top_k=10,
    filter={
        "$and": [
            {"published": {"$eq": True}},  # 先过滤掉未发布
            {"category": {"$eq": "tech"}}  # 再过滤类别
        ]
    }
)

定价和限制

Serverless 定价

存储: $0.33/GB/月
查询: $0.07/1000 次查询
写入: $0.05/1000 次写入

限制

限制项	值
单索引最大向量数	无限制
向量维度	最大 20,000
元数据大小	最大 40KB/向量
批量插入	最大 100 条/次
查询 top_k	最大 10,000

常见问题

Q: Serverless vs Pod 索引如何选择？

特性	Serverless	Pod
计费方式	按使用量	按资源
自动扩展	是	手动
适用场景	大多数应用	大规模稳定负载
成本控制	按需付费	预留资源

建议: 新项目优先使用 Serverless

Q: 如何迁移数据？

# 导出数据
results = index.fetch(ids=all_ids)
vectors = list(results["vectors"].values())

# 保存到文件
import json
with open("backup.json", "w") as f:
    json.dump(vectors, f)

# 导入到新索引
new_index.upsert(vectors=vectors)

Q: 如何监控使用情况？

# 查看索引统计
stats = index.describe_index_stats()
print(f"总向量数: {stats['total_vector_count']}")
print(f"维度: {stats['dimension']}")

# 命名空间统计
for namespace, count in stats["namespaces"].items():
    print(f"命名空间 {namespace}: {count} 条")

资源链接

官方文档: https://docs.pinecone.io/
Python SDK: https://github.com/pinecone-io/pinecone-python-client
示例代码: https://github.com/pinecone-io/examples

概述​

为什么选择 Pinecone​

适用场景​

快速开始​

1. 注册和获取 API Key​

2. 安装 SDK​

3. 第一个示例​

核心概念​

Index（索引）​

Namespace（命名空间）​

Metadata（元数据）​

数据操作​

插入和更新​

查询​

基本向量搜索​

带过滤条件的搜索​

批量查询​

获取和删除数据​

更新元数据​

与嵌入模型集成​

OpenAI Embedding​

LangChain 集成​

RAG 应用完整示例​

性能优化​

批量操作​

命名空间策略​

元数据过滤优化​

定价和限制​

Serverless 定价​

限制​

常见问题​

Q: Serverless vs Pod 索引如何选择？​

Q: 如何迁移数据？​

Q: 如何监控使用情况？​

资源链接​

概述

为什么选择 Pinecone

适用场景

快速开始

1. 注册和获取 API Key

2. 安装 SDK

3. 第一个示例

核心概念

Index（索引）

Namespace（命名空间）

Metadata（元数据）

数据操作

插入和更新

查询

基本向量搜索

带过滤条件的搜索

批量查询

获取和删除数据

更新元数据

与嵌入模型集成

OpenAI Embedding

LangChain 集成

RAG 应用完整示例

性能优化

批量操作

命名空间策略

元数据过滤优化

定价和限制

Serverless 定价

限制

常见问题

Q: Serverless vs Pod 索引如何选择？

Q: 如何迁移数据？

Q: 如何监控使用情况？

资源链接