零停机嵌入迁移：在生产环境中将 text-embedding-004 切换到 text-embedding-3-large

I’m happy to translate the article for you, but I’ll need the text you’d like translated. Could you please paste the content (or the portion you want translated) here? I’ll keep the source link and all formatting exactly as you requested.

情况

服务： 使用 PostgreSQL 上的 pgvector 的 RAG 检索服务
旧模型： text-embedding-004（已弃用）
新模型： text-embedding-3-large（768 维）
数据量： 数千个已嵌入的文档
约束条件： 零停机时间、零数据丢失，生产流量必须持续运行

第一步：使模型可配置

在做任何其他事情之前，停止硬编码模型名称：

# Before (hardcoded in 6 places)
response = openai.embeddings.create(
    model="text-embedding-004",
    input=text,
)

# After (configured once)
EMBED_MODEL = os.getenv("EMBED_MODEL", "text-embedding-3-large")
EMBED_DIMENSIONS = int(os.getenv("EMBED_DIMENSIONS", "768"))

response = openai.embeddings.create(
    model=EMBED_MODEL,
    input=text,
    dimensions=EMBED_DIMENSIONS,
)

两个环境变量决定了是两天的迁移还是两周的迁移。

第2步：添加新列（不替换）

-- Migration: add new embedding column alongside the old one
ALTER TABLE documents 
ADD COLUMN embedding_v2 vector(768);

CREATE INDEX CONCURRENTLY idx_documents_embedding_v2 
ON documents USING ivfflat (embedding_v2 vector_cosine_ops)
WITH (lists = 100);

使用 CONCURRENTLY 构建索引时不会锁定表，因此生产环境的读取可以不中断地继续进行。

第3步：批量重新嵌入脚本

import asyncio
from tqdm import tqdm

async def re_embed_batch(session, documents, batch_size=50):
    """Re‑embed documents in batches with progress tracking."""
    for i in tqdm(range(0, len(documents), batch_size)):
        batch = documents[i:i + batch_size]
        texts = [doc.content for doc in batch]

        # Batch embedding call
        response = await openai.embeddings.create(
            model=EMBED_MODEL,
            input=texts,
            dimensions=EMBED_DIMENSIONS,
        )

        for doc, embedding in zip(batch, response.data):
            doc.embedding_v2 = embedding.embedding

        await session.commit()

        # Rate limiting
        await asyncio.sleep(0.5)

关键特性

• 批处理 – 不要一次只嵌入一个文档。
• 进度条 – 需要了解整个过程需要多长时间。
• 速率限制 – 嵌入 API 有调用限制。
• 每批提交 – 不要为 10 K 文档保持一个事务。

第4步：干运行验证

在切换生产流量之前：

async def validate_migration(session, sample_size=100):
    """Compare search results between old and new embeddings."""
    test_queries = get_random_queries(sample_size)

    overlaps = []
    for query in test_queries:
        old_results = await search(session, query, column="embedding")
        new_results = await search(session, query, column="embedding_v2")

        # Check overlap
        old_ids = {r.id for r in old_results[:10]}
        new_ids = {r.id for r in new_results[:10]}
        overlap = len(old_ids & new_ids) / len(old_ids)
        overlaps.append(overlap)

        if overlap  :query_vec) AS similarity
        FROM documents
        ORDER BY {column}  :query_vec
        LIMIT :top_k
    """), {"query_vec": str(query_embedding), "top_k": top_k})

    return results.fetchall()

使用 USE_V2_EMBEDDINGS=false 部署。确认一切正常后切换为 true。如果出现任何问题，立即切回。

第6步：清理

在使用 v2 运行了一周且没有问题后：

ALTER TABLE documents DROP COLUMN embedding;
ALTER TABLE documents RENAME COLUMN embedding_v2 TO embedding;
DROP INDEX idx_documents_embedding;
ALTER INDEX idx_documents_embedding_v2 RENAME TO idx_documents_embedding;

Lessons Learned

• 始终抽象化嵌入提供者。 两个环境变量帮助我们避免了多文件重构。
• 为存储的向量添加模型版本跟踪。 我们没有做到；本应该做到。
• 在需要之前构建迁移工具。 批处理脚本和验证工具是可复用的。
• 并行列 > 原位替换。 回滚过程瞬间完成。
• 所有操作都进行干运行。 我们的验证捕获了三个重叠度低的查询，需要进一步调查。

Total impact: 48 hours, zero downtime, zero data loss.

Read the full migration story on my blog. Part of my “Production GCP Patterns” series — find me at humzakt.github.io.