Fix feedback table in server

Author: nik

adala/environments/servers/base.py

@@ -10,18 +10,11 @@
 STORAGE_DB = 'feedback.db'
 
 
-class GroundTruth(BaseModel):
+class Feedback(BaseModel):
     prediction_id: int
-    skill_output: str
-    gt_match: Optional[bool] = None
-    gt_data: Optional[str] = None
-
-
-class Prediction(BaseModel):
-    id: int
-    input: Dict[str, Any]
-    skill_name: str
-    output: str
+    prediction_column: str
+    fb_match: Optional[bool] = None
+    fb_message: Optional[str] = None
 
 
 router = APIRouter()
@@ -38,34 +31,34 @@ async def init_db(self):
         print(f'Initializing database {STORAGE_DB}...')
         async with aiosqlite.connect(STORAGE_DB) as db:
             await db.execute('''
-                CREATE TABLE IF NOT EXISTS ground_truth (
+                CREATE TABLE IF NOT EXISTS feedback (
                     prediction_id INTEGER NOT NULL,
-                    skill_name TEXT NOT NULL,
-                    gt_match BOOLEAN,
-                    gt_data TEXT,
-                    PRIMARY KEY (prediction_id, skill_name)
+                    prediction_column TEXT NOT NULL,
+                    fb_match BOOLEAN,
+                    fb_message TEXT,
+                    PRIMARY KEY (prediction_id, prediction_column)
                 )
             ''')
             await db.commit()
 
     async def request_feedback(
         self,
-        predictions: List[Prediction],
+        predictions: List[Dict[str, Any]],
         skills: List[Dict[str, Any]],
         db: aiosqlite.Connection
     ):
         raise NotImplementedError
 
-    async def retrieve_ground_truth(self, db: aiosqlite.Connection):
-        cursor = await db.execute('SELECT prediction_id, skill_name, gt_match, gt_data FROM ground_truth')
+    async def retrieve_feedback(self, db: aiosqlite.Connection):
+        cursor = await db.execute('SELECT prediction_id, prediction_column, fb_match, fb_message FROM feedback')
         rows = await cursor.fetchall()
-        return [GroundTruth(prediction_id=row[0], skill_name=row[1], gt_match=row[2], gt_data=row[3]) for row in rows]
+        return [Feedback(prediction_id=row[0], prediction_column=row[1], fb_match=row[2], fb_message=row[3]) for row in rows]
 
-    async def store_ground_truths(self, ground_truths: List[GroundTruth], db: aiosqlite.Connection):
+    async def store_feedback(self, feedbacks: List[Feedback], db: aiosqlite.Connection):
         await db.executemany('''
-            INSERT OR REPLACE INTO ground_truth (prediction_id, skill_name, gt_match, gt_data)
+            INSERT OR REPLACE INTO feedback (prediction_id, prediction_column, fb_match, fb_message)
             VALUES (?, ?, ?, ?)
-        ''', [(gt.prediction_id, gt.skill_name, gt.gt_match, gt.gt_data) for gt in ground_truths])
+        ''', [(fb.prediction_id, fb.prediction_column, fb.fb_match, fb.fb_message) for fb in feedbacks])
         await db.commit()
 
 
@@ -79,23 +72,23 @@ async def get_db() -> aiosqlite.Connection:
 app = BaseAPI()
 
 
-@router.post("/feedback")
-async def create_feedback(
+@router.post("/request-feedback")
+async def request_feedback(
     request: Request,
     predictions: List[Dict[str, Any]],
     skills: List[Dict[str, Any]],
     db: aiosqlite.Connection = Depends(get_db)
 ):
     app = request.app
     await app.request_feedback(predictions, skills, db)
-    return {"message": "Feedback received successfully"}
+    return {"message": "Feedback requested successfully"}
 
 
-@router.get("/ground-truth", response_model=List[GroundTruth])
-async def get_ground_truth(request: Request, db: aiosqlite.Connection = Depends(get_db)):
+@router.get("/feedback", response_model=List[Feedback])
+async def get_feedback(request: Request, db: aiosqlite.Connection = Depends(get_db)):
     app = request.app
-    ground_truths = await app.retrieve_ground_truth(db)
-    return ground_truths
+    fb = await app.retrieve_feedback(db)
+    return fb
 
 
 @app.on_event("startup")

adala/environments/servers/discord_bot.py

@@ -7,7 +7,7 @@
 from typing import List, Dict, Any
 from discord.ext import commands
 from discord.ui import View
-from adala.environments.servers.base import BaseAPI, Prediction, GroundTruth, STORAGE_DB
+from adala.environments.servers.base import BaseAPI, Feedback, STORAGE_DB
 
 
 intents = discord.Intents.default()
@@ -43,17 +43,17 @@ async def on_message(message):
         initial_message_id = message.reference.message_id
         print(f'Got reply in thread for message id: {initial_message_id}', message)
         async with aiosqlite.connect(STORAGE_DB) as db:
-            async with db.execute('SELECT * FROM discord_gt_message WHERE message_id = ?', (initial_message_id,)) as cursor:
+            async with db.execute('SELECT * FROM discord_fb_message WHERE message_id = ?', (initial_message_id,)) as cursor:
                 row = await cursor.fetchone()
                 if row is None:
                     print(f'No ground truth message found for message id: {initial_message_id}')
                     return
-                prediction_id, skill_name = int(row[1]), row[2]
+                prediction_id, prediction_column = int(row[1]), row[2]
 
             # update ground truth with reply
-            await db.execute('UPDATE ground_truth SET gt_data = ? '
-                             'WHERE prediction_id = ? AND skill_name = ?',
-                             (message.content, prediction_id, skill_name))
+            await db.execute('UPDATE feedback SET fb_message = ? '
+                             'WHERE prediction_id = ? AND prediction_column = ?',
+                             (message.content, prediction_id, prediction_column))
             await db.commit()
 
     # Process other messages normally
@@ -63,46 +63,46 @@ async def on_message(message):
 @bot.event
 async def on_interaction(interaction: discord.Interaction):
 
-    async def update_ground_truth_match(prediction_id: int, skill_name: str, match: bool):
+    async def update_feedback_match(prediction_id: int, prediction_column: str, match: bool):
         async with aiosqlite.connect(STORAGE_DB) as db:
-            await db.execute('UPDATE ground_truth SET gt_match = ?, gt_data = NULL '
-                             'WHERE prediction_id = ? AND skill_name = ?',
-                             (match, prediction_id, skill_name))
+            await db.execute('UPDATE feedback SET fb_match = ?, fb_message = NULL '
+                             'WHERE prediction_id = ? AND prediction_column = ?',
+                             (match, prediction_id, prediction_column))
             await db.commit()
             print(f'Updated ground truth for prediction id: {prediction_id} with match: {match}')
 
     if interaction.type == discord.InteractionType.component:
         await interaction.response.defer(ephemeral=True)
 
         custom_id = interaction.data['custom_id']
-        action, prediction_id_str, skill_name = custom_id.split(':')
+        action, prediction_id_str, prediction_column = custom_id.split(':')
         prediction_id = int(prediction_id_str)  # Convert prediction_id to int
 
         if action == 'accept':
             # Handle the accept action
-            await update_ground_truth_match(prediction_id, skill_name, True)
+            await update_feedback_match(prediction_id, prediction_column, True)
             # React with a checkmark emoji to the message
             await interaction.message.add_reaction('✅')
         elif action == 'reject':
             # Handle the reject action
-            await update_ground_truth_match(prediction_id, skill_name, False)
+            await update_feedback_match(prediction_id, prediction_column, False)
             # React with a cross mark emoji to the message
             await interaction.message.add_reaction('❌')
 
 
 class AcceptRejectView(View):
 
-    def __init__(self, prediction_id: int, skill_name: str, *args, **kwargs):
+    def __init__(self, prediction_id: int, prediction_column: str, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.add_item(discord.ui.Button(
             label='Accept',
             style=discord.ButtonStyle.success,
-            custom_id=f'accept:{prediction_id}:{skill_name}'
+            custom_id=f'accept:{prediction_id}:{prediction_column}'
         ))
         self.add_item(discord.ui.Button(
             label='Reject',
             style=discord.ButtonStyle.danger,
-            custom_id=f'reject:{prediction_id}:{skill_name}'
+            custom_id=f'reject:{prediction_id}:{prediction_column}'
         ))
 
     async def interaction_check(self, interaction: discord.Interaction) -> bool:
@@ -115,10 +115,10 @@ class DiscordAPI(BaseAPI):
     async def init_db_gt_message(self):
         async with aiosqlite.connect(STORAGE_DB) as db:
             await db.execute('''
-                CREATE TABLE IF NOT EXISTS discord_gt_message (
+                CREATE TABLE IF NOT EXISTS discord_fb_message (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
                     prediction_id INTEGER NOT NULL,
-                    skill_name TEXT NOT NULL,
+                    prediction_column TEXT NOT NULL,
                     message_id INTEGER NOT NULL
                 )
             ''')
@@ -140,29 +140,29 @@ async def request_feedback(
         channel = bot.get_channel(CHANNEL_ID)
         if not channel:
             raise Exception(f'Channel with id {CHANNEL_ID} not found')
-        ground_truths = []
+        fbs = []
         skill_outputs = sum([skill['outputs'] for skill in skills], [])
         for skill_output in skill_outputs:
             for prediction in predictions:
                 text = '========================\n'
                 text += '\n'.join(f'**{k}**: {v}' for k, v in prediction.items() if k not in skill_outputs + ['index'])
                 text += f'\n\n__**{skill_output}**__: {prediction[skill_output]}'
-                ground_truth = GroundTruth(prediction_id=prediction['index'], skill_name=skill_output)
-
+                text = text[:2000]
+                fb = Feedback(prediction_id=prediction['index'], prediction_column=skill_output)
                 message = await channel.send(
                     text, view=AcceptRejectView(
-                        prediction_id=ground_truth.prediction_id,
-                        skill_name=ground_truth.skill_name)
+                        prediction_id=fb.prediction_id,
+                        prediction_column=fb.prediction_column)
                 )
-                ground_truths.append(ground_truth)
+                fbs.append(fb)
                 await db.execute('''
-                    INSERT INTO discord_gt_message (prediction_id, skill_name, message_id)
+                    INSERT INTO discord_fb_message (prediction_id, prediction_column, message_id)
                     VALUES (?, ?, ?)
-                ''', (ground_truth.prediction_id, ground_truth.skill_name, message.id))
+                ''', (fb.prediction_id, fb.prediction_column, message.id))
                 await db.commit()
 
         # TODO: do we need to store it in advance?
-        await self.store_ground_truths(ground_truths, db)
+        await self.store_feedback(fbs, db)
 
 
 app = DiscordAPI()

adala/environments/web.py

@@ -1,8 +1,8 @@
 import requests
 import time
 from typing import Optional
-from .base import StaticEnvironment
-from .servers.base import GroundTruth
+from .base import StaticEnvironment, EnvironmentFeedback
+from .servers.base import Feedback
 from adala.skills import SkillSet
 from adala.utils.internal_data import InternalDataFrame, InternalSeries
 from collections import defaultdict
@@ -13,17 +13,22 @@ class WebStaticEnvironment(StaticEnvironment):
     """
     Web environment interacts with server API to request feedback and retrieve ground truth.
     Following endpoints are expected:
-    - POST /feedback
-    - GET /ground-truth
+    - POST /request-feedback
+    - GET /feedback
     """
     url: str
 
+    def _get_fb_records(self):
+        fb_records = requests.get(f'{self.url}/feedback', timeout=3).json()
+        fb_records = [Feedback(**r) for r in fb_records]
+        return fb_records
+
     def get_feedback(
         self,
         skills: SkillSet,
         predictions: InternalDataFrame,
         num_feedbacks: Optional[int] = None,
-    ):
+    ) -> EnvironmentFeedback:
         """
         Request feedback for the predictions.
 
@@ -45,30 +50,35 @@ def get_feedback(
             'predictions': predictions.reset_index().to_dict(orient='records')
         }
 
-        requests.post(f'{self.url}/feedback', json=payload, timeout=3)
+        requests.post(f'{self.url}/request-feedback', json=payload, timeout=3)
 
         # wait for feedback
         with Progress() as progress:
             task = progress.add_task(f"Waiting for feedback...", total=3600)
-            gt_records = []
-            while len(gt_records) < num_feedbacks:
+            fb_records = []
+            while len(fb_records) < num_feedbacks:
                 progress.advance(task, 10)
                 time.sleep(10)
-                gt_records = self.get_gt_records()
+                fb_records = self._get_fb_records()
+                print('ZZZZ', fb_records)
 
-        if not gt_records:
+        if not fb_records:
             raise RuntimeError('No ground truth found.')
 
-        gt = defaultdict(dict)
-        for g in gt_records:
-            gt[g.skill_output][g.prediction_id] = g.gt_data or True
+        match = defaultdict(list)
+        feedback = defaultdict(list)
+        index = []
+        for f in fb_records:
+            match[f.prediction_column].append(f.fb_match)
+            feedback[f.prediction_column].append(f.fb_message)
+            index.append(f.prediction_id)
+
+        print(11111, match)
+        print(2222, feedback)
+        print(3333, fb_records)
 
-        df = InternalDataFrame({skill: InternalSeries(g) for skill, g in gt.items()})
+        match = InternalDataFrame(match, index=index)
+        feedback = InternalDataFrame(feedback, index=index)
 
-        return df
 
-    def get_gt_records(self):
-        gt_records = requests.get(f'{self.url}/ground-truth', timeout=3).json()
-        gt_records = [GroundTruth(**r) for r in gt_records]
-        gt_records = [r for r in gt_records if r.gt_data or r.gt_match]
-        return gt_records
+        return EnvironmentFeedback(match=match, feedback=feedback)