DynamoDB Data Layer

This data layer also supports the BaseStorageClient that enables you to store your elements into AWS S3 or Azure Blob Storage.

Example

Here is an example of setting up this data layer. First install boto3:

pip install boto3

Import the custom data layer and storage client, and set the cl_data._data_layer variable at the beginning of your Chainlit app.

import chainlit.data as cl_data
from chainlit.data.dynamodb import DynamoDBDataLayer
from chainlit.data.storage_clients.s3 import S3StorageClient

storage_client = S3StorageClient(bucket="<Your Bucket>")

cl_data._data_layer = DynamoDBDataLayer(table_name="<Your Table>", storage_provider=storage_client)

Table structure

Here is the Cloudformation used to create the dynamo table:

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Resources": {
    "DynamoDBTable": {
      "Type": "AWS::DynamoDB::Table",
      "Properties": {
        "TableName": "<YOUR-TABLE-NAME>",
        "AttributeDefinitions": [
          {
            "AttributeName": "PK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "SK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "UserThreadPK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "UserThreadSK",
            "AttributeType": "S"
          }
        ],
        "KeySchema": [
          {
            "AttributeName": "PK",
            "KeyType": "HASH"
          },
          {
            "AttributeName": "SK",
            "KeyType": "RANGE"
          }
        ],
        "GlobalSecondaryIndexes": [
          {
            "IndexName": "UserThread",
            "KeySchema": [
              {
                "AttributeName": "UserThreadPK",
                "KeyType": "HASH"
              },
              {
                "AttributeName": "UserThreadSK",
                "KeyType": "RANGE"
              }
            ],
            "Projection": {
              "ProjectionType": "INCLUDE",
              "NonKeyAttributes": ["id", "name"]
            }
          }
        ],
        "BillingMode": "PAY_PER_REQUEST"
      }
    }
  }
}

Logging

DynamoDB data layer defines a child of chainlit logger.

import logging
from chainlit import logger

logger.getChild("DynamoDB").setLevel(logging.DEBUG)

Limitations

Filtering by positive/negative feedback is not supported. The data layer methods are not async. Boto3 is not async and therefore the data layer uses non-async blocking io.

Design

This implementation uses Single Table Design. There are 4 different entity types in one table identified by the prefixes in PK & SK. Here are the entity types:

type User = {
    PK: "USER#{user.identifier}"
    SK: "USER"
    // ...PersistedUser
}

type Thread = {
    PK: f"THREAD#{thread_id}"
    SK: "THREAD"
    // GSI: UserThread for querying in list_threads
    UserThreadPK: f"USER#{user_id}"
    UserThreadSK: f"TS#{ts}"
    // ...ThreadDict
}

type Step = {
    PK: f"THREAD#{threadId}"
    SK: f"STEP#{stepId}"
    // ...StepDict

    // feedback is stored as part of step. 
    // NOTE: feedback.value is stored as Decimal in dynamo which is not json serializable
    feedback?: Feedback
}

type Element = {
    "PK": f"THREAD#{threadId}"
    "SK": f"ELEMENT#{element.id}"
    // ...ElementDict
}

​Example

​Table structure

​Logging

​Limitations

​Design

Example

Table structure

Logging

Limitations

Design