add missing dependencies

document_qa_engine.py

@@ -12,8 +12,6 @@ from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain.vectorstores import Chroma
 from tqdm import tqdm
 
-from commons.annotations_utils import GrobidProcessor
-
 
 class DocumentQAEngine:
     llm = None

grobid_processors.py

@@ -0,0 +1,726 @@
+import re
+from collections import OrderedDict
+from html import escape
+from pathlib import Path
+
+import dateparser
+import grobid_tei_xml
+from bs4 import BeautifulSoup
+from tqdm import tqdm
+
+from commons import supermat_tei_parser
+
+
+def get_span_start(type, title=None):
+    title_ = ' title="' + title + '"' if title is not None else ""
+    return '<span class="label ' + type + '"' + title_ + '>'
+
+
+def get_span_end():
+    return '</span>'
+
+
+def get_rs_start(type):
+    return '<rs type="' + type + '">'
+
+
+def get_rs_end():
+    return '</rs>'
+
+
+def has_space_between_value_and_unit(quantity):
+    return quantity['offsetEnd'] < quantity['rawUnit']['offsetStart']
+
+
+def decorate_text_with_annotations(text, spans, tag="span"):
+    """
+        Decorate a text using spans, using two style defined by the tag:
+            - "span" generated HTML like annotated text
+            - "rs" generate XML like annotated text (format SuperMat)
+    """
+    sorted_spans = list(sorted(spans, key=lambda item: item['offset_start']))
+    annotated_text = ""
+    start = 0
+    for span in sorted_spans:
+        type = span['type'].replace("<", "").replace(">", "")
+        if 'unit_type' in span and span['unit_type'] is not None:
+            type = span['unit_type'].replace(" ", "_")
+        annotated_text += escape(text[start: span['offset_start']])
+        title = span['quantified'] if 'quantified' in span else None
+        annotated_text += get_span_start(type, title) if tag == "span" else get_rs_start(type)
+        annotated_text += escape(text[span['offset_start']: span['offset_end']])
+        annotated_text += get_span_end() if tag == "span" else get_rs_end()
+
+        start = span['offset_end']
+    annotated_text += escape(text[start: len(text)])
+    return annotated_text
+
+
+def extract_quantities(client, x_all, column_text_index):
+    # relevant_items = ['magnetic field strength', 'magnetic induction', 'maximum energy product',
+    #                   "magnetic flux density", "magnetic flux"]
+    # property_keywords = ['coercivity', 'remanence']
+
+    output_data = []
+
+    for idx, example in tqdm(enumerate(x_all), desc="extract quantities"):
+        text = example[column_text_index]
+        spans = GrobidQuantitiesProcessor(client).extract_quantities(text)
+
+        data_record = {
+            "id": example[0],
+            "filename": example[1],
+            "passage_id": example[2],
+            "text": text,
+            "spans": spans
+        }
+
+        output_data.append(data_record)
+
+    return output_data
+
+
+def extract_materials(client, x_all, column_text_index):
+    output_data = []
+
+    for idx, example in tqdm(enumerate(x_all), desc="extract materials"):
+        text = example[column_text_index]
+        spans = GrobidMaterialsProcessor(client).extract_materials(text)
+        data_record = {
+            "id": example[0],
+            "filename": example[1],
+            "passage_id": example[2],
+            "text": text,
+            "spans": spans
+        }
+
+        output_data.append(data_record)
+
+    return output_data
+
+
+def get_parsed_value_type(quantity):
+    if 'parsedValue' in quantity and 'structure' in quantity['parsedValue']:
+        return quantity['parsedValue']['structure']['type']
+
+
+class BaseProcessor(object):
+    # def __init__(self, grobid_superconductors_client=None, grobid_quantities_client=None):
+    #     self.grobid_superconductors_client = grobid_superconductors_client
+    #     self.grobid_quantities_client = grobid_quantities_client
+
+    patterns = [
+        r'\d+e\d+'
+    ]
+
+    def post_process(self, text):
+        output = text.replace('À', '-')
+        output = output.replace('¼', '=')
+        output = output.replace('þ', '+')
+        output = output.replace('Â', 'x')
+        output = output.replace('$', '~')
+        output = output.replace('−', '-')
+        output = output.replace('–', '-')
+
+        for pattern in self.patterns:
+            output = re.sub(pattern, lambda match: match.group().replace('e', '-'), output)
+
+        return output
+
+
+class GrobidProcessor(BaseProcessor):
+    def __init__(self, grobid_client):
+        # super().__init__()
+        self.grobid_client = grobid_client
+
+    def process_structure(self, input_path):
+        pdf_file, status, text = self.grobid_client.process_pdf("processFulltextDocument",
+                                                                input_path,
+                                                                consolidate_header=True,
+                                                                consolidate_citations=False,
+                                                                segment_sentences=False,
+                                                                tei_coordinates=False,
+                                                                include_raw_citations=False,
+                                                                include_raw_affiliations=False,
+                                                                generateIDs=True)
+
+        if status != 200:
+            return
+
+        output_data = self.parse_grobid_xml(text)
+        output_data['filename'] = Path(pdf_file).stem.replace(".tei", "")
+
+        return output_data
+
+    def process_single(self, input_file):
+        doc = self.process_structure(input_file)
+
+        for paragraph in doc['passages']:
+            entities = self.process_single_text(paragraph['text'])
+            paragraph['spans'] = entities
+
+        return doc
+
+    def parse_grobid_xml(self, text):
+        output_data = OrderedDict()
+
+        doc_biblio = grobid_tei_xml.parse_document_xml(text)
+        biblio = {
+            "doi": doc_biblio.header.doi if doc_biblio.header.doi is not None else "",
+            "authors": ", ".join([author.full_name for author in doc_biblio.header.authors]),
+            "title": doc_biblio.header.title,
+            "hash": doc_biblio.pdf_md5
+        }
+        try:
+            year = dateparser.parse(doc_biblio.header.date).year
+            biblio["year"] = year
+        except:
+            pass
+
+        output_data['biblio'] = biblio
+
+        passages = []
+        output_data['passages'] = passages
+        # if biblio['title'] is not None and len(biblio['title']) > 0:
+        #     passages.append({
+        #         "text": self.post_process(biblio['title']),
+        #         "type": "paragraph",
+        #         "section": "<header>",
+        #         "subSection": "<title>",
+        #         "passage_id": "title0"
+        #     })
+
+        if doc_biblio.abstract is not None and len(doc_biblio.abstract) > 0:
+            passages.append({
+                "text": self.post_process(doc_biblio.abstract),
+                "type": "paragraph",
+                "section": "<header>",
+                "subSection": "<abstract>",
+                "passage_id": "abstract0"
+            })
+
+        soup = BeautifulSoup(text, 'xml')
+        text_blocks_body = get_children_body(soup, verbose=False)
+
+        passages.extend([
+            {
+                "text": self.post_process(''.join(text for text in sentence.find_all(text=True) if
+                                                  text.parent.name != "ref" or (
+                                                          text.parent.name == "ref" and text.parent.attrs[
+                                                      'type'] != 'bibr'))),
+                "type": "paragraph",
+                "section": "<body>",
+                "subSection": "<paragraph>",
+                "passage_id": str(paragraph_id) + str(sentence_id)
+            }
+            for paragraph_id, paragraph in enumerate(text_blocks_body) for
+            sentence_id, sentence in enumerate(paragraph)
+        ])
+
+        text_blocks_figures = get_children_figures(soup, verbose=False)
+
+        passages.extend([
+            {
+                "text": self.post_process(''.join(text for text in sentence.find_all(text=True) if
+                                                  text.parent.name != "ref" or (
+                                                          text.parent.name == "ref" and text.parent.attrs[
+                                                      'type'] != 'bibr'))),
+                "type": "paragraph",
+                "section": "<body>",
+                "subSection": "<figure>",
+                "passage_id": str(paragraph_id) + str(sentence_id)
+            }
+            for paragraph_id, paragraph in enumerate(text_blocks_figures) for
+            sentence_id, sentence in enumerate(paragraph)
+        ])
+
+        return output_data
+
+
+class GrobidQuantitiesProcessor(BaseProcessor):
+    def __init__(self, grobid_quantities_client):
+        self.grobid_quantities_client = grobid_quantities_client
+
+    def extract_quantities(self, text):
+        status, result = self.grobid_quantities_client.process_text(text.strip())
+
+        if status != 200:
+            result = {}
+
+        spans = []
+
+        if 'measurements' in result:
+            found_measurements = self.parse_measurements_output(result)
+
+            for m in found_measurements:
+                item = {
+                    "text": text[m['offset_start']:m['offset_end']],
+                    'offset_start': m['offset_start'],
+                    'offset_end': m['offset_end']
+                }
+
+                if 'raw' in m and m['raw'] != item['text']:
+                    item['text'] = m['raw']
+
+                if 'quantified_substance' in m:
+                    item['quantified'] = m['quantified_substance']
+
+                if 'type' in m:
+                    item["unit_type"] = m['type']
+
+                item['type'] = 'property'
+                # if 'raw_value' in m:
+                #     item['raw_value'] = m['raw_value']
+
+                spans.append(item)
+
+        return spans
+
+    @staticmethod
+    def parse_measurements_output(result):
+        measurements_output = []
+
+        for measurement in result['measurements']:
+            type = measurement['type']
+            measurement_output_object = {}
+            quantity_type = None
+            has_unit = False
+            parsed_value_type = None
+
+            if 'quantified' in measurement:
+                if 'normalizedName' in measurement['quantified']:
+                    quantified_substance = measurement['quantified']['normalizedName']
+                    measurement_output_object["quantified_substance"] = quantified_substance
+
+            if 'measurementOffsets' in measurement:
+                measurement_output_object["offset_start"] = measurement["measurementOffsets"]['start']
+                measurement_output_object["offset_end"] = measurement["measurementOffsets"]['end']
+            else:
+                # If there are no offsets we skip the measurement
+                continue
+
+            # if 'measurementRaw' in measurement:
+            #     measurement_output_object['raw_value'] = measurement['measurementRaw']
+
+            if type == 'value':
+                quantity = measurement['quantity']
+
+                parsed_value = GrobidQuantitiesProcessor.get_parsed(quantity)
+                if parsed_value:
+                    measurement_output_object['parsed'] = parsed_value
+
+                normalized_value = GrobidQuantitiesProcessor.get_normalized(quantity)
+                if normalized_value:
+                    measurement_output_object['normalized'] = normalized_value
+
+                raw_value = GrobidQuantitiesProcessor.get_raw(quantity)
+                if raw_value:
+                    measurement_output_object['raw'] = raw_value
+
+                if 'type' in quantity:
+                    quantity_type = quantity['type']
+
+                if 'rawUnit' in quantity:
+                    has_unit = True
+
+                parsed_value_type = get_parsed_value_type(quantity)
+
+            elif type == 'interval':
+                if 'quantityMost' in measurement:
+                    quantityMost = measurement['quantityMost']
+                    if 'type' in quantityMost:
+                        quantity_type = quantityMost['type']
+
+                    if 'rawUnit' in quantityMost:
+                        has_unit = True
+
+                    parsed_value_type = get_parsed_value_type(quantityMost)
+
+                if 'quantityLeast' in measurement:
+                    quantityLeast = measurement['quantityLeast']
+
+                    if 'type' in quantityLeast:
+                        quantity_type = quantityLeast['type']
+
+                    if 'rawUnit' in quantityLeast:
+                        has_unit = True
+
+                    parsed_value_type = get_parsed_value_type(quantityLeast)
+
+            elif type == 'listc':
+                quantities = measurement['quantities']
+
+                if 'type' in quantities[0]:
+                    quantity_type = quantities[0]['type']
+
+                if 'rawUnit' in quantities[0]:
+                    has_unit = True
+
+                parsed_value_type = get_parsed_value_type(quantities[0])
+
+            if quantity_type is not None or has_unit:
+                measurement_output_object['type'] = quantity_type
+
+            if parsed_value_type is None or parsed_value_type not in ['ALPHABETIC', 'TIME']:
+                measurements_output.append(measurement_output_object)
+
+        return measurements_output
+
+    @staticmethod
+    def get_parsed(quantity):
+        parsed_value = parsed_unit = None
+        if 'parsedValue' in quantity and 'parsed' in quantity['parsedValue']:
+            parsed_value = quantity['parsedValue']['parsed']
+        if 'parsedUnit' in quantity and 'name' in quantity['parsedUnit']:
+            parsed_unit = quantity['parsedUnit']['name']
+
+        if parsed_value and parsed_unit:
+            if has_space_between_value_and_unit(quantity):
+                return str(parsed_value) + str(parsed_unit)
+            else:
+                return str(parsed_value) + " " + str(parsed_unit)
+
+    @staticmethod
+    def get_normalized(quantity):
+        normalized_value = normalized_unit = None
+        if 'normalizedQuantity' in quantity:
+            normalized_value = quantity['normalizedQuantity']
+        if 'normalizedUnit' in quantity and 'name' in quantity['normalizedUnit']:
+            normalized_unit = quantity['normalizedUnit']['name']
+
+        if normalized_value and normalized_unit:
+            if has_space_between_value_and_unit(quantity):
+                return str(normalized_value) + " " + str(normalized_unit)
+            else:
+                return str(normalized_value) + str(normalized_unit)
+
+    @staticmethod
+    def get_raw(quantity):
+        raw_value = raw_unit = None
+        if 'rawValue' in quantity:
+            raw_value = quantity['rawValue']
+        if 'rawUnit' in quantity and 'name' in quantity['rawUnit']:
+            raw_unit = quantity['rawUnit']['name']
+
+        if raw_value and raw_unit:
+            if has_space_between_value_and_unit(quantity):
+                return str(raw_value) + " " + str(raw_unit)
+            else:
+                return str(raw_value) + str(raw_unit)
+
+
+class GrobidMaterialsProcessor(BaseProcessor):
+    def __init__(self, grobid_superconductors_client):
+        self.grobid_superconductors_client = grobid_superconductors_client
+
+    def extract_materials(self, text):
+        status, result = self.grobid_superconductors_client.process_text(text.strip(), "processText_disable_linking")
+
+        if status != 200:
+            result = {}
+
+        spans = []
+
+        if 'passages' in result:
+            materials = self.parse_superconductors_output(result, text)
+
+            for m in materials:
+                item = {"text": text[m['offset_start']:m['offset_end']]}
+
+                item['offset_start'] = m['offset_start']
+                item['offset_end'] = m['offset_end']
+
+                if 'formula' in m:
+                    item["formula"] = m['formula']
+
+                item['type'] = 'material'
+                item['raw_value'] = m['text']
+
+                spans.append(item)
+
+        return spans
+
+    def parse_materials(self, text):
+        status, result = self.grobid_superconductors_client.process_texts(text.strip(), "parseMaterials")
+
+        if status != 200:
+            result = []
+
+        results = []
+        for position_material in result:
+            compositions = []
+            for material in position_material:
+                if 'resolvedFormulas' in material:
+                    for resolved_formula in material['resolvedFormulas']:
+                        if 'formulaComposition' in resolved_formula:
+                            compositions.append(resolved_formula['formulaComposition'])
+                elif 'formula' in material:
+                    if 'formulaComposition' in material['formula']:
+                        compositions.append(material['formula']['formulaComposition'])
+            results.append(compositions)
+
+        return results
+
+    def parse_material(self, text):
+        status, result = self.grobid_superconductors_client.process_text(text.strip(), "parseMaterial")
+
+        if status != 200:
+            result = []
+
+        compositions = []
+        for material in result:
+            if 'resolvedFormulas' in material:
+                for resolved_formula in material['resolvedFormulas']:
+                    if 'formulaComposition' in resolved_formula:
+                        compositions.append(resolved_formula['formulaComposition'])
+            elif 'formula' in material:
+                if 'formulaComposition' in material['formula']:
+                    compositions.append(material['formula']['formulaComposition'])
+
+        return compositions
+
+    @staticmethod
+    def parse_superconductors_output(result, original_text):
+        materials = []
+
+        for passage in result['passages']:
+            sentence_offset = original_text.index(passage['text'])
+            if 'spans' in passage:
+                spans = passage['spans']
+                for material_span in filter(lambda s: s['type'] == '<material>', spans):
+                    text_ = material_span['text']
+
+                    base_material_information = {
+                        "text": text_,
+                        "offset_start": sentence_offset + material_span['offset_start'],
+                        'offset_end': sentence_offset + material_span['offset_end']
+                    }
+
+                    materials.append(base_material_information)
+
+        return materials
+
+
+class GrobidAggregationProcessor(GrobidProcessor, GrobidQuantitiesProcessor, GrobidMaterialsProcessor):
+    def __init__(self, grobid_client, grobid_quantities_client=None, grobid_superconductors_client=None):
+        GrobidProcessor.__init__(self, grobid_client)
+        GrobidQuantitiesProcessor.__init__(self, grobid_quantities_client)
+        GrobidMaterialsProcessor.__init__(self, grobid_superconductors_client)
+
+    def process_single_text(self, text):
+        extracted_quantities_spans = extract_quantities(self.grobid_quantities_client, text)
+        extracted_materials_spans = extract_materials(self.grobid_superconductors_client, text)
+        all_entities = extracted_quantities_spans + extracted_materials_spans
+        entities = self.prune_overlapping_annotations(all_entities)
+        return entities
+
+    @staticmethod
+    def prune_overlapping_annotations(entities: list) -> list:
+        # Sorting by offsets
+        sorted_entities = sorted(entities, key=lambda d: d['offset_start'])
+
+        if len(entities) <= 1:
+            return sorted_entities
+
+        to_be_removed = []
+
+        previous = None
+        first = True
+
+        for current in sorted_entities:
+            if first:
+                first = False
+                previous = current
+                continue
+
+            if previous['offset_start'] < current['offset_start'] \
+                    and previous['offset_end'] < current['offset_end'] \
+                    and (previous['offset_end'] < current['offset_start'] \
+                         and not (previous['text'] == "-" and current['text'][0].isdigit())):
+                previous = current
+                continue
+
+            if previous['offset_end'] < current['offset_end']:
+                if current['type'] == previous['type']:
+                    # Type is the same
+                    if current['offset_start'] == previous['offset_end']:
+                        if current['type'] == 'property':
+                            if current['text'].startswith("."):
+                                print(
+                                    f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+                                # current entity starts with a ".", suspiciously look like a truncated value
+                                to_be_removed.append(previous)
+                                current['text'] = previous['text'] + current['text']
+                                current['raw_value'] = current['text']
+                                current['offset_start'] = previous['offset_start']
+                            elif previous['text'].endswith(".") and current['text'][0].isdigit():
+                                print(
+                                    f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+                                # previous entity ends with ".", current entity starts with a number
+                                to_be_removed.append(previous)
+                                current['text'] = previous['text'] + current['text']
+                                current['raw_value'] = current['text']
+                                current['offset_start'] = previous['offset_start']
+                            elif previous['text'].startswith("-"):
+                                print(
+                                    f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+                                # previous starts with a `-`, sherlock this is another truncated value
+                                current['text'] = previous['text'] + current['text']
+                                current['raw_value'] = current['text']
+                                current['offset_start'] = previous['offset_start']
+                                to_be_removed.append(previous)
+                            else:
+                                print("Other cases to be considered: ", previous, current)
+                        else:
+                            if current['text'].startswith("-"):
+                                print(
+                                    f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+                                # previous starts with a `-`, sherlock this is another truncated value
+                                current['text'] = previous['text'] + current['text']
+                                current['raw_value'] = current['text']
+                                current['offset_start'] = previous['offset_start']
+                                to_be_removed.append(previous)
+                            else:
+                                print("Other cases to be considered: ", previous, current)
+
+                    elif previous['text'] == "-" and current['text'][0].isdigit():
+                        print(
+                            f"Merging. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+                        # previous starts with a `-`, sherlock this is another truncated value
+                        current['text'] = previous['text'] + " " * (current['offset_start'] - previous['offset_end']) + \
+                                          current['text']
+                        current['raw_value'] = current['text']
+                        current['offset_start'] = previous['offset_start']
+                        to_be_removed.append(previous)
+                    else:
+                        print(
+                            f"Overlapping. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+
+                        # take the largest one
+                        if len(previous['text']) > len(current['text']):
+                            to_be_removed.append(current)
+                        elif len(previous['text']) < len(current['text']):
+                            to_be_removed.append(previous)
+                        else:
+                            to_be_removed.append(previous)
+                elif current['type'] != previous['type']:
+                    print(
+                        f"Overlapping. {current['text']} <{current['type']}> with {previous['text']} <{previous['type']}>")
+
+                    if len(previous['text']) > len(current['text']):
+                        to_be_removed.append(current)
+                    elif len(previous['text']) < len(current['text']):
+                        to_be_removed.append(previous)
+                    else:
+                        if current['type'] == "material":
+                            to_be_removed.append(previous)
+                        else:
+                            to_be_removed.append(current)
+                previous = current
+
+            elif previous['offset_end'] > current['offset_end']:
+                to_be_removed.append(current)
+                # the previous goes after the current, so we keep the previous and we discard the current
+            else:
+                if current['type'] == "material":
+                    to_be_removed.append(previous)
+                else:
+                    to_be_removed.append(current)
+                previous = current
+
+        new_sorted_entities = [e for e in sorted_entities if e not in to_be_removed]
+
+        return new_sorted_entities
+
+
+class XmlProcessor(BaseProcessor):
+    def __init__(self, grobid_superconductors_client, grobid_quantities_client):
+        super().__init__(grobid_superconductors_client, grobid_quantities_client)
+
+    def process_structure(self, input_file):
+        text = ""
+        with open(input_file, encoding='utf-8') as fi:
+            text = fi.read()
+
+        output_data = self.parse_xml(text)
+        output_data['filename'] = Path(input_file).stem.replace(".tei", "")
+
+        return output_data
+
+    def process_single(self, input_file):
+        doc = self.process_structure(input_file)
+
+        for paragraph in doc['passages']:
+            entities = self.process_single_text(paragraph['text'])
+            paragraph['spans'] = entities
+
+        return doc
+
+    def parse_xml(self, text):
+        output_data = OrderedDict()
+        soup = BeautifulSoup(text, 'xml')
+        text_blocks_children = supermat_tei_parser.get_children_list(soup, verbose=False)
+
+        passages = []
+        output_data['passages'] = passages
+        passages.extend([
+            {
+                "text": self.post_process(''.join(text for text in sentence.find_all(text=True) if
+                                                  text.parent.name != "ref" or (
+                                                          text.parent.name == "ref" and text.parent.attrs[
+                                                      'type'] != 'bibr'))),
+                "type": "paragraph",
+                "section": "<body>",
+                "subSection": "<paragraph>",
+                "passage_id": str(paragraph_id) + str(sentence_id)
+            }
+            for paragraph_id, paragraph in enumerate(text_blocks_children) for
+            sentence_id, sentence in enumerate(paragraph)
+        ])
+
+        return output_data
+
+
+def get_children_list(soup: object, use_paragraphs: object = True, verbose: object = False) -> object:
+    children = []
+
+    child_name = "p" if use_paragraphs else "s"
+    for child in soup.TEI.children:
+        if child.name == 'teiHeader':
+            pass
+            # children.extend(child.find_all("title", attrs={"level": "a"}, limit=1))
+            # children.extend([subchild.find_all(child_name) for subchild in child.find_all("abstract")])
+        elif child.name == 'text':
+            children.extend([subchild.find_all(child_name) for subchild in child.find_all("body")])
+            children.extend([subchild.find_all("figDesc") for subchild in child.find_all("body")])
+
+    if verbose:
+        print(str(children))
+
+    return children
+
+
+def get_children_body(soup: object, use_paragraphs: object = True, verbose: object = False) -> object:
+    children = []
+    child_name = "p" if use_paragraphs else "s"
+    for child in soup.TEI.children:
+        if child.name == 'text':
+            children.extend([subchild.find_all(child_name) for subchild in child.find_all("body")])
+
+    if verbose:
+        print(str(children))
+
+    return children
+
+
+def get_children_figures(soup: object, use_paragraphs: object = True, verbose: object = False) -> object:
+    children = []
+    child_name = "p" if use_paragraphs else "s"
+    for child in soup.TEI.children:
+        if child.name == 'text':
+            children.extend([subchild.find_all("figDesc") for subchild in child.find_all("body")])
+
+    if verbose:
+        print(str(children))
+
+    return children

streamlit_app.py

@@ -1,17 +1,17 @@
 import os
-from datetime import datetime
 from hashlib import blake2b
 from tempfile import NamedTemporaryFile
 
 import dotenv
+
+dotenv.load_dotenv(override=True)
+
 import streamlit as st
 from langchain.chat_models import PromptLayerChatOpenAI
 from langchain.embeddings import OpenAIEmbeddings
 
 from document_qa_engine import DocumentQAEngine
 
-dotenv.load_dotenv(override=True)
-
 if 'rqa' not in st.session_state:
     st.session_state['rqa'] = None