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

Merged in feature/prov-group-tin (pull request #584)

Browse Source 
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
This commit is contained in:
Katon Minhas
2025-06-26 19:35:32 +00:00
parent 6d27b17819
commit 1531ad6bac
7 changed files with 352 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/investment/file_processing.py
+4 -22
View File
@@ -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
fieldExtraction/src/investment/one_to_one_funcs.py
+2 -57
View File
@@ -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.
fieldExtraction/src/investment/tin_npi_funcs.py
+111 -79
View File
@@ -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
fieldExtraction/src/prompts/investment_prompts.json
+10 -4
View File
@@ -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'."
},
{
fieldExtraction/src/prompts/investment_prompts.py
+8 -5
View File
@@ -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",
}}
]
fieldExtraction/src/testbed/testbed_utils.py
+78 -68
View File
@@ -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])
fieldExtraction/tests/test_tin_npi_funcs.py
+139 -182
View File
@@ -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
@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)