jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/reimbursement-level-tins (pull request #379)

Browse Source 
Initialize Reimbursement-Level TINs

* Initialized Reimbursement-Level TIN and NPI

* Update unit tests

* Docstrings and cleanup

* Update test


Approved-by: Alex Galarce
This commit is contained in:
Katon Minhas
2025-02-03 21:50:50 +00:00
parent 55f96e77fc
commit 71e0eb5c45
7 changed files with 75 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/investment/file_processing.py
+6 -2
View File
@@ -14,7 +14,7 @@ from src import config
import src.utils.string_utils as string_utils
import src.utils.io_utils as io_utils
from src.config import FIELD_JSON_PATH
from src.investment.one_to_n_funcs import dynamic_and_exhibit_level, reimbursement_level, combine_exhibit_and_reimbursement
from src.investment.one_to_n_funcs import dynamic_and_exhibit_level, reimbursement_level, combine_one_to_n_answers, reimbursement_tin_npi
def process_file(file_object, run_timestamp):
filename, contract_text = file_object
@@ -92,11 +92,15 @@ def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mappi
for exhibit_page in exhibit_pages:
exhibit_chunk = string_utils.get_exhibit_chunk(text_dict, exhibit_chunk_mapping, exhibit_page)
if string_utils.contains_reimbursement(exhibit_chunk):
################## INITIALIZE FIELDS ##################
reimbursement_level_fields = FieldSet(relationship="one_to_n", field_type="reimbursement_level", file_path=FIELD_JSON_PATH)
exhibit_level_fields = FieldSet(relationship="one_to_n", field_type="exhibit_level", file_path=FIELD_JSON_PATH)
dynamic_fields = FieldSet(relationship="one_to_n", field_type="dynamic", file_path=FIELD_JSON_PATH)
################## GET REIMBURSEMENT TIN/NPI ##################
tin_npi_answers, reimbursement_level_fields = reimbursement_tin_npi(exhibit_chunk, reimbursement_level_fields)
################## GET EXHIBIT-LEVEL ANSWERS (INCLUDING DYNAMIC) ##################
exhibit_level_answers= dynamic_and_exhibit_level(exhibit_chunk, filename, reimbursement_level_fields, exhibit_level_fields, dynamic_fields)
exhibit_level_answers['EXHIBIT_PAGE'] = exhibit_page
@@ -105,7 +109,7 @@ def run_one_to_n_prompts(filename, text_dict, exhibit_pages, exhibit_chunk_mappi
reimbursement_level_answers = reimbursement_level(exhibit_chunk, filename, reimbursement_level_fields) # Return list of dictionaries
################## COMBINE ANSWERS ##################
full_answer_dict = combine_exhibit_and_reimbursement(exhibit_level_answers, reimbursement_level_answers)
full_answer_dict = combine_one_to_n_answers(exhibit_level_answers, reimbursement_level_answers, tin_npi_answers)
one_to_n_results += full_answer_dict
fieldExtraction/src/investment/one_to_n_funcs.py
+27 -4
View File
@@ -7,9 +7,10 @@ import src.utils.llm_utils as llm_utils
import src.utils.string_utils as string_utils
from src import config, postprocessing_funcs
from src.enums.delimiters import Delimiter
from src.prompts.investment_prompts import FieldSet
from src.prompts.investment_prompts import FieldSet, Field
import src.prompts.investment_prompts as investment_prompts
from src.config import FIELD_JSON_PATH
from src.regex.regex_utils import find_regex_matches
def get_exhibit_header(exhibit_chunk, filename):
"""
@@ -191,6 +192,26 @@ def get_methodology_breakout(reimbursement_primary_answers, filename):
methodology_breakout_answers.append({**answer_dict, **breakout_dict})
return methodology_breakout_answers
def reimbursement_tin_npi(exhibit_chunk: str, reimbursement_level_fields: FieldSet):
answer_dict = {}
fields = [("PROV_REIMBURSEMENT_TIN", r"\b\d{2}-\d{7}\b|\b\d{9}\b"),
("PROV_REIMBURSEMENT_NPI", r"\b\d{10}\b")]
for field_tuple in fields:
field_name, field_pattern = field_tuple[0], field_tuple[1]
field_matches, page_list = find_regex_matches(field_pattern, {'1' : exhibit_chunk})
if not field_matches:
answer_dict[field_name] = "N/A"
elif len(field_matches) == 1:
answer_dict['PROV_REIMBURSEMENT_TIN'] = field_matches[0]
else: # Multiple tins
reimbursement_field = Field.load_from_file(FIELD_JSON_PATH, field_name)
reimbursement_field.update_valid_values(field_matches)
reimbursement_level_fields.add_field(reimbursement_field)
return answer_dict, reimbursement_level_fields
def reimbursement_level(exhibit_text, filename, reimbursement_level_fields):
@@ -206,31 +227,33 @@ def reimbursement_level(exhibit_text, filename, reimbursement_level_fields):
Returns:
dict: The parsed LLM response as a list of dictionaries
"""
reimbursement_primary_answers = get_reimbursement_primary(reimbursement_level_fields, exhibit_text, filename) # Returns list of dicts
carveout_answers = get_carveout_ind(reimbursement_primary_answers, filename)
# Later, we will only run methodology breakout for those without certain carveout indicators. We will run carveout-specific breakouts for the others
methodology_breakout_answers = get_methodology_breakout(carveout_answers, filename)
return methodology_breakout_answers
def combine_exhibit_and_reimbursement(exhibit_level_answers: dict, reimbursement_level_answers: list[dict]) -> list[dict]:
def combine_one_to_n_answers(exhibit_level_answers: dict, reimbursement_level_answers: list[dict], tin_npi_answers: dict) -> list[dict]:
"""
Combines exhibit-level answers with each dictionary in the reimbursement-level answers.
Args:
exhibit_level_answers (dict): A dictionary containing exhibit-level key-value pairs.
reimbursement_level_answers (list[dict]): A list of dictionaries containing reimbursement-level answers.
tin_npi_answers (dict): A dictionary containing exhibit-level key-value pairs, or nothing, depending on regex dynamic assignment
Returns:
list[dict]: A new list of dictionaries where each dictionary combines exhibit-level answers
with the respective reimbursement-level answers.
"""
# Add exhibit-level answers to each reimbursement dictionary
combined_answers = [
{**reimbursement_dict, **exhibit_level_answers}
{**reimbursement_dict, **tin_npi_answers, **exhibit_level_answers}
for reimbursement_dict in reimbursement_level_answers
]
fieldExtraction/src/investment/one_to_one_funcs.py
+1 -1
View File
@@ -15,7 +15,7 @@ def run_regex_fields(one_to_one_fields: FieldSet,
top_sheet_dict: str,
filename: str):
# Fill NPI/TIN regex here
regex_fields = one_to_one_fields.filter(field_type="regex")
# regex_fields = one_to_one_fields.filter(relationship="one_to_one", field_type="regex")
# One to One TIN Fields
tin_answers = get_tin_fields(filename, text_dict, top_sheet_dict)
fieldExtraction/src/prompts/investment_prompts.json
+12
View File
@@ -143,6 +143,18 @@
"field_type": "regex",
"prompt": "TBD"
},
{
"field_name": "PROV_REIMBURSEMENT_TIN",
"relationship": "one_to_n",
"field_type": "regex",
"prompt": "What TIN (Tax ID Number) does the reimbursement term apply to? Choose only from the following: {valid_values}."
},
{
"field_name": "PROV_REIMBURSEMENT_NPI",
"relationship": "one_to_n",
"field_type": "regex",
"prompt": "What NPI (National Provider Identifier) does the reimbursement term apply to? Choose only from the following: {valid_values}."
},
{
"field_name": "PROV_GROUP_NAME_FULL",
"relationship": "one_to_one",
fieldExtraction/src/prompts/investment_prompts.py
+25 -2
View File
@@ -32,14 +32,34 @@ class Field:
self.retrieved_text = None
self.format = field_dict['format'] if 'format' in field_dict else None
@classmethod
def load_from_file(cls, file_path, field_name):
"""
Reads a JSON file and creates a Field object if field_name is present in the JSON.
:param file_path: Path to the JSON file.
:param field_name: The field name to look for in the JSON data.
:return: A Field object if found, otherwise raises a ValueError.
"""
with open(file_path, 'r') as file:
data = json.load(file)
for field_dict in data:
if field_dict.get("field_name") == field_name:
return cls(field_dict)
raise ValueError(f"Field '{field_name}' not found in {file_path}")
@classmethod
def from_values(cls, **kwargs):
"""Creates a Field instance from given keyword arguments."""
return cls(kwargs)
def to_dict(self):
"""Converts the Field instance to a dictionary."""
return {
"field_name": self.field_name,
"base_field":self.base_field,
"base_field": self.base_field,
"relationship": self.relationship,
"field_type": self.field_type,
"prompt": self.prompt,
@@ -47,9 +67,11 @@ class Field:
"keywords": self.keywords,
"methodology": self.methodology,
"case_sensitive": self.case_sensitive,
"regex_pattern": self.regex_pattern
"regex_pattern": self.regex_pattern,
"format": self.format
}
def print(self):
"""Prints the details of a FieldInfo object in a readable format."""
print(f"Field Name: {self.field_name}")
@@ -63,6 +85,7 @@ class Field:
print(f"Regex Pattern: {'None specified' if not self.regex_pattern else self.regex_pattern}")
def get_prompt_dict(self):
"""Returns a dictionary with the field name as the key and its prompt as the value."""
return {self.field_name : self.get_prompt()}
def get_prompt(self):
fieldExtraction/src/regex/regex_utils.py
+1 -3
View File
@@ -92,7 +92,7 @@ def npi_pattern_match(text_dict: dict[str, str]) -> dict[str, str]:
for pattern in patterns:
matches, page_list = find_regex_matches(pattern, text_dict)
matches = [s for s in matches if s[0] in ("1", "2")]
matches = [s for s in matches if s[0] in ("1", "2")] # NPI can only start with 1 or 2
matches = list(set(matches))
if len(matches) == 1:
NPI_dict["PROV_GROUP_NPI"] = matches[0]
@@ -129,8 +129,6 @@ def get_npi_fields(
#################################################################################################################################################
# IRS
def check_tin_in_filename(filename):
"""
Checks for the pattern match in filename
fieldExtraction/tests/one_to_n_test.py
+3 -2
View File
@@ -1,6 +1,6 @@
import unittest
from unittest.mock import patch, MagicMock
from src.investment.one_to_n_funcs import get_exhibit_header, prompt_dynamic, process_dynamic, dynamic_and_exhibit_level, reimbursement_level, combine_exhibit_and_reimbursement
from src.investment.one_to_n_funcs import get_exhibit_header, prompt_dynamic, process_dynamic, dynamic_and_exhibit_level, reimbursement_level, combine_one_to_n_answers
class TestOneToN(unittest.TestCase):
@@ -40,8 +40,9 @@ class TestOneToN(unittest.TestCase):
def test_combine_exhibit_and_reimbursement(self):
exhibit_level_answers = {"EXHIBIT_NAME": "Exhibit Header", "field1": "value1"}
reimbursement_level_answers = [{"field2": "value2"}, {"field3": "value3"}]
tin_npi_answers = {}
result = combine_exhibit_and_reimbursement(exhibit_level_answers, reimbursement_level_answers)
result = combine_one_to_n_answers(exhibit_level_answers, reimbursement_level_answers, tin_npi_answers)
self.assertEqual(result, [
{"field2": "value2", "EXHIBIT_NAME": "Exhibit Header", "field1": "value1"},