Merged in bugfix/one_to_one_and_prov_info (pull request #808)
Bugfix/one to one and prov info * Add post-processing logic to update PROV_INFO_JSON if needed; Updated prompt to verify the number of digitis in TIN and NPI * Added update_prov_info_with_group_name to postprocessing_funcs.py * Updated FIRST_PAGE_PROVIDER_NAME_TEMPLATE to ask LLM to explain resoning before answering * Updated logic to use LLM to check if provider name matches * Use only Hybrid Smart Chunking to get provider group name * Cleaned up code * Removed commented out code * Updated unit test * Merged main into bugfix/one_to_one_and_prov_info * Edited prompt to distinguish provider names and to enclose answer in pipes * Updated test_tin_npi_funcs Approved-by: Katon Minhas
This commit is contained in:
Sha Brown
@@ -149,6 +149,8 @@ def run_one_to_one_prompts(
|
||||
elif key not in one_to_one_results:
|
||||
# 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, filename)
|
||||
|
||||
################## ADD AD FIELDS ################
|
||||
one_to_one_results = one_to_one_funcs.get_aarete_derived_dates(one_to_one_results, text_dict, filename)
|
||||
|
||||
@@ -250,8 +250,8 @@ def run_full_context_fields(
|
||||
)
|
||||
|
||||
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"], "IS_GROUP": "Y"}])
|
||||
full_context_answers_dict["PROV_INFO_JSON_FORMATTED"] = json.dumps([{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": full_context_answers_dict["PROV_GROUP_NAME_FULL"], "IS_GROUP": "Y"}])
|
||||
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:
|
||||
|
||||
@@ -50,8 +50,6 @@ def postprocess(df, constants: Constants):
|
||||
if "CPT" in col:
|
||||
df[col] = df[col].apply(postprocessing_funcs.normalize_cpt_fields)
|
||||
|
||||
# Replace PROV_GROUP_NAME_FULL with PROVIDER_NAME extracted from Hybrid Smart Chunking when not empty/N/A
|
||||
df = postprocessing_funcs.replace_provider_name_from_hybrid_smart_chunking(df)
|
||||
|
||||
# Deduplicate provider fields - remove GROUP values from OTHER fields and deduplicate OTHER lists
|
||||
df = postprocessing_funcs.deduplicate_provider_columns(df)
|
||||
|
||||
@@ -4,7 +4,7 @@ import logging
|
||||
import os
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
import json
|
||||
import pandas as pd
|
||||
import src.prompts.prompt_templates as prompt_templates
|
||||
import src.utils.llm_utils as llm_utils
|
||||
|
||||
@@ -776,54 +776,41 @@ def prompt_lesser_of_check(service_term: str, reimb_term: str, filename: str):
|
||||
return True # Default to True - prefer extra rows to missing
|
||||
|
||||
|
||||
def extract_provider_name_from_first_page(text_dict: dict, filename: str) -> str:
|
||||
"""
|
||||
Extracts the provider group name from the first page of the contract using LLM.
|
||||
def provider_name_match_check(
|
||||
provider_name: str,
|
||||
hybrid_smart_chunking_provider_group_name: str,
|
||||
filename: str
|
||||
) -> bool:
|
||||
"""Check if the provider name from npi/tin page match provider group name"""
|
||||
if string_utils.is_empty(provider_name) or string_utils.is_empty(hybrid_smart_chunking_provider_group_name):
|
||||
return False
|
||||
|
||||
This is an additional check to identify provider group names that may be present
|
||||
on the first page but not captured through other extraction methods.
|
||||
|
||||
Args:
|
||||
text_dict (dict): Dictionary containing the contract text with page numbers as keys.
|
||||
filename (str): The name of the file being processed.
|
||||
|
||||
Returns:
|
||||
str: The extracted provider name or "N/A" if not found.
|
||||
"""
|
||||
# Get the first page
|
||||
pages = list(text_dict.keys())
|
||||
pages.sort(key=string_utils.page_key_sort)
|
||||
|
||||
if not pages:
|
||||
logging.warning(f"No pages found in text_dict for {filename}")
|
||||
return "N/A"
|
||||
|
||||
first_page = pages[0]
|
||||
first_page_text = text_dict[first_page]
|
||||
if hybrid_smart_chunking_provider_group_name == "N/A":
|
||||
return False
|
||||
|
||||
# Create prompt for LLM
|
||||
FIRST_PAGE_PROVIDER_NAME_PROMPT = prompt_templates.FIRST_PAGE_PROVIDER_NAME_TEMPLATE(
|
||||
first_page_text=first_page_text
|
||||
CHECK_PROVIDER_NAME_MATCH_PROMPT = prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
|
||||
provider_name=provider_name,
|
||||
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name
|
||||
)
|
||||
|
||||
# Get response from LLM
|
||||
response = llm_utils.invoke_claude(FIRST_PAGE_PROVIDER_NAME_PROMPT, "legacy_sonnet", filename)
|
||||
|
||||
try:
|
||||
# Extract the provider name from delimiters
|
||||
provider_name = string_utils.extract_text_from_delimiters(
|
||||
response, Delimiter.PIPE
|
||||
).strip()
|
||||
# Adjust this based on your LLM client
|
||||
response = llm_utils.invoke_claude(CHECK_PROVIDER_NAME_MATCH_PROMPT, "sonnet_latest", filename)
|
||||
|
||||
# Validate the response
|
||||
if provider_name and not string_utils.is_empty(provider_name):
|
||||
logging.info(f"First page provider name extracted for {filename}: {provider_name}")
|
||||
return provider_name
|
||||
# Extract last character (should be Y or N)
|
||||
response = string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
|
||||
|
||||
if response=='Y':
|
||||
logging.debug(f"[is_provider_name_match_llm] ✓ LLM matched: '{provider_name}' ≈ '{hybrid_smart_chunking_provider_group_name}'")
|
||||
return True
|
||||
elif response=='N':
|
||||
logging.debug(f"[is_provider_name_match_llm] ✗ LLM no match: '{provider_name}' ≠ '{hybrid_smart_chunking_provider_group_name}'")
|
||||
return False
|
||||
else:
|
||||
logging.debug(f"No valid provider name found on first page for {filename}")
|
||||
return "N/A"
|
||||
logging.warning(f"[is_provider_name_match_llm] Unexpected response: {response}, defaulting to False")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error extracting provider name from first page: {str(e)}")
|
||||
logging.error(f"Raw response: {response}")
|
||||
return "N/A"
|
||||
logging.error(f"[is_provider_name_match_llm] Error calling LLM: {e}, defaulting to False")
|
||||
return False
|
||||
@@ -149,15 +149,14 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
|
||||
|
||||
Args:
|
||||
provider_info (list[dict]): A list of provider information dictionaries. Each
|
||||
dictionary should be keyed by TIN, NPI, NAME, IS_GROUP.
|
||||
dictionary should be keyed by TIN, NPI, NAME
|
||||
|
||||
Returns:
|
||||
list: Cleaned provider information
|
||||
"""
|
||||
cleaned_providers = []
|
||||
for provider in provider_info:
|
||||
# Create a new provider dict to hold the cleaned data
|
||||
cleaned_provider = {"IS_GROUP": provider["IS_GROUP"]}
|
||||
cleaned_provider = {}
|
||||
|
||||
# Clean TIN - remove hyphens and dots, ensure it's 9 digits
|
||||
if "TIN" in provider and provider["TIN"]:
|
||||
@@ -190,7 +189,7 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
|
||||
return cleaned_providers
|
||||
|
||||
|
||||
def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) -> dict:
|
||||
def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename):
|
||||
"""
|
||||
Merges provider_info into one_to_one_results.
|
||||
|
||||
@@ -210,9 +209,20 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) ->
|
||||
other_tins = []
|
||||
other_npis = []
|
||||
other_names = []
|
||||
|
||||
provider_name = one_to_one_results.get('PROVIDER_NAME')
|
||||
provider_info = one_to_one_results['PROV_INFO_JSON']
|
||||
provider_list = json.loads(provider_info) if isinstance(provider_info, str) else provider_info
|
||||
|
||||
# Track if we found any name matches
|
||||
found_match = False
|
||||
|
||||
for provider in provider_list:
|
||||
name_matches = prompt_calls.provider_name_match_check(provider_name, provider.get('NAME', 'UNKNOWN'), filename)
|
||||
|
||||
for provider in provider_info:
|
||||
if provider.get("IS_GROUP", False) == "Y":
|
||||
if name_matches:
|
||||
found_match = True
|
||||
provider['IS_GROUP'] = 'Y'
|
||||
if provider.get("TIN"):
|
||||
group_tins.add(provider["TIN"])
|
||||
if provider.get("NPI"):
|
||||
@@ -220,6 +230,7 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) ->
|
||||
if provider.get("NAME"):
|
||||
group_names.add(provider["NAME"])
|
||||
else:
|
||||
provider['IS_GROUP'] = 'N'
|
||||
if provider.get("TIN"):
|
||||
other_tins.append(provider["TIN"])
|
||||
if provider.get("NPI"):
|
||||
@@ -227,6 +238,17 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) ->
|
||||
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 and not found_match:
|
||||
new_provider = {
|
||||
'NAME': provider_name,
|
||||
'TIN': 'UNKNOWN',
|
||||
'NPI': 'UNKNOWN',
|
||||
'IS_GROUP': 'Y'
|
||||
}
|
||||
provider_list.append(new_provider)
|
||||
group_names.add(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"]
|
||||
@@ -251,6 +273,12 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) ->
|
||||
name for name in other_names if name not in group_names and name != "UNKNOWN"
|
||||
] # remove any group Names from other Names
|
||||
|
||||
# Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags
|
||||
one_to_one_results["PROV_INFO_JSON"] = json.dumps(provider_list)
|
||||
one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join(
|
||||
[json.dumps(provider) for provider in provider_list]
|
||||
)
|
||||
|
||||
# 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 ""
|
||||
@@ -335,104 +363,6 @@ def reimbursement_tin_npi(
|
||||
return answer_dict, reimbursement_level_fields
|
||||
|
||||
|
||||
def identify_group_provider(
|
||||
text_dict: dict, providers: list[dict], filename: str
|
||||
) -> Optional[dict]:
|
||||
"""Identifies the main contracting group provider from a list of providers.
|
||||
Focuses on first/last parts of the contract, where group identification is typically found
|
||||
|
||||
Args:
|
||||
text_dict (dict): A dictionary containing text data, organized by pages
|
||||
providers (list[dict]): A list of deduplicated provider information dictionaries
|
||||
filename (str): The name of the file being processed
|
||||
|
||||
Returns:
|
||||
Optional[dict]: The provider that should be marked as IS_GROUP=true,
|
||||
or None if no group provider is found
|
||||
"""
|
||||
# Extract first 3 pages and signature pages
|
||||
# Sort pages to ensure consistent order
|
||||
# Pages come in as strings, so convert to int for sorting
|
||||
pages = list(text_dict.keys())
|
||||
pages.sort(key=string_utils.page_key_sort)
|
||||
|
||||
# I want to get the first 3 pages and the signature pages. But if they overlap, I want to avoid duplicates
|
||||
first_pages = pages[:3]
|
||||
signature_pages = string_utils.extract_signature_page(
|
||||
text_dict, filename
|
||||
).keys() # Use textract "This page has X signature" as a marker
|
||||
|
||||
# Extract text from the first 3 pages and the signature pages
|
||||
first_section = "\n".join(
|
||||
[text_dict[page] for page in first_pages if page not in signature_pages]
|
||||
) # Avoid duplicates
|
||||
signature_section = "\n".join(
|
||||
[text_dict[page] for page in signature_pages]
|
||||
) # Use textract "This page has X signature" as a marker
|
||||
|
||||
# Create focused sections for group identification
|
||||
key_sections = first_section + "\n\n" + signature_section
|
||||
|
||||
# Format provider list for inclusion in prompt
|
||||
provider_list = json.dumps(providers, indent=2)
|
||||
|
||||
# Construct prompt
|
||||
prompt = prompt_templates.GROUP_TIN_NPI_TEMPLATE(
|
||||
key_sections=key_sections, provider_list=provider_list
|
||||
)
|
||||
|
||||
# Get response and parse
|
||||
response = llm_utils.invoke_claude(prompt, "legacy_sonnet", filename)
|
||||
try:
|
||||
group_info = json.loads(
|
||||
string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
|
||||
)
|
||||
|
||||
# Match with an existing provider
|
||||
for provider in providers:
|
||||
if provider.get("TIN") == group_info.get("TIN") and provider.get(
|
||||
"NPI"
|
||||
) == group_info.get("NPI"):
|
||||
# Always use the group-provided name when available
|
||||
# This is a fallback in case the group provider name is not in the original provider list
|
||||
if group_info.get("NAME") is not None and group_info.get(
|
||||
"NAME"
|
||||
) not in ["", "null", "None", "N/A", "UNKNOWN"]:
|
||||
provider["NAME"] = group_info.get("NAME")
|
||||
# Mark as group
|
||||
provider["IS_GROUP"] = True
|
||||
return provider
|
||||
|
||||
# If we got here, no match was found, but we have group info from the LLM
|
||||
# Add the group provider to the list if it has valid information
|
||||
if (
|
||||
(group_info.get("TIN", "") not in ["", "null", "None", "N/A", "UNKNOWN"])
|
||||
or (group_info.get("NPI", "") not in ["", "null", "None", "N/A", "UNKNOWN"])
|
||||
or (
|
||||
group_info.get("NAME", "") not in ["", "null", "None", "N/A", "UNKNOWN"]
|
||||
)
|
||||
):
|
||||
# Create a new provider with the group info
|
||||
new_group_provider = {
|
||||
"TIN": group_info.get("TIN", "UNKNOWN"),
|
||||
"NPI": group_info.get("NPI", "UNKNOWN"),
|
||||
"NAME": group_info.get("NAME", "UNKNOWN"),
|
||||
"IS_GROUP": True,
|
||||
}
|
||||
# Add to the provider list
|
||||
providers.append(new_group_provider)
|
||||
return new_group_provider
|
||||
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"Error parsing group provider response: {response}. Exception: {e}"
|
||||
)
|
||||
return None
|
||||
|
||||
# Fallback: if no match found, return None
|
||||
return None
|
||||
|
||||
|
||||
def deduplicate_providers(
|
||||
provider_info: list[dict],
|
||||
) -> list[dict]: # TODO: add unit test
|
||||
@@ -479,88 +409,10 @@ def deduplicate_providers(
|
||||
valid_providers.append(provider)
|
||||
return valid_providers
|
||||
|
||||
|
||||
def is_provider_name_match(
|
||||
provider_name: str,
|
||||
first_page_provider_group_name: str,
|
||||
threshold: float = 0.75
|
||||
) -> bool:
|
||||
"""Check if provider_name matches first_page_provider_group_name (handles d/b/a)."""
|
||||
if string_utils.is_empty(provider_name) or string_utils.is_empty(first_page_provider_group_name):
|
||||
return False
|
||||
|
||||
if first_page_provider_group_name == "N/A":
|
||||
return False
|
||||
|
||||
provider_clean = provider_name.lower().strip()
|
||||
expected_clean = first_page_provider_group_name.lower().strip()
|
||||
|
||||
# Strategy 1: Exact match
|
||||
if provider_clean == expected_clean:
|
||||
logging.debug(f"[is_provider_name_match] ✓ Exact match")
|
||||
return True
|
||||
|
||||
# Strategy 2: Fuzzy similarity
|
||||
similarity = SequenceMatcher(None, provider_clean, expected_clean).ratio()
|
||||
if similarity >= threshold:
|
||||
logging.debug(f"[is_provider_name_match] ✓ Fuzzy match ({similarity:.2f})")
|
||||
return True
|
||||
|
||||
# Strategy 3: Contains match
|
||||
if expected_clean in provider_clean or provider_clean in expected_clean:
|
||||
logging.debug(f"[is_provider_name_match] ✓ Contains match")
|
||||
return True
|
||||
|
||||
# Strategy 4: Word overlap
|
||||
common_words = {"llc", "inc", "corp", "ltd", "the", "and", "dba", "d/b/a", "pa", "pc"}
|
||||
provider_words = set([w for w in provider_clean.split() if len(w) > 3 and w not in common_words])
|
||||
expected_words = set([w for w in expected_clean.split() if len(w) > 3 and w not in common_words])
|
||||
|
||||
if len(provider_words) > 0 and len(expected_words) > 0:
|
||||
overlap = len(provider_words & expected_words)
|
||||
overlap_ratio = overlap / min(len(provider_words), len(expected_words))
|
||||
|
||||
if overlap_ratio >= 0.6:
|
||||
logging.debug(f"[is_provider_name_match] ✓ Word overlap ({overlap_ratio:.2f})")
|
||||
return True
|
||||
|
||||
# Strategy 5: D/B/A parsing
|
||||
if "d/b/a" in expected_clean or "dba" in expected_clean:
|
||||
dba_pattern = r'd\.?/?b\.?/?a\.?|doing\s+business\s+as'
|
||||
parts = re.split(dba_pattern, expected_clean, flags=re.IGNORECASE)
|
||||
|
||||
if len(parts) >= 2:
|
||||
# Check legal name (before d/b/a)
|
||||
legal_name = parts[0].strip()
|
||||
if provider_clean in legal_name or legal_name in provider_clean:
|
||||
logging.debug(f"[is_provider_name_match] ✓ Legal name match")
|
||||
return True
|
||||
|
||||
# Check d/b/a section
|
||||
dba_section = parts[1].strip()
|
||||
if provider_clean in dba_section or dba_section in provider_clean:
|
||||
logging.debug(f"[is_provider_name_match] ✓ D/B/A section match")
|
||||
return True
|
||||
|
||||
# Word overlap with d/b/a section
|
||||
dba_words = set([w for w in dba_section.split() if len(w) > 3 and w not in common_words])
|
||||
if len(provider_words) > 0 and len(dba_words) > 0:
|
||||
overlap = len(provider_words & dba_words)
|
||||
overlap_ratio = overlap / min(len(provider_words), len(dba_words))
|
||||
|
||||
if overlap_ratio >= 0.6:
|
||||
logging.debug(f"[is_provider_name_match] ✓ D/B/A word overlap ({overlap_ratio:.2f})")
|
||||
return True
|
||||
|
||||
logging.debug(f"[is_provider_name_match] ✗ No match")
|
||||
return False
|
||||
|
||||
|
||||
def get_provider_info(
|
||||
text_dict: dict,
|
||||
page_num: str,
|
||||
filename: str,
|
||||
first_page_provider_group_name: str = None
|
||||
filename: str
|
||||
) -> list:
|
||||
"""
|
||||
Extracts provider information from a specified page in a document using a language model.
|
||||
@@ -587,7 +439,7 @@ def get_provider_info(
|
||||
chunk, provider_fields.print_prompt_dict()
|
||||
)
|
||||
claude_answer_raw = llm_utils.invoke_claude(
|
||||
prompt, "legacy_sonnet", filename, max_tokens=8192, cache=True, instruction=instruction
|
||||
prompt, "sonnet_latest", filename, max_tokens=8192, cache=True, instruction=instruction
|
||||
) # Sometimes rosters can have 50+ provider
|
||||
|
||||
logging.debug(
|
||||
@@ -611,8 +463,7 @@ def get_provider_info(
|
||||
{
|
||||
"TIN": "UNKNOWN",
|
||||
"NPI": "UNKNOWN",
|
||||
"NAME": "PARSING_ERROR",
|
||||
"ON_SIGNATURE_PAGE": "N",
|
||||
"NAME": "PARSING_ERROR"
|
||||
}
|
||||
]
|
||||
|
||||
@@ -666,21 +517,6 @@ def get_provider_info(
|
||||
f"LLM missed identifiers on page {page_num} of {filename}. Missed TINs: {missed_tins}, Missed NPIs: {missed_npis}"
|
||||
)
|
||||
|
||||
for provider in providers:
|
||||
provider_name = provider.get("NAME", "")
|
||||
|
||||
name_matches = False
|
||||
if first_page_provider_group_name and first_page_provider_group_name != "N/A" and not string_utils.is_empty(provider_name):
|
||||
name_matches = is_provider_name_match(provider_name, first_page_provider_group_name)
|
||||
|
||||
if name_matches:
|
||||
provider["IS_GROUP"] = "Y"
|
||||
else:
|
||||
provider["IS_GROUP"] = "N"
|
||||
|
||||
if "ON_SIGNATURE_PAGE" in provider:
|
||||
del provider["ON_SIGNATURE_PAGE"]
|
||||
|
||||
return providers
|
||||
|
||||
except ValueError as e: # Handle JSON parsing error
|
||||
@@ -691,8 +527,7 @@ def get_provider_info(
|
||||
{
|
||||
"TIN": "UNKNOWN",
|
||||
"NPI": "UNKNOWN",
|
||||
"NAME": "PARSING_ERROR",
|
||||
"IS_GROUP": "N",
|
||||
"NAME": "PARSING_ERROR"
|
||||
}
|
||||
]
|
||||
|
||||
@@ -738,10 +573,6 @@ def run_provider_info_fields(
|
||||
logging.debug(f"Extracted TINs: {all_tins}")
|
||||
logging.debug(f"Extracted NPIs: {all_npis}")
|
||||
|
||||
# Extract expected group name from first page
|
||||
first_page_provider_group_name = prompt_calls.extract_provider_name_from_first_page(text_dict, filename)
|
||||
logging.info(f"[run_provider_info_fields] Expected group name: '{first_page_provider_group_name}'")
|
||||
|
||||
# Log OCR corrections for monitoring
|
||||
exact_tins = [tin for tin, quality in tin_matches if quality == "EXACT"]
|
||||
ocr_corrected_tins = [
|
||||
@@ -795,14 +626,13 @@ def run_provider_info_fields(
|
||||
"TIN": "UNKNOWN",
|
||||
"NPI": "UNKNOWN",
|
||||
"NAME": "NO_IDENTIFIERS_FOUND",
|
||||
"IS_GROUP": "N",
|
||||
}
|
||||
]
|
||||
else:
|
||||
# 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, first_page_provider_group_name)
|
||||
page_provider_info = get_provider_info(text_dict, page_num, filename)
|
||||
combined_provider_info.extend(page_provider_info)
|
||||
|
||||
# Clean and standardize results
|
||||
@@ -822,28 +652,6 @@ def run_provider_info_fields(
|
||||
one_to_one_results["NPI_EXTRACTION_QUALITY"] = (
|
||||
f"{len(exact_npis)} exact, {len(ocr_corrected_npis)} OCR-corrected"
|
||||
)
|
||||
one_to_one_results = merge_provider_info(
|
||||
one_to_one_results, deduplicated_provider_info
|
||||
) # Add Group TIN
|
||||
|
||||
# Override PROV_GROUP_NAME_FULL with first page extraction if found
|
||||
if first_page_provider_group_name != "N/A":
|
||||
logging.info(f"Overriding PROV_GROUP_NAME_FULL with first page extraction: {first_page_provider_group_name}")
|
||||
one_to_one_results["PROV_GROUP_NAME_FULL"] = first_page_provider_group_name
|
||||
|
||||
# Fallback: What if the TIN/NPI regexes don't find anything? Add PROV_GROUP_NAME_FULL to full context
|
||||
if (
|
||||
string_utils.is_empty(one_to_one_results.get("PROV_OTHER_NAME_FULL")) and
|
||||
string_utils.is_empty(one_to_one_results.get("PROV_GROUP_NAME_FULL"))
|
||||
):
|
||||
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 full name of the group provider name associated with this contract? provider name is usually present in the first three pages or in the signature page. 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"
|
||||
|
||||
# Add FILENAME_TIN
|
||||
filename_tin = get_all_matches(filename, regex_patterns.FILE_NAME_TIN_PATTERN)
|
||||
|
||||
@@ -261,7 +261,7 @@
|
||||
"field_name": "PROVIDER_NAME",
|
||||
"relationship": "one_to_one",
|
||||
"field_type": "smart_chunked",
|
||||
"prompt": "Extract the PROVIDER organization name from the retrieved contract text.\n\nThe provider is the healthcare organization entering into the agreement (NOT the payer/insurance company like Highmark, Aetna, UnitedHealthcare, Blue Cross Blue Shield, etc.).\n\nGUIDELINES:\n1. Look for the provider name in phrases like:\n - 'This Agreement is between [PAYER] and [PROVIDER]'\n - 'Agreement between [PAYER] and [PROVIDER]'\n - 'by and between [PAYER] and [PROVIDER]'\n - '[PROVIDER] (hereinafter referred to as \"Provider\")'\n - 'Provider: [PROVIDER]'\n2. The provider name often appears in the opening paragraph or preamble of the contract.\n3. Return the EXACT provider name as it appears in the document, including legal suffixes (LLC, Inc., P.C., etc.).\n4. If the provider name includes 'd/b/a' (doing business as), return only the legal entity name before 'd/b/a'.\n - Example: 'Middletown Universal, LLC d/b/a American Kidney Care' → return 'Middletown Universal, LLC'\n5. Do NOT return:\n - The payer/insurance company name\n - Individual physician names (unless that is the practice name)\n - Department names only\n - Addresses or other metadata\n6. If multiple potential provider names appear, return the one from the primary agreement clause (usually in the preamble).\n7. If the provider name cannot be determined from the retrieved text, return 'N/A'.\n\nReturn ONLY the provider name, nothing else.",
|
||||
"prompt": "Extract the PROVIDER organization name(s) from the retrieved contract text.\n\nThe provider is the healthcare organization entering into the agreement (NOT the payer/insurance company like Highmark, Aetna, UnitedHealthcare, Blue Cross Blue Shield, etc.).\n\nGUIDELINES:\n1. Look for the provider name in phrases like:\n - 'This Agreement is between [PAYER] and [PROVIDER]'\n - 'Agreement between [PAYER] and [PROVIDER]'\n - 'by and between [PAYER] and [PROVIDER]'\n - '[PROVIDER] (hereinafter referred to as \"Provider\")'\n - 'Provider: [PROVIDER]'\n2. The provider name often appears in the opening paragraph or preamble of the contract.\n3. Return the EXACT provider name as it appears in the document, including legal suffixes (LLC, Inc., P.C., etc.).\n4. If the provider name includes 'd/b/a' (doing business as), return the FULL name including the d/b/a portion.\n5. If MULTIPLE provider organization names appear in the contract (e.g., multiple entities entering the agreement), return ALL of them separated by ' | '.\n - Example: 'Provider Group A, LLC | Provider Group B, P.C. | Provider Group C d/b/a City Medical'\n6. Do NOT return:\n - The payer/insurance company name\n - Individual physician names (unless that is the practice name)\n - Department names only\n - Addresses or other metadata\n7. If the provider name cannot be determined from the retrieved text, return 'N/A'.\n\nBriefly explain your reasoning. Return ONLY the provider name(s) separated by ' | ' if multiple, nothing else.",
|
||||
"retrieval_question": "What is the provider organization name? Agreement between payer and provider. This agreement is made between. By and between. Hereinafter referred to as Provider. Provider name in contract preamble. Provider party to the agreement. Healthcare organization entering agreement."
|
||||
},
|
||||
{
|
||||
|
||||
@@ -880,10 +880,13 @@ Briefly explain your reasoning. After your explanation, include the heading "FIN
|
||||
[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.
|
||||
- Once TIN, NPI or NAME are identified, the check if the same provider is on the signature page. If so, set "ON_SIGNATURE_PAGE" to "Y", otherwise "N"
|
||||
- IMPORTANT VALIDATION REQUIREMENTS:
|
||||
* TIN must be exactly 9 digits (no more, no less)
|
||||
* NPI must be exactly 10 digits (no more, no less)
|
||||
* Double-check the digit count for each TIN and NPI before including them in the response
|
||||
- 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", "ON_SIGNATURE_PAGE"
|
||||
- All provider objects must have these exact keys: "TIN", "NPI", "NAME"
|
||||
- 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
|
||||
@@ -895,14 +898,12 @@ FINAL PROVIDER INFO:
|
||||
{{
|
||||
"TIN": "123456789",
|
||||
"NPI": "1234567890",
|
||||
"NAME": "Main Provider Group",
|
||||
"ON_SIGNATURE_PAGE": "Y",
|
||||
"NAME": "Main Provider Group"
|
||||
}},
|
||||
{{
|
||||
"TIN": "987654321",
|
||||
"NPI": null,
|
||||
"NAME": "Dr. John Smith",
|
||||
"ON_SIGNATURE_PAGE": "N",
|
||||
}}
|
||||
]
|
||||
|
||||
@@ -960,18 +961,30 @@ def FULL_CONTEXT_ADDITIONAL_INSTRUCTIONS(allow_na):
|
||||
else:
|
||||
return " Do NOT leave this field N/A. Choose the most appropriate answer based on the context. If there are multiple answers, return them in a pipe (|) separated list."
|
||||
|
||||
def FIRST_PAGE_PROVIDER_NAME_TEMPLATE(first_page_text: str) -> str:
|
||||
return f"""Extract the PROVIDER organization name from this contract first page.
|
||||
|
||||
The provider is the healthcare organization entering into the agreement (NOT the payer/insurance company like Highmark, Aetna, UnitedHealthcare, etc.). Look for patterns like 'This AGREEMENT is made between [PAYER] and [PROVIDER]'. Return the EXACT provider name as it appears in the contract.
|
||||
FIRST PAGE TEXT:
|
||||
{first_page_text}
|
||||
def CHECK_PROVIDER_NAME_MATCH_PROMPT(provider_name: str, hybrid_smart_chunking_provider_group_name: str) -> str:
|
||||
"""Create prompt for LLM-based provider name matching."""
|
||||
return f"""Determine if the provider name matches any of the healthcare organizations listed.
|
||||
|
||||
CRITICAL: Return ONLY the provider name wrapped in || delimiters, with NO explanation, NO preamble, NO additional text.
|
||||
PROVIDER NAME A: {provider_name}
|
||||
PROVIDER NAME B: {hybrid_smart_chunking_provider_group_name}
|
||||
|
||||
If no provider found, return ONLY: ||N/A||
|
||||
Note: PROVIDER NAME B is typically a single name but may contain multiple names separated by ' | '. Determine if PROVIDER NAME A matches PROVIDER NAME B (or ANY name if multiple are present).
|
||||
|
||||
Your response must start with || and end with ||"""
|
||||
MATCHING RULES:
|
||||
1. PROVIDER NAME A must appear in PROVIDER NAME B (either as an exact match or close variation).
|
||||
2. If PROVIDER NAME B contains multiple names (separated by ' | '), check each name individually.
|
||||
3. Legal entity suffixes (LLC, Inc, PA, PC, etc.) can be ignored when matching.
|
||||
4. D/B/A / "doing business as" variations count as matches (e.g., "Lakeview Hospital" matches "Hospital Corporation of Utah d.b.a. Lakeview Hospital").
|
||||
5. Acronyms that clearly represent the same entity count as matches (e.g., "MHS" matches "Memorial Hospital System").
|
||||
6. Common word variations count as matches (e.g., "St. Mark's" matches "Saint Mark's").
|
||||
|
||||
CRITICAL - DO NOT MATCH:
|
||||
- Parent company or holding company names that do NOT appear in PROVIDER NAME B, even if they own the listed entities.
|
||||
- Health system names that are NOT in PROVIDER NAME B, even if the listed hospitals belong to that system.
|
||||
- Corporate umbrella names unless they are explicitly listed in PROVIDER NAME B.
|
||||
|
||||
Briefly explain your reasoning. Then return ONLY 'Y' (match found) or 'N' (no match found). Enclose your final answer in |pipes|"""
|
||||
|
||||
def ONE_TO_ONE_MULTI_FIELD_TEMPLATE(context, questions: dict[str, str]):
|
||||
return f"""The following is a contract (or excerpt thereof). Answer the following questions: {questions}
|
||||
|
||||
@@ -67,41 +67,44 @@ class TestTinNpiFuncs:
|
||||
assert result[1]["NPI"] == "UNKNOWN"
|
||||
assert result[1]["NAME"] == "UNKNOWN"
|
||||
|
||||
def test_merge_provider_info(self):
|
||||
one_to_one_results = {}
|
||||
provider_info = [
|
||||
{
|
||||
"TIN": "123456789",
|
||||
"NPI": "1234567890",
|
||||
"NAME": "Group Provider",
|
||||
"IS_GROUP": "Y",
|
||||
},
|
||||
{
|
||||
"TIN": "987654321",
|
||||
"NPI": "9876543210",
|
||||
"NAME": "Other Provider",
|
||||
"IS_GROUP": "N",
|
||||
},
|
||||
]
|
||||
@patch("src.investment.prompt_calls.provider_name_match_check")
|
||||
def test_merge_provider_info_with_hybrid_smart_chunking(self, mock_name_match_check):
|
||||
"""Test merge_provider_info_with_hybrid_smart_chunking with name matching logic"""
|
||||
# Mock the name matching to return True for "Group Provider"
|
||||
def name_match_side_effect(provider_name, name, filename):
|
||||
return name == "Group Provider"
|
||||
|
||||
mock_name_match_check.side_effect = name_match_side_effect
|
||||
|
||||
result = tin_npi_funcs.merge_provider_info(one_to_one_results, provider_info)
|
||||
one_to_one_results = {
|
||||
"PROVIDER_NAME": "Group Provider",
|
||||
"PROV_INFO_JSON": json.dumps([
|
||||
{
|
||||
"TIN": "123456789",
|
||||
"NPI": "1234567890",
|
||||
"NAME": "Group Provider",
|
||||
},
|
||||
{
|
||||
"TIN": "987654321",
|
||||
"NPI": "9876543210",
|
||||
"NAME": "Other Provider",
|
||||
},
|
||||
])
|
||||
}
|
||||
|
||||
result = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, "test.pdf")
|
||||
|
||||
# Check that IS_GROUP flags were added correctly
|
||||
prov_info = json.loads(result["PROV_INFO_JSON"])
|
||||
assert prov_info[0]["IS_GROUP"] == "Y"
|
||||
assert prov_info[1]["IS_GROUP"] == "N"
|
||||
|
||||
# Check group and other fields
|
||||
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_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"
|
||||
|
||||
@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(
|
||||
@@ -109,9 +112,7 @@ class TestTinNpiFuncs:
|
||||
{
|
||||
"TIN": "123456789",
|
||||
"NPI": "1234567890",
|
||||
"NAME": "Test Provider",
|
||||
"ON_SIGNATURE_PAGE": "N",
|
||||
"IS_GROUP": "Y",
|
||||
"NAME": "Test Provider"
|
||||
|
||||
}
|
||||
]
|
||||
@@ -119,7 +120,6 @@ class TestTinNpiFuncs:
|
||||
|
||||
result = tin_npi_funcs.get_provider_info(sample_text_dict, "1", "test.pdf")
|
||||
assert len(result) == 1
|
||||
assert result[0]["IS_GROUP"] == "N" # Page 1 should not be marked as intro
|
||||
|
||||
@patch("src.utils.llm_utils.invoke_claude")
|
||||
def test_run_provider_info_fields(self, mock_invoke_claude, sample_text_dict):
|
||||
|
||||
Reference in New Issue
Block a user