BedrockEmbeddings批量生成向量嵌入遇ModelTimeoutException求解决方案

关于自动重试功能 #

LangChain的BedrockEmbeddings本身没有内置超时自动重试，但可以通过配置底层依赖的boto3客户端来实现重试逻辑，针对ModelTimeoutException等可重试错误进行自动重试。

具体解决步骤 #

• 配置boto3重试策略
  初始化BedrockEmbeddings时传入自定义的boto3客户端，通过botocore.config.Config设置重试次数、目标错误类型：

  from langchain_community.embeddings import BedrockEmbeddings
  import boto3
  from botocore.config import Config
  
  # 定义重试配置：最多重试3次，针对超时、限流等可重试异常
  retry_config = Config(
      retries={
          'max_attempts': 3,
          'mode': 'standard',
          'retryable_exceptions': [
              'ModelTimeoutException',
              'ThrottlingException',
              'ServiceUnavailableException'
          ]
      }
  )
  
  # 创建带重试配置的Bedrock Runtime客户端
  bedrock_client = boto3.client(
      'bedrock-runtime',
      region_name='us-east-1',
      profile_name='default',
      config=retry_config
  )
  
  # 初始化嵌入工具
  embeddings = BedrockEmbeddings(
      client=bedrock_client,
      model_id="amazon.titan-embed-text-v2:0"
  )
  

• 减小批量处理规模
  当前单批次250个文档可能过大，尝试将批量拆分为更小的分组（比如50-100个文档一组），分多次调用embed_documents：

  def chunked_embed(embeddings, docs, chunk_size=100):
      all_embeddings = []
      for i in range(0, len(docs), chunk_size):
          chunk = docs[i:i+chunk_size]
          all_embeddings.extend(embeddings.embed_documents(chunk))
      return all_embeddings
  
  # 用分块方式处理原批量文档
  final_embeddings = chunked_embed(embeddings, batch, chunk_size=100)
  

• 优化单文档长度
  如果存在过长的文档，先对其进行分段拆分（按字符数或语义拆分），生成嵌入后再根据需求合并或保留分段结果，避免单文档处理超时。

• 切换区域或检查资源
  偶尔的超时可能是目标区域（us-east-1）资源紧张导致，可以尝试切换到其他支持Amazon Titan Embeddings的区域（如us-west-2），观察是否缓解超时问题。

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