From ee85884dce00ec5f76309b718ff83f569e46ecbe Mon Sep 17 00:00:00 2001 From: Katon Minhas Date: Wed, 4 Feb 2026 02:23:55 -0500 Subject: [PATCH] Refactor llm call to prompt_calls --- src/pipelines/saas/prompts/prompt_calls.py | 74 +++++- src/pipelines/saas/test.py | 14 ++ .../shared/extraction/tin_npi_funcs.py | 228 +++++++----------- 3 files changed, 169 insertions(+), 147 deletions(-) create mode 100644 src/pipelines/saas/test.py diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index c1e34fc..c8dcb99 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -4,10 +4,8 @@ import src.config as config import src.prompts.prompt_templates as prompt_templates from src.constants.constants import Constants from src.prompts.fieldset import Field, FieldSet -from src.utils import llm_utils, string_utils, timing_utils, rag_utils -from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import ( - format_chunks_for_llm, -) +from src.utils import llm_utils, string_utils +from src.pipelines.shared.extraction import tin_npi_funcs import json @@ -910,3 +908,71 @@ def prompt_split_reimb_dates(date_range: str, filename: str) -> dict: # Extract the effective and termination dates from the LLM output llm_answer_final = _parser(llm_answer_raw) return llm_answer_final + + +def prompt_provider_info( + text_dict: dict, page_num: str, filename: str, payer_name: str +) -> list: + """ + Extracts provider information from a specified page in a document using a language model. + + Args: + 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. + first_page_provider_group_name (str, optional): Expected group name from page 1 for name verification. + + Returns: + 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" + ) + prompt, _parser = prompt_templates.TIN_NPI_TEMPLATE( + chunk, provider_fields.print_prompt_dict(), payer_name + ) + llm_answer_raw = llm_utils.invoke_claude( + prompt, + "sonnet_latest", + filename, + max_tokens=8192, + cache=True, + instruction=prompt_templates.TIN_NPI_TEMPLATE_INSTRUCTION(), + ) # Sometimes rosters can have 50+ provider + + logging.debug( + f"Raw LLM response for provider info on page {page_num}: {llm_answer_raw}" + ) + + # Defensive programming for any json parsing errors + try: + 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 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"]}] + + # VALIDATION LAYER - Cross-check with regex findings + page_text = text_dict[page_num] + tin_npi_funcs.flag_missed_prov_info(page_text, providers, page_num, filename) + + return providers + + except ValueError as e: # Handle JSON parsing error + 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"]}] \ No newline at end of file diff --git a/src/pipelines/saas/test.py b/src/pipelines/saas/test.py new file mode 100644 index 0000000..70f7a50 --- /dev/null +++ b/src/pipelines/saas/test.py @@ -0,0 +1,14 @@ + +import src.pipelines.shared.extraction.tin_npi_funcs as tin_npi_funcs + +prov_info_json = [{'TIN': '971098746', 'NPI': "N/A", 'NAME': 'Midwest Medical Equipment Solutions, Inc.'}] + +print("Before Clean: ", prov_info_json) + +# Clean and standardize results +prov_info_json_clean = tin_npi_funcs.clean_provider_info(prov_info_json) +print("After Clean: ", prov_info_json_clean) + +# Deduplicate providers based on TIN and NPI +prov_info_json_clean = tin_npi_funcs.deduplicate_providers(prov_info_json_clean) +print("After dedup: ", prov_info_json_clean) \ No newline at end of file diff --git a/src/pipelines/shared/extraction/tin_npi_funcs.py b/src/pipelines/shared/extraction/tin_npi_funcs.py index 4e704b4..0dba81f 100644 --- a/src/pipelines/shared/extraction/tin_npi_funcs.py +++ b/src/pipelines/shared/extraction/tin_npi_funcs.py @@ -171,40 +171,41 @@ def clean_provider_info(prov_info_json: list[dict]) -> list[dict]: for provider in prov_info_json: cleaned_provider = {} - # ---- Clean TIN (list, digits only, 9 chars) ---- - cleaned_tins = [] - if isinstance(provider.get("TIN"), list): - for tin in provider["TIN"]: - if not tin: - continue - tin_str = str(tin).replace("-", "").replace(".", "") - if tin_str.isdigit() and len(tin_str) == 9: - cleaned_tins.append(tin_str) - cleaned_provider["TIN"] = cleaned_tins if cleaned_tins else "UNKNOWN" - - # ---- Clean NPI (list, digits only, 10 chars) ---- - cleaned_npis = [] - if isinstance(provider.get("NPI"), list): - for npi in provider["NPI"]: - if not npi: - continue - npi_str = str(npi).replace("-", "").replace(".", "") - if npi_str.isdigit() and len(npi_str) == 10: - cleaned_npis.append(npi_str) - cleaned_provider["NPI"] = cleaned_npis if cleaned_npis else "UNKNOWN" + # ---- 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) ---- - cleaned_names = [] - if isinstance(provider.get("NAME"), list): - for name in provider["NAME"]: - if not name: - continue - name_str = " ".join(str(name).split()).strip().rstrip(".") - if name_str: - cleaned_names.append(name_str) - cleaned_provider["NAME"] = cleaned_names if cleaned_names else "UNKNOWN" + 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 @@ -446,19 +447,19 @@ def deduplicate_providers( # Deduplicate if all three fields match for provider in provider_info: - tins = provider.get("TIN", []) - npis = provider.get("NPI", []) - names = provider.get("NAME", []) + 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 [] + # # 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"]) + (not tins or tins == "UNKNOWN") + and (not npis or npis == "UNKNOWN") + and (not names or names == "UNKNOWN") ): continue key = ( @@ -473,113 +474,48 @@ def deduplicate_providers( return valid_providers +def flag_missed_prov_info(page_text, providers, page_num, filename): + # Get all TINs and NPIs found via regex on this page + regex_tins = [ + tin + for tin, _ in get_all_matches_with_ocr( + page_text, + regex_patterns.TIN_PATTERN, + regex_patterns.TIN_OCR_PATTERN, + identifier_type="TIN", + ) + ] + regex_npis = [ + npi + for npi, _ in get_all_matches_with_ocr( + page_text, + regex_patterns.NPI_PATTERN, + regex_patterns.NPI_OCR_PATTERN, + identifier_type="NPI", + ) + ] -def get_provider_info( - text_dict: dict, page_num: str, filename: str, payer_name: str -) -> list: - """ - Extracts provider information from a specified page in a document using a language model. + # Collect all TINs/NPIs that LLM found + llm_found_tins = [] + llm_found_npis = [] + for provider in providers: + tin = provider["TIN"] + if tin and tin != "UNKNOWN": + llm_found_tins.append(tin.replace("-", "").replace(".", "")) - Args: - 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. - first_page_provider_group_name (str, optional): Expected group name from page 1 for name verification. + npi = provider["NPI"] + if npi and npi != "UNKNOWN": + llm_found_npis.append(npi.replace("-", "").replace(".", "")) - Returns: - list: A list of extracted provider information dictionaries, each containing TIN, NPI, - NAME, and flags for presence on signature or intro pages. + # 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_npis = [npi for npi in regex_npis if npi not in llm_found_npis] + if missed_tins or missed_npis: + # WARN ONLY if there are missed identifiers + logging.warning( + f"LLM missed identifiers on page {page_num} of {filename}. Missed TINs: {missed_tins}, Missed NPIs: {missed_npis}" + ) - 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" - ) - prompt, _parser = prompt_templates.TIN_NPI_TEMPLATE( - chunk, provider_fields.print_prompt_dict(), payer_name - ) - llm_answer_raw = llm_utils.invoke_claude( - prompt, - "sonnet_latest", - filename, - max_tokens=8192, - cache=True, - instruction=prompt_templates.TIN_NPI_TEMPLATE_INSTRUCTION(), - ) # Sometimes rosters can have 50+ provider - - logging.debug( - f"Raw LLM response for provider info on page {page_num}: {llm_answer_raw}" - ) - - # Defensive programming for any json parsing errors - try: - 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 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"]}] - - # VALIDATION LAYER - Cross-check with regex findings - page_text = text_dict[page_num] - - # Get all TINs and NPIs found via regex on this page - regex_tins = [ - tin - for tin, _ in get_all_matches_with_ocr( - page_text, - regex_patterns.TIN_PATTERN, - regex_patterns.TIN_OCR_PATTERN, - identifier_type="TIN", - ) - ] - regex_npis = [ - npi - for npi, _ in get_all_matches_with_ocr( - page_text, - regex_patterns.NPI_PATTERN, - regex_patterns.NPI_OCR_PATTERN, - identifier_type="NPI", - ) - ] - - # Collect all TINs/NPIs that LLM found - llm_found_tins = [] - llm_found_npis = [] - for provider in providers: - for tin in provider["TIN"] or []: - if tin and tin != "UNKNOWN": - llm_found_tins.append(tin.replace("-", "").replace(".", "")) - - for npi in provider["NPI"] or []: - if npi and npi != "UNKNOWN": - llm_found_npis.append(npi.replace("-", "").replace(".", "")) - - # 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_npis = [npi for npi in regex_npis if npi not in llm_found_npis] - if missed_tins or missed_npis: - # WARN ONLY if there are missed identifiers - logging.warning( - f"LLM missed identifiers on page {page_num} of {filename}. Missed TINs: {missed_tins}, Missed NPIs: {missed_npis}" - ) - - return providers - - except ValueError as e: # Handle JSON parsing error - 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"]}] def get_relevant_pages(all_tins, all_npis, text_dict, filename): @@ -630,11 +566,15 @@ 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 @@ -700,10 +640,12 @@ def run_provider_info_fields( """ all_tins, all_npis, tin_matches, npi_matches = get_all_tins_and_npis(contract_text) + print("All Tins: ", all_tins) 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) + print("Relevant Pages: ", relevant_pages) prov_info_json_clean = get_prov_info_json(relevant_pages, text_dict, filename, payer_name)