Spaces:
Runtime error
Runtime error
File size: 5,350 Bytes
82a7a28 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
import pandas as pd
from typing import Union, List
from tinytroupe.extraction import logger
from tinytroupe import openai_utils
import tinytroupe.utils as utils
class Normalizer:
"""
A mechanism to normalize passages, concepts and other textual elements.
"""
def __init__(self, elements:List[str], n:int, verbose:bool=False):
"""
Normalizes the specified elements.
Args:
elements (list): The elements to normalize.
n (int): The number of normalized elements to output.
verbose (bool, optional): Whether to print debug messages. Defaults to False.
"""
# ensure elements are unique
self.elements = list(set(elements))
self.n = n
self.verbose = verbose
# a JSON-based structure, where each output element is a key to a list of input elements that were merged into it
self.normalized_elements = None
# a dict that maps each input element to its normalized output. This will be used as cache later.
self.normalizing_map = {}
rendering_configs = {"n": n,
"elements": self.elements}
messages = utils.compose_initial_LLM_messages_with_templates("normalizer.system.mustache", "normalizer.user.mustache",
base_module_folder="extraction",
rendering_configs=rendering_configs)
next_message = openai_utils.client().send_message(messages, temperature=0.1)
debug_msg = f"Normalization result message: {next_message}"
logger.debug(debug_msg)
if self.verbose:
print(debug_msg)
result = utils.extract_json(next_message["content"])
logger.debug(result)
if self.verbose:
print(result)
self.normalized_elements = result
def normalize(self, element_or_elements:Union[str, List[str]]) -> Union[str, List[str]]:
"""
Normalizes the specified element or elements.
This method uses a caching mechanism to improve performance. If an element has been normalized before,
its normalized form is stored in a cache (self.normalizing_map). When the same element needs to be
normalized again, the method will first check the cache and use the stored normalized form if available,
instead of normalizing the element again.
The order of elements in the output will be the same as in the input. This is ensured by processing
the elements in the order they appear in the input and appending the normalized elements to the output
list in the same order.
Args:
element_or_elements (Union[str, List[str]]): The element or elements to normalize.
Returns:
str: The normalized element if the input was a string.
list: The normalized elements if the input was a list, preserving the order of elements in the input.
"""
if isinstance(element_or_elements, str):
denormalized_elements = [element_or_elements]
elif isinstance(element_or_elements, list):
denormalized_elements = element_or_elements
else:
raise ValueError("The element_or_elements must be either a string or a list.")
normalized_elements = []
elements_to_normalize = []
for element in denormalized_elements:
if element not in self.normalizing_map:
elements_to_normalize.append(element)
if elements_to_normalize:
rendering_configs = {"categories": self.normalized_elements,
"elements": elements_to_normalize}
messages = utils.compose_initial_LLM_messages_with_templates("normalizer.applier.system.mustache", "normalizer.applier.user.mustache",
base_module_folder="extraction",
rendering_configs=rendering_configs)
next_message = openai_utils.client().send_message(messages, temperature=0.1)
debug_msg = f"Normalization result message: {next_message}"
logger.debug(debug_msg)
if self.verbose:
print(debug_msg)
normalized_elements_from_llm = utils.extract_json(next_message["content"])
assert isinstance(normalized_elements_from_llm, list), "The normalized element must be a list."
assert len(normalized_elements_from_llm) == len(elements_to_normalize), "The number of normalized elements must be equal to the number of elements to normalize."
for i, element in enumerate(elements_to_normalize):
normalized_element = normalized_elements_from_llm[i]
self.normalizing_map[element] = normalized_element
for element in denormalized_elements:
normalized_elements.append(self.normalizing_map[element])
return normalized_elements
|