已设15分钟超时与500MB+内存，AWS Lambda仍在嵌入索引阶段超时

我使用AWS Lambda函数处理存储在S3存储桶中的PDF文件，流程为：从S3读取文件→解析为文本→通过OpenAI的text-embedding-3-large模型生成嵌入向量→用LlamaIndex将数据索引并存储到EFS。尽管已将Lambda超时时间设为最大值15分钟，内存分配提升至500MB以上，函数仍持续在嵌入索引阶段超时。CloudWatch日志显示该阶段耗时约6-7分钟，但仍无法在15分钟内完成。

已尝试的操作 #

• 将Lambda超时设为15分钟上限
• 提升内存至500MB以上，避免内存受限
• 配置VPC与IAM角色，确保网络权限与访问能力

CloudWatch日志信息 #

• 初始化阶段耗时10006.53ms后超时
• 请求阶段执行至嵌入步骤后提示超时
• 计费时长约485秒，内存最大使用量为404MB

核心代码片段 #

import boto3
import os
import magic
import json
# from urllib.parse import unquote_plus
import logging
from tempfile import NamedTemporaryFile
from llama_index.core import SimpleDirectoryReader
from llama_index.core import (
    VectorStoreIndex,
    Document,
    StorageContext,
    load_index_from_storage,
)
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.storage.index_store import SimpleIndexStore
from llama_index.core.vector_stores import SimpleVectorStore
from llama_index.core.node_parser import SimpleNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding


# Setup basic logging
logging.basicConfig(level=logging.INFO)

storage_dir = '/mnt/storage'
new_folder = 'TRANE_S123'
full_storage_path = os.path.join(storage_dir, new_folder)
os.environ["OPENAI_API_KEY"] = "******"

def process_files(content: bytes, file_path: str):
    try:
        file_type = magic.from_buffer(content, mime=True)
        logging.info("File type: %s", file_type)

        if file_type == "application/pdf":
            print("Its a pdf file..........")
            return process_pdf(content, file_path)
        else:
            raise Exception("Unsupported file type")
    except Exception as e:
        logging.error("Error processing file: %s", e)
        raise

def process_pdf(content: bytes, file_path: str):
    try:
        print("Processing PDF file...")
        with NamedTemporaryFile(delete=False, suffix=".pdf") as temp_file:
            temp_file.write(content)
            temp_file_path = temp_file.name
        
        pages = load_pdf(temp_file_path, file_path)

        # Clean up temporary file
        os.remove(temp_file_path)
        
        return pages
    except Exception as e:
        logging.error("Error extracting text from PDF: %s", e)
        raise

def load_pdf(pdf_path: str, file_path: str):
    try:
        print("Loading PDF...")
        reader = SimpleDirectoryReader(input_files=[pdf_path])
        documents = reader.load_data()
        pages = []
        
        for i, doc in enumerate(documents, start=1):
            pages.append({
                "content": doc.text,
                "page_number": i,
                "filename": os.path.basename(file_path)  # Get filename using os.path
            })
        
        return pages
    except Exception as e:
        logging.error("Error loading PDF: %s", e)
        print("error loading pdf   ")
        raise

# Main Lambda handler
def lambda_handler(event, context):
    # Extract bucket name and file key from the event
    print("Received event:", event)
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    file_path = event['Records'][0]['s3']['object']['key']
    file_key = event['Records'][0]['s3']['object']['key'].replace(f"s3://{bucket_name}/", "")

    print("Bucket name : ", bucket_name)
    print("File key : ", file_key)
    print("File path : ", file_path)

    if not bucket_name or not file_key:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'bucket_name and file_key are required'})
        }

    # Initialize the S3 client
    s3_client = boto3.client('s3')

    try:
        # Read file content from S3 bucket
        response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
        content = response['Body'].read()

        # Process the file
        pages = process_files(content, file_path)

        print("after processing filessss.......", full_storage_path)
        # Create the storage directory if it doesn't exist
        if not os.path.exists(full_storage_path):
            print("Creating directory....")
            os.makedirs(full_storage_path, exist_ok=True)

        # Set up the storage context for EFS
        print("settuppting storage context for efs..")
        storage_context = StorageContext.from_defaults(
            persist_dir = full_storage_path,
            docstore=SimpleDocumentStore(),
            vector_store=SimpleVectorStore(),
            index_store=SimpleIndexStore(),
        )
        index = VectorStoreIndex([], storage_context=storage_context)

        print("Creating document.............")
        # Create new documents from the pages
        documents = [
            Document(
                text=page["content"],
                metadata={
                    "file_name": page["filename"],
                    "page_number": page["page_number"],
                },
            )
            for page in pages
        ]

        print("Parsing dcument into nodes.......")
        # Parse documents into nodes
        parser = SimpleNodeParser.from_defaults(chunk_size=1000, chunk_overlap=200)
        nodes = parser.get_nodes_from_documents(documents)

        # Add page number to metadata
        print("Adding page number to metadata.........")
        for node in nodes:
            node.metadata["page_number"] = node.metadata.get("page_number", "Unknown")

        # Embedding and Indexing
        print("Embedding and indexing...............")
        embed_model = OpenAIEmbedding(model="text-embedding-3-large")
        print("after embedding .....")
        index.insert_nodes(nodes, embed_model=embed_model)

        # Persist updated index
        print("Persist updated index........")
        index.storage_context.persist(persist_dir=full_storage_path)

        return {
            'statusCode': 200,
            'body': json.dumps({'message': 'File processed successfully'})
        }

    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

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