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

Refactor llm call to prompt_calls

Browse Source
This commit is contained in:
Katon Minhas
2026-02-04 02:23:55 -05:00
parent 2c76b8a92b
commit ee85884dce
3 changed files with 169 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/saas/prompts/prompt_calls.py
+70 -4
View File
@@ -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"]}]
src/pipelines/saas/test.py
+14
View File
@@ -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)
src/pipelines/shared/extraction/tin_npi_funcs.py
+85 -143
View File
@@ -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)