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

Update prov_info

Browse Source
This commit is contained in:
Katon Minhas
2026-02-04 03:37:23 -05:00
parent f0da944ade
commit f8783ae968
12 changed files with 101 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/constants/investment_columns.py
-1
View File
@@ -20,7 +20,6 @@ COLUMN_ORDER = [
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
"PROV_INFO_JSON",
"PROV_INFO_JSON_FORMATTED",
"EFFECTIVE_DT",
"AARETE_DERIVED_EFFECTIVE_DT",
"TERMINATION_DT",
src/pipelines/clients/bcbs_promise/file_processing.py
+1 -1
View File
@@ -238,7 +238,7 @@ def run_one_to_one_prompts(
# Add HSC's N/A if field doesn't exist yet
one_to_one_results[key] = value
one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(
one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one(
one_to_one_results, filename
)
src/pipelines/clients/clover/file_processing.py
+1 -1
View File
@@ -244,7 +244,7 @@ def run_one_to_one_prompts(
# Add HSC's N/A if field doesn't exist yet
one_to_one_results[key] = value
one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(
one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one(
one_to_one_results, filename
)
src/pipelines/saas/file_processing.py
+2 -1
View File
@@ -228,7 +228,8 @@ def run_one_to_one_prompts(
# Add HSC's N/A if field doesn't exist yet
one_to_one_results[key] = value
one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(
one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one(
one_to_one_results, filename
)
src/pipelines/saas/prompts/prompt_calls.py
+5 -9
View File
@@ -929,13 +929,12 @@ def prompt_provider_info(
Raises:
ValueError: If the response from the language model cannot be parsed as valid JSON.
"""
chunk = text_dict[page_num]
page_text = text_dict[page_num]
provider_fields = FieldSet(
file_path=config.FIELD_JSON_PATH, field_type="provider_info"
)
prompt, _parser = prompt_templates.TIN_NPI_TEMPLATE(
chunk, provider_fields.print_prompt_dict(), payer_name
page_text, provider_fields.print_prompt_dict(), payer_name
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
@@ -955,15 +954,12 @@ def prompt_provider_info(
providers = _parser(llm_answer_raw)
logging.debug(f"Parsed provider info on page {page_num}: {providers}")
# Ensure we have a list of providers (not a dict or other type)
if isinstance(providers, dict):
# Single provider returned as dict - wrap in list
providers = [providers]
elif not isinstance(providers, list):
if not isinstance(providers, list):
# If not a list or dict, create a default list with error info
logging.warning(
f"Warning: Unexpected format in provider info response: {type(providers)}"
)
providers = [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}]
providers = [{"TIN": "N/A", "NPI": "N/A", "NAME": "PARSING_ERROR"}]
# VALIDATION LAYER - Cross-check with regex findings
page_text = text_dict[page_num]
@@ -975,4 +971,4 @@ def prompt_provider_info(
logging.error(f"Error parsing JSON response from LLM: {str(e)}")
logging.error(f"Raw response: {llm_answer_raw}")
# Return a default value
return [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}]
return [{"TIN": "N/A", "NPI": "N/A", "NAME": "N/A"}]
src/pipelines/saas/test.py
-12
View File
@@ -1,12 +0,0 @@
import src.pipelines.shared.extraction.tin_npi_funcs as tin_npi_funcs
one_to_one_results = {'PROV_INFO_JSON': [{'TIN': '971098746', 'NPI': 'UNKNOWN', 'NAME': 'Midwest Medical Equipment Solutions, Inc'}], 'PROV_INFO_JSON_FORMATTED': '{"TIN": "971098746", "NPI": "UNKNOWN", "NAME": "Midwest Medical Equipment Solutions, Inc"}', 'TIN_EXTRACTION_QUALITY': '1 exact, 0 OCR-corrected', 'NPI_EXTRACTION_QUALITY': '0 exact, 0 OCR-corrected', 'FILENAME_TIN': 'N/A', 'PAYER_STATE': ['UNKNOWN'], 'TERMINATION_DT': 'N/A', 'AUTO_RENEWAL_TERM': 'N/A', 'CONTRACT_AMENDMENT_NUM': 'UNKNOWN', 'CONTRACT_TITLE': 'MEDICARE PRODUCT ATTACHMENT COMPENSATION SCHEDULE EXHIBIT ANCILLARY SERVICES DURABLE MEDICAL EQUIPMENT', 'PAYER_NAME': 'IlliniCare Health', 'PROVIDER_NAME': 'Midwest Medical Equipment Solutions, Inc.', 'NUM_AVAILABLE_SIGNATORY_LINE_COUNT': '0', 'PROVIDER_STATE': ['IL'], 'AUTO_RENEWAL_IND': 'N', 'EFFECTIVE_DT': 'January 1, 2018', 'AARETE_DERIVED_EFFECTIVE_DT': '2018/01/01'}
one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename="")
print("\nFinal: ", one_to_one_results)
src/pipelines/shared/extraction/one_to_one_funcs.py
-20
View File
@@ -196,26 +196,6 @@ def run_full_context_fields(
contract_text, full_context_fields, constants, filename
)
if "PROV_GROUP_NAME_FULL" in full_context_answers_dict:
full_context_answers_dict["PROV_INFO_JSON"] = json.dumps(
[
{
"TIN": "UNKNOWN",
"NPI": "UNKNOWN",
"NAME": full_context_answers_dict["PROV_GROUP_NAME_FULL"],
}
]
)
full_context_answers_dict["PROV_INFO_JSON_FORMATTED"] = json.dumps(
[
{
"TIN": "UNKNOWN",
"NPI": "UNKNOWN",
"NAME": full_context_answers_dict["PROV_GROUP_NAME_FULL"],
}
]
)
# Normalize PAYER_STATE to two-letter abbreviation
if "PAYER_STATE" in full_context_answers_dict:
full_context_answers_dict["PAYER_STATE"] = (
src/pipelines/shared/extraction/tin_npi_funcs.py
+88 -265
View File
@@ -10,6 +10,70 @@ from src.pipelines.saas.prompts import prompt_calls
from src.prompts.fieldset import Field, FieldSet
def run_provider_info_fields(
contract_text: str, text_dict: dict, filename: str, payer_name: str
):
"""
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.
"""
one_to_one_results = {}
all_tins, all_npis, tin_matches, npi_matches = get_all_tins_and_npis(contract_text)
exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis = get_tin_extraction_quality(tin_matches, npi_matches, filename)
relevant_pages = get_relevant_pages(all_tins, all_npis, text_dict, filename)
prov_info_json = get_prov_info_json(relevant_pages, text_dict, filename, payer_name)
# Create columns
one_to_one_results["PROV_INFO_JSON"] = prov_info_json
one_to_one_results["TIN_EXTRACTION_QUALITY"] = (
f"{len(exact_tins)} exact, {len(ocr_corrected_tins)} OCR-corrected"
)
one_to_one_results["NPI_EXTRACTION_QUALITY"] = (
f"{len(exact_npis)} exact, {len(ocr_corrected_npis)} OCR-corrected"
)
# Add FILENAME_TIN
filename_tin = get_all_matches(filename, regex_patterns.FILE_NAME_TIN_PATTERN)
one_to_one_results["FILENAME_TIN"] = filename_tin[0] if filename_tin else "N/A"
return one_to_one_results
def get_all_tins_and_npis(contract_text):
tin_matches = get_all_matches_with_ocr(
contract_text,
regex_patterns.TIN_PATTERN,
regex_patterns.TIN_OCR_PATTERN,
identifier_type="TIN",
)
npi_matches = get_all_matches_with_ocr(
contract_text,
regex_patterns.NPI_PATTERN,
regex_patterns.NPI_OCR_PATTERN,
identifier_type="NPI",
)
# Extract just the values from the (value, match_quality) tuples for downstream processing
all_tins = [tin for tin, quality in tin_matches]
all_npis = [npi for npi, quality in npi_matches]
logging.debug(f"Extracted TINs: {all_tins}")
logging.debug(f"Extracted NPIs: {all_npis}")
return all_tins, all_npis, tin_matches, npi_matches
def get_all_matches(text: str, pattern: str) -> list[str]:
"""
Extracts all unique matches of a given pattern from the provided text.
@@ -138,77 +202,6 @@ def attempt_ocr_correction(candidate: str, identifier_type: str) -> tuple[str, b
return candidate, False
def chunk_on_matches(matches: list[str], text_dict: dict) -> str:
"""
Extracts and concatenates text from a dictionary of pages where any of the specified matches are found.
Args:
matches (list[str]): A list of strings to search for in the text.
text_dict (dict): A dictionary where keys are identifiers (e.g., page numbers) and values are strings of text.
Returns:
str: A single string containing the concatenated text from all pages where at least one match is found,
separated by newline characters.
"""
valid_pages = []
for page_text in text_dict.values():
if any(match in page_text for match in matches):
valid_pages.append(page_text)
return "\n".join(valid_pages)
def clean_provider_info(prov_info_json: list[dict]) -> list[dict]:
"""Cleans provider information by standardizing Name/TIN/NPI formats.
Args:
provider_info (list[dict]): A list of provider information dictionaries. Each
dictionary should be keyed by TIN, NPI, NAME
Returns:
list: Cleaned provider information
"""
prov_info_json_clean = []
for provider in prov_info_json:
cleaned_provider = {}
# ---- Clean TIN (digits only, 9 chars) ----
tin = provider.get("TIN")
if tin:
tin_str = str(tin).replace("-", "").replace(".", "")
if tin_str.isdigit() and len(tin_str) == 9:
cleaned_provider["TIN"] = tin_str
else:
cleaned_provider["TIN"] = "UNKNOWN"
else:
cleaned_provider["TIN"] = "UNKNOWN"
# ---- Clean NPI (digits only, 10 chars) ----
npi = provider.get("NPI")
if npi:
npi_str = str(npi).replace("-", "").replace(".", "")
if npi_str.isdigit() and len(npi_str) == 10:
cleaned_provider["NPI"] = npi_str
else:
cleaned_provider["NPI"] = "UNKNOWN"
else:
cleaned_provider["NPI"] = "UNKNOWN"
# ---- Clean NAME (list) ----
name = provider.get("NAME")
if name:
name_str = " ".join(str(name).split()).strip().rstrip(".")
if name_str:
cleaned_provider["NAME"] = name_str
else:
cleaned_provider["NAME"] = "UNKNOWN"
else:
cleaned_provider["NAME"] = "UNKNOWN"
prov_info_json_clean.append(cleaned_provider)
return prov_info_json_clean
def normalize_to_lists(provider: dict):
tins = provider.get("TIN")
@@ -234,28 +227,28 @@ def normalize_to_lists(provider: dict):
def deunknown_and_deduplicate(group_tins, group_npis, group_names, other_tins, other_npis, other_names):
# 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"]
group_tins = [v for v in group_tins if not string_utils.is_empty(v)]
group_npis = [v for v in group_npis if not string_utils.is_empty(v)]
group_names = [v for v in group_names if not string_utils.is_empty(v)]
# Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other
group_tins = list(dict.fromkeys(group_tins))
other_tins = list(dict.fromkeys(other_tins))
other_tins = [v for v in other_tins if v not in group_tins and v != "UNKNOWN"]
other_tins = [v for v in other_tins if v not in group_tins and not string_utils.is_empty(v)]
group_npis = list(dict.fromkeys(group_npis))
other_npis = list(dict.fromkeys(other_npis))
other_npis = [v for v in other_npis if v not in group_npis and v != "UNKNOWN"]
other_npis = [v for v in other_npis if v not in group_npis and not string_utils.is_empty(v)]
group_names = list(dict.fromkeys(group_names))
other_names = list(dict.fromkeys(other_names))
other_names = [v for v in other_names if v not in group_names and v != "UNKNOWN"]
other_names = [v for v in other_names if v not in group_names and not string_utils.is_empty(v)]
return group_tins, group_npis, group_names, other_tins, other_npis, other_names
def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, filename: str):
def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str):
"""
Merges provider_info into one_to_one_results.
@@ -268,14 +261,11 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil
dict: Updated one_to_one_results with merged values from provider_info.
"""
# Extract group and non-group providers
group_tins = set()
group_npis = set()
group_names = set()
other_tins = []
other_npis = []
other_names = []
group_tins, group_npis, group_names = set(), set(), set()
other_tins, other_npis, other_names = [], [], []
provider_name = one_to_one_results.get("PROVIDER_NAME") # Can be List or string
# Normalize provider_name to list: if string, convert to list to avoid character splitting
if isinstance(provider_name, str):
provider_name = [provider_name]
@@ -292,14 +282,12 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil
found_match = False
for provider_dict in prov_info_json_list:
print("Provider dict: ", provider_dict)
tins, npis, names = normalize_to_lists(provider_dict)
name_matches = prompt_calls.provider_name_match_check(
names, provider_name, filename
) # Returns Boolean
print("Name matches: ", name_matches)
if name_matches:
found_match = True
@@ -338,18 +326,12 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil
for p in prov_info_json_list
if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME", []))
]
print("Before dedup: ", prov_info_json_list)
# Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values
prov_info_json_list = deduplicate_providers_by_name(prov_info_json_list)
print("After dedup: ", prov_info_json_list)
# Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags
one_to_one_results["PROV_INFO_JSON"] = prov_info_json_list
one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join(
[json.dumps(provider) for provider in prov_info_json_list]
)
# Merge group provider information into one_to_one_results
one_to_one_results["PROV_GROUP_TIN"] = group_tins if group_tins else []
@@ -363,77 +345,8 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil
return one_to_one_results
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.
Args:
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.
Returns:
tuple[dict, FieldSet]: A dictionary containing extracted TIN and NPI values, and an
updated FieldSet with additional fields if multiple values are found.
Fields Added: REIMB_PROV_NAME,
REIMB_PROV_TIN,
REIMB_PROV_NPI
"""
def get_list(field_name, filtered_providers):
# Get all values and filter out invalid ones
all_values = [
provider.get(field_name, "N/A") for provider in filtered_providers
]
valid_values = [
val
for val in all_values
if not string_utils.is_empty(val) and val != "UNKNOWN"
]
return list(set(valid_values)) # Remove duplicates
answer_dict = {}
# Extract all providers from the exhibit chunk
providers = prompt_calls.prompt_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))
field_configs = [
("REIMB_PROV_TIN", "TIN"),
("REIMB_PROV_NPI", "NPI"),
("REIMB_PROV_NAME", "NAME"),
]
# Always add providers to reimbursement_level fields for proper term attribution
for field_name, provider_key in field_configs:
if len(filtered_providers) > 0:
field = Field.load_from_file(config.FIELD_JSON_PATH, field_name)
value_list = get_list(provider_key, filtered_providers)
if len(value_list) == 0:
answer_dict[field_name] = "N/A"
else:
field.update_valid_values(value_list)
reimbursement_level_fields.add_field(field)
else:
# If no providers are found, set default values
answer_dict[field_name] = "N/A"
return answer_dict, reimbursement_level_fields
def deduplicate_providers(
provider_info: list[dict],
prov_info_json: list[dict],
) -> list[dict]: # TODO: add unit test
"""Deduplicates provider information based on TIN, NPI, and NAME fields.
This function checks for duplicates in the provider information list by comparing
@@ -451,31 +364,19 @@ def deduplicate_providers(
valid_providers = []
# Deduplicate if all three fields match
for provider in provider_info:
tins = provider.get("TIN")
npis = provider.get("NPI")
names = provider.get("NAME")
for prov_info_dict in prov_info_json:
tin = prov_info_dict.get("TIN")
npi = prov_info_dict.get("NPI")
name = prov_info_dict.get("NAME")
# # Normalize missing values
tins = tins if isinstance(tins, list) else []
npis = npis if isinstance(npis, list) else []
names = names if isinstance(names, list) else []
# Skip providers where all critical identification fields are unknown
if (
(not tins or tins == "UNKNOWN")
and (not npis or npis == "UNKNOWN")
and (not names or names == "UNKNOWN")
):
if (string_utils.is_empty(tin) and string_utils.is_empty(npi) and string_utils.is_empty(name)):
continue
key = (
tuple(tins),
tuple(npis),
tuple(names),
)
key = (tin, npi, name)
if key not in seen:
seen.add(key)
valid_providers.append(provider)
valid_providers.append(prov_info_dict)
return valid_providers
@@ -526,10 +427,7 @@ def flag_missed_prov_info(page_text, providers, page_num, filename):
def get_relevant_pages(all_tins, all_npis, text_dict, filename):
# If there are no identifiers, return early with default values
if not all_tins and not all_npis:
# if there are no TINs or NPIs, we need to extract signature pages
relevant_pages = list(
string_utils.extract_signature_page(text_dict, filename).keys()
)
return []
else:
# Identify relevant pages containing either NPIs or TINs (with or without hyphen formatting, e.g., 68-0640053)
relevant_pages = [
@@ -555,10 +453,10 @@ def get_relevant_pages(all_tins, all_npis, text_dict, filename):
def get_prov_info_json(relevant_pages, text_dict, filename, payer_name):
if not relevant_pages:
prov_info_json_clean = [
prov_info_json = [
{
"TIN": "UNKNOWN",
"NPI": "UNKNOWN",
"TIN": "N/A",
"NPI": "N/A",
"NAME": "NO_IDENTIFIERS_FOUND",
}
]
@@ -571,41 +469,12 @@ def get_prov_info_json(relevant_pages, text_dict, filename, payer_name):
)
prov_info_json.extend(page_provider_info)
print("Before Clean: ", prov_info_json)
# Clean and standardize results
prov_info_json_clean = clean_provider_info(prov_info_json)
print("After Clean: ", prov_info_json_clean)
# Deduplicate providers based on TIN and NPI
prov_info_json_clean = deduplicate_providers(prov_info_json_clean)
print("After dedup: ", prov_info_json_clean)
return prov_info_json_clean
prov_info_json = deduplicate_providers(prov_info_json)
def get_all_tins_and_npis(contract_text):
tin_matches = get_all_matches_with_ocr(
contract_text,
regex_patterns.TIN_PATTERN,
regex_patterns.TIN_OCR_PATTERN,
identifier_type="TIN",
)
return prov_info_json
npi_matches = get_all_matches_with_ocr(
contract_text,
regex_patterns.NPI_PATTERN,
regex_patterns.NPI_OCR_PATTERN,
identifier_type="NPI",
)
# Extract just the values from the (value, match_quality) tuples for downstream processing
all_tins = [tin for tin, quality in tin_matches]
all_npis = [npi for npi, quality in npi_matches]
logging.debug(f"Extracted TINs: {all_tins}")
logging.debug(f"Extracted NPIs: {all_npis}")
return all_tins, all_npis, tin_matches, npi_matches
def get_tin_extraction_quality(tin_matches, npi_matches, filename):
# Log OCR corrections for monitoring
@@ -627,52 +496,6 @@ def get_tin_extraction_quality(tin_matches, npi_matches, filename):
return exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis
def run_provider_info_fields(
contract_text: str, text_dict: dict, filename: str, payer_name: str
):
"""
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, tin_matches, npi_matches = get_all_tins_and_npis(contract_text)
exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis = get_tin_extraction_quality(tin_matches, npi_matches, filename)
relevant_pages = get_relevant_pages(all_tins, all_npis, text_dict, filename)
prov_info_json_clean = get_prov_info_json(relevant_pages, text_dict, filename, payer_name)
# Create columns
one_to_one_results = {}
one_to_one_results["PROV_INFO_JSON"] = prov_info_json_clean
one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join(
[json.dumps(provider) for provider in prov_info_json_clean]
)
one_to_one_results["TIN_EXTRACTION_QUALITY"] = (
f"{len(exact_tins)} exact, {len(ocr_corrected_tins)} OCR-corrected"
)
one_to_one_results["NPI_EXTRACTION_QUALITY"] = (
f"{len(exact_npis)} exact, {len(ocr_corrected_npis)} OCR-corrected"
)
# Add FILENAME_TIN
filename_tin = get_all_matches(filename, regex_patterns.FILE_NAME_TIN_PATTERN)
one_to_one_results["FILENAME_TIN"] = filename_tin[0] if filename_tin else "N/A"
print("run_provider_info_fields: ", one_to_one_results)
return one_to_one_results
def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]:
"""
Deduplicates a list of provider dictionaries by NAME, preferring records with actual TIN/NPI values.
src/prompts/prompt_templates.py
+2 -2
View File
@@ -1453,8 +1453,8 @@ Briefly explain your reasoning. After your explanation, include the heading "FIN
- 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"
- Use null (not strings like "null" or "N/A") for missing values
- A provider record is considered valid as long as either a TIN, NPI, or NAME is present. If any one or two of those fields are missing, use null for those fields.
- Use "N/A" for any missing values
- A provider record is considered valid as long as either a TIN, NPI, or NAME is present. If any one or two of those fields are missing, use "N/A" for those fields.
- 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
- Do not include any JSON notation elsewhere in your response
src/tests/test_tin_npi_funcs.py
+1 -38
View File
@@ -30,42 +30,6 @@ class TestTinNpiFuncs:
# Test no matches
assert tin_npi_funcs.get_all_matches("No TINs here", pattern) == []
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_clean_provider_info(self):
providers = [
{
"TIN": "12-345.6789",
"NPI": "12.34567890",
"NAME": "Test Provider",
"IS_GROUP": "Y",
},
{
"TIN": "",
"NPI": "invalid",
"NAME": "",
"IS_GROUP": "N",
},
]
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"
@patch("src.pipelines.saas.prompts.prompt_calls.provider_name_match_check")
def test_merge_provider_info_with_hybrid_smart_chunking(
@@ -97,7 +61,7 @@ class TestTinNpiFuncs:
),
}
result = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(
result = tin_npi_funcs.merge_provider_info_with_one_to_one(
one_to_one_results, "test.pdf"
)
@@ -142,7 +106,6 @@ class TestTinNpiFuncs:
)
assert "PROV_INFO_JSON" in results
assert "PROV_INFO_JSON_FORMATTED" in results
assert isinstance(results["PROV_INFO_JSON"], str)
def test_get_all_matches_with_ocr_exact_matches(self):
src/utils/qa_qc_utils.py
+1 -1
View File
@@ -678,7 +678,7 @@ QC_PROV_OTHER_TIN = "PROV_OTHER_TIN"
QC_REIMB_PROV_TIN = "REIMB_PROV_TIN"
# JSON columns that should not be scanned for hallucinations
QC_JSON_OK_COLS = {"PROV_INFO_JSON", "PROV_INFO_JSON_FORMATTED"}
QC_JSON_OK_COLS = {"PROV_INFO_JSON"}
# Date pattern strings for extraction
QC_DATE_PATTERNS = [
src/utils/string_utils.py
-1
View File
@@ -895,7 +895,6 @@ def normalize_one_to_one_answers_dict(answers_dict: dict[str, str | list]) -> di
"""
DO_NOT_NORMALIZE = [
"PROV_INFO_JSON",
"PROV_INFO_JSON_FORMATTED",
"PAYER_STATE",
"PROVIDER_STATE"
]