从SQL数据库向Pinecone插入向量时的性能异常问题排查

我有一个SQL的profiles表，包含约50列、仅244行数据。已创建包含ID和content两列的视图，content列将其他所有列的数据拼接成类似以下格式：

FirstName: John. LastName: Smith. Age: 70, Likes: Gardening, Painting. Dislikes: Soccer

我编写了Python代码将视图内容索引到Pinecone，代码能运行但出现异常：已生成超2000个向量仍未完成，初始迭代速度快，但当前每次迭代耗时超18秒，预计完成upsert需超40分钟（仅244行数据）。请问是操作有误，还是这种情况正常？

pinecone.init(
        api_key=PINECONE_API_KEY,  # find at app.pinecone.io
        environment=PINECONE_ENV  # next to api key in console
    )

    import streamlit as st
    st.title('Work in progress')
    embed = OpenAIEmbeddings(deployment=OPENAI_EMBEDDING_DEPLOYMENT_NAME, model=OPENAI_EMBEDDING_MODEL_NAME, chunk_size=1)
   
    cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+DATABASE_SERVER+'.database.windows.net;DATABASE='+DATABASE_DB+';UID='+DATABASE_USERNAME+';PWD='+ DATABASE_PASSWORD)
    query = "SELECT * from views.vwprofiles2;"
    df = pd.read_sql(query, cnxn)
    index = pinecone.Index("default")
   
    batch_limit = 100

    texts = []
    metadatas = []

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=400,
        chunk_overlap=20,
        length_function=tiktoken_len,
        separators=["\n\n", "\n", " ", ""]
    )
    

    for _, record in stqdm(df.iterrows(), total=len(df)):
        # First get metadata fields for this record
        metadata = {
            'IdentityId': str(record['IdentityId'])
        }
        # Now we create chunks from the record text
        record_texts = text_splitter.split_text(record['content'])
        # Create individual metadata dicts for each chunk
        record_metadatas = [{
            "chunk": j, "text": text, **metadata
        } for j, text in enumerate(record_texts)]
        # Append these to the current batches
        texts.extend(record_texts)
        metadatas.extend(record_metadatas)
        # If we have reached the batch_limit, we can add texts
        if len(texts) >= batch_limit:
            ids = [str(uuid4()) for _ in range(len(texts))]
            embeds = embed.embed_documents(texts)
            index.upsert(vectors=zip(ids, embeds, metadatas))    
            texts = []
            metadatas = []

        if len(texts) > 0:
            ids = [str(uuid4()) for _ in range(len(texts))]
            embeds = embed.embed_documents(texts)
            index.upsert(vectors=zip(ids, embeds, metadatas))

1. 向量数量过多的原因 #

你的代码中使用了RecursiveCharacterTextSplitter对每条record的content进行文本拆分，哪怕单条content本身很短（比如示例中的内容仅几十token），也会被拆分（甚至可能拆成多个小片段）。244行生成2000+向量就是拆分后的结果，这不是技术异常，但完全没必要——这种短文本拆分后会破坏用户信息的完整性，后续检索时无法拿到完整的用户特征。

2. 速度极慢的核心原因 #

• Embedding调用效率极低：OpenAIEmbeddings的chunk_size设为1，意味着每次API调用仅处理1条文本，频繁的API请求会大幅增加耗时。默认chunk_size是1000，调大后能减少调用次数，提升效率。
• 循环内重复执行upsert：代码最后一个if len(texts) > 0的缩进错误，它被放在了for循环内部，导致每次循环都会执行一次upsert（哪怕只有几条文本）。频繁的小批量upsert会触发大量Pinecone API请求，严重拖慢速度。

优化后的代码 #

pinecone.init(
    api_key=PINECONE_API_KEY,
    environment=PINECONE_ENV
)

import streamlit as st
st.title('Work in progress')
# 调大chunk_size，提升Embedding效率
embed = OpenAIEmbeddings(
    deployment=OPENAI_EMBEDDING_DEPLOYMENT_NAME, 
    model=OPENAI_EMBEDDING_MODEL_NAME, 
    chunk_size=1000  # 修改为默认值或更大的合理值
)

cnxn = pyodbc.connect(
    'DRIVER={ODBC Driver 17 for SQL Server};SERVER='+DATABASE_SERVER+'.database.windows.net;DATABASE='+DATABASE_DB+';UID='+DATABASE_USERNAME+';PWD='+ DATABASE_PASSWORD
)
query = "SELECT * from views.vwprofiles2;"
df = pd.read_sql(query, cnxn)
index = pinecone.Index("default")

batch_limit = 100

texts = []
metadatas = []

# 短文本无需拆分，直接移除文本拆分逻辑

for _, record in stqdm(df.iterrows(), total=len(df)):
    metadata = {
        'IdentityId': str(record['IdentityId']),
        'text': record['content']  # 把完整文本存入metadata，方便后续查看
    }
    # 直接使用完整的content，不拆分
    texts.append(record['content'])
    metadatas.append(metadata)
    
    # 积累到batch_limit再执行upsert
    if len(texts) >= batch_limit:
        ids = [str(uuid4()) for _ in range(len(texts))]
        embeds = embed.embed_documents(texts)
        index.upsert(vectors=zip(ids, embeds, metadatas))    
        texts = []
        metadatas = []

# 循环结束后处理剩余的文本，注意缩进在循环外部
if len(texts) > 0:
    ids = [str(uuid4()) for _ in range(len(texts))]
    embeds = embed.embed_documents(texts)
    index.upsert(vectors=zip(ids, embeds, metadatas))

额外建议 #

• 如果你的content偶尔有超长文本（超过Embedding模型的token限制，比如text-embedding-ada-002限制8191token），再考虑针对超长文本做拆分，短文本直接跳过拆分步骤。
• 可以在代码中添加日志，打印每次upsert的数量，方便监控进度。

内容的提问来源于stack exchange，提问作者Luis Valencia