Amazon OpenSearch Service

Amazon OpenSearch Serverless is a fully managed, auto-scaling version of OpenSearch. It eliminates operational overhead (you don't have to manage nodes, clusters, or shards), automatically adjusts capacity based on your usage so you only pay for what you use without the risk of timeouts, and provides the best performance as it's specifically optimized for the low-latency vector search that applications need for a responsive user experience. It is a production-ready service with a robust feature set.

Amazon OpenSearch Service is designed for search and analytics workloads, providing capabilities for querying and analyzing large datasets.

1

Define AWSTemplateFormatVersion and Description.

Standard CloudFormation template metadata.

AWSTemplateFormatVersion: '2010-09-09'
Description: OpenSearch Serverless collection for RAG.

2

Define Parameters.

To make the collection name configurable.

Parameters:
  CollectionName:
    Type: String
    Default: pdf-conversation-vectors
    Description: The name for the OpenSearch Serverless collection.

3

Create the Encryption Policy for the collection.

This policy must be created before the collection itself to ensure data encryption.

EncryptionPolicy:
  Type: AWS::OpenSearchServerless::SecurityPolicy
  Properties:
    Type: encryption
    Name: !Sub '${CollectionName}-enc'
    Policy: !Sub |
      {
        "Rules": [
          {
            "ResourceType": "collection",
            "Resource": [
              "collection/${CollectionName}"
            ]
          }
        ],
        "AWSOwnedKey": true
      }

4

Create the Network Access Policy for the collection.

This policy defines network access rules, allowing access from the public internet for development purposes.

NetworkAccessPolicy:
  Type: AWS::OpenSearchServerless::SecurityPolicy
  Properties:
    Type: network
    Name: !Sub '${CollectionName}-net'
    Policy: !Sub |
      [
        {
          "Rules": [
            {
              "ResourceType": "collection",
              "Resource": [
                "collection/${CollectionName}"
              ]
            },
            {
              "ResourceType": "dashboard",
              "Resource": [
                "collection/${CollectionName}"
              ]
            }
          ],
          "AllowFromPublic": true
        }
      ]

5

Create the Data Access Policy for the collection.

This policy grants specific permissions (e.g., ReadDocument, WriteDocument, CreateIndex, DeleteIndex) to a specified AWS Principal (e.g., an IAM role) for interacting with indices within the collection.

DataAccessPolicy:
  Type: AWS::OpenSearchServerless::AccessPolicy
  Properties:
    Type: data
    Name: !Sub '${CollectionName}-data'
    Policy: !Sub |
      [
        {
          "Rules": [
            {
              "ResourceType": "index",
              "Resource": ["index/${CollectionName}/*"],
              "Permission": [
                "aoss:ReadDocument",
                "aoss:WriteDocument",
                "aoss:CreateIndex",
                "aoss:DeleteIndex"
              ]
            }
          ],
          "Principal": ["arn:aws:sts::198945929229:assumed-role/AWSReservedSSO_PDFConversationPermissionSet_27c1e49c62b9097d/PDFConvo"]
        }
      ]

6

Create the OpenSearch Serverless collection for vector search.

This defines the core OpenSearch Serverless collection, explicitly depending on the security policies to ensure they are created first.

VectorSearchCollection:
  Type: AWS::OpenSearchServerless::Collection
  DependsOn:
    - EncryptionPolicy
    - NetworkAccessPolicy
    - DataAccessPolicy
  Properties:
    Name: !Ref CollectionName
    Type: VECTORSEARCH
    Description: Collection for storing document vector embeddings.

7

Define Outputs.

To easily retrieve important attributes of the deployed collection, such as its name, endpoint, and ARN.

Outputs:
  CollectionName:
    Value: !Ref VectorSearchCollection
  CollectionEndpoint:
    Value: !GetAtt VectorSearchCollection.CollectionEndpoint
  CollectionArn:
    Description: The ARN of the collection.
    Value: !GetAtt VectorSearchCollection.Arn

1

Configure endpoint, region, index name, and vector dimensions.

These constants define the target OpenSearch collection and the specifics of the vector index to be created.

COLLECTION_ENDPOINT = "6sxorgnw362or9e7bw0b.us-west-2.aoss.amazonaws.com"
REGION_NAME = "us-west-2"
INDEX_NAME = "document-vectors"
VECTOR_DIMENSIONS = 1536

2

Set up AWS authentication.

Authenticates the Python script to securely interact with the OpenSearch Serverless collection using AWS SigV4.

session = boto3.Session(profile_name=AWS_PROFILE, region_name=REGION_NAME)
credentials = session.get_credentials()
auth = AWSV4SignerAuth(credentials, REGION_NAME, 'aoss')

3

Create the OpenSearch client.

Initializes the client object used to send requests to the OpenSearch Serverless endpoint.

client = OpenSearch(
    hosts=[{'host': COLLECTION_ENDPOINT, 'port': 443}],
    http_auth=auth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

4

Define the index body with KNN settings and mappings.

This JSON structure specifies the index settings to enable k-Nearest Neighbors (KNN) search and defines the data types and properties for fields like the vector_field, text, and document_id.

index_body = {
    'settings': {
        'index': {
            'knn': 'true',
            'knn.algo_param.ef_search': 100
        }
    },
    'mappings': {
        'properties': {
            'vector_field': {
                'type': 'knn_vector',
                'dimension': VECTOR_DIMENSIONS,
                'method': {
                    'engine': 'faiss',
                    'name': 'hnsw',
                    'parameters': {
                        'ef_construction': 100,
                        'm': 16
                    }
                }
            },
            'text': {
                'type': 'text'
            },
            'document_id': {
                'type': 'keyword'
            }
        }
    }
}

5

Check for existing index and create if not present.

Ensures idempotency; the index is only created if it doesn't already exist, preventing errors on re-runs.

if not client.indices.exists(index=INDEX_NAME):
    print(f"Creating index '{INDEX_NAME}'...")
    response = client.indices.create(index=INDEX_NAME, body=index_body)
    print(f"Index creation response: {response}")
else:
    print(f"Index '{INDEX_NAME}' already exists.")

1

Configure Lambda environment variables and clients.

These define the necessary endpoints, index, embedding model, and S3 bucket, and initialize AWS service clients for S3 and Bedrock.

OPENSEARCH_ENDPOINT = os.environ.get("OPENSEARCH_ENDPOINT")
REGION_NAME = os.environ.get("REGION_NAME", "us-west-2")
OPENSEARCH_INDEX_NAME = "document-vectors"
EMBEDDING_MODEL_ID = "cohere.embed-multilingual-v3"
PROCESSED_BUCKET = os.environ.get("PROCESSED_BUCKET", "pdf-conversation-digests")

s3_client = boto3.client('s3')
bedrock_client = boto3.client('bedrock-runtime', region_name=REGION_NAME)

2

Implement OpenSearch client initialization.

A helper function (`get_opensearch_client`) is used to initialize an authenticated OpenSearch client instance, using `AWSV4SignerAuth` for secure communication.

def get_opensearch_client():
    credentials = boto3.Session().get_credentials()
    auth = AWSV4SignerAuth(credentials, REGION_NAME, 'aoss')

    return OpenSearch(
        hosts=[{'host': OPENSEARCH_ENDPOINT, 'port': 443}],
        http_auth=auth,
        use_ssl=True,
        verify_certs=True,
        connection_class=RequestsHttpConnection
    )

3

Implement text chunking logic.

The `chunk_text(text)` function splits long raw text into smaller, manageable chunks based on sentence boundaries, to respect the token limits of the Bedrock embedding model (e.g., a simple character count limit of 500).

def chunk_text(text):
    sentences = deque(text.split('. '))
    chunks = []
    current_chunk = ""

    while sentences:
        sentence = sentences.popleft()
        if len(current_chunk) + len(sentence) < 500: # Simple character count limit
            current_chunk += sentence + '. '
        else:
            chunks.append(current_chunk.strip())
            current_chunk = sentence + '. '
    
    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

4

Implement vector embedding retrieval from Bedrock.

The `get_vector_embeddings(text)` function calls the Bedrock `invoke_model` API with the specified embedding model ID (e.g., `cohere.embed-multilingual-v3`) and `search_document` input type to obtain the numerical vector representation of a text chunk.

def get_vector_embeddings(text):
    try:
        response = bedrock_client.invoke_model(
            body=json.dumps({
                "texts": [text],
                "input_type": "search_document"
            }),
            modelId=EMBEDDING_MODEL_ID,
            accept='application/json',
            contentType='application/json'
        )
        response_body = json.loads(response.get('body').read())
        return response_body['embeddings'][0]
    except ClientError as e:
        print(f"Error invoking Bedrock embedding model: {e}")
        raise e

5

Define the main Lambda handler (`lambda_handler`).

This function orchestrates the entire ingestion process: reading the processed JSON from S3, extracting document metadata, chunking the text, generating embeddings, and preparing documents for OpenSearch.

def lambda_handler(event, context):
    s3_key = event['s3_key_processed']
    bucket_name = PROCESSED_BUCKET

    try:
        file_content = s3_client.get_object(Bucket=bucket_name, Key=s3_key)['Body'].read().decode('utf-8')
        processed_data = json.loads(file_content)
        raw_text = processed_data.get('raw_text', '')

        key_parts = urllib.parse.unquote_plus(s3_key).split('/')
        user_id = key_parts[1]
        document_id = os.path.splitext(os.path.basename(s3_key))[0]

        text_chunks = chunk_text(raw_text)
        os_client = get_opensearch_client()

        # ... (ingestion logic in next step) ...

6

Process chunks and ingest into OpenSearch.

For each text chunk, a unique ID is created, a vector embedding is obtained from Bedrock, and a document with `vector_field`, `text`, `document_id`, `user_id`, and `chunk_id` is indexed into the OpenSearch collection.

        for i, chunk in enumerate(text_chunks):
            chunk_id = f"{document_id}_{i}"
            vector_embedding = get_vector_embeddings(chunk)
            os_document = {
                "vector_field": vector_embedding,
                "text": chunk,
                "document_id": document_id,
                "user_id": user_id,
                "chunk_id": chunk_id
            }

            os_client.index(
                index=OPENSEARCH_INDEX_NAME,
                body=os_document,
                id=chunk_id
            )
        
        print(f"Successfully processed {len(text_chunks)} chunks for document {document_id}")
        
        return {
            'statusCode': 200,
            'body': json.dumps({'document_id': document_id, 'user_id': user_id, 'chunks_processed': len(text_chunks)})
        }

Hot storage is designed for high performance, low latency, frequently accessed, and mutable data. It incurs significant overhead due to replicas, Linux reserved space, and OpenSearch reserved space. For example, 100 GB of data with one replica can require approximately 290 GB of hot storage.

Use Cases:

• Data requiring frequent searches, high performance, and frequent updates.

UltraWarm storage is for less frequently accessed, read-only data. It achieves a hot-like experience for aggregations and visualizations by combining Amazon S3 with nodes powered by the AWS Nitro System. S3 contributes durable, cost-effective storage, eliminating the requirement for replicas with its 11 nines durability, enabling each UltraWarm node to utilize its entire available storage capacity. These nodes are equipped with query processing optimizations and an advanced caching solution that proactively fetches data.

Use Cases:

• Less frequent searches
• Read-only data
• Warm access to data

Cold Storage is optimized for rarely accessed, read-only audit log data that is searched periodically. Unlike UltraWarm storage which have compute nodes, Cold Storage separates compute from storage, thereby lowering costs further. You can easily attach cold storage to UltraWarm nodes when you need to run search queries on your data.

Use Cases:

• Rarely accessed read-only audit log data
• Periodic searches
• Lowest cost storage tier