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

Merged in bugfix/list-of-lists (pull request #866)

Browse Source 
Bugfix/list of lists

* Revert tin_npi_funcs.py to DEV

* update bill type mapping

* Remove test.py

* Prep for merge
This commit is contained in:
Katon Minhas
2026-02-04 04:27:03 +00:00
committed by Praneel Panchigar
parent cff0472717
commit 0882a79b5a
4 changed files with 126 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+2 -2
View File
@@ -510,7 +510,7 @@ def fill_bill_type(
llm_answer_final = _parser(llm_answer_raw) llm_answer_final = _parser(llm_answer_raw)
# Second attempt: Service term + reimbursement term + exhibit context (if first attempt failed and context available) # Second attempt: Service term + reimbursement term + exhibit context (if first attempt failed and context available)
if not llm_answer_final and exhibit_text: if string_utils.is_empty(llm_answer_final) and exhibit_text:
prompt, _parser = prompt_templates.FILL_BILL_TYPE( prompt, _parser = prompt_templates.FILL_BILL_TYPE(
service, service,
valid_bill_type, valid_bill_type,
@@ -534,7 +534,7 @@ def fill_bill_type(
bill_codes, bill_descs = [], [] bill_codes, bill_descs = [], []
for description in llm_answer_final: for description in llm_answer_final:
if description in BILL_TYPE_REVERSE_MAPPING: if description in BILL_TYPE_REVERSE_MAPPING:
bill_codes.append(BILL_TYPE_REVERSE_MAPPING[description]) bill_codes += BILL_TYPE_REVERSE_MAPPING[description]
bill_descs.append(description) bill_descs.append(description)
if bill_codes: if bill_codes:
src/pipelines/saas/prompts/prompt_calls.py
+1
View File
@@ -481,6 +481,7 @@ def prompt_dynamic(text: str, field_prompts, filename):
cache=True, cache=True,
instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(), instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(),
) # Returns dictionary of lists ) # Returns dictionary of lists
logging.debug(f"Dynamic answer for {filename}: {llm_answer_raw}") logging.debug(f"Dynamic answer for {filename}: {llm_answer_raw}")
llm_answer_final = _parser(llm_answer_raw) llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final return llm_answer_final
src/pipelines/saas/test.py
-19
View File
@@ -1,19 +0,0 @@
from src.pipelines.shared.postprocessing import aarete_derived
from src.constants.constants import Constants
from src.utils import io_utils
from src.pipelines.saas.prompts import prompt_calls
from src.pipelines.shared.extraction.one_to_n_funcs import reimbursement_level
# constants = Constants()
service_term = "Covered Services provided to KIDSfirst Members"
reimb_term = "One hundred percent (100%) of the prevailing yearly and current Medicaid fee schedule for the State of Texas or the Participating Ancillary Services Provider's usual and customary charge, whichever is less."
exhibit_title = "Attachment B SECTION 1 - PCHP MEDICAID STAR HMO SERVICES"
filename = ""
lesser_of_answer_dict = prompt_calls.prompt_lesser_of_check(
service_term, reimb_term, exhibit_title, filename
)
print(lesser_of_answer_dict)
src/pipelines/shared/extraction/tin_npi_funcs.py
+123 -153
View File
@@ -2,12 +2,16 @@ import json
import logging import logging
import re import re
from typing import Optional
import src.constants.regex_patterns as regex_patterns import src.constants.regex_patterns as regex_patterns
import src.config as config import src.config as config
import src.prompts.prompt_templates as prompt_templates import src.prompts.prompt_templates as prompt_templates
from src.utils import llm_utils, string_utils from src.utils import llm_utils, string_utils
from src.pipelines.saas.prompts import prompt_calls from src.pipelines.saas.prompts import prompt_calls
from src.constants.delimiters import Delimiter
from src.prompts.fieldset import Field, FieldSet from src.prompts.fieldset import Field, FieldSet
from difflib import SequenceMatcher
def get_all_matches(text: str, pattern: str) -> list[str]: def get_all_matches(text: str, pattern: str) -> list[str]:
@@ -171,38 +175,34 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
for provider in provider_info: for provider in provider_info:
cleaned_provider = {} cleaned_provider = {}
# ---- Clean TIN (list, digits only, 9 chars) ---- # Clean TIN - remove hyphens and dots, ensure it's 9 digits
cleaned_tins = [] if "TIN" in provider and provider["TIN"]:
if isinstance(provider.get("TIN"), list): cleaned_provider["TIN"] = provider["TIN"].replace("-", "").replace(".", "")
for tin in provider["TIN"]: # Validate that TIN only contains digits or "|"
if not tin: if not all(c.isdigit() or c == "|" for c in cleaned_provider["TIN"]):
continue cleaned_provider["TIN"] = "UNKNOWN"
tin_str = str(tin).replace("-", "").replace(".", "") else:
if tin_str.isdigit() and len(tin_str) == 9: cleaned_provider["TIN"] = "UNKNOWN"
cleaned_tins.append(tin_str)
cleaned_provider["TIN"] = cleaned_tins if cleaned_tins else ["UNKNOWN"]
# ---- Clean NPI (list, digits only, 10 chars) ---- # Clean NPI - remove hyphens and dots, ensure it's 10 digits
cleaned_npis = [] if "NPI" in provider and provider["NPI"]:
if isinstance(provider.get("NPI"), list): cleaned_provider["NPI"] = provider["NPI"].replace("-", "").replace(".", "")
for npi in provider["NPI"]: # Validate that NPI only contains digits or "|"
if not npi: if not all(c.isdigit() or c == "|" for c in cleaned_provider["NPI"]):
continue cleaned_provider["NPI"] = "UNKNOWN"
npi_str = str(npi).replace("-", "").replace(".", "") else:
if npi_str.isdigit() and len(npi_str) == 10: cleaned_provider["NPI"] = "UNKNOWN"
cleaned_npis.append(npi_str)
cleaned_provider["NPI"] = cleaned_npis if cleaned_npis else ["UNKNOWN"]
# ---- Clean NAME (list) ---- # Clean Name - remove extra spaces and ensure it's not empty
cleaned_names = [] if "NAME" in provider and provider["NAME"]:
if isinstance(provider.get("NAME"), list): cleaned_provider["NAME"] = (
for name in provider["NAME"]: " ".join(provider["NAME"].split()).strip().rstrip(".")
if not name: )
continue # If name is empty after cleaning, set to "UNKNOWN"
name_str = " ".join(str(name).split()).strip().rstrip(".") if not cleaned_provider["NAME"]:
if name_str: cleaned_provider["NAME"] = "UNKNOWN"
cleaned_names.append(name_str) else:
cleaned_provider["NAME"] = cleaned_names if cleaned_names else ["UNKNOWN"] cleaned_provider["NAME"] = "UNKNOWN"
cleaned_providers.append(cleaned_provider) cleaned_providers.append(cleaned_provider)
return cleaned_providers return cleaned_providers
@@ -220,6 +220,7 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
Returns: Returns:
dict: Updated one_to_one_results with merged values from provider_info. dict: Updated one_to_one_results with merged values from provider_info.
""" """
# Extract group and non-group providers # Extract group and non-group providers
group_tins = set() group_tins = set()
group_npis = set() group_npis = set()
@@ -228,72 +229,48 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
other_npis = [] other_npis = []
other_names = [] other_names = []
provider_name = one_to_one_results.get("PROVIDER_NAME") # Can be List or string provider_name = one_to_one_results.get("PROVIDER_NAME")
# Normalize provider_name to list: if string, convert to list to avoid character splitting provider_info = one_to_one_results["PROV_INFO_JSON"]
if isinstance(provider_name, str): provider_list = (
provider_name = [provider_name]
elif not isinstance(provider_name, list):
provider_name = []
provider_info = one_to_one_results[
"PROV_INFO_JSON"
] # Returns string representation of list
prov_info_json_list = (
json.loads(provider_info) if isinstance(provider_info, str) else provider_info json.loads(provider_info) if isinstance(provider_info, str) else provider_info
) )
# Track if we found any name matches # Track if we found any name matches
found_match = False found_match = False
for provider in prov_info_json_list: for provider in provider_list:
# Ensure list format (FORMAT FIX ONLY)
tins = provider.get("TIN") or []
npis = provider.get("NPI") or []
names = provider.get("NAME") or []
# Normalize to lists: if string, convert to list to avoid character splitting
if isinstance(tins, str):
tins = [tins]
elif not isinstance(tins, list):
tins = []
if isinstance(npis, str):
npis = [npis]
elif not isinstance(npis, list):
npis = []
if isinstance(names, str):
names = [names]
elif not isinstance(names, list):
names = []
name_matches = prompt_calls.provider_name_match_check( name_matches = prompt_calls.provider_name_match_check(
names, provider_name, filename provider_name, provider.get("NAME", "UNKNOWN"), filename
) # Returns Boolean )
if name_matches: if name_matches:
found_match = True found_match = True
provider["IS_GROUP"] = "Y" provider["IS_GROUP"] = "Y"
for tin in tins: if provider.get("TIN"):
group_tins.add(tin) group_tins.add(provider["TIN"])
for npi in npis: if provider.get("NPI"):
group_npis.add(npi) group_npis.add(provider["NPI"])
for name in names: if provider.get("NAME"):
group_names.add(name) group_names.add(provider["NAME"])
else: else:
provider["IS_GROUP"] = "N" provider["IS_GROUP"] = "N"
other_tins.extend(tins) if provider.get("TIN"):
other_npis.extend(npis) other_tins.append(provider["TIN"])
other_names.extend(names) if provider.get("NPI"):
other_npis.append(provider["NPI"])
if provider.get("NAME"):
other_names.append(provider["NAME"])
# If provider_name is not null and no match was found, add a new record # If provider_name is not null and no match was found, add a new record
if provider_name and not found_match: if provider_name and not found_match:
new_provider = { new_provider = {
"NAME": provider_name, "NAME": provider_name,
"TIN": [], "TIN": "UNKNOWN",
"NPI": [], "NPI": "UNKNOWN",
"IS_GROUP": "Y", "IS_GROUP": "Y",
} }
prov_info_json_list.append(new_provider) provider_list.append(new_provider)
for name in provider_name: group_names.add(provider_name)
group_names.add(name)
# De-Unknown GROUP fields # De-Unknown GROUP fields
group_tins = [v for v in group_tins if v != "UNKNOWN"] group_tins = [v for v in group_tins if v != "UNKNOWN"]
@@ -301,42 +278,50 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
group_names = [v for v in group_names if v != "UNKNOWN"] group_names = [v for v in group_names if v != "UNKNOWN"]
# Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other # Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other
group_tins = list(dict.fromkeys(group_tins)) group_tins = list(dict.fromkeys("|".join(group_tins).split("|")))
other_tins = list(dict.fromkeys(other_tins)) other_tins = list(dict.fromkeys("|".join(other_tins).split("|")))
other_tins = [v for v in other_tins if v not in group_tins and v != "UNKNOWN"] other_tins = [
tin for tin in other_tins if tin not in group_tins and tin != "UNKNOWN"
] # remove any group TINs from other TINs
group_npis = list(dict.fromkeys(group_npis)) group_npis = list(dict.fromkeys("|".join(group_npis).split("|")))
other_npis = list(dict.fromkeys(other_npis)) other_npis = list(dict.fromkeys("|".join(other_npis).split("|")))
other_npis = [v for v in other_npis if v not in group_npis and v != "UNKNOWN"] other_npis = [
npi for npi in other_npis if npi not in group_npis and npi != "UNKNOWN"
] # remove any group NPIs from other NPIs
group_names = list(dict.fromkeys(group_names)) group_names = list(dict.fromkeys("|".join(group_names).split("|")))
other_names = list(dict.fromkeys(other_names)) other_names = list(dict.fromkeys("|".join(other_names).split("|")))
other_names = [v for v in other_names if v not in group_names and v != "UNKNOWN"] other_names = [
name for name in other_names if name not in group_names and name != "UNKNOWN"
] # remove any group Names from other Names
# Filter out records with NO_IDENTIFIERS_FOUND # Filter out records with NO_IDENTIFIERS_FOUND
prov_info_json_list = [ provider_list = [
p p for p in provider_list if p.get("NAME") != "NO_IDENTIFIERS_FOUND"
for p in prov_info_json_list
if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME") or [])
] ]
# Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values # Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values
prov_info_json_list = deduplicate_providers_by_name(prov_info_json_list) provider_list = deduplicate_providers_by_name(provider_list)
# Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags # Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags
one_to_one_results["PROV_INFO_JSON"] = json.dumps(prov_info_json_list) one_to_one_results["PROV_INFO_JSON"] = json.dumps(provider_list)
one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join( one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join(
[json.dumps(provider) for provider in prov_info_json_list] [json.dumps(provider) for provider in provider_list]
) )
# Merge group provider information into one_to_one_results # Merge group provider information into one_to_one_results
one_to_one_results["PROV_GROUP_TIN"] = group_tins if group_tins else [] one_to_one_results["PROV_GROUP_TIN"] = "|".join(group_tins) if group_tins else ""
one_to_one_results["PROV_GROUP_NPI"] = group_npis if group_npis else [] one_to_one_results["PROV_GROUP_NPI"] = "|".join(group_npis) if group_npis else ""
one_to_one_results["PROV_GROUP_NAME_FULL"] = group_names if group_names 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 # Merge other provider information into one_to_one_results
one_to_one_results["PROV_OTHER_TIN"] = other_tins if other_tins else [] one_to_one_results["PROV_OTHER_TIN"] = "|".join(other_tins) if other_tins else ""
one_to_one_results["PROV_OTHER_NPI"] = other_npis if other_npis else [] one_to_one_results["PROV_OTHER_NPI"] = "|".join(other_npis) if other_npis else ""
one_to_one_results["PROV_OTHER_NAME_FULL"] = other_names if other_names else [] one_to_one_results["PROV_OTHER_NAME_FULL"] = (
"|".join(other_names) if other_names else ""
)
return one_to_one_results return one_to_one_results
@@ -429,31 +414,30 @@ def deduplicate_providers(
# Deduplicate if all three fields match # Deduplicate if all three fields match
for provider in provider_info: for provider in provider_info:
tins = provider.get("TIN", [])
npis = provider.get("NPI", [])
names = provider.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 # Skip providers where all critical identification fields are unknown
if ( if (
(not tins or tins == ["UNKNOWN"]) (
and (not npis or npis == ["UNKNOWN"]) string_utils.is_empty(provider.get("TIN", ""))
and (not names or names == ["UNKNOWN"]) or provider.get("TIN", "") == "UNKNOWN"
)
and (
string_utils.is_empty(provider.get("NPI", ""))
or provider.get("NPI", "") == "UNKNOWN"
)
and (
string_utils.is_empty(provider.get("NAME", ""))
or provider.get("NAME", "") == "UNKNOWN"
)
): ):
continue continue
key = ( key = (
tuple(tins), provider.get("TIN", ""),
tuple(npis), provider.get("NPI", ""),
tuple(names), provider.get("NAME", ""),
) )
if key not in seen: if key not in seen:
seen.add(key) seen.add(key)
valid_providers.append(provider) valid_providers.append(provider)
return valid_providers return valid_providers
@@ -510,7 +494,7 @@ def get_provider_info(
logging.warning( logging.warning(
f"Warning: Unexpected format in provider info response: {type(providers)}" f"Warning: Unexpected format in provider info response: {type(providers)}"
) )
providers = [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}]
# VALIDATION LAYER - Cross-check with regex findings # VALIDATION LAYER - Cross-check with regex findings
page_text = text_dict[page_num] page_text = text_dict[page_num]
@@ -539,13 +523,19 @@ def get_provider_info(
llm_found_tins = [] llm_found_tins = []
llm_found_npis = [] llm_found_npis = []
for provider in providers: for provider in providers:
for tin in provider["TIN"] or []: if provider.get("TIN") and provider["TIN"] != "UNKNOWN":
if tin and tin != "UNKNOWN": # Normalize for comparison (remove hyphens/dots)
llm_found_tins.append(tin.replace("-", "").replace(".", "")) raw_tins = [tin.strip() for tin in provider["TIN"].split("|")]
normalized_tins = [
for npi in provider["NPI"] or []: tin.replace("-", "").replace(".", "") for tin in raw_tins
if npi and npi != "UNKNOWN": ]
llm_found_npis.append(npi.replace("-", "").replace(".", "")) llm_found_tins.extend(normalized_tins)
if provider.get("NPI") and provider["NPI"] != "UNKNOWN":
raw_npis = [npi.strip() for npi in provider["NPI"].split("|")]
normalized_npis = [
npi.replace("-", "").replace(".", "") for npi in raw_npis
]
llm_found_npis.extend(normalized_npis)
# Note: regex_tins and regex_npis are already normalized in get_all_matches_with_ocr # Note: regex_tins and regex_npis are already normalized in get_all_matches_with_ocr
missed_tins = [tin for tin in regex_tins if tin not in llm_found_tins] missed_tins = [tin for tin in regex_tins if tin not in llm_found_tins]
@@ -562,7 +552,7 @@ def get_provider_info(
logging.error(f"Error parsing JSON response from LLM: {str(e)}") logging.error(f"Error parsing JSON response from LLM: {str(e)}")
logging.error(f"Raw response: {llm_answer_raw}") logging.error(f"Raw response: {llm_answer_raw}")
# Return a default value # Return a default value
return [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}]
def run_provider_info_fields( def run_provider_info_fields(
@@ -652,9 +642,9 @@ def run_provider_info_fields(
if not relevant_pages: if not relevant_pages:
deduplicated_provider_info = [ deduplicated_provider_info = [
{ {
"TIN": ["UNKNOWN"], "TIN": "UNKNOWN",
"NPI": ["UNKNOWN"], "NPI": "UNKNOWN",
"NAME": ["NO_IDENTIFIERS_FOUND"], "NAME": "NO_IDENTIFIERS_FOUND",
} }
] ]
else: else:
@@ -704,32 +694,12 @@ def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]:
seen_names = set() seen_names = set()
deduplicated_list = [] deduplicated_list = []
# Sort so providers with actual TIN/NPI values come first # Sort so providers with actual TIN/NPI values come first
sorted_providers = sorted( for provider in sorted(
provider_list, provider_list,
key=lambda p: ( key=lambda p: (p.get("TIN") == "UNKNOWN", p.get("NPI") == "UNKNOWN"),
not p.get("TIN") or p.get("TIN") == ["UNKNOWN"], ):
not p.get("NPI") or p.get("NPI") == ["UNKNOWN"], name = provider.get("NAME")
), if name not in seen_names:
)
for provider in sorted_providers:
names = provider.get("NAME", [])
# Skip invalid NAME values
if not isinstance(names, list) or not names or names == ["UNKNOWN"]:
continue
# Check if any name was already seen
is_duplicate = False
for name in names:
if name in seen_names:
is_duplicate = True
break
if is_duplicate:
continue
# Mark all names as seen
for name in names:
seen_names.add(name) seen_names.add(name)
deduplicated_list.append(provider) deduplicated_list.append(provider)
return deduplicated_list return deduplicated_list