from __future__ import annotations # For self-referencing annotations import json import os import shutil import sqlite3 import sys from huggingface_hub import Repository from random import choices from typing import List, Dict, Optional from src.architectures import Architecture from src.common import data_dir class TestGenerator: """ Wrapper class to hold testing questions and serve up examples """ questions: List[str] = None @classmethod def load_questions(cls, reload=False) -> None: """ Load the available questions from the json file. Default to not re-loading if already done, but allow for the option to do so """ if cls.questions is not None and not reload: return question_file = os.path.join(data_dir, 'json', 'test_questions.json') with open(question_file, 'r') as f: question_json = json.load(f) cls.questions = question_json['questions'] @classmethod def question_count(cls) -> int: """ The total number of questions in the question set """ cls.load_questions() return len(cls.questions) @classmethod def get_random_questions(cls, n: int): """ Return n random questions """ cls.load_questions() return choices(cls.questions, k=n) class ArchitectureRequestRecord: """ Representation of the test data associated with each invocation of an architecture """ all: List[ArchitectureRequestRecord] = None class ArchStep: """ Inner class to just hold this data """ def __init__(self, name: str, start: int, end: int): self.name = name self.start = start self.end = end self.elapsed = end - start def __init__(self, arch: str, request: str, response: str, response_len: int, start: int, end: int, elapsed: int, tags: List[str], test_group: Optional[str], comment: str, steps: List[ArchitectureRequestRecord.ArchStep]): self.arch = arch self.request = request self.response = response self.response_len = response_len self.start = start self.end = end self.elapsed = elapsed self.tags = tags self.test_group = test_group self.comment = comment self.steps = steps @classmethod def from_dict(cls, test: Dict) -> ArchitectureRequestRecord: arch = test['architecture'] request = test['request']['request_evolution'][0] response = "" if len(test['request']['response_evolution']) == 0: response_len = 0 else: response_len = len(test['request']['response_evolution'][-1]) response = test['request']['response_evolution'][-1] start = test['trace']['steps'][0]['start_ms'] end = test['trace']['steps'][-1]['end_ms'] elapsed = end - start tags = test['test_tags'] test_group = None for tag in tags: if tag.startswith("TestGroup"): test_group = tag comment = test['test_comment'] steps = [] for s in test['trace']['steps']: steps.append(ArchitectureRequestRecord.ArchStep(s['name'], s['start_ms'], s['end_ms'])) return ArchitectureRequestRecord(arch, request, response, response_len, start, end, elapsed, tags, test_group, comment, steps) @classmethod def load_all(cls, reload=False) -> None: """ Load all the traces from json trace log """ if cls.all is None or reload: records = [] test_traces = Architecture.get_trace_records() for trace in test_traces: records.append(ArchitectureRequestRecord.from_dict(trace)) cls.all = records class TestGroup: """ A class representing a single batch run of tests from the UI. Identified by the tag which was assigned from the UI when the test was run, and including summary items (start, end, elapse) for convenience """ all: Dict[str, TestGroup] = None def __init__(self, test_group:str): self.arch_request_records: List[ArchitectureRequestRecord] = [] self.test_group = test_group self.comment = None self.start = None self.end = None self.elapsed = None self.architectures = set() @property def num_archs(self) -> int: """ The number of LLM Architectures which were included in this test run from the UI """ return len(self.architectures) @property def num_tests(self) -> int: """ The total number of Architecture tests (inferences) done in this test run from the UI """ return len(self.arch_request_records) @property def num_tests_per_arch(self) -> int: """ The calculated number of tests run through each architecture (simple divide as the UI forces each architecture to get the same number of requests) """ # Should always be an even number but cast to int just in case return int(self.num_tests / self.num_archs) def arch_request_records_by_arch(self) -> Dict[List[ArchitectureRequestRecord]]: """ Get all the tests ArchitectureRequestRecords grouped by the architecture. :return: dict keyed by the architecture name containing a list of ArchitectureRequestRecords detailing the tests run through that architecture. Note - the keys are intended to be used for display purposes - attempting to use them to load the original architecture will be dependent on the availability of that architecture at look up time and changes in architecture config could cause that lookup to fail (i.e. the tested architecture is no longer configured). """ grouped = {} for arr in self.arch_request_records: if arr.arch not in grouped: grouped[arr.arch] = [] grouped[arr.arch].append(arr) return grouped def summary_stats_by_arch(self) -> List[Dict]: """ Get a pack of statistics for use in the UI, detailing this TestGroup. :return: a list, sorted by architecture name, of statistics per architecture. Each list item is a dict of information (arch_name, elapsed[list of elapsed times in ms], response_len[list of the lengthe of the final response in characters], steps[list of the individual architecture steps each containing dict of name, mean_elapsed(ms)]) """ arch_records = self.arch_request_records_by_arch() arch_names = list(arch_records.keys()) arch_names.sort() stats = [] for a in arch_names: stat_pack = {'arch_name': a, 'elapsed': [rec.elapsed for rec in arch_records[a]], 'response_len': [rec.response_len for rec in arch_records[a]], 'steps': [], 'q_and_a': {}} for rec in arch_records[a]: stat_pack['q_and_a'][rec.request] = rec.response for i in range(len(arch_records[a][0].steps)): stat_pack['steps'].append({'step_name': arch_records[a][0].steps[i].name}) num_recs = len(arch_records[a]) total_elapsed = 0 for j in range(num_recs): total_elapsed += arch_records[a][j].steps[i].elapsed stat_pack['steps'][-1]['mean_elapsed'] = total_elapsed / num_recs stats.append(stat_pack) return stats def add_record(self, arr: ArchitectureRequestRecord) -> None: """ Add an ArchitectureRequestRecord into this test group. Update the TestGroup level start, end and elapsed with the new data """ if arr.test_group != self.test_group: raise ValueError("Attempted to group a test record into the wrong group") self.arch_request_records.append(arr) self.architectures.add(arr.arch) if self.comment is None: self.comment = arr.comment if self.start is None or self.start > arr.start: self.start = arr.start if self.end is None or self.end < arr.end: self.end = arr.end self.elapsed = self.end - self.start @classmethod def load_json_test_groups(cls, reload: bool = False) -> List[TestGroup]: """ Load all the test groups from the local json file, reloading from the HF Hub if requested """ ArchitectureRequestRecord.load_all(reload=reload) test_groups: Dict[str, TestGroup] = {} for arr in ArchitectureRequestRecord.all: if arr.test_group is not None: if arr.test_group not in test_groups: test_groups[arr.test_group] = TestGroup(arr.test_group) test_groups[arr.test_group].add_record(arr) return list(test_groups.values()) @classmethod def load_db_test_groups(cls) -> List[TestGroup]: """ Load all the test groups from the DataBase """ db_file = os.path.join(data_dir, 'sqlite', 'test_records.db') con = sqlite3.connect(db_file) cur = con.cursor() sql = "SELECT id, test_group from test_groups" cur.execute(sql) tg_id_names = [(r[0], r[1]) for r in cur.fetchall()] test_groups: List[TestGroup] = [] for tg_id, tg_name in tg_id_names: tg = TestGroup(tg_name) sql = f"SELECT id, arch_name, request, response, response_len, start, end, comment FROM arch_requests WHERE test_group_id={tg_id}" cur.execute(sql) arch_requests = [(r[0], r[1], r[2], r[3], r[4], r[5], r[6], r[7]) for r in cur.fetchall()] for ar_id, ar_arch_name, ar_req, ar_resp, ar_resp_len, ar_start, ar_end, ar_comment in arch_requests: sql = f"SELECT name, start, end FROM arch_req_steps WHERE arch_req_id={ar_id}" cur.execute(sql) steps = [ArchitectureRequestRecord.ArchStep(r[0], r[1], r[2]) for r in cur.fetchall()] arch_req_record = ArchitectureRequestRecord(ar_arch_name, ar_req, ar_resp, ar_resp_len, ar_start, ar_end, (ar_end - ar_start), [tg_name], tg_name, f"(DB persisted) {ar_comment}", steps) tg.add_record(arch_req_record) test_groups.append(tg) return test_groups @classmethod def load_all(cls, reload: bool = False): """ Load all the available TestGroups, from both the json file and the DB into the class variable - for efficiency do not reload unless requested """ if cls.all is None or reload: working_test_groups = {} json_tgs = cls.load_json_test_groups(reload=reload) for test_group in json_tgs: working_test_groups[test_group.test_group] = test_group db_tgs = cls.load_db_test_groups() for test_group in db_tgs: working_test_groups[test_group.test_group] = test_group cls.all = working_test_groups @classmethod def for_test_group_tag(cls, test_group_tag: str) -> TestGroup: """ Get a single TestGroup based on the test_group_tag which was assigned when the test was run """ cls.load_all() return cls.all[test_group_tag] def move_test_records_to_db(hf_hub_token: str) -> None: """ This is an offline utility to move the transaction logs from a flat file to a database. To keep things simpler transaction logs are initially stored into a json file as a Hugging Face dataset, but this can get cumbersome, so this utility will move the records into an sqlite database. It can be run periodically just to move them across. """ def download_latest_json_file(hf_hub_token: str) -> Repository: """ Wipe any local version of the json file and re-downlad from the HF Hub """ if os.path.exists(Architecture.trace_dir): shutil.rmtree(Architecture.trace_dir) return Repository(local_dir=Architecture.trace_dir, clone_from=Architecture.save_repo_url, token=hf_hub_token) def create_local_db(): """ Create the local database if it does not exist """ db_file = os.path.join(data_dir, 'sqlite', 'test_records.db') con = sqlite3.connect(db_file) sql = "CREATE TABLE test_groups (id INTEGER PRIMARY KEY AUTOINCREMENT, test_group TEXT NOT NULL, start INTEGER NOT NULL, end INTEGER NOT NULL);" con.execute(sql) sql = "CREATE TABLE arch_requests (id INTEGER PRIMARY KEY AUTOINCREMENT, arch_name TEXT NOT NULL, request TEXT NOT NULL, response TEXT NOT NULL, response_len INTEGER NOT NULL, start INTEGER NOT NULL, end INTEGER NOT NULL, comment TEXT NOT NULL, test_group_id INTEGER NOT NULL, FOREIGN KEY (test_group_id) REFERENCES test_groups (id))" con.execute(sql) sql = "CREATE TABLE arch_req_steps (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, start INTEGER NOT NULL, end INTEGER NOT NULL, arch_req_id INTEGER NOT NULL, FOREIGN KEY (arch_req_id) REFERENCES arch_requests(id))" con.execute(sql) def get_local_db() -> sqlite3.Connection: """ Get a connection to the local database and create it if it is not already there """ db_file = os.path.join(data_dir, 'sqlite', 'test_records.db') if not os.path.exists(db_file): create_local_db() return sqlite3.connect(db_file) def load_test_group_to_db(test_group: TestGroup, con: sqlite3.Connection) -> None: """ Load a single TestGroup object into the DB, decomposing to the TestGroup, ArchitectureRequest within that and ArchitectureRequestSteps within those """ cur = con.cursor() sql = f'SELECT count(*) from test_groups where test_group ="{test_group.test_group}"' cur.execute(sql) tg_not_in_db = cur.fetchall()[0][0] == 0 if tg_not_in_db: sql = f'INSERT into test_groups (test_group, start, end) VALUES ("{test_group.test_group}", {test_group.start}, {test_group.end})' tg_id = con.execute(sql).lastrowid for arr in test_group.arch_request_records: sql = f'INSERT INTO arch_requests (arch_name, request, response, response_len, start, end, comment, test_group_id) VALUES ("{arr.arch}", ?, ?, {arr.response_len}, {arr.start}, {arr.end}, "{arr.comment}", {tg_id})' arr_id = con.execute(sql, (arr.request, arr.response)).lastrowid for s in arr.steps: sql= f'INSERT INTO arch_req_steps (name, start, end, arch_req_id) VALUES ("{s.name}", {s.start}, {s.end}, {arr_id})' con.execute(sql) con.commit() else: print(f"Warning TestGroup {test_group.test_group} was not added to the DB as it already existed there") def load_all_test_groups_to_db(con: sqlite3.Connection) -> None: """ Load a list of TestGroups to the DB, one at a time """ TestGroup.load_all() for tg in TestGroup.all.values(): load_test_group_to_db(tg, con) """ Main control flow using utility nested functions above for better structure and readability """ download_latest_json_file(hf_hub_token) conn = get_local_db() load_all_test_groups_to_db(conn) Architecture.wipe_trace(hf_hub_token) print("REMINDER: need to commit the local sqlite file to make it available to the server") if __name__ == "__main__": # Expected to only be directly called for the json to db transfer - arg should be the HF token hf_token = sys.argv[1] move_test_records_to_db(hf_token)