如何检查Pinecone命名空间的向量是否完成上传（替代time.sleep）

可以通过查询Pinecone索引的实时统计数据替代time.sleep()，直接验证目标namespace的向量数量是否与待上传文档总数一致，确保上传完成后再返回PineconeVectorStore。

实现步骤 #

1. 提前获取待上传文档总数（len(docs)）作为预期向量数；
2. 循环调用Pinecone索引的describe_index_stats()方法，获取目标namespace的实际向量数；
3. 当实际数量等于预期数量时结束等待，超时未达标则抛出异常或做其他处理。

修改后的代码 #

from typing import List
import time
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
import pinecone

# 初始化Pinecone客户端（替换为你的API密钥和环境）
pinecone.init(api_key="YOUR_PINECONE_API_KEY", environment="YOUR_PINECONE_ENV")

embed_model = OpenAIEmbeddings(
    model='text-embedding-3-small'
)

def upload_data(docs: List[Document], embed_model, index_name: str, namespace: str) -> PineconeVectorStore:
    expected_vector_count = len(docs)
    vector_store = PineconeVectorStore.from_documents(
        documents=docs,
        embedding=embed_model,
        index_name=index_name,
        namespace=namespace
    )
    
    pc_index = pinecone.Index(index_name)
    check_interval = 1  # 每秒检查一次
    max_wait_time = 30  # 最长等待30秒，可按需调整
    elapsed_time = 0

    while elapsed_time < max_wait_time:
        index_stats = pc_index.describe_index_stats()
        # 获取目标namespace的当前向量数，不存在则默认0
        current_vector_count = index_stats['namespaces'].get(namespace, {}).get('vector_count', 0)
        
        if current_vector_count == expected_vector_count:
            break
        
        time.sleep(check_interval)
        elapsed_time += check_interval
    else:
        # 超时处理，可根据业务需求调整
        raise TimeoutError(
            f"向量上传超时：预期{expected_vector_count}个，当前仅上传{current_vector_count}个"
        )
    
    return vector_store

注意事项 #

• 如果文档存在重复ID，上传后实际向量数会小于len(docs)，需提前处理ID去重逻辑，或调整预期数量的计算方式；
• 可根据索引规模和网络情况，调整check_interval（检查间隔）和max_wait_time（最长等待时间）；
• 该方法依赖Pinecone实时统计数据，比固定时长的time.sleep()更可靠，能确保所有向量完成上传后再执行后续逻辑。

内容的提问来源于stack exchange，提问作者H.Sdq