[CAI-183] Chatbot Patch query api (#1198)

* feat(chatbot): session GSI

* feat(chatbot): docker compose

* fix(chatbot): dynamodb and redis for local development with docker compose

* chore(chatbot):remove duplicate imports

* chore(chatbot): linting

* fix(chatbot):create index in docker

* chore(chatbot): llamaindex index id

* fix(chatbot): create vector index with all docs

* chore(chatbot): terraform lint

* fix(chatbot): terraform syntax

* chore(chatbot): remove dynamodb options

* chore(chatbot): from global to local secondary index

* feat(chatbot): find or create session

* feat(chatbot): PATCH /sessions/{sessionId}/queries/{id} feedback API

* chore(chatbot): lint

* chore: remove old var

* Update apps/chatbot/docker/compose.yaml

Co-authored-by: marcobottaro <[email protected]>

* chore: remove logs

* fix(chatbot): compose vars

* Update modules

* Update config prompts

* Update env example

* redis admin port

* chore: add env example var

* dynamodb utility scripts for local development

* page params

* CHB_SESSION_MAX_DURATION_DAYS env var

* chore: dynamodb local scripts

* fix(chatbot): query list filter

* chore(chatbot): remove query detail API

* feat(chatbot): queries in ascending order

---------

Co-authored-by: marcobottaro <[email protected]>
Co-authored-by: mdciri <[email protected]>

apps/chatbot/.env.example

@@ -26,3 +26,4 @@ CHB_ENGINE_USE_STREAMING=...
 CHB_QUERY_TABLE_PREFIX=chatbot-local
 CHB_DYNAMODB_URL=http://locahost:8080
 CHB_USE_PRESIDIO=True
+CHB_SESSION_MAX_DURATION_DAYS=1

apps/chatbot/docker/docker-compose-run-create_index.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+docker compose -f docker/compose.yaml -p chatbot run create_index

apps/chatbot/scripts/dynamodb-create-table-queries-local.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+aws dynamodb create-table \
+	--table-name chatbot-local-queries \
+	--key-schema \
+	  AttributeName=sessionId,KeyType=HASH \
+	  AttributeName=id,KeyType=RANGE \
+	--attribute-definitions \
+	  AttributeName=id,AttributeType=S \
+	  AttributeName=sessionId,AttributeType=S \
+	  AttributeName=createdAt,AttributeType=S \
+  --local-secondary-indexes '[
+    {
+      "IndexName": "QueriesByCreatedAtIndex",
+      "KeySchema": [
+        {
+          "AttributeName": "sessionId",
+          "KeyType": "HASH"
+        },
+        {
+          "AttributeName": "createdAt",
+          "KeyType": "RANGE"
+        }
+      ],
+      "Projection": {
+        "ProjectionType": "ALL"
+      }
+    }
+  ]' \
+	--table-class STANDARD \
+	--provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
+	--endpoint-url http://localhost:8000 \
+  --region eu-south-1 \
+  --profile dummy

apps/chatbot/scripts/dynamodb-create-table-sessions-local.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+aws dynamodb create-table \
+	--table-name chatbot-local-sessions \
+	--key-schema \
+	  AttributeName=userId,KeyType=HASH \
+	  AttributeName=id,KeyType=RANGE \
+	--attribute-definitions \
+	  AttributeName=id,AttributeType=S \
+	  AttributeName=userId,AttributeType=S \
+	  AttributeName=createdAt,AttributeType=S \
+  --local-secondary-indexes '[
+    {
+      "IndexName": "SessionsByCreatedAtIndex",
+      "KeySchema": [
+        {
+          "AttributeName": "userId",
+          "KeyType": "HASH"
+        },
+        {
+          "AttributeName": "createdAt",
+          "KeyType": "RANGE"
+        }
+      ],
+      "Projection": {
+        "ProjectionType": "ALL"
+      }
+    }
+  ]' \
+	--table-class STANDARD \
+	--provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
+	--endpoint-url http://localhost:8000 \
+  --region eu-south-1 \
+  --profile dummy

apps/chatbot/scripts/dynamodb-delete-table-queries-local.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+aws dynamodb delete-table \
+	--table-name chatbot-local-queries \
+	--endpoint-url http://localhost:8000 \
+  --region eu-south-1 \
+  --profile dummy

apps/chatbot/scripts/dynamodb-delete-table-sessions-local.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+aws dynamodb delete-table \
+  --table-name chatbot-local-sessions \
+  --endpoint-url http://localhost:8000 \
+  --region eu-south-1 \
+  --profile dummy

apps/chatbot/scripts/dynamodb-scan-queries.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+aws dynamodb scan \
+  --table-name chatbot-local-queries \
+  --region eu-south-1 \
+  --endpoint-url http://localhost:8000 \
+  --profile dummy

apps/chatbot/scripts/dynamodb-scan-sessions.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+aws dynamodb scan \
+  --table-name chatbot-local-sessions \
+  --region eu-south-1 \
+  --endpoint-url http://localhost:8000 \
+  --profile dummy

apps/chatbot/src/app/main.py

@@ -28,6 +28,9 @@ class Query(BaseModel):
   question: str
   queriedAt: str | None = None
 
+class QueryFeedback(BaseModel):
+  badAnswer: bool
+
 boto3_session = boto3.session.Session(
   region_name=AWS_DEFAULT_REGION
 )
@@ -85,7 +88,8 @@ async def query_creation (
     "question": query.question,
     "answer": answer,
     "createdAt": now.isoformat(),
-    "queriedAt": queriedAt
+    "queriedAt": queriedAt,
+    "badAnswer": False
   }
 
   try:
@@ -115,12 +119,12 @@ def find_or_create_session(userId: str, now: datetime.datetime):
   if userId is None:
     userId = '-'
 
-  SESSION_MAX_DURATION_DAYS = int(os.getenv('SESSION_MAX_DURATION_DAYS', '1'))
+  SESSION_MAX_DURATION_DAYS = float(os.getenv('CHB_SESSION_MAX_DURATION_DAYS', '1'))
   datetimeLimit = now - datetime.timedelta(SESSION_MAX_DURATION_DAYS - 1)
   startOfDay = datetime.datetime.combine(datetimeLimit, datetime.time.min)
   # trovare una sessione con createdAt > datetimeLimit
   try:
-    db_response = table_sessions.query(
+    dbResponse = table_sessions.query(
       KeyConditionExpression=Key("userId").eq(userId) &
         Key('createdAt').gt(startOfDay.isoformat()),
       IndexName='SessionsByCreatedAtIndex',
@@ -130,7 +134,7 @@ def find_or_create_session(userId: str, now: datetime.datetime):
   except (BotoCoreError, ClientError) as e:
     raise HTTPException(status_code=422, detail=f"[find_or_create_session] userId: {userId}, error: {e}")
 
-  items = db_response.get('Items', [])
+  items = dbResponse.get('Items', [])
   if len(items) == 0:
     body = {
       "id": f'{uuid.uuid4()}',
@@ -149,29 +153,44 @@ def find_or_create_session(userId: str, now: datetime.datetime):
   return body
 
 
-@app.get("/queries/{id}")
-async def query_fetching(id: str):
-  # TODO: dynamoDB integration
-  body = {
-    "id": id,
-    "sessionId": "",
-    "question": "",
-    "answer": "",
-    "createdAt": "",
-    "queriedAt": ""
-  }
-  return body
+@app.get("/queries")
+async def queries_fetching(
+  sessionId: str | None = None,
+  page: int | None = 1,
+  pageSize: int | None = 10,
+  authorization: Annotated[str | None, Header()] = None
+):
+  userId = current_user_id(authorization)
+  if sessionId is None:
+    sessionId = last_session_id(userId)
+
+  if sessionId is None:
+    result = []
+  else:
+    try:
+      dbResponse = table_queries.query(
+        KeyConditionExpression=Key('sessionId').eq(sessionId),
+        IndexName='QueriesByCreatedAtIndex',
+        ScanIndexForward=True
+      )
+    except (BotoCoreError, ClientError) as e:
+      raise HTTPException(status_code=422, detail=f"[queries_fetching] sessionId: {sessionId}, error: {e}")
+    result = dbResponse.get('Items', [])
+
+  return result
 
 
 # retrieve sessions of current user
 @app.get("/sessions")
 async def sessions_fetching(
+  page: int = 1,
+  pageSize: int = 10,
   authorization: Annotated[str | None, Header()] = None
 ):
   userId = current_user_id(authorization)
 
   try:
-    db_response = table_sessions.query(
+    dbResponse = table_sessions.query(
       KeyConditionExpression=Key("userId").eq(userId),
       IndexName='SessionsByCreatedAtIndex',
       ScanIndexForward=False
@@ -180,7 +199,7 @@ async def sessions_fetching(
     raise HTTPException(status_code=422, detail=f"[sessions_fetching] userId: {userId}, error: {e}")
 
   # TODO: pagination
-  items = db_response.get('Items', [])
+  items = dbResponse.get('Items', [])
   result = {
     "items": items,
     "page": 1,
@@ -200,12 +219,12 @@ async def session_delete(
     "id": id,
   }
   try:
-    db_response_queries = table_queries.query(
+    dbResponse_queries = table_queries.query(
       KeyConditionExpression=Key("sessionId").eq(id)
     )
     # TODO: use batch writer
 #    with table_sessions.batch_writer() as batch:
-    for query in db_response_queries['Items']:
+    for query in dbResponse_queries['Items']:
       table_queries.delete_item(
         Key={
           "id": query["id"],
@@ -225,50 +244,47 @@ async def session_delete(
 
   return body
 
-@app.get("/queries")
-async def queries_fetching(
-  sessionId: str | None = None,
-  authorization: Annotated[str | None, Header()] = None
-):
-  userId = current_user_id(authorization)
-  if sessionId is None:
-    sessionId = last_session_id(userId)
-
-  try:
-    db_response = table_queries.query(
-      KeyConditionExpression=Key("sessionId").eq(sessionId) &
-        Key("id").eq(userId)
-    )
-  except (BotoCoreError, ClientError) as e:
-    raise HTTPException(status_code=422, detail=f"[queries_fetching] sessionId: {sessionId}, error: {e}")
-
-  result = db_response.get('Items', [])
-  return result
-
 
 def last_session_id(userId: str):
-  db_response = table_sessions.query(
+  dbResponse = table_sessions.query(
     IndexName='SessionsByCreatedAtIndex',
     KeyConditionExpression=Key('userId').eq(userId),
     ScanIndexForward=False,
     Limit=1
   )
-  items = db_response.get('Items', [])
-  return items[0] if items else None
+  items = dbResponse.get('Items', [])
+  return items[0].get('id', None) if items else None
 
-@app.patch("/queries/{id}")
-async def query_feedback (badAnswer: bool):
-  # TODO: dynamoDB integration
-  body = {
-    "id": "",
-    "sessionId": "",
-    "question": "",
-    "answer": "",
-    "badAnswer": badAnswer,
-    "createdAt": "",
-    "queriedAt": ""
-  }
-  return body
+@app.patch("/sessions/{sessionId}/queries/{id}")
+async def query_feedback (
+  id: str,
+  sessionId: str,
+  query: QueryFeedback,
+  authorization: Annotated[str | None, Header()] = None
+):
+
+  try:
+    dbResponse = table_queries.update_item(
+      Key={
+        'sessionId': sessionId,
+        'id': id
+      },
+      UpdateExpression='SET #badAnswer = :badAnswer',
+      ExpressionAttributeNames={
+        '#badAnswer': 'badAnswer'
+      },
+      ExpressionAttributeValues={
+        ':badAnswer': query.badAnswer
+      },
+      ReturnValues='ALL_NEW'
+    )
+  except (BotoCoreError, ClientError) as e:
+    raise HTTPException(status_code=422, detail=f"[query_feedback] id: {id}, sessionId: {sessionId}, error: {e}")
+
+  if 'Attributes' in dbResponse:
+    return dbResponse.get('Attributes')
+  else:
+    raise HTTPException(status_code=404, detail="Record not found")
 
 handler = mangum.Mangum(app, lifespan="off")
 

apps/chatbot/src/modules/chatbot.py

@@ -175,6 +175,7 @@ def generate(self, query_str: str) -> str:
                     response_str = "Mi dispiace, ma non posso fornire informazioni che potrebbero essere pericolose o dannose."
                     logging.info("Gemini Safety: blocked query because retrieved DANGEROUS_CONTENT in it.")
             else:
-                logging.info(exception_str)
+                response_str = ""
+                logging.info(f"[chatbot.py] generate: exception_str = {exception_str}")
 
         return response_str

apps/infrastructure/src/modules/chatbot/dynamodb.tf

@@ -18,6 +18,20 @@ module "dynamodb_chatbot_queries" {
       name = "sessionId"
       type = "S"
     },
+    {
+      name = "createdAt"
+      type = "S"
+    },
+  ]
+
+  # LSI for query on created_at
+  local_secondary_indexes = [
+    {
+      name            = "QueriesByCreatedAtIndex"
+      hash_key        = "userId"
+      range_key       = "createdAt"
+      projection_type = "ALL"
+    }
   ]
 }