Pinecone 使用教程¶

pip install pinecone  # 安装 Pinecone Python 客户端

1. 访问 https://app.pinecone.io 注册账号
2. 登录后在 "API Keys" 页面找到你的 Key
3. 设置环境变量（推荐）或直接在代码中使用

# 设置环境变量
export PINECONE_API_KEY="pcsk_xxxxxxxxxxxx"  # 你的 API Key

from pinecone import Pinecone  # 导入 Pinecone

# 方式一：使用环境变量（推荐）
pc = Pinecone()  # 自动读取 PINECONE_API_KEY

# 方式二：直接传入 API Key
pc = Pinecone(api_key="pcsk_xxxxxxxxxxxx")

from pinecone import Pinecone, ServerlessSpec  # 导入必要的类

pc = Pinecone()

# 创建 Serverless 索引
pc.create_index(
    name="bioinfo-papers",           # 索引名（全局唯一）
    dimension=384,                    # 向量维度
    metric="cosine",                  # 距离度量：cosine（余弦）
    spec=ServerlessSpec(
        cloud="aws",                  # 云提供商
        region="us-east-1"            # 区域
    )
)

# 等待索引就绪
import time
while not pc.describe_index("bioinfo-papers").status["ready"]:
    time.sleep(1)  # 每秒检查一次
print("索引已就绪！")

# 获取索引引用
index = pc.Index("bioinfo-papers")

# 插入向量（upsert = insert or update）
index.upsert(
    vectors=[
        {
            "id": "paper-001",                    # 唯一 ID
            "values": [0.1, 0.2, ...],             # 384 维向量
            "metadata": {                          # 元数据（可过滤）
                "title": "Metagenomics of T2D",
                "year": 2024,
                "topic": "metagenomics"
            }
        },
        {
            "id": "paper-002",
            "values": [0.3, 0.1, ...],
            "metadata": {
                "title": "16S rRNA Analysis Methods",
                "year": 2023,
                "topic": "amplicon"
            }
        }
    ],
    namespace="research"  # 命名空间（可选，用于数据隔离）
)

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

# 基本查询
results = index.query(
    vector=[0.15, 0.25, ...],   # 查询向量
    top_k=5,                     # 返回前 5 个
    include_metadata=True,       # 包含元数据
    namespace="research"         # 命名空间
)

for match in results["matches"]:
    print(f"ID: {match['id']}")
    print(f"相似度: {match['score']:.4f}")
    print(f"标题: {match['metadata']['title']}")
    print("---")

# 组合向量搜索 + 元数据过滤
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=5,
    include_metadata=True,
    filter={
        "year": {"$gte": 2024},           # 年份 >= 2024
        "topic": {"$eq": "metagenomics"}  # 主题 = metagenomics
    }
)

# 更复杂的过滤
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=5,
    filter={
        "$and": [
            {"year": {"$gte": 2023}},
            {"$or": [
                {"topic": {"$eq": "metagenomics"}},
                {"topic": {"$eq": "amplicon"}}
            ]}
        ]
    }
)

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone()

# 创建集成嵌入索引（Pinecone 自带嵌入模型）
pc.create_index(
    name="integrated-index",
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    # 指定内置嵌入模型
    # Pinecone 会自动将文本转为向量
)

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

# 直接用文本 upsert（自动嵌入）
# 注意：集成嵌入功能需要特定的 API 版本和配置

from pinecone import Pinecone

pc = Pinecone()

# 使用 Pinecone 的重排序模型
# pinecone-rerank-v0 在 BEIR 基准上比行业领先模型高 9%

# 先进行向量搜索获取候选结果
results = index.query(vector=[...], top_k=20)

# 然后对候选结果重排序
reranked = pc.inference.rerank(
    model="pinecone-rerank-v0",           # Pinecone 重排序模型
    query="宏基因组分析最佳实践",          # 原始查询
    documents=[
        m["metadata"]["title"] for m in results["matches"]
    ],
    top_n=5                                # 返回重排后前 5 个
)

# Pinecone Assistant 可以基于你的数据构建问答系统

# 创建 Assistant
assistant = pc.assistant.create_assistant(
    assistant_name="bioinfo-qa",          # 助手名称
    instructions="你是一个生物信息学专家助手",  # 指令
    model="gpt-4o-mini"                   # 底层模型
)

# 上传知识文档
assistant.upload_file(
    file_path="metagenomics_guide.pdf",   # 上传 PDF 文档
    metadata={"topic": "metagenomics"}     # 元数据
)

# 提问
response = assistant.chat(
    messages=[{"role": "user", "content": "什么是MAGs？"}]
)
print(response.message.content)  # AI 回答（基于上传的文档）

# 批量 upsert（每次最多 1000 条，2MB）
import itertools

def chunks(iterable, batch_size=100):
    """将列表分成批次"""
    it = iter(iterable)
    chunk = list(itertools.islice(it, batch_size))
    while chunk:
        yield chunk
        chunk = list(itertools.islice(it, batch_size))

# 假设有大量向量要插入
all_vectors = [{"id": f"vec-{i}", "values": [...]} for i in range(10000)]

for batch in chunks(all_vectors, batch_size=100):
    index.upsert(vectors=batch)  # 每批 100 条

# === Pinecone Python SDK 速查 ===
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="xxx")

# 索引操作
pc.create_index("name", dimension=384, metric="cosine", spec=ServerlessSpec(...))
pc.list_indexes()
pc.describe_index("name")
pc.delete_index("name")

# 数据操作
index = pc.Index("name")
index.upsert(vectors=[{"id": "1", "values": [...], "metadata": {...}}])
index.fetch(ids=["1", "2"])
index.delete(ids=["1"])
index.update(id="1", set_metadata={"key": "new_value"})

# 搜索
index.query(vector=[...], top_k=5, include_metadata=True)
index.query(vector=[...], top_k=5, filter={"key": {"$eq": "value"}})

# 统计
index.describe_index_stats()