Implementacja DSM wraz z testami

attributes.json

@@ -1,9 +1,6 @@
 {
-    "addr": null,
-    "confirm": [
-        "true",
-        "false"
-    ],
+    "address": null,
+    "order-complete": false,
     "dough": ["thick"],
     "drink": ["pepsi", "cola", "water"],
     "food": ["pizza"],

requirements.txt

@@ -2,4 +2,5 @@ flair==0.13.1
 conllu==4.5.3
 pandas==1.5.3
 numpy==1.26.4
-torch==2.3.0
+torch==2.3.0
+convlab==3.0.2a0

src/main.py

@@ -18,10 +18,10 @@ while True:
     # print(frame)
 
     # DSM
-    # monitor.append(frame)
+    monitor.update(frame)
 
     # DP
-    # print(dialog_policy.next_dialogue_act(monitor.get_all()).act)
+    # print(dialog_policy.next_dialogue_act(monitor.read()).act)
 
     # NLG
     act, slots = parse_frame(frame)

src/service/dialog_state_monitor.py

@@ -1,14 +1,55 @@
-from model.frame import Frame
+from src.model.frame import Frame
+from convlab.dst.dst import DST
+import json
+import copy
 
 
 class DialogStateMonitor:
-    dialog = []
+    def __init__(self, initial_state_file: str = '../attributes.json'):
+        DST.__init__(self)
+        with open(initial_state_file) as file:
+            self.__initial_state = json.load(file)
+        self.__memory = copy.deepcopy(self.__initial_state)
 
-    def append(self, frame: Frame):
-        self.dialog.append(frame)
+    # def __access_memory__(self, path: str) -> str | int | float | None:
+    #     result = self.state['memory']
+    #     for segment in path.split('.'):
+    #         if segment not in result:
+    #             return None
+    #         result = result[segment]
+    #     return result
 
-    def get_all(self) -> [Frame]:
-        return self.dialog
+    def update(self, frame: Frame):
+        if frame.source != 'user':
+            return
+        if frame.act == 'inform/order':
+            new_order = dict()
+            for slot in frame.slots:
+                new_order[slot.name] = slot.value
+            self.__memory['order'].append(new_order)
+        elif frame.act == 'inform/address':
+            for slot in frame.slots:
+                self.__memory['address'][slot.name] = slot.value
+        elif frame.act == 'inform/phone':
+            for slot in frame.slots:
+                self.__memory['phone'][slot.name] = slot.value
+        elif frame.act == 'inform/order-complete':
+            self.__memory['order-complete'] = True
+        elif frame.act == 'inform/delivery':
+            for slot in frame.slots:
+                self.__memory['delivery'][slot.name] = slot.value
+        elif frame.act == 'inform/payment':
+            for slot in frame.slots:
+                self.__memory['payment'][slot.name] = slot.value
+        elif frame.act == 'inform/time':
+            for slot in frame.slots:
+                self.__memory['time'][slot.name] = slot.value
+        elif frame.act == 'inform/name':
+            for slot in frame.slots:
+                self.__memory['name'][slot.name] = slot.value
 
-    def get_last(self) -> Frame:
-        return self.dialog[len(self.dialog) - 1]
+    def read(self) -> dict:
+        return self.__memory
+
+    def reset(self):
+        self.__memory = copy.deepcopy(self.__initial_state)

src/test/dialog_state_monitor.py

@@ -0,0 +1,20 @@
+from src.service.dialog_state_monitor import DialogStateMonitor
+from src.model.frame import Frame
+from src.model.slot import Slot
+
+dst = DialogStateMonitor()
+
+assert dst.read()['pizza']['capri']['price'] == 25
+
+dst.update(Frame('user', 'inform/order', [Slot('B-pizza', 'margaritta'), Slot('B-sauce', 'ketchup')]))
+dst.update(Frame('user', 'inform/order', [Slot('B-pizza', 'carbonara')]))
+dst.update(Frame('user', 'inform/order-complete', []))
+
+assert dst.read()['order'][0]['B-pizza'] == 'margaritta'
+assert dst.read()['order'][0]['B-sauce'] == 'ketchup'
+assert dst.read()['order-complete'] == True
+
+dst.reset()
+
+assert dst.read()['order'] == []
+assert dst.read()['order-complete'] == False