From 1531ad6bac23c2c819ed4880e0ebfbf315f69437 Mon Sep 17 00:00:00 2001 From: Katon Minhas Date: Thu, 26 Jun 2025 19:35:32 +0000 Subject: [PATCH] Merged in feature/prov-group-tin (pull request #584) Feature/prov group tin * don't prompt for INTRO_PAGE * test complete * test 2 * Update test script * Merge branch 'main' into feature/prov-group-tin * Merged main into feature/prov-group-tin * refactor * cleaning * Update save * Fix return type * Update for reimbursement_tin_npi * Comment out reimbursement-tin-npi * Pass unit tests * Merge branch 'main' into feature/prov-group-tin * Merged main into feature/prov-group-tin * Remove length validation * Merge branch 'main' into feature/prov-group-tin * Remove ON_INTRO from prompt * Restore * Remove test * Move reimb prov info to dynamic reimb info * e2e test * Fix for unit test * remove prints * update docstrings and typehints Approved-by: Alex Galarce --- .../src/investment/file_processing.py | 26 +- .../src/investment/one_to_one_funcs.py | 59 +--- .../src/investment/tin_npi_funcs.py | 190 ++++++----- .../src/prompts/investment_prompts.json | 14 +- .../src/prompts/investment_prompts.py | 13 +- fieldExtraction/src/testbed/testbed_utils.py | 146 ++++---- fieldExtraction/tests/test_tin_npi_funcs.py | 321 ++++++++---------- 7 files changed, 352 insertions(+), 417 deletions(-) diff --git a/fieldExtraction/src/investment/file_processing.py b/fieldExtraction/src/investment/file_processing.py index cce8c13..c75ee06 100644 --- a/fieldExtraction/src/investment/file_processing.py +++ b/fieldExtraction/src/investment/file_processing.py @@ -9,6 +9,7 @@ import src.investment.dynamic_funcs as dynamic_funcs import src.investment.investment_postprocessing_funcs as investment_postprocessing_funcs import src.investment.one_to_n_funcs as one_to_n_funcs import src.investment.one_to_one_funcs as one_to_one_funcs +import src.investment.tin_npi_funcs as tin_npi_funcs import src.investment.postprocess as postprocess import src.investment.preprocess as preprocess import src.investment.row_funcs as row_funcs @@ -74,24 +75,8 @@ def run_one_to_one_prompts(filename, contract_text, text_dict, top_sheet_dict, d ################## INITIALIZE FIELDS ################## one_to_one_fields = FieldSet(relationship="one_to_one", file_path=config.FIELD_JSON_PATH).combine(dynamic_one_to_one_fields) - ################## RUN REGEX ################## - regex_answers_dict = one_to_one_funcs.run_provider_info_fields(contract_text, text_dict, filename) - one_to_one_results = {} - one_to_one_results['PROV_INFO_JSON'] = json.dumps(regex_answers_dict) - one_to_one_results['PROV_INFO_JSON_FORMATTED'] = '\n'.join([json.dumps(provider) for provider in regex_answers_dict]) - - one_to_one_results = merge_provider_info(one_to_one_results, regex_answers_dict) - - # What if the TIN/NPI regexes don't find anything? Add PROV_GROUP_NAME_FULL to full context - if one_to_one_results.get("PROV_OTHER_NAME_FULL") == "NO_IDENTIFIERS_FOUND": # This is how we know the regex didn't find anything - other_provider_name_field = Field.from_values( - field_name="PROV_GROUP_NAME_FULL", - relationship="one_to_one", - field_type="full_context", - prompt="What is the name of the group provider associated with this contract? Return the full name as it appears in the document." - ) - one_to_one_fields.add_field(other_provider_name_field) - one_to_one_results["PROV_OTHER_NAME_FULL"] = "UNKNOWN" + ################## RUN PROVIDER INFO ################## + one_to_one_results, one_to_one_fields = tin_npi_funcs.run_provider_info_fields(contract_text, one_to_one_fields, text_dict, filename) ################## RUN SMART CHUNKED PROMPTS ################## smart_chunked_answers_dict = one_to_one_funcs.run_smart_chunked_fields( @@ -136,10 +121,7 @@ def run_one_to_n_prompts(filename, exhibit_dict, all_exhibit_headers, all_datase ################## INITIALIZE FIELDS ################## reimbursement_level_fields = FieldSet(relationship="one_to_n", field_type="reimbursement_level", file_path=FIELD_JSON_PATH) - ################## GET REIMBURSEMENT TIN/NPI ################## - # tin_npi_answers, reimbursement_level_fields = reimbursement_tin_npi(exhibit_text, reimbursement_level_fields, filename) - - ################## GET EXHIBIT-LEVEL ANSWERS ################## + ################# GET EXHIBIT-LEVEL ANSWERS ################## exhibit_level_answers = one_to_n_funcs.get_exhibit_level_answers(exhibit_text, filename) exhibit_level_answers['EXHIBIT_PAGE'] = exhibit_page exhibit_level_answers['EXHIBIT_TITLE'] = exhibit_header diff --git a/fieldExtraction/src/investment/one_to_one_funcs.py b/fieldExtraction/src/investment/one_to_one_funcs.py index a24fc00..bc90d11 100644 --- a/fieldExtraction/src/investment/one_to_one_funcs.py +++ b/fieldExtraction/src/investment/one_to_one_funcs.py @@ -5,74 +5,19 @@ import pandas as pd import src.prompts.investment_prompts as prompts import src.utils.llm_utils as llm_utils import src.utils.string_utils as string_utils + from rapidfuzz import fuzz from src import config from src import postprocessing_funcs as generic_postprocessing_funcs from src.enums.delimiters import Delimiter from src.investment import investment_postprocessing_funcs from src.investment.one_to_n_funcs import METHODOLOGY_BREAKOUT_QUESTIONS -from src.investment.smart_chunking_funcs import (field_context, group_fields, - parse_chunk, stitch_chunks) -from src.investment.tin_npi_funcs import (clean_provider_info, - deduplicate_providers, - extract_identifiers, - get_provider_info, - identify_group_provider) +from src.investment.smart_chunking_funcs import (field_context, group_fields, stitch_chunks) from src.investment.vision_funcs import get_image_array_based_answer from src.prompts.investment_prompts import REIMB_TERM_BREAKOUT, Field, FieldSet MAX_CONTEXT_LENGTH = 100000 -def run_provider_info_fields(contract_text: str, - text_dict: dict, - filename: str, - # payer_name: str # This is not being used right now, but it might be useful for filtering providers later - ) -> list[dict]: - """ - Extracts provider information from a contract text using regex-based identifier extraction, - processes the relevant text chunks, and cleans the results. - Args: - contract_text (str): The full text of the contract to process. - text_dict (dict): A dictionary containing text data, organized by pages - filename (str): The name of the file being processed, used for logging or reference. - payer_name (str): The name of the payer extracted from the full context answers. - This is not being used right now, but it might be useful for filtering providers later. - Returns: - list[dict]: A cleaned and standardized list containing provider information extracted - from the contract text. - """ - all_tins, all_npis = extract_identifiers(contract_text, filename) - - # If there are no identifiers, return early with default values - if not all_tins and not all_npis: - return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "NO_IDENTIFIERS_FOUND", "IS_GROUP": False}] - - # Find pages with TINs or NPIs - relevant_pages = {} - for page_num, page_text in text_dict.items(): - if any(tin in page_text for tin in all_tins) or any(npi in page_text for npi in all_npis): - relevant_pages[page_num] = page_text - - # Process each page separately and combine results - combined_provider_info = [] - for page_num, page_text in relevant_pages.items(): - page_provider_info = get_provider_info(page_text, filename) - combined_provider_info.extend(page_provider_info) - - # Clean and standardize results - cleaned_provider_info = clean_provider_info(combined_provider_info) - - # Deduplicate providers based on TIN and NPI - deduplicated_provider_info = deduplicate_providers(cleaned_provider_info) - - if len(deduplicated_provider_info) > 1: - group_provider = identify_group_provider(text_dict, deduplicated_provider_info, filename) - elif len(deduplicated_provider_info) == 1: - # If only one provider is found, mark it as the group provider - deduplicated_provider_info[0]["IS_GROUP"] = True - - return deduplicated_provider_info - def extract_global_lesser_of(contract_text: str, filename:str) -> dict: """Extract global "lesser of" statements that apply contract-wide. diff --git a/fieldExtraction/src/investment/tin_npi_funcs.py b/fieldExtraction/src/investment/tin_npi_funcs.py index 7451e93..d8d9ecc 100644 --- a/fieldExtraction/src/investment/tin_npi_funcs.py +++ b/fieldExtraction/src/investment/tin_npi_funcs.py @@ -5,14 +5,13 @@ from typing import Optional import src.config as config import src.prompts.investment_prompts as investment_prompts + import src.regex.regex_patterns as regex_patterns -import src.regex.regex_utils as regex_utils import src.utils.llm_utils as llm_utils import src.utils.string_utils as string_utils from src.enums.delimiters import Delimiter from src.prompts.investment_prompts import Field, FieldSet - def get_all_matches(text, pattern, filename): """ Extracts all unique matches of a given pattern from the provided text. @@ -63,13 +62,13 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]: cleaned_providers = [] for provider in provider_info: # Create a new provider dict to hold the cleaned data - cleaned_provider = {} + cleaned_provider = {"ON_SIGNATURE_PAGE" : provider["ON_SIGNATURE_PAGE"], "ON_INTRO_PAGE" : provider["ON_INTRO_PAGE"]} # Clean TIN - remove hyphens and dots, ensure it's 9 digits if "TIN" in provider and provider["TIN"]: cleaned_provider["TIN"] = provider["TIN"].replace("-", "").replace(".", "") - # Validate length - if len(cleaned_provider["TIN"]) != 9: + # Validate that TIN only contains digits or "|" + if not all(c.isdigit() or c == "|" for c in cleaned_provider["TIN"]): cleaned_provider["TIN"] = "UNKNOWN" else: cleaned_provider["TIN"] = "UNKNOWN" @@ -77,8 +76,8 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]: # Clean NPI - remove hyphens and dots, ensure it's 10 digits if "NPI" in provider and provider["NPI"]: cleaned_provider["NPI"] = provider["NPI"].replace("-", "").replace(".", "") - # Validate length - if len(cleaned_provider["NPI"]) != 10: + # Validate that NPI only contains digits or "|" + if not all(c.isdigit() or c == "|" for c in cleaned_provider["NPI"]): cleaned_provider["NPI"] = "UNKNOWN" else: cleaned_provider["NPI"] = "UNKNOWN" @@ -92,25 +91,6 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]: else: cleaned_provider["NAME"] = "UNKNOWN" - # Ensure IS_GROUP is a boolean, handling string representations correctly - if "IS_GROUP" in provider: - value = provider["IS_GROUP"] - if isinstance(value, bool): - cleaned_provider["IS_GROUP"] = value - elif isinstance(value, str): - # Handle string representations of boolean values - cleaned_provider["IS_GROUP"] = value.lower() not in ("false", "no", "0", "n", "f", "") - elif isinstance(value, (int, float)): - if value == 0: - cleaned_provider["IS_GROUP"] = False - else: - cleaned_provider["IS_GROUP"] = bool(value) - else: - # For None, empty lists, etc. - cleaned_provider["IS_GROUP"] = False - else: - cleaned_provider["IS_GROUP"] = False - cleaned_providers.append(cleaned_provider) return cleaned_providers @@ -126,23 +106,23 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) -> Returns: dict: Updated one_to_one_results with merged values from provider_info. """ - + # Extract group and non-group providers - group_tins = [] - group_npis = [] - group_names = [] + group_tins = set() + group_npis = set() + group_names = set() other_tins = [] other_npis = [] other_names = [] for provider in provider_info: - if provider.get("IS_GROUP", False): + if provider.get("ON_SIGNATURE_PAGE", False) == "Y" or provider.get("ON_INTRO_PAGE", False) == "Y" : if provider.get("TIN"): - group_tins.append(provider["TIN"]) + group_tins.add(provider["TIN"]) if provider.get("NPI"): - group_npis.append(provider["NPI"]) + group_npis.add(provider["NPI"]) if provider.get("NAME"): - group_names.append(provider["NAME"]) + group_names.add(provider["NAME"]) else: if provider.get("TIN"): other_tins.append(provider["TIN"]) @@ -151,24 +131,29 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) -> if provider.get("NAME"): other_names.append(provider["NAME"]) + # De-Unknown GROUP fields + group_tins = [v for v in group_tins if v != "UNKNOWN"] + group_npis = [v for v in group_npis if v != "UNKNOWN"] + group_names = [v for v in group_names if v != "UNKNOWN"] + # Merge group provider information into one_to_one_results - one_to_one_results["PROV_GROUP_TIN"] = ", ".join(group_tins) if group_tins else "" - one_to_one_results["PROV_GROUP_NPI"] = ", ".join(group_npis) if group_npis else "" - one_to_one_results["PROV_GROUP_NAME_FULL"] = ", ".join(group_names) if group_names else "" + one_to_one_results["PROV_GROUP_TIN"] = "|".join(group_tins) if group_tins else "" + one_to_one_results["PROV_GROUP_NPI"] = "|".join(group_npis) if group_npis else "" + one_to_one_results["PROV_GROUP_NAME_FULL"] = "|".join(group_names) if group_names else "" # Merge other provider information into one_to_one_results one_to_one_results["PROV_OTHER_TIN"] = "|".join(other_tins) if other_tins else "" one_to_one_results["PROV_OTHER_NPI"] = "|".join(other_npis) if other_npis else "" one_to_one_results["PROV_OTHER_NAME_FULL"] = "|".join(other_names) if other_names else "" return one_to_one_results -def reimbursement_tin_npi(exhibit_chunk: str, reimbursement_level_fields: FieldSet, filename: str) -> tuple[dict, FieldSet]: +def reimbursement_tin_npi(exhibit_dict: dict[str, str], exhibit_page: str, reimbursement_level_fields: FieldSet, filename: str) -> tuple[dict, FieldSet]: + """ Extracts Reimbursement-Level TIN and NPI values from the given exhibit chunk using an LLM. - When multiple providers are found, they're added as valid values for LLM selection - of these fields during reimbursement primary processing. - + Args: - exhibit_chunk (str): The text chunk containing reimbursement details. + exhibit_dict (dict[str, str]): Dictionary containing the exhibit text with page numbers as keys. + exhibit_page (str): The page number containing reimbursement details. reimbursement_level_fields (FieldSet): The FieldSet object to store extracted field values. filename (str): The name of the file being processed. @@ -188,7 +173,9 @@ def reimbursement_tin_npi(exhibit_chunk: str, reimbursement_level_fields: FieldS answer_dict = {} # Extract all providers from the exhibit chunk - providers = get_provider_info(exhibit_chunk, filename) + providers = get_provider_info(text_dict=exhibit_dict, + page_num=exhibit_page, + filename=filename) # Clean, standardize, and deduplicate extracted providers filtered_providers = deduplicate_providers(clean_provider_info(providers)) @@ -329,48 +316,28 @@ def deduplicate_providers(provider_info: list[dict]) -> list[dict]: # TODO: add return valid_providers -def extract_identifiers(contract_text: str, filename: str) -> tuple: +def get_provider_info(text_dict: dict, page_num : str, filename: str) -> list: """ - Extracts Taxpayer Identification Numbers (TINs) and National Provider Identifiers (NPIs) - from the given contract text using predefined patterns. + Extracts provider information from a specified page in a document using a language model. Args: - contract_text (str): The text content of the contract to search for identifiers. - filename (str): The name of the file being processed, used for logging or debugging. - - Returns: - tuple: A tuple containing two lists: - - A list of all matched TINs found in the contract text. - - A list of all matched NPIs found in the contract text. - """ - - - # Get all matches for TIN and NPI - all_tins = get_all_matches(contract_text, regex_patterns.TIN_PATTERN, filename) - all_npis = get_all_matches(contract_text, regex_patterns.NPI_PATTERN, filename) - - return all_tins, all_npis - - -def get_provider_info(chunk: str, filename: str) -> list: - """ - Extracts provider information from a given text chunk using a language model. - - Args: - chunk (str): The text chunk containing provider information to be extracted. + text_dict (dict): Dictionary of text content with page numbers as keys. + page_num (str): The page number to extract provider information from. filename (str): The name of the file being processed, used for logging or tracking purposes. Returns: - list: A list of extracted provider information in JSON format, parsed from the language model's response. + list: A list of extracted provider information dictionaries, each containing TIN, NPI, + NAME, and flags for presence on signature or intro pages. Raises: ValueError: If the response from the language model cannot be parsed as valid JSON. """ + + chunk = text_dict[page_num] + provider_fields = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="provider_info") - # Generate prompt - prompt = investment_prompts.TIN_NPI_TEMPLATE(chunk, provider_fields.get_prompt_dict()) - # Get response from LLM - claude_answer_raw = llm_utils.invoke_claude(prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=8192) # Sometimes rosters can have 50+ providers + prompt = investment_prompts.TIN_NPI_TEMPLATE(chunk, provider_fields.print_prompt_dict()) + claude_answer_raw = llm_utils.invoke_claude(prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=8192) # Sometimes rosters can have 50+ provider # Defensive programming for any json parsing errors try: @@ -383,17 +350,82 @@ def get_provider_info(chunk: str, filename: str) -> list: elif not isinstance(providers, list): # If not a list or dict, create a default list with error info print(f"Warning: Unexpected format in provider info response: {type(providers)}") - providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] - - # For all providers, set IS_GROUP to False + providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "ON_SIGNATURE_PAGE": "N", "ON_INTRO_PAGE" : "N"}] + + # Set ON_INTRO_PAGE flag based on page number for provider in providers: - provider["IS_GROUP"] = False + provider["ON_INTRO_PAGE"] = "Y" if int(page_num) <= 2 else "N" return providers except ValueError as e: # Handle JSON parsing error print(f"Error parsing JSON response from LLM: {str(e)}") print(f"Raw response: {claude_answer_raw}") # Return a default value - return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] + return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "ON_SIGNATURE_PAGE": "N", "ON_INTRO_PAGE" : "N"}] + + +def run_provider_info_fields(contract_text: str, + one_to_one_fields: FieldSet, + text_dict: dict, + filename: str, + # payer_name: str # This is not being used right now, but it might be useful for filtering providers later + ): + """ + Extracts provider information from a contract text using regex-based identifier extraction, + processes the relevant text chunks, and cleans the results. + Args: + contract_text (str): The full text of the contract to process. + text_dict (dict): A dictionary containing text data, organized by pages + filename (str): The name of the file being processed, used for logging or reference. + payer_name (str): The name of the payer extracted from the full context answers. + This is not being used right now, but it might be useful for filtering providers later. + Returns: + list[dict]: A cleaned and standardized list containing provider information extracted + from the contract text. + """ + all_tins = get_all_matches(contract_text, regex_patterns.TIN_PATTERN, filename) + all_npis = get_all_matches(contract_text, regex_patterns.NPI_PATTERN, filename) + + # If there are no identifiers, return early with default values + if not all_tins and not all_npis: + deduplicated_provider_info = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "NO_IDENTIFIERS_FOUND", "ON_SIGNATURE_PAGE": "N", "ON_INTRO_PAGE" : "N"}] + else: + # Find pages with TINs or NPIs + relevant_pages = [page_num for page_num, page_text in text_dict.items() if any(tin in page_text for tin in all_tins) or any(npi in page_text for npi in all_npis)] + + # Stopgap for when the TIN is written in the header or footer of every page (or nearly every page) + if len(relevant_pages) >= len(text_dict) - 1 and len(all_tins) == 1: + relevant_pages = [relevant_pages[0]] + + # Process each page separately and combine results + combined_provider_info = [] + for page_num in relevant_pages: + page_provider_info = get_provider_info(text_dict, page_num, filename) + combined_provider_info.extend(page_provider_info) + + # Clean and standardize results + cleaned_provider_info = clean_provider_info(combined_provider_info) + + # Deduplicate providers based on TIN and NPI + deduplicated_provider_info = deduplicate_providers(cleaned_provider_info) + + # Create columns + one_to_one_results = {} + one_to_one_results['PROV_INFO_JSON'] = json.dumps(deduplicated_provider_info) + one_to_one_results['PROV_INFO_JSON_FORMATTED'] = '\n'.join([json.dumps(provider) for provider in deduplicated_provider_info]) + one_to_one_results = merge_provider_info(one_to_one_results, deduplicated_provider_info) # Add Group TIN + + # What if the TIN/NPI regexes don't find anything? Add PROV_GROUP_NAME_FULL to full context + if one_to_one_results.get("PROV_OTHER_NAME_FULL") == "NO_IDENTIFIERS_FOUND": # This is how we know the regex didn't find anything + other_provider_name_field = Field.from_values( + field_name="PROV_GROUP_NAME_FULL", + relationship="one_to_one", + field_type="full_context", + prompt="What is the name of the group provider associated with this contract? Return the full name as it appears in the document." + ) + one_to_one_fields.add_field(other_provider_name_field) + one_to_one_results["PROV_OTHER_NAME_FULL"] = "UNKNOWN" + + return one_to_one_results, one_to_one_fields diff --git a/fieldExtraction/src/prompts/investment_prompts.json b/fieldExtraction/src/prompts/investment_prompts.json index 632802d..92684ee 100644 --- a/fieldExtraction/src/prompts/investment_prompts.json +++ b/fieldExtraction/src/prompts/investment_prompts.json @@ -226,7 +226,7 @@ "field_name": "TIN", "relationship": "one_to_one", "field_type": "provider_info", - "prompt": "What is the Tax Identification Number (TIN)? A TIN is typically a 9-digit number that may be formatted as XXX-XX-XXXX or XXXXXXXXX. Look for it near labels like 'Tax ID', 'EIN', 'Employer ID', or 'TIN'." + "prompt": "What is the Tax Identification Number (TIN)? A TIN is typically a 9-digit number that may be formatted as XXX-XX-XXXX or XXXXXXXXX. Look for it near labels like 'Tax ID', 'EIN', 'Employer ID', or 'TIN'. It may also unlabeled, in the header or footer of the page." }, { "field_name": "NPI", @@ -240,22 +240,28 @@ "field_type": "provider_info", "prompt": "What is the name of the provider associated with this TIN and NPI? Return the full name as it appears in the document." }, + { + "field_name": "ON_SIGNATURE_PAGE", + "relationship": "one_to_one", + "field_type": "provider_info", + "prompt": "Is this TIN, NPI, or Name found on the Signature page of the contract? Signature pages are those which contain a signature section. Write 'Y' or 'N'." + }, { "field_name": "REIMB_PROV_TIN", "relationship": "one_to_n", - "field_type": "dynamic_prov_info", + "field_type": "dynamic_reimb_info", "prompt": "What TIN (Tax ID Number) does the reimbursement term apply to? When the term applies to multiple providers simultaneously, list ALL applicable TINs separated by pipes (|). Look for language patterns indicating multiple providers (phrases like 'following providers', 'these facilities', 'listed providers', etc.)." }, { "field_name": "REIMB_PROV_NPI", "relationship": "one_to_n", - "field_type": "dynamic_prov_info", + "field_type": "dynamic_reimb_info", "prompt": "What NPI (National Provider Identifier) does the reimbursement term apply to? When the term applies to multiple providers simultaneously, list ALL applicable NPIs separated by pipes (|). Look for language patterns indicating multiple providers (phrases like 'following providers', 'these facilities', 'listed providers', etc.)." }, { "field_name": "REIMB_PROV_NAME", "relationship": "one_to_n", - "field_type": "dynamic_prov_info", + "field_type": "dynamic_reimb_info", "prompt": "Identify any Provider Names that this reimbursement applies to. When the term applies to multiple providers simultaneously, list ALL applicable names separated by pipes (|). Look for language patterns indicating multiple providers. Extract only the official names of healthcare providers (doctors, hospitals, clinics, etc.) - not payers or insurance plans. If unable to determine applicable provider names, respond with 'N/A'." }, { diff --git a/fieldExtraction/src/prompts/investment_prompts.py b/fieldExtraction/src/prompts/investment_prompts.py index 6d18a57..acb1ece 100644 --- a/fieldExtraction/src/prompts/investment_prompts.py +++ b/fieldExtraction/src/prompts/investment_prompts.py @@ -812,15 +812,16 @@ IMPORTANT DISTINCTION: - Payers: Insurance companies, health plans who PAY for healthcare services For EACH provider entity found (group or individual practitioner), extract: -- {'\n- '.join(f'{field} - {question}' for (field, question) in questions.items())} +{questions} -Feel free to explain your reasoning as you work through the document. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output. +Briefly explain your reasoning. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output. [EXTRACTION AND FORMATTING INSTRUCTIONS] - First, analyze the text thoroughly and identify all provider entities +- Return TIN and NPIs that are found on the page even if they aren't explicitly labeled as such. - In your response, include ONLY ONE SINGLE properly formatted JSON array, after your explanation - Each provider must be an object within this array -- All provider objects must have these exact keys: "TIN", "NPI", "NAME" +- All provider objects must have these exact keys: "TIN", "NPI", "NAME", "ON_SIGNATURE_PAGE" - Use null (not strings like "null" or "N/A") for missing values - Your final JSON array must begin with an opening bracket '[' and end with a closing bracket ']' - Make sure the JSON array is correctly formatted with commas between objects @@ -832,12 +833,14 @@ FINAL PROVIDER INFO: {{ "TIN": "123456789", "NPI": "1234567890", - "NAME": "Main Provider Group" + "NAME": "Main Provider Group", + "ON_SIGNATURE_PAGE": "Y", }}, {{ "TIN": "987654321", "NPI": null, - "NAME": "Dr. John Smith" + "NAME": "Dr. John Smith", + "ON_SIGNATURE_PAGE": "N", }} ] diff --git a/fieldExtraction/src/testbed/testbed_utils.py b/fieldExtraction/src/testbed/testbed_utils.py index 7f416b6..f22c919 100644 --- a/fieldExtraction/src/testbed/testbed_utils.py +++ b/fieldExtraction/src/testbed/testbed_utils.py @@ -93,6 +93,70 @@ def testbed_preprocess(df): return df + +def is_positive(val): + return not string_utils.is_empty(val) + +def is_negative(val): + return string_utils.is_empty(val) + +def normalize_term(text: str) -> str: + """Normalize term text for comparison""" + replacements = { + "ONE (1)": "1", + "ONE": "1", + "TWO": "2", + "THREE (3)": "3", + "THREE": "3", + "FOUR": "4", + "FIVE": "5", + "YEAR TO YEAR": "ANNUAL", + "FROM THE CONTRACT EFFECTIVE DATE": "", + "FROM CONTRACT EFFECTIVE DATE": "", + "TERM": "", + "YEAR(S)": "YEAR", + "YEARS": "YEAR", + "-YEAR": " YEAR", + " ": " " + } + + text = text.upper() + for old, new in replacements.items(): + text = text.replace(old, new) + + return text.strip() + +def is_fuzzy_match(pred, truth): + """Compare strings using fuzzy matching with preprocessing + + Args: + pred: Predicted value + truth: Ground truth value + + Returns: + bool: True if strings match within threshold + """ + if is_negative(pred) or is_negative(truth): + return False + + # Normalize both strings + pred_norm = normalize_term(str(pred)) + truth_norm = normalize_term(str(truth)) + + # First try exact match on normalized strings + if pred_norm == truth_norm or pred_norm in truth_norm or truth_norm in pred_norm: + return True + + # Then try fuzzy match + match_ratio = fuzz.ratio(pred_norm, truth_norm) + + # # Debugging line + # if match_ratio < 100 and field == "field of interest": + # print(f"Fuzzy match ratio for '{pred_norm}' | '{truth_norm}': {match_ratio}") + + return match_ratio >= 80 + + def calculate_field_metrics(merged, field): """Calculate precision, recall, and accuracy for a single field using fuzzy matching. @@ -103,67 +167,6 @@ def calculate_field_metrics(merged, field): Returns: tuple: (precision, recall, accuracy) """ - def is_positive(val): - return not string_utils.is_empty(val) - - def is_negative(val): - return string_utils.is_empty(val) - - def normalize_term(text: str) -> str: - """Normalize term text for comparison""" - replacements = { - "ONE (1)": "1", - "ONE": "1", - "TWO": "2", - "THREE (3)": "3", - "THREE": "3", - "FOUR": "4", - "FIVE": "5", - "YEAR TO YEAR": "ANNUAL", - "FROM THE CONTRACT EFFECTIVE DATE": "", - "FROM CONTRACT EFFECTIVE DATE": "", - "TERM": "", - "YEAR(S)": "YEAR", - "YEARS": "YEAR", - "-YEAR": " YEAR", - " ": " " - } - - text = text.upper() - for old, new in replacements.items(): - text = text.replace(old, new) - - return text.strip() - - def is_fuzzy_match(pred, truth): - """Compare strings using fuzzy matching with preprocessing - - Args: - pred: Predicted value - truth: Ground truth value - - Returns: - bool: True if strings match within threshold - """ - if is_negative(pred) or is_negative(truth): - return False - - # Normalize both strings - pred_norm = normalize_term(str(pred)) - truth_norm = normalize_term(str(truth)) - - # First try exact match on normalized strings - if pred_norm == truth_norm or pred_norm in truth_norm or truth_norm in pred_norm: - return True - - # Then try fuzzy match - match_ratio = fuzz.ratio(pred_norm, truth_norm) - - # # Debugging line - # if match_ratio < 100 and field == "field of interest": - # print(f"Fuzzy match ratio for '{pred_norm}' | '{truth_norm}': {match_ratio}") - - return match_ratio >= 80 # True Positive: both pred and truth are non-empty and fuzzy match >= 90% TP = sum( @@ -196,8 +199,6 @@ def calculate_field_metrics(merged, field): truth for pred, truth in zip(merged[f'{field}_pred'], merged[f'{field}_truth']) if is_negative(pred) and is_positive(truth) ] - # print(field) - # print(f"TP: {TP}, FP: {FP}, TN: {TN}, FN: {FN}") # Calculate metrics precision = TP / (TP + FP) if (TP + FP) > 0 else 0 @@ -1126,7 +1127,7 @@ def get_one_to_one_analysis(testbed, results): one_to_one_fields.extend(['PROV_GROUP_TIN', 'PROV_GROUP_NPI', 'PROV_GROUP_NAME_FULL']) - one_to_one_fields = [f for f in one_to_one_fields if f not in {'TIN', 'NPI', 'NAME', "CLIENT_NAME"}] + one_to_one_fields = [f for f in one_to_one_fields if f not in {'TIN', 'NPI', 'NAME', "CLIENT_NAME", "GLOBAL_LESSER_OF_STATEMENT"}] # Create filtered dataframes with only one-to-one fields one_to_one_columns = ['FILE_NAME'] + [f for f in one_to_one_fields if f in testbed.columns and f in results.columns] @@ -1134,17 +1135,26 @@ def get_one_to_one_analysis(testbed, results): results_one_to_one = results[one_to_one_columns].drop_duplicates() metrics_df = pd.DataFrame(columns=['Field', 'Precision', 'Recall', 'Accuracy', 'FN_list']) - comparison_df = pd.DataFrame(index=testbed_one_to_one['FILE_NAME'].unique()) + comparison_df = pd.DataFrame(index=testbed['FILE_NAME'].unique()) - for field in one_to_one_fields: + # for field in one_to_one_fields: + for field in ['PROV_GROUP_TIN']: + if field in testbed.columns and field in results.columns and field != 'FILE_NAME': # Merge and compare values (no need to drop_duplicates here since we already did) merged = pd.merge(testbed_one_to_one[['FILE_NAME', field]], results_one_to_one[['FILE_NAME', field]], on='FILE_NAME', suffixes=('_truth', '_pred')) - # Compare values - merged[field] = (merged[f"{field}_truth"] == merged[f"{field}_pred"]) & (merged[f"{field}_truth"] != '') + + + # Normalize strings by stripping whitespace and converting to same case + merged[f"{field}_truth_norm"] = merged[f"{field}_truth"].astype(str).str.strip().str.upper() + merged[f"{field}_pred_norm"] = merged[f"{field}_pred"].astype(str).str.strip().str.upper() + + # Compare values using normalized strings + merged[field] = (merged[f"{field}_truth_norm"] == merged[f"{field}_pred_norm"]) & (merged[f"{field}_truth"] != '') + # Drop duplicate rows merged = merged.drop_duplicates(subset=['FILE_NAME', field]) diff --git a/fieldExtraction/tests/test_tin_npi_funcs.py b/fieldExtraction/tests/test_tin_npi_funcs.py index 0b5b39d..077a478 100644 --- a/fieldExtraction/tests/test_tin_npi_funcs.py +++ b/fieldExtraction/tests/test_tin_npi_funcs.py @@ -1,194 +1,151 @@ import pytest -from src.investment.tin_npi_funcs import clean_provider_info, get_all_matches, chunk_on_matches, deduplicate_providers +from unittest.mock import patch, MagicMock +import json -def test_clean_provider_info_valid_data(): - provider_info = [ - {"TIN": "12-3456789", "NPI": "1234567890", "NAME": "John Doe", "IS_GROUP": True}, - {"TIN": "98.7654321", "NPI": "0987654321", "NAME": " Jane Smith ", "IS_GROUP": False} - ] - expected_output = [ - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe", "IS_GROUP": True}, - {"TIN": "987654321", "NPI": "0987654321", "NAME": "Jane Smith", "IS_GROUP": False} - ] - assert clean_provider_info(provider_info) == expected_output +import src.investment.tin_npi_funcs as tin_npi_funcs +from src.prompts.investment_prompts import FieldSet -def test_clean_provider_info_invalid_tin_npi(): - provider_info = [ - {"TIN": "12345", "NPI": "12345", "NAME": "John Doe", "IS_GROUP": True}, - {"TIN": None, "NPI": None, "NAME": "Jane Smith", "IS_GROUP": False} - ] - expected_output = [ - {"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "John Doe", "IS_GROUP": True}, - {"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "Jane Smith", "IS_GROUP": False} - ] - assert clean_provider_info(provider_info) == expected_output +class TestTinNpiFuncs: + @pytest.fixture + def sample_text_dict(self): + return { + "1": "First page with TIN 12-3456789", + "2": "Second page with NPI 1234567890", + "2.1": "Continuation with same TIN 12-3456789", + "3": "Page with both TIN 98-7654321 and NPI 9876543210" + } -def test_clean_provider_info_missing_fields(): - provider_info = [ - {"TIN": "12-3456789", "NAME": "John Doe"}, - {"NPI": "1234567890", "IS_GROUP": True} - ] - expected_output = [ - {"TIN": "123456789", "NPI": "UNKNOWN", "NAME": "John Doe", "IS_GROUP": False}, - {"TIN": "UNKNOWN", "NPI": "1234567890", "NAME": "UNKNOWN", "IS_GROUP": True} - ] - assert clean_provider_info(provider_info) == expected_output + def test_get_all_matches(self): + text = "TIN: 12-3456789 and another TIN: 98-7654321" + pattern = r'\d{2}[-\s]?\d{7}' + result = tin_npi_funcs.get_all_matches(text, pattern, "test.pdf") + assert sorted(result) == sorted(['12-3456789', '98-7654321']) -def test_clean_provider_info_empty_input(): - provider_info = [] - expected_output = [] - assert clean_provider_info(provider_info) == expected_output + # Test empty pattern + assert tin_npi_funcs.get_all_matches(text, "", "test.pdf") == [] + + # Test no matches + assert tin_npi_funcs.get_all_matches("No TINs here", pattern, "test.pdf") == [] -def test_clean_provider_info_whitespace_name(): - provider_info = [ - {"TIN": "12-3456789", "NPI": "1234567890", "NAME": " ", "IS_GROUP": True} - ] - expected_output = [ - {"TIN": "123456789", "NPI": "1234567890", "NAME": "UNKNOWN", "IS_GROUP": True} - ] - assert clean_provider_info(provider_info) == expected_output + def test_chunk_on_matches(self): + text_dict = { + "1": "Page with match1", + "2": "Page without matches", + "3": "Page with match2" + } + matches = ["match1", "match2"] + result = tin_npi_funcs.chunk_on_matches(matches, text_dict) + assert "match1" in result + assert "match2" in result + assert "without matches" not in result -def test_get_all_matches_single_match(): - text = "The TIN is 123456789." - pattern = r"\b\d{9}\b" - filename = "test_file.txt" - expected_output = ["123456789"] - assert get_all_matches(text, pattern, filename) == expected_output + def test_clean_provider_info(self): + providers = [ + { + "TIN": "12-345.6789", + "NPI": "12.34567890", + "NAME": "Test Provider", + "ON_SIGNATURE_PAGE": "Y", + "ON_INTRO_PAGE": "N" + }, + { + "TIN": "", + "NPI": "invalid", + "NAME": "", + "ON_SIGNATURE_PAGE": "N", + "ON_INTRO_PAGE": "Y" + } + ] + result = tin_npi_funcs.clean_provider_info(providers) + + assert result[0]["TIN"] == "123456789" + assert result[0]["NPI"] == "1234567890" + assert result[0]["NAME"] == "Test Provider" + + assert result[1]["TIN"] == "UNKNOWN" + assert result[1]["NPI"] == "UNKNOWN" + assert result[1]["NAME"] == "UNKNOWN" -def test_get_all_matches_multiple_matches(): - text = "TINs: 123456789, 987654321, and 123456789." - pattern = r"\b\d{9}\b" - filename = "test_file.txt" - expected_output = ["123456789", "987654321"] - assert sorted(get_all_matches(text, pattern, filename)) == sorted(expected_output) + def test_merge_provider_info(self): + one_to_one_results = {} + provider_info = [ + { + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Group Provider", + "ON_SIGNATURE_PAGE": "Y" + }, + { + "TIN": "987654321", + "NPI": "9876543210", + "NAME": "Other Provider", + "ON_SIGNATURE_PAGE": "N" + } + ] + + result = tin_npi_funcs.merge_provider_info(one_to_one_results, provider_info) + + assert result["PROV_GROUP_TIN"] == "123456789" + assert result["PROV_OTHER_TIN"] == "987654321" + assert "Group Provider" in result["PROV_GROUP_NAME_FULL"] + assert "Other Provider" in result["PROV_OTHER_NAME_FULL"] -def test_get_all_matches_no_matches(): - text = "No valid TINs here." - pattern = r"\b\d{9}\b" - filename = "test_file.txt" - expected_output = [] - assert get_all_matches(text, pattern, filename) == expected_output -def test_get_all_matches_special_characters(): - text = "TINs: 123-45-6789 and 987.65.4321." - pattern = r"\b\d{3}-\d{2}-\d{4}\b|\b\d{3}\.\d{2}\.\d{4}\b" - filename = "test_file.txt" - expected_output = ["123-45-6789", "987.65.4321"] - assert sorted(get_all_matches(text, pattern, filename)) == sorted(expected_output) + def test_deduplicate_providers(self): + providers = [ + { + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Provider A" + }, + { + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Provider A" + }, + { + "TIN": "UNKNOWN", + "NPI": "UNKNOWN", + "NAME": "UNKNOWN" + } + ] + + result = tin_npi_funcs.deduplicate_providers(providers) + assert len(result) == 1 + assert result[0]["TIN"] == "123456789" -def test_get_all_matches_empty_text(): - text = "" - pattern = r"\b\d{9}\b" - filename = "test_file.txt" - expected_output = [] - assert get_all_matches(text, pattern, filename) == expected_output + @patch('src.utils.llm_utils.invoke_claude') + def test_get_provider_info(self, mock_invoke_claude, sample_text_dict): + mock_invoke_claude.return_value = json.dumps([{ + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Test Provider", + "ON_SIGNATURE_PAGE": "N" + }]) + + result = tin_npi_funcs.get_provider_info(sample_text_dict, "1", "test.pdf") + assert len(result) == 1 + assert result[0]["ON_INTRO_PAGE"] == "Y" # Page 1 should be marked as intro -def test_get_all_matches_empty_pattern(): - text = "The TIN is 123456789." - pattern = "" - filename = "test_file.txt" - expected_output = [] - assert get_all_matches(text, pattern, filename) == expected_output - -def test_chunk_on_matches_single_match(): - matches = ["match1"] - text_dict = { - "page1": "This is a text with match1.", - "page2": "This is another text without matches." - } - expected_output = "This is a text with match1." - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_multiple_matches(): - matches = ["match1", "match2"] - text_dict = { - "page1": "This is a text with match1.", - "page2": "This is another text with match2.", - "page3": "No matches here." - } - expected_output = "This is a text with match1.\nThis is another text with match2." - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_no_matches(): - matches = ["match1"] - text_dict = { - "page1": "No relevant text here.", - "page2": "Still no matches." - } - expected_output = "" - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_empty_matches(): - matches = [] - text_dict = { - "page1": "This is some text.", - "page2": "This is more text." - } - expected_output = "" - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_empty_text_dict(): - matches = ["match1"] - text_dict = {} - expected_output = "" - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_partial_match(): - matches = ["match"] - text_dict = { - "page1": "This is a text with match1.", - "page2": "This is another text with match2.", - "page3": "But this page you won't find one." - } - expected_output = "This is a text with match1.\nThis is another text with match2." - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_chunk_on_matches_whitespace_handling(): - matches = ["match1"] - text_dict = { - "page1": " This is a text with match1. ", - "page2": "This is another text without matches." - } - expected_output = " This is a text with match1. " - assert chunk_on_matches(matches, text_dict) == expected_output - -def test_deduplicate_providers_removes_exact_duplicates(): - providers = [ - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - ] - result = deduplicate_providers(providers) - assert len(result) == 1 - assert result[0]["NAME"] == "John Doe" - -def test_deduplicate_providers_same_tin_npi_different_name(): - providers = [ - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - {"TIN": "123456789", "NPI": "1234567890", "NAME": "Jane Smith"}, - ] - result = deduplicate_providers(providers) - assert len(result) == 2 - -def test_deduplicate_providers_skips_empty_and_unknown(): - providers = [ - {"TIN": "", "NPI": "", "NAME": ""}, - {"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "UNKNOWN"}, - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - ] - result = deduplicate_providers(providers) - assert len(result) == 1 - assert result[0]["NAME"] == "John Doe" - -def test_deduplicate_providers_mixed_cases(): - providers = [ - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - {"TIN": "123456789", "NPI": "1234567890", "NAME": "John Doe"}, - {"TIN": "", "NPI": "", "NAME": ""}, - {"TIN": "987654321", "NPI": "0987654321", "NAME": "Jane Smith"}, - {"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "UNKNOWN"}, - ] - result = deduplicate_providers(providers) - assert len(result) == 2 - names = {p["NAME"] for p in result} - assert "John Doe" in names - assert "Jane Smith" in names \ No newline at end of file + @patch('src.utils.llm_utils.invoke_claude') + def test_run_provider_info_fields(self, mock_invoke_claude, sample_text_dict): + mock_invoke_claude.return_value = json.dumps([{ + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Test Provider", + "ON_SIGNATURE_PAGE": "N" + }]) + + one_to_one_fields = FieldSet(relationship="one_to_one") + contract_text = "Contract with TIN 12-3456789" + + results, fields = tin_npi_funcs.run_provider_info_fields( + contract_text, + one_to_one_fields, + sample_text_dict, + "test.pdf" + ) + + assert "PROV_INFO_JSON" in results + assert "PROV_INFO_JSON_FORMATTED" in results + assert isinstance(results["PROV_INFO_JSON"], str) \ No newline at end of file