使用OpenAI+Pinecone构建YouTube语义搜索遇429速率限制错误求助

云原生阿岚

2026-7-27

问题描述

基于OpenAI + Pinecone实现YouTube视频语义搜索，执行命令npx tsx src/bin/process-yt-playlist.ts预处理字幕、获取嵌入向量并插入Pinecone索引时，持续触发OpenAI API的429请求超限错误。
已尝试大幅增加API调用间隔，问题仍未解决，相关代码如下：

import pMap from 'p-map'
import unescape from 'unescape'

import * as config from '@/lib/config'

import * as types from './types'

import pMemoize from 'p-memoize'
import pRetry from 'p-retry'
import pThrottle from 'p-throttle'

// TODO: enforce max OPENAI_EMBEDDING_CTX_LENGTH of 8191

// https://platform.openai.com/docs/guides/rate-limits/what-are-the-rate-limits-for-our-api
// TODO: enforce TPM
const throttleRPM = pThrottle({
  // 3k per minute instead of 3.5k per minute to add padding
  limit: 3000,
  interval: 60 * 1000,
  strict: true
})

type PineconeCaptionVectorPending = {
  id: string
  input: string
  metadata: types.PineconeCaptionMetadata
}

export async function getEmbeddingsForVideoTranscript({
  transcript,
  title,
  openai,
  model = config.openaiEmbeddingModel,
  maxInputTokens = 100, // TODO???
  concurrency = 1
}: {
  transcript: types.Transcript
  title: string
  openai: types.OpenAIApi
  model?: string
  maxInputTokens?: number
  concurrency?: number
}) {
  const { videoId } = transcript

  let pendingVectors: PineconeCaptionVectorPending[] = []
  let currentStart = ''
  let currentNumTokensEstimate = 0
  let currentInput = ''
  let currentPartIndex = 0
  let currentVectorIndex = 0
  let isDone = false

  // const createEmbedding = pMemoize(throttleRPM(createEmbeddingImpl))

  // Pre-compute the embedding inputs, making sure none of them are too long
  do {
    isDone = currentPartIndex >= transcript.parts.length

    const part = transcript.parts[currentPartIndex]
    const text = unescape(part?.text)
      .replaceAll('[Music]', '')
      .replaceAll(/[\t\n]/g, ' ')
      .replaceAll('  ', ' ')
      .trim()
    const numTokens = getNumTokensEstimate(text)

    if (!isDone && currentNumTokensEstimate + numTokens < maxInputTokens) {
      if (!currentStart) {
        currentStart = part.start
      }

      currentNumTokensEstimate += numTokens
      currentInput = `${currentInput} ${text}`

      ++currentPartIndex
    } else {
      currentInput = currentInput.trim()
      if (isDone && !currentInput) {
        break
      }

      const currentVector: PineconeCaptionVectorPending = {
        id: `${videoId}:${currentVectorIndex++}`,
        input: currentInput,
        metadata: {
          title,
          videoId,
          text: currentInput,
          start: currentStart
        }
      }

      pendingVectors.push(currentVector)

      // reset current batch
      currentNumTokensEstimate = 0
      currentStart = ''
      currentInput = ''
    }
  } while (!isDone)
  let index = 0;

  console.log("Entering embeddings calculation")
  // Evaluate all embeddings with a max concurrency
  // const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
  const vectors: types.PineconeCaptionVector[] = await pMap(
    pendingVectors,
    async (pendingVector) => {
      // await delay(6000); // add a delay of 1 second before each iteration
      console.log(pendingVector.input + " " + model)


      // const { data: embed } = await openai.createEmbedding({
      //   input: pendingVector.input,
      //   model
      // })

      async function createEmbeddingImpl({
        input = pendingVector.input,
        model = 'text-embedding-ada-002'
      }: {
        input: string
        model?: string
      }): Promise<number[]> {
        const res = await pRetry(
          () =>
            openai.createEmbedding({
              input,
              model
            }),
          {
            retries: 4,
            minTimeout: 1000,
            factor: 2.5
          }
        )
      
        return res.data.data[0].embedding
      }

      const embedding = await pMemoize(throttleRPM(createEmbeddingImpl));
      

      const vector: types.PineconeCaptionVector = {
        id: pendingVector.id,
        metadata: pendingVector.metadata,
        values: await embedding(pendingVector)
      }
      console.log(index + " THIS IS THE NUMBER OF CALLS TO OPENAI Embedding: " + embedding)
      index++;
      return vector
    },
    {
      concurrency
    }
  )

  return vectors
}

function getNumTokensEstimate(input: string): number {
  const numTokens = (input || '')
    .split(/\s/)
    .map((token) => token.trim())
    .filter(Boolean).length

  return numTokens
}

问题分析与解决方案

你的代码节流、重试逻辑存在核心问题，导致429错误无法缓解，以下是针对性修复步骤：

1. 修复节流逻辑的作用域错误

你在pMap循环内部每次创建新的createEmbeddingImpl和embedding实例，导致节流器throttleRPM无法跨请求共享状态，等于没生效。把节流、重试、memoize逻辑移到函数外部，确保所有请求共用同一个限制实例：

// 移到函数外部，全局复用
async function createEmbeddingImpl({
  input,
  model = 'text-embedding-ada-002',
  openai
}: {
  input: string
  model?: string
  openai: types.OpenAIApi
}): Promise<number[]> {
  const res = await pRetry(
    () =>
      openai.createEmbedding({
        input,
        model
      }),
    {
      retries: 4,
      minTimeout: 1000,
      factor: 2.5,
      // 新增失败日志，方便排查
      onFailedAttempt: (error) => {
        console.log(`请求失败，重试次数 ${error.attemptNumber}，原因：${error.message}`);
      }
    }
  )

  return res.data.data[0].embedding
}

// 全局节流实例，确保所有请求共享RPM限制
const throttledCreateEmbedding = throttleRPM(createEmbeddingImpl);
const memoizedEmbedding = pMemoize(throttledCreateEmbedding);

然后在pMap回调中直接调用这个全局函数：

const vectors: types.PineconeCaptionVector[] = await pMap(
  pendingVectors,
  async (pendingVector) => {
    console.log(pendingVector.input + " " + model)
    // 直接调用全局的节流+缓存函数
    const embedding = await memoizedEmbedding({
      input: pendingVector.input,
      model,
      openai
    });
    console.log(`${index} 已完成OpenAI嵌入调用`)
    index++;
    return {
      id: pendingVector.id,
      metadata: pendingVector.metadata,
      values: embedding
    }
  },
  {
    concurrency: 1 // 保持并发为1，避免叠加请求突破限制
  }
)

2. 修复令牌数估算偏差

你的getNumTokensEstimate用空格分割单词估算令牌数，和OpenAI实际计算方式偏差极大，可能导致单请求令牌数超过TPM（每分钟令牌数）限制触发429。改用官方tiktoken库准确计算：

首先安装依赖：

npm install tiktoken

然后修改估算函数：

import { encoding_for_model } from 'tiktoken';

function getNumTokensEstimate(input: string, model: string = 'text-embedding-ada-002'): number {
  const encoding = encoding_for_model(model);
  const numTokens = encoding.encode(input).length;
  encoding.free();
  return numTokens;
}

3. 额外排查点

登录OpenAI控制台确认API密钥的配额（RPM、TPM）是否足够，若配额本身不足，节流也无法解决问题；
检查是否有其他进程同时使用同一API密钥发起请求，导致总请求量超限。

最终整合后的完整代码

import pMap from 'p-map'
import unescape from 'unescape'
import { encoding_for_model } from 'tiktoken';

import * as config from '@/lib/config'
import * as types from './types'
import pMemoize from 'p-memoize'
import pRetry from 'p-retry'
import pThrottle from 'p-throttle'

// TODO: enforce max OPENAI_EMBEDDING_CTX_LENGTH of 8191

// https://platform.openai.com/docs/guides/rate-limits/what-are-the-rate-limits-for-our-api
// TODO: enforce TPM
const throttleRPM = pThrottle({
  // 3k per minute instead of 3.5k per minute to add padding
  limit: 3000,
  interval: 60 * 1000,
  strict: true
})

async function createEmbeddingImpl({
  input,
  model = 'text-embedding-ada-002',
  openai
}: {
  input: string
  model?: string
  openai: types.OpenAIApi
}): Promise<number[]> {
  const res = await pRetry(
    () =>
      openai.createEmbedding({
        input,
        model
      }),
    {
      retries: 4,
      minTimeout: 1000,
      factor: 2.5,
      onFailedAttempt: (error) => {
        console.log(`请求失败，重试次数 ${error.attemptNumber}，原因：${error.message}`);
      }
    }
  )

  return res.data.data[0].embedding
}

const throttledCreateEmbedding = throttleRPM(createEmbeddingImpl);
const memoizedEmbedding = pMemoize(throttledCreateEmbedding);

type PineconeCaptionVectorPending = {
  id: string
  input: string
  metadata: types.PineconeCaptionMetadata
}

export async function getEmbeddingsForVideoTranscript({
  transcript,
  title,
  openai,
  model = config.openaiEmbeddingModel,
  maxInputTokens = 100,
  concurrency = 1
}: {
  transcript: types.Transcript
  title: string
  openai: types.OpenAIApi
  model?: string
  maxInputTokens?: number
  concurrency?: number
}) {
  const { videoId } = transcript

  let pendingVectors: PineconeCaptionVectorPending[] = []
  let currentStart = ''
  let currentNumTokensEstimate = 0
  let currentInput = ''
  let currentPartIndex = 0
  let currentVectorIndex = 0
  let isDone = false

  // Pre-compute the embedding inputs, making sure none of them are too long
  do {
    isDone = currentPartIndex >= transcript.parts.length

    const part = transcript.parts[currentPartIndex]
    const text = unescape(part?.text)
      .replaceAll('[Music]', '')
      .replaceAll(/[\t\n]/g, ' ')
      .replaceAll('  ', ' ')
      .trim()
    const numTokens = getNumTokensEstimate(text, model)

    if (!isDone && currentNumTokensEstimate + numTokens < maxInputTokens) {
      if (!currentStart) {
        currentStart = part.start
      }

      currentNumTokensEstimate += numTokens
      currentInput = `${currentInput} ${text}`

      ++currentPartIndex
    } else {
      currentInput = currentInput.trim()
      if (isDone && !currentInput) {
        break
      }

      const currentVector: PineconeCaptionVectorPending = {
        id: `${videoId}:${currentVectorIndex++}`,
        input: currentInput,
        metadata: {
          title,
          videoId,
          text: currentInput,
          start: currentStart
        }
      }

      pendingVectors.push(currentVector)

      // reset current batch
      currentNumTokensEstimate = 0
      currentStart = ''
      currentInput = ''
    }
  } while (!isDone)
  let index = 0;

  console.log("Entering embeddings calculation")
  const vectors: types.PineconeCaptionVector[] = await pMap(
    pendingVectors,
    async (pendingVector) => {
      console.log(pendingVector.input + " " + model)
      const embedding = await memoizedEmbedding({
        input: pendingVector.input,
        model,
        openai
      });
      console.log(`${index} 已完成OpenAI嵌入调用`)
      index++;
      return {
        id: pendingVector.id,
        metadata: pendingVector.metadata,
        values: embedding
      }
    },
    {
      concurrency
    }
  )

  return vectors
}

function getNumTokensEstimate(input: string, model: string = 'text-embedding-ada-002'): number {
  const encoding = encoding_for_model(model);
  const numTokens = encoding.encode(input).length;
  encoding.free();
  return numTokens;
}

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

使用OpenAI+Pinecone构建YouTube语义搜索遇429速率限制错误求助

1. 修复节流逻辑的作用域错误 #

2. 修复令牌数估算偏差 #

3. 额外排查点 #

最终整合后的完整代码 #

1. 修复节流逻辑的作用域错误

2. 修复令牌数估算偏差

3. 额外排查点

最终整合后的完整代码