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

Merged in hotfix/revert-standardize-list-formats (pull request #861)

Browse Source 
Revert "Merged in feature/standardize-list-formats (pull request #844)"

* Revert "Merged in feature/standardize-list-formats (pull request #844)"

This reverts commit faf707265d.
This commit is contained in:
Katon Minhas
2026-02-02 18:27:37 +00:00
parent 0efb0e5369
commit 02d10e711e
22 changed files with 679 additions and 810 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+14 -14
View File
@@ -178,7 +178,10 @@ def code_implicit_special(service, filename):
claude_answer_raw = llm_utils.invoke_claude(
prompt_templates.CODE_IMPLICIT_SPECIAL(service), "sonnet_latest", filename
)
claude_answer_final = string_utils.universal_json_load(claude_answer_raw)[0]
claude_answer_final = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
special_case_mapping = {
"Drugs": ["J0000-J9999"],
"Vaccines": [
@@ -348,7 +351,7 @@ def code_implicit_rag(service, implicit_run_dict, filename, constants):
"sonnet_latest",
filename,
)
logging.debug(f"Implicit RAG Claude Response: {claude_answer_raw}")
try:
claude_answer_final = string_utils.universal_json_load(claude_answer_raw)
except Exception as e:
@@ -429,8 +432,9 @@ def code_last_check(service, filename):
claude_answer_raw = llm_utils.invoke_claude(
prompt_templates.CODE_LAST_CHECK(service), "sonnet_latest", filename
)
claude_answer_final = string_utils.universal_json_load(claude_answer_raw)[0]
claude_answer_final = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
return claude_answer_final
@@ -490,16 +494,12 @@ def fill_bill_type(
bill_codes, bill_descs = [], []
for description in claude_answer_final:
if description in BILL_TYPE_REVERSE_MAPPING:
code_val = BILL_TYPE_REVERSE_MAPPING[description]
if isinstance(code_val, list):
bill_codes.extend(code_val)
else:
bill_codes.append(str(code_val))
bill_codes.append(BILL_TYPE_REVERSE_MAPPING[description])
bill_descs.append(description)
if bill_codes:
answer_dict["BILL_TYPE_CD"] = sorted(list(set(bill_codes)))
answer_dict["BILL_TYPE_CD_DESC"] = bill_descs
answer_dict["BILL_TYPE_CD"] = "|".join(bill_codes)
answer_dict["BILL_TYPE_CD_DESC"] = "|".join(bill_descs)
return answer_dict
@@ -551,7 +551,7 @@ def fill_grouper_cd_desc(answer_dict, constants: Constants):
)
if grouper_descs:
answer_dict["GROUPER_CD_DESC"] = list(set(grouper_descs))
answer_dict["GROUPER_CD_DESC"] = "|".join(list(set(grouper_descs)))
return answer_dict
@@ -740,7 +740,7 @@ def fill_claim_type(answer_dicts):
list[dict]: List of dictionaries with AARETE_DERIVED_CLAIM_TYPE
"""
claim_types = [
d.get("AARETE_DERIVED_CLAIM_TYPE_CD")[0]
d.get("AARETE_DERIVED_CLAIM_TYPE_CD")
for d in answer_dicts
if d.get("AARETE_DERIVED_CLAIM_TYPE_CD")
]
@@ -750,7 +750,7 @@ def fill_claim_type(answer_dicts):
mode_claim_type = max(set(claim_types), key=claim_types.count)
for answer_dict in answer_dicts:
if not answer_dict.get("AARETE_DERIVED_CLAIM_TYPE_CD"):
answer_dict["AARETE_DERIVED_CLAIM_TYPE_CD"] = [mode_claim_type]
answer_dict["AARETE_DERIVED_CLAIM_TYPE_CD"] = mode_claim_type
return answer_dicts
src/crosswalk/crosswalk_builder.py
+6 -4
View File
@@ -147,13 +147,14 @@ class CrosswalkBuilder:
df = pd.DataFrame(self.mapping.items(), columns=[from_col, to_col])
df.to_csv(output_path, index=False, quoting=csv.QUOTE_ALL, encoding="utf-8-sig")
def create_reverse_mapping(self) -> dict[str, list[str]]:
def create_reverse_mapping(self) -> dict[str, str]:
"""Create a reverse mapping where values that map to the same key are concatenated
Example:
{"A": "1", "B": "1", "C": "2"} becomes {"1": ["A", "B"], "2": ["C"]}
{"A": "1", "B": "1", "C": "2"} becomes {"1": "A, B", "2": "C"}
Returns:
dict[str, list[str]]: Reverse mapping with list of sources for each target
dict[str, str]: Reverse mapping with concatenated values
"""
reverse_dict: dict[str, list[str]] = {}
@@ -164,7 +165,8 @@ class CrosswalkBuilder:
else:
reverse_dict[target] = [str(source)]
return reverse_dict
# Then join the lists with commas
return {target: "|".join(sources) for target, sources in reverse_dict.items()}
def map_value(self, key_value):
return self.mapping.get(key_value)
src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py
+67 -103
View File
@@ -23,15 +23,9 @@ def prompt_exhibit_level(
exhibit_text, exhibit_level_fields.print_prompt_dict(constants)
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
filename,
max_tokens=8192,
cache=True,
instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(),
usage_label="EXHIBIT_LEVEL",
prompt, "sonnet_latest", filename, max_tokens=8192
)
logging.debug(f"Exhibit Level LLM raw output for {filename}: {llm_answer_raw}")
logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}")
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
@@ -71,9 +65,7 @@ def prompt_exhibit_level_breakout(
llm_answer_raw = llm_utils.invoke_claude(
prompt=prompt, model_id="sonnet_latest", filename=filename
)
logging.debug(
f"LLM raw output for FACILITY_ADJUSTMENT_BREAKOUT in {filename}: {llm_answer_raw}"
)
print("Facility Adjustment: ", llm_answer_raw)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
exhibit_level_answers.update(llm_answer_final)
@@ -92,10 +84,14 @@ def prompt_dynamic_primary(
cache=True,
instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(),
)
logging.debug(
f"Dynamic Primary Claude answer for {filename}; {field}: {llm_answer_raw}"
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, string_utils.Delimiter.PIPE
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
if "," in llm_answer_final:
llm_answer_final = "|".join(
[item.strip() for item in llm_answer_final.split(",")]
)
return llm_answer_final
@@ -118,9 +114,7 @@ def prompt_reimbursement_primary(
instruction=prompt_templates.REIMBURSEMENT_PRIMARY_INSTRUCTION(),
usage_label="REIMBURSEMENT_PRIMARY",
)
logging.debug(
f"""Reimbursement Primary LLM raw output for {filename}: {llm_answer_raw}"""
)
logging.debug(f"""LLM raw output for {filename}: {llm_answer_raw}""")
# Check for the special "no results" case
if "NO_REIMBURSEMENT_TERMS_FOUND" in llm_answer_raw:
@@ -181,13 +175,8 @@ def prompt_fee_schedule_breakout(
prompt,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.FEE_SCHEDULE_BREAKOUT_INSTRUCTION(),
usage_label="FEE_SCHEDULE_BREAKOUT",
)
logging.debug(
f"FEE Schedule Breakout LLM Response for {filename}: {llm_answer_raw}"
)
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
fs_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except ValueError as e:
@@ -212,11 +201,8 @@ def prompt_grouper_breakout(
prompt,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.GROUPER_BREAKOUT_INSTRUCTION(),
usage_label="GROUPER_BREAKOUT",
)
logging.debug(f"Grouper Breakout LLM Response for {filename}: {llm_answer_raw}")
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
grouper_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except:
@@ -246,8 +232,10 @@ def prompt_carveout_check(
instruction=prompt_templates.CARVEOUT_CHECK_INSTRUCTION(),
usage_label="CARVEOUT_CHECK",
)
logging.debug(f"Carveout Check LLM raw output for {filename}: {llm_answer_raw}")
carveout_answer = string_utils.universal_json_load(llm_answer_raw)[0]
carveout_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return carveout_answer
@@ -303,7 +291,9 @@ def prompt_lob_relationship(
cache=True,
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final
@@ -322,10 +312,12 @@ def prompt_special_case_assignment(
for key in non_term_keys
):
answer_dict = special_case_dicts[0].copy()
answer_dict[special_case_term] = [
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
answer_dict[special_case_term] = "|".join(
[
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
)
return answer_dict
# Otherwise use LLM
@@ -335,11 +327,10 @@ def prompt_special_case_assignment(
),
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.SPECIAL_CASE_ASSIGNMENT_INSTRUCTION(),
usage_label="SPECIAL_CASE_ASSIGNMENT",
)
index_answer = string_utils.universal_json_load(llm_answer_raw)[0]
index_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
try:
if not string_utils.is_empty(index_answer):
return special_case_dicts[int(index_answer)]
@@ -376,9 +367,6 @@ def prompt_full_context(
cache=True,
instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(),
)
logging.debug(
f"Full Context LLM raw output for {filename}: {claude_answer_raw}"
)
full_context_answers_dict = string_utils.universal_json_load(claude_answer_raw)
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
@@ -424,7 +412,11 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
logging.debug(
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
)
final_answer = string_utils.universal_json_load(llm_response)[0]
final_answer = (
string_utils.extract_text_from_delimiters(llm_response, Delimiter.PIPE)
.strip()
.upper()
)
return final_answer == "YES"
@@ -469,18 +461,10 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
str: LLM response indicating if the exhibit has changed.
"""
prompt = prompt_templates.EXHIBIT_LINKAGE(page_header, previous_header)
claude_answer_raw = llm_utils.invoke_claude(
prompt,
"legacy_sonnet",
filename,
cache=True,
instruction=prompt_templates.EXHIBIT_LINKAGE_INSTRUCTION(),
usage_label="EXHIBIT_LINKAGE",
claude_answer_raw = llm_utils.invoke_claude(prompt, "legacy_sonnet", filename)
exhibit_is_different = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
logging.debug(
f"Claude response for exhibit linkage in {filename}: {claude_answer_raw}"
)
exhibit_is_different = string_utils.universal_json_load(claude_answer_raw)[0]
return exhibit_is_different
@@ -502,17 +486,11 @@ def prompt_exhibit_header(page_content, EXHIBIT_HEADER_MARKERS, filename):
page_content[0:400], EXHIBIT_HEADER_MARKERS
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
filename,
max_tokens=300,
cache=True,
instruction=prompt_templates.EXHIBIT_HEADER_INSTRUCTION(),
usage_label="EXHIBIT_HEADER",
prompt, "sonnet_latest", filename, max_tokens=300
)
llm_answer_extracted = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(f"LLM raw output for exhibit header in {filename}: {llm_answer_raw}")
llm_answer_extracted = string_utils.universal_json_load(llm_answer_raw)[0]
return llm_answer_extracted
@@ -528,15 +506,8 @@ def prompt_date_fix(date: str) -> str:
if string_utils.is_empty(date):
return "N/A"
prompt = prompt_templates.DATE_FIX_PROMPT(date)
response = llm_utils.invoke_claude(
prompt,
"haiku_latest",
"date_fix",
cache=True,
instruction=prompt_templates.DATE_FIX_INSTRUCTION(),
usage_label="DATE_FIX",
)
return string_utils.universal_json_load(response)[0]
response = llm_utils.invoke_claude(prompt, "haiku_latest", "date_fix")
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_derived_term_date(
@@ -557,15 +528,8 @@ def prompt_derived_term_date(
prompt = prompt_templates.DERIVED_TERM_DATE_PROMPT(
effective_date, termination_information
)
response = llm_utils.invoke_claude(
prompt,
"haiku_latest",
"derive_term_date",
cache=True,
instruction=prompt_templates.DERIVED_TERM_DATE_INSTRUCTION(),
usage_label="DERIVED_TERM_DATE",
)
return string_utils.universal_json_load(response)[0]
response = llm_utils.invoke_claude(prompt, "haiku_latest", "derive_term_date")
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_dynamic_assignment(
@@ -684,7 +648,15 @@ def prompt_lesser_of_distribution(
)
return reimb_term # Return original term unchanged
else:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
f"{reimb_term} → {llm_answer_final[:60]}..."
)
return llm_answer_final
@@ -722,12 +694,14 @@ def prompt_lesser_of_check(
instruction=prompt_templates.LESSER_OF_CHECK_INSTRUCTION(),
usage_label="LESSER_OF_CHECK",
)
logging.debug(f"LESSER_OF_CHECK LLM raw output for {filename}: {llm_answer_raw}")
try:
# Extract JSON from pipes
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(f"LESSER_OF_CHECK extracted: {llm_answer_final}")
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
# Parse JSON
lesser_of_answer_dict = json.loads(llm_answer_final)
@@ -796,18 +770,13 @@ def prompt_exhibit_title_match(
prompt = prompt_templates.EXHIBIT_TITLE_MATCH(
current_exhibit_title, target_exhibit_reference
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.EXHIBIT_TITLE_MATCH_INSTRUCTION(),
usage_label="EXHIBIT_TITLE_MATCH",
)
logging.debug(f"LLM raw output for Lesser Of Check in {filename}: {llm_answer_raw}")
llm_answer_raw = llm_utils.invoke_claude(prompt, "sonnet_latest", filename)
try:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
return llm_answer_final == "Y"
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final.strip().upper() # Return "YES" or "NO" as string
except:
return "NO" # Default to "NO" string
@@ -833,16 +802,11 @@ def provider_name_match_check(
try:
# Adjust this based on your LLM client
response = llm_utils.invoke_claude(
CHECK_PROVIDER_NAME_MATCH_PROMPT,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.CHECK_PROVIDER_NAME_MATCH_INSTRUCTION(),
usage_label="CHECK_PROVIDER_NAME_MATCH",
CHECK_PROVIDER_NAME_MATCH_PROMPT, "sonnet_latest", filename
)
# Extract last character (should be Y or N)
response = string_utils.universal_json_load(response)[0]
response = string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
if response == "Y":
logging.debug(
src/pipelines/clients/clover/prompts/prompt_calls.py
+79 -42
View File
@@ -31,7 +31,7 @@ def prompt_exhibit_level(
instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(),
usage_label="EXHIBIT_LEVEL",
)
logging.debug(f"Exhibit Level LLM raw output for {filename}: {llm_answer_raw}")
logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}")
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
@@ -71,9 +71,7 @@ def prompt_exhibit_level_breakout(
llm_answer_raw = llm_utils.invoke_claude(
prompt=prompt, model_id="sonnet_latest", filename=filename
)
logging.debug(
f"LLM raw output for FACILITY_ADJUSTMENT_BREAKOUT in {filename}: {llm_answer_raw}"
)
print("Facility Adjustment: ", llm_answer_raw)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
exhibit_level_answers.update(llm_answer_final)
@@ -92,10 +90,14 @@ def prompt_dynamic_primary(
cache=True,
instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(),
)
logging.debug(
f"Dynamic Primary Claude answer for {filename}; {field}: {llm_answer_raw}"
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, string_utils.Delimiter.PIPE
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
if "," in llm_answer_final:
llm_answer_final = "|".join(
[item.strip() for item in llm_answer_final.split(",")]
)
return llm_answer_final
@@ -118,9 +120,7 @@ def prompt_reimbursement_primary(
instruction=prompt_templates.REIMBURSEMENT_PRIMARY_INSTRUCTION(),
usage_label="REIMBURSEMENT_PRIMARY",
)
logging.debug(
f"""Reimbursement Primary LLM raw output for {filename}: {llm_answer_raw}"""
)
logging.debug(f"""LLM raw output for {filename}: {llm_answer_raw}""")
# Check for the special "no results" case
if "NO_REIMBURSEMENT_TERMS_FOUND" in llm_answer_raw:
@@ -185,9 +185,7 @@ def prompt_fee_schedule_breakout(
instruction=prompt_templates.FEE_SCHEDULE_BREAKOUT_INSTRUCTION(),
usage_label="FEE_SCHEDULE_BREAKOUT",
)
logging.debug(
f"FEE Schedule Breakout LLM Response for {filename}: {llm_answer_raw}"
)
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
fs_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except ValueError as e:
@@ -216,7 +214,7 @@ def prompt_grouper_breakout(
instruction=prompt_templates.GROUPER_BREAKOUT_INSTRUCTION(),
usage_label="GROUPER_BREAKOUT",
)
logging.debug(f"Grouper Breakout LLM Response for {filename}: {llm_answer_raw}")
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
grouper_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except:
@@ -246,8 +244,10 @@ def prompt_carveout_check(
instruction=prompt_templates.CARVEOUT_CHECK_INSTRUCTION(),
usage_label="CARVEOUT_CHECK",
)
logging.debug(f"Carveout Check LLM raw output for {filename}: {llm_answer_raw}")
carveout_answer = string_utils.universal_json_load(llm_answer_raw)[0]
carveout_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return carveout_answer
@@ -303,7 +303,9 @@ def prompt_lob_relationship(
cache=True,
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final
@@ -322,10 +324,12 @@ def prompt_special_case_assignment(
for key in non_term_keys
):
answer_dict = special_case_dicts[0].copy()
answer_dict[special_case_term] = [
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
answer_dict[special_case_term] = "|".join(
[
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
)
return answer_dict
# Otherwise use LLM
@@ -339,7 +343,9 @@ def prompt_special_case_assignment(
instruction=prompt_templates.SPECIAL_CASE_ASSIGNMENT_INSTRUCTION(),
usage_label="SPECIAL_CASE_ASSIGNMENT",
)
index_answer = string_utils.universal_json_load(llm_answer_raw)[0]
index_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
try:
if not string_utils.is_empty(index_answer):
return special_case_dicts[int(index_answer)]
@@ -349,6 +355,22 @@ def prompt_special_case_assignment(
return special_case_dicts[0]
def extract_and_parse(text):
import re
import json
pattern = r"\{[^{}]+\}"
match = re.search(pattern, text, re.DOTALL)
if not match:
return {}
try:
return json.loads(match.group(0))
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
def prompt_full_context(
contract_text: str,
full_context_fields: FieldSet,
@@ -376,10 +398,10 @@ def prompt_full_context(
cache=True,
instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(),
)
logging.debug(
f"Full Context LLM raw output for {filename}: {claude_answer_raw}"
)
full_context_answers_dict = string_utils.universal_json_load(claude_answer_raw)
if not isinstance(full_context_answers_dict, dict):
full_context_answers_dict = extract_and_parse(claude_answer_raw)
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
full_context_answers_dict = {}
@@ -424,7 +446,11 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
logging.debug(
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
)
final_answer = string_utils.universal_json_load(llm_response)[0]
final_answer = (
string_utils.extract_text_from_delimiters(llm_response, Delimiter.PIPE)
.strip()
.upper()
)
return final_answer == "YES"
@@ -477,10 +503,9 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
instruction=prompt_templates.EXHIBIT_LINKAGE_INSTRUCTION(),
usage_label="EXHIBIT_LINKAGE",
)
logging.debug(
f"Claude response for exhibit linkage in {filename}: {claude_answer_raw}"
exhibit_is_different = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
exhibit_is_different = string_utils.universal_json_load(claude_answer_raw)[0]
return exhibit_is_different
@@ -510,9 +535,9 @@ def prompt_exhibit_header(page_content, EXHIBIT_HEADER_MARKERS, filename):
instruction=prompt_templates.EXHIBIT_HEADER_INSTRUCTION(),
usage_label="EXHIBIT_HEADER",
)
logging.debug(f"LLM raw output for exhibit header in {filename}: {llm_answer_raw}")
llm_answer_extracted = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_extracted = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_extracted
@@ -536,7 +561,7 @@ def prompt_date_fix(date: str) -> str:
instruction=prompt_templates.DATE_FIX_INSTRUCTION(),
usage_label="DATE_FIX",
)
return string_utils.universal_json_load(response)[0]
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_derived_term_date(
@@ -565,7 +590,7 @@ def prompt_derived_term_date(
instruction=prompt_templates.DERIVED_TERM_DATE_INSTRUCTION(),
usage_label="DERIVED_TERM_DATE",
)
return string_utils.universal_json_load(response)[0]
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_dynamic_assignment(
@@ -684,7 +709,15 @@ def prompt_lesser_of_distribution(
)
return reimb_term # Return original term unchanged
else:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
f"{reimb_term} → {llm_answer_final[:60]}..."
)
return llm_answer_final
@@ -722,12 +755,14 @@ def prompt_lesser_of_check(
instruction=prompt_templates.LESSER_OF_CHECK_INSTRUCTION(),
usage_label="LESSER_OF_CHECK",
)
logging.debug(f"LESSER_OF_CHECK LLM raw output for {filename}: {llm_answer_raw}")
try:
# Extract JSON from pipes
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(f"LESSER_OF_CHECK extracted: {llm_answer_final}")
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
# Parse JSON
lesser_of_answer_dict = json.loads(llm_answer_final)
@@ -804,10 +839,12 @@ def prompt_exhibit_title_match(
instruction=prompt_templates.EXHIBIT_TITLE_MATCH_INSTRUCTION(),
usage_label="EXHIBIT_TITLE_MATCH",
)
logging.debug(f"LLM raw output for Lesser Of Check in {filename}: {llm_answer_raw}")
try:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
return llm_answer_final == "Y"
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final.strip().upper() # Return "YES" or "NO" as string
except:
return "NO" # Default to "NO" string
@@ -842,7 +879,7 @@ def provider_name_match_check(
)
# Extract last character (should be Y or N)
response = string_utils.universal_json_load(response)[0]
response = string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
if response == "Y":
logging.debug(
src/pipelines/saas/prompts/prompt_calls.py
+63 -48
View File
@@ -31,7 +31,7 @@ def prompt_exhibit_level(
instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(),
usage_label="EXHIBIT_LEVEL",
)
logging.debug(f"Exhibit Level LLM raw output for {filename}: {llm_answer_raw}")
logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}")
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
@@ -71,9 +71,6 @@ def prompt_exhibit_level_breakout(
llm_answer_raw = llm_utils.invoke_claude(
prompt=prompt, model_id="sonnet_latest", filename=filename
)
logging.debug(
f"LLM raw output for FACILITY_ADJUSTMENT_BREAKOUT in {filename}: {llm_answer_raw}"
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
exhibit_level_answers.update(llm_answer_final)
@@ -92,10 +89,14 @@ def prompt_dynamic_primary(
cache=True,
instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(),
)
logging.debug(
f"Dynamic Primary Claude answer for {filename}; {field}: {llm_answer_raw}"
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, string_utils.Delimiter.PIPE
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
if "," in llm_answer_final:
llm_answer_final = "|".join(
[item.strip() for item in llm_answer_final.split(",")]
)
return llm_answer_final
@@ -118,9 +119,7 @@ def prompt_reimbursement_primary(
instruction=prompt_templates.REIMBURSEMENT_PRIMARY_INSTRUCTION(),
usage_label="REIMBURSEMENT_PRIMARY",
)
logging.debug(
f"""Reimbursement Primary LLM raw output for {filename}: {llm_answer_raw}"""
)
logging.debug(f"""LLM raw output for {filename}: {llm_answer_raw}""")
# Check for the special "no results" case
if "NO_REIMBURSEMENT_TERMS_FOUND" in llm_answer_raw:
@@ -185,9 +184,7 @@ def prompt_fee_schedule_breakout(
instruction=prompt_templates.FEE_SCHEDULE_BREAKOUT_INSTRUCTION(),
usage_label="FEE_SCHEDULE_BREAKOUT",
)
logging.debug(
f"FEE Schedule Breakout LLM Response for {filename}: {llm_answer_raw}"
)
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
fs_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except ValueError as e:
@@ -216,7 +213,7 @@ def prompt_grouper_breakout(
instruction=prompt_templates.GROUPER_BREAKOUT_INSTRUCTION(),
usage_label="GROUPER_BREAKOUT",
)
logging.debug(f"Grouper Breakout LLM Response for {filename}: {llm_answer_raw}")
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
grouper_breakout_dict = string_utils.universal_json_load(llm_answer_raw)
except:
@@ -246,8 +243,10 @@ def prompt_carveout_check(
instruction=prompt_templates.CARVEOUT_CHECK_INSTRUCTION(),
usage_label="CARVEOUT_CHECK",
)
logging.debug(f"Carveout Check LLM raw output for {filename}: {llm_answer_raw}")
carveout_answer = string_utils.universal_json_load(llm_answer_raw)[0]
carveout_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return carveout_answer
@@ -303,7 +302,9 @@ def prompt_lob_relationship(
cache=True,
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final
@@ -322,10 +323,12 @@ def prompt_special_case_assignment(
for key in non_term_keys
):
answer_dict = special_case_dicts[0].copy()
answer_dict[special_case_term] = [
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
answer_dict[special_case_term] = "|".join(
[
special_case_dict[special_case_term]
for special_case_dict in special_case_dicts
]
)
return answer_dict
# Otherwise use LLM
@@ -339,7 +342,9 @@ def prompt_special_case_assignment(
instruction=prompt_templates.SPECIAL_CASE_ASSIGNMENT_INSTRUCTION(),
usage_label="SPECIAL_CASE_ASSIGNMENT",
)
index_answer = string_utils.universal_json_load(llm_answer_raw)[0]
index_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
try:
if not string_utils.is_empty(index_answer):
return special_case_dicts[int(index_answer)]
@@ -376,9 +381,6 @@ def prompt_full_context(
cache=True,
instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(),
)
logging.debug(
f"Full Context LLM raw output for {filename}: {claude_answer_raw}"
)
full_context_answers_dict = string_utils.universal_json_load(claude_answer_raw)
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
@@ -424,7 +426,11 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
logging.debug(
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
)
final_answer = string_utils.universal_json_load(llm_response)[0]
final_answer = (
string_utils.extract_text_from_delimiters(llm_response, Delimiter.PIPE)
.strip()
.upper()
)
return final_answer == "YES"
@@ -477,10 +483,9 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
instruction=prompt_templates.EXHIBIT_LINKAGE_INSTRUCTION(),
usage_label="EXHIBIT_LINKAGE",
)
logging.debug(
f"Claude response for exhibit linkage in {filename}: {claude_answer_raw}"
exhibit_is_different = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
exhibit_is_different = string_utils.universal_json_load(claude_answer_raw)[0]
return exhibit_is_different
@@ -510,9 +515,9 @@ def prompt_exhibit_header(page_content, EXHIBIT_HEADER_MARKERS, filename):
instruction=prompt_templates.EXHIBIT_HEADER_INSTRUCTION(),
usage_label="EXHIBIT_HEADER",
)
logging.debug(f"LLM raw output for exhibit header in {filename}: {llm_answer_raw}")
llm_answer_extracted = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_extracted = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_extracted
@@ -536,7 +541,7 @@ def prompt_date_fix(date: str) -> str:
instruction=prompt_templates.DATE_FIX_INSTRUCTION(),
usage_label="DATE_FIX",
)
return string_utils.universal_json_load(response)[0]
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_derived_term_date(
@@ -565,7 +570,7 @@ def prompt_derived_term_date(
instruction=prompt_templates.DERIVED_TERM_DATE_INSTRUCTION(),
usage_label="DERIVED_TERM_DATE",
)
return string_utils.universal_json_load(response)[0]
return string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
def prompt_dynamic_assignment(
@@ -605,9 +610,7 @@ def prompt_dynamic_assignment(
instruction=prompt_templates.DYNAMIC_ASSIGNMENT_INSTRUCTION(),
usage_label="DYNAMIC_ASSIGNMENT",
)
logging.debug(
f"LLM raw output for dynamic assignment of {field_name} in {filename}: {llm_answer_raw}"
)
try:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
return llm_answer_final
@@ -686,7 +689,15 @@ def prompt_lesser_of_distribution(
)
return reimb_term # Return original term unchanged
else:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
f"{reimb_term} → {llm_answer_final[:60]}..."
)
return llm_answer_final
@@ -724,16 +735,18 @@ def prompt_lesser_of_check(
instruction=prompt_templates.LESSER_OF_CHECK_INSTRUCTION(),
usage_label="LESSER_OF_CHECK",
)
logging.debug(f"LESSER_OF_CHECK LLM raw output for {filename}: {llm_answer_raw}")
try:
# Extract JSON from LLM response
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
logging.debug(f"LESSER_OF_CHECK extracted: {llm_answer_final}")
try:
# Extract JSON from pipes
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
# Parse JSON
lesser_of_answer_dict = json.loads(llm_answer_final)
print("lesser_of_answer_dict:", lesser_of_answer_dict)
# Validate required fields
required_fields = [
"service_term",
@@ -806,10 +819,12 @@ def prompt_exhibit_title_match(
instruction=prompt_templates.EXHIBIT_TITLE_MATCH_INSTRUCTION(),
usage_label="EXHIBIT_TITLE_MATCH",
)
logging.debug(f"LLM raw output for Lesser Of Check in {filename}: {llm_answer_raw}")
try:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)[0]
return llm_answer_final == "Y"
llm_answer_final = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
return llm_answer_final.strip().upper() # Return "YES" or "NO" as string
except:
return "NO" # Default to "NO" string
@@ -844,7 +859,7 @@ def provider_name_match_check(
)
# Extract last character (should be Y or N)
response = string_utils.universal_json_load(response)[0]
response = string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)
if response == "Y":
logging.debug(
src/pipelines/shared/extraction/dynamic_funcs.py
+1 -1
View File
@@ -52,7 +52,7 @@ def dynamic_primary(
)
# If there is at least ONE Non-N/A answers in the Exhibit
if not string_utils.is_empty(exhibit_text_answer):
field.update_valid_values(exhibit_text_answer.append("N/A"))
field.update_valid_values(exhibit_text_answer + " | N/A")
dynamic_reimbursement_fields.add_field(field)
dynamic_primary_fields.remove_field(field.field_name)
# If the field is not found, add to Exhibit-Level answers (the answer will be N/A or similar)
src/pipelines/shared/extraction/one_to_one_funcs.py
-55
View File
@@ -17,61 +17,6 @@ from src.pipelines.shared.extraction.vision_funcs import get_image_array_based_a
from src.prompts.fieldset import Field, FieldSet
def extract_global_lesser_of(
text_dict: str, first_reimbursement_page: str, filename: str
) -> dict:
"""Extract global "lesser of" statements that apply contract-wide.
Args:
contract_text (str): The full text of the contract to process.
filename (str): The name of the file being processed.
Returns:
dict: A dictionary containing the extracted "lesser of" statement.
"""
global_lesser_of_field = Field.load_from_file(
file_path=config.FIELD_JSON_PATH, field_name="GLOBAL_LESSER_OF_STATEMENT"
)
context = "\n".join(
[
page_text
for page_num, page_text in text_dict.items()
if int(page_num) < int(first_reimbursement_page)
]
)
# Process using existing full context logic
prompt = prompt_templates.GLOBAL_LESSER_OF_STATEMENT_TEMPLATE(
context,
field_prompt=global_lesser_of_field.prompt,
)
logging.debug(
f"Extracting global lesser of statement for {filename} with prompt: {prompt}"
)
response = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION(),
usage_label="ONE_TO_ONE_SINGLE_FIELD",
)
logging.debug(f"Response for global lesser of statement extraction: {response}")
parsed_global_lesser_of = string_utils.universal_json_load(response).get(
"GLOBAL_LESSER_OF_STATEMENT", "N/A"
)
return {
"GLOBAL_LESSER_OF_STATEMENT": (
parsed_global_lesser_of if parsed_global_lesser_of else "N/A" or "N/A"
)
}
def pass_smart_chunked_to_full_context(answers_dict, one_to_one_fields):
# even after running the vision api, if the effective date is still N/A, we need to run the prompt again with full context
if "EFFECTIVE_DT" in answers_dict and string_utils.is_empty(
src/pipelines/shared/extraction/page_funcs.py
+2 -9
View File
@@ -299,7 +299,7 @@ def parse_table_rows(table_text: str, page_num: str = "") -> tuple[list[str], in
if num_columns == 0:
num_columns = len(row)
# Join list items with a separator
row_str = [str(item) for item in row]
row_str = " | ".join(str(item) for item in row)
rows.append(row_str)
else:
# Non-list row - log warning but include it
@@ -360,13 +360,6 @@ def split_table_by_cell_count(
chunks = []
for i in range(0, len(table_rows), max_rows):
chunk = table_rows[i : i + max_rows]
chunk_strs = []
for row in chunk:
if isinstance(row, list):
chunk_strs.append(" ".join(str(cell) for cell in row))
else:
chunk_strs.append(str(row))
chunks.append("\n".join(chunk_strs))
chunks.append("\n".join(chunk))
return chunks
src/pipelines/shared/extraction/tin_npi_funcs.py
+125 -141
View File
@@ -165,47 +165,44 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
"""Cleans provider information by standardizing Name/TIN/NPI formats.
Args:
provider_info (list[dict]): A list of provider information dictionaries.
TIN, NPI, and NAME values are expected to be lists.
provider_info (list[dict]): A list of provider information dictionaries. Each
dictionary should be keyed by TIN, NPI, NAME
Returns:
list[dict]: Cleaned provider information
list: Cleaned provider information
"""
cleaned_providers = []
for provider in provider_info:
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 - remove hyphens and dots, ensure it's 9 digits
if "TIN" in provider and provider["TIN"]:
cleaned_provider["TIN"] = provider["TIN"].replace("-", "").replace(".", "")
# Validate that TIN only contains digits or "|"
if not all(c.isdigit() or c == "|" for c in cleaned_provider["TIN"]):
cleaned_provider["TIN"] = "UNKNOWN"
else:
cleaned_provider["TIN"] = "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"]
# Clean NPI - remove hyphens and dots, ensure it's 10 digits
if "NPI" in provider and provider["NPI"]:
cleaned_provider["NPI"] = provider["NPI"].replace("-", "").replace(".", "")
# Validate that NPI only contains digits or "|"
if not all(c.isdigit() or c == "|" for c in cleaned_provider["NPI"]):
cleaned_provider["NPI"] = "UNKNOWN"
else:
cleaned_provider["NPI"] = "UNKNOWN"
# Clean Name - remove extra spaces and ensure it's not empty
if "NAME" in provider and provider["NAME"]:
cleaned_provider["NAME"] = (
" ".join(provider["NAME"].split()).strip().rstrip(".")
)
# If name is empty after cleaning, set to "UNKNOWN"
if not cleaned_provider["NAME"]:
cleaned_provider["NAME"] = "UNKNOWN"
else:
cleaned_provider["NAME"] = "UNKNOWN"
cleaned_providers.append(cleaned_provider)
return cleaned_providers
@@ -233,7 +230,6 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
other_names = []
provider_name = one_to_one_results.get("PROVIDER_NAME")
print(f"Provider name for matching: {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
@@ -243,61 +239,66 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
found_match = False
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 []
name_matches = prompt_calls.provider_name_match_check(
names, provider_name, filename
provider_name, provider.get("NAME", "UNKNOWN"), filename
)
if name_matches:
found_match = True
provider["IS_GROUP"] = "Y"
for tin in tins:
group_tins.add(tin)
for npi in npis:
group_npis.add(npi)
for name in names:
group_names.add(name)
if provider.get("TIN"):
group_tins.add(provider["TIN"])
if provider.get("NPI"):
group_npis.add(provider["NPI"])
if provider.get("NAME"):
group_names.add(provider["NAME"])
else:
provider["IS_GROUP"] = "N"
other_tins.extend(tins)
other_npis.extend(npis)
other_names.extend(names)
if provider.get("TIN"):
other_tins.append(provider["TIN"])
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 and not found_match:
new_provider = {
"NAME": provider_name,
"TIN": [],
"NPI": [],
"TIN": "UNKNOWN",
"NPI": "UNKNOWN",
"IS_GROUP": "Y",
}
provider_list.append(new_provider)
for name in provider_name:
group_names.add(name)
group_names.add(provider_name)
# Remove UNKNOWN
# 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"]
# Deduplicate and remove overlaps
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"]
# Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other
group_tins = list(dict.fromkeys("|".join(group_tins).split("|")))
other_tins = list(dict.fromkeys("|".join(other_tins).split("|")))
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))
other_npis = list(dict.fromkeys(other_npis))
other_npis = [v for v in other_npis if v not in group_npis and v != "UNKNOWN"]
group_npis = list(dict.fromkeys("|".join(group_npis).split("|")))
other_npis = list(dict.fromkeys("|".join(other_npis).split("|")))
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))
other_names = list(dict.fromkeys(other_names))
other_names = [v for v in other_names if v not in group_names and v != "UNKNOWN"]
group_names = list(dict.fromkeys("|".join(group_names).split("|")))
other_names = list(dict.fromkeys("|".join(other_names).split("|")))
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 NO_IDENTIFIERS_FOUND
# Filter out records with NO_IDENTIFIERS_FOUND
provider_list = [
p for p in provider_list if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME") or [])
p for p in provider_list if p.get("NAME") != "NO_IDENTIFIERS_FOUND"
]
# Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values
@@ -310,13 +311,17 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
)
# 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_NPI"] = 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_TIN"] = "|".join(group_tins) if group_tins else ""
one_to_one_results["PROV_GROUP_NPI"] = "|".join(group_npis) if group_npis 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
one_to_one_results["PROV_OTHER_TIN"] = 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_NAME_FULL"] = other_names if other_names else []
one_to_one_results["PROV_OTHER_TIN"] = "|".join(other_tins) if other_tins else ""
one_to_one_results["PROV_OTHER_NPI"] = "|".join(other_npis) if other_npis else ""
one_to_one_results["PROV_OTHER_NAME_FULL"] = (
"|".join(other_names) if other_names else ""
)
return one_to_one_results
@@ -389,15 +394,16 @@ def reimbursement_tin_npi(
return answer_dict, reimbursement_level_fields
def deduplicate_providers(provider_info: list[dict]) -> list[dict]:
def deduplicate_providers(
provider_info: list[dict],
) -> list[dict]: # TODO: add unit test
"""Deduplicates provider information based on TIN, NPI, and NAME fields.
Each provider dict contains TIN, NPI, and NAME as lists.
If all three fields are empty or ['UNKNOWN'], the provider is skipped.
Providers are considered duplicates if all three fields match exactly.
This function checks for duplicates in the provider information list by comparing
the TIN, NPI, and NAME fields. If all three fields match, only one instance of the
provider is kept. If all three fields are unknown, the provider is skipped.
Args:
provider_info (list[dict]): A list of provider information dictionaries.
provider_info (list[dict]): A list of provider information dictionaries to be deduplicated.
Returns:
list[dict]: A list of deduplicated provider information dictionaries.
@@ -406,35 +412,32 @@ def deduplicate_providers(provider_info: list[dict]) -> list[dict]:
seen = set()
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", [])
# 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 provider if all identifiers are unknown or empty
# 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"])
(
string_utils.is_empty(provider.get("TIN", ""))
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
# Convert lists to tuples so they can be hashed
key = (
tuple(tins),
tuple(npis),
tuple(names),
provider.get("TIN", ""),
provider.get("NPI", ""),
provider.get("NAME", ""),
)
if key not in seen:
seen.add(key)
valid_providers.append(provider)
return valid_providers
@@ -491,7 +494,7 @@ def get_provider_info(
logging.warning(
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
page_text = text_dict[page_num]
@@ -520,13 +523,19 @@ def get_provider_info(
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(".", ""))
if provider.get("TIN") and provider["TIN"] != "UNKNOWN":
# Normalize for comparison (remove hyphens/dots)
raw_tins = [tin.strip() for tin in provider["TIN"].split("|")]
normalized_tins = [
tin.replace("-", "").replace(".", "") for tin in raw_tins
]
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
missed_tins = [tin for tin in regex_tins if tin not in llm_found_tins]
@@ -543,7 +552,7 @@ def get_provider_info(
logging.error(f"Error parsing JSON response from LLM: {str(e)}")
logging.error(f"Raw response: {claude_answer_raw}")
# Return a default value
return [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}]
return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}]
def run_provider_info_fields(
@@ -646,6 +655,7 @@ def run_provider_info_fields(
text_dict, page_num, filename, payer_name
)
combined_provider_info.extend(page_provider_info)
# Clean and standardize results
cleaned_provider_info = clean_provider_info(combined_provider_info)
# Deduplicate providers based on TIN and NPI
@@ -673,49 +683,23 @@ def run_provider_info_fields(
def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]:
"""
Deduplicates a list of provider dictionaries by NAME (list-aware),
preferring records with actual TIN/NPI values.
Deduplicates a list of provider dictionaries by NAME, preferring records with actual TIN/NPI values.
Args:
provider_list (list[dict]): List of provider dictionaries with NAME, TIN, NPI as lists.
provider_list (list[dict]): List of provider dictionaries with NAME, TIN, NPI keys.
Returns:
list[dict]: Deduplicated list with one entry per unique NAME.
list[dict]: Deduplicated list with one entry per unique NAME, preferring non-UNKNOWN values.
"""
seen_names = set()
deduplicated_list = []
# Sort so providers with actual TIN/NPI values come first
sorted_providers = sorted(
for provider in sorted(
provider_list,
key=lambda p: (
not p.get("TIN") or p.get("TIN") == ["UNKNOWN"],
not p.get("NPI") or p.get("NPI") == ["UNKNOWN"],
),
)
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:
key=lambda p: (p.get("TIN") == "UNKNOWN", p.get("NPI") == "UNKNOWN"),
):
name = provider.get("NAME")
if name not in seen_names:
seen_names.add(name)
deduplicated_list.append(provider)
deduplicated_list.append(provider)
return deduplicated_list
src/pipelines/shared/postprocessing/aarete_derived.py
+7 -3
View File
@@ -102,7 +102,7 @@ def fill_na_mapping(answer_dicts):
# Set the merged, deduplicated value
if all_lob_values:
answer_dict["AARETE_DERIVED_LOB"] = sorted(all_lob_values)
answer_dict["AARETE_DERIVED_LOB"] = "|".join(sorted(all_lob_values))
# Set relationship flags if values came from PROGRAM or PRODUCT
# Only set if field is empty or "N/A" (preserve LLM-determined Inclusive/Exclusive)
if program_lob_values:
@@ -156,11 +156,15 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
elif "," in from_field_value:
from_field_value_list = from_field_value.split(",")
else:
from_field_value_list = from_field_value
from_field_value_list = [from_field_value]
# Map each value in the list
to_field_answer_list = []
for individual_from_field_value in from_field_value_list:
individual_from_field_value = (
individual_from_field_value.strip()
)
if individual_from_field_value in crosswalk.mapping.keys():
# Value is a key in the mapping - map it to the target value
to_field_answer_list.append(
@@ -177,6 +181,6 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
individual_from_field_value
)
)
answer_dict[to_field_name] = to_field_answer_list
answer_dict[to_field_name] = "|".join(to_field_answer_list)
return answer_dicts
src/pipelines/shared/postprocessing/postprocessing_funcs.py
+66 -61
View File
@@ -73,20 +73,18 @@ def format_rate_fields_with_commas(value: str) -> str:
return ""
def remove_hyphens(value):
def remove_hyphens(value: str) -> str:
"""
Remove hyphens from the input value.
Remove hyphens from the input string.
Args:
value : The input might be list or string
value (str): The input string.
Returns:
str: The cleaned value with hyphens removed.
str: The cleaned string with hyphens removed.
"""
if isinstance(value, str):
return value.replace("-", "")
elif isinstance(value, list):
return [v.replace("-", "") if isinstance(v, str) else v for v in value]
return value
if not isinstance(value, str):
return ""
return value.replace("-", "")
def flatten_singleton_string_list(list_str: str) -> str:
@@ -548,16 +546,16 @@ def update_lob_for_duals(answer_dicts):
prompt, model_id="legacy_sonnet", filename="", max_tokens=200
)
logging.debug(f"Claude response for dual LOB check: {claude_answer_raw}")
claude_answer_extracted = string_utils.universal_json_load(
claude_answer_raw
)[0]
claude_answer_extracted = string_utils.extract_text_from_delimiters(
claude_answer_raw, Delimiter.PIPE
)
if (
claude_answer_extracted.lower() == "neither"
or claude_answer_extracted.lower() == "n/a"
): # If neither, take original AARETE_DERIVED_LOB
pass
else:
answer_dict["AARETE_DERIVED_LOB"] = [claude_answer_extracted]
answer_dict["AARETE_DERIVED_LOB"] = claude_answer_extracted
final_answer_dicts.append(answer_dict)
return final_answer_dicts
@@ -678,8 +676,8 @@ def deduplicate_provider_columns(df: pd.DataFrame) -> pd.DataFrame:
This function removes PROV_GROUP_* values from the PROV_OTHER_* columns if they are already present.
In addition, it deduplicates PROV_OTHER_* values.
E.g. If PROV_GROUP_TIN is ["123456789"] and PROV_OTHER_TIN is ["123456789", "987654321", "987654321"],
then PROV_OTHER_TIN will be updated to ["987654321"].
E.g. If PROV_GROUP_TIN is "123456789" and PROV_OTHER_TIN is "123456789 | 987654321 | 987654321",
then PROV_OTHER_TIN will be updated to "987654321".
Args:
@@ -696,44 +694,62 @@ def deduplicate_provider_columns(df: pd.DataFrame) -> pd.DataFrame:
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
]
if not all(col in df.columns for col in provider_columns):
return df
other_columns = ["PROV_OTHER_TIN", "PROV_OTHER_NPI", "PROV_OTHER_NAME_FULL"]
# Check if all provider columns exist in the DataFrame
for index, row in df.iterrows():
group_values = {
"PROV_OTHER_TIN": row.get("PROV_GROUP_TIN"),
"PROV_OTHER_NPI": row.get("PROV_GROUP_NPI"),
"PROV_OTHER_NAME_FULL": row.get("PROV_GROUP_NAME_FULL"),
}
if all(col in df.columns for col in provider_columns):
# Process each row
for index, row in df.iterrows():
# Get GROUP values for this row
group_tin = str(row.get("PROV_GROUP_TIN", "")).strip()
group_npi = str(row.get("PROV_GROUP_NPI", "")).strip()
group_name = str(row.get("PROV_GROUP_NAME_FULL", "")).strip()
for other_col, group_vals in group_values.items():
other_vals = row.get(other_col)
# Simple mapping of OTHER columns to their corresponding GROUP values
column_mappings = {
"PROV_OTHER_TIN": group_tin,
"PROV_OTHER_NPI": group_npi,
"PROV_OTHER_NAME_FULL": group_name,
}
if not other_vals:
# None, NaN, or empty list
df.at[index, other_col] = []
continue
# Process each OTHER column
for other_col, group_val in column_mappings.items():
other_value = str(row.get(other_col, "")).strip()
# Normalize group values (list or empty)
group_vals = group_vals or []
if (
other_value
and not string_utils.is_empty(other_value)
and other_value != "UNKNOWN"
):
# Split by pipe and strip whitespace
other_list = [
v.strip() for v in other_value.split("|") if v.strip()
]
# Remove GROUP values
filtered = [
v for v in other_vals if v not in group_vals and v and v != "UNKNOWN"
]
# Remove corresponding GROUP value if present
if (
group_val
and not string_utils.is_empty(group_val)
and group_val != "UNKNOWN"
):
other_list = [v for v in other_list if v != group_val]
# Deduplicate while preserving order
seen = set()
deduped = []
for v in filtered:
if v not in seen:
seen.add(v)
deduped.append(v)
df.at[index, other_col] = deduped
# Remove duplicates while preserving order
seen = set()
deduped_list = []
for item in other_list:
if (
item not in seen
and not string_utils.is_empty(item)
and item != "UNKNOWN"
):
seen.add(item)
deduped_list.append(item)
# Update the DataFrame
df.at[index, other_col] = (
" | ".join(deduped_list) if deduped_list else ""
)
return df
@@ -827,17 +843,12 @@ def fill_empty_dynamic(df):
# Get non-NA LOB values
column = "AARETE_DERIVED_LOB"
non_na_values = [
tuple(val) if isinstance(val, list) else val
for val in group[column]
if not string_utils.is_empty(val)
val for val in group[column].unique() if not string_utils.is_empty(val)
]
non_na_values = pd.unique(pd.Series(non_na_values))
# If there's exactly one unique non-NA value, fill NA values with it
if len(non_na_values) == 1:
fill_value = non_na_values[0]
if isinstance(fill_value, tuple):
fill_value = list(fill_value)
# Only apply to rows with this FILE_NAME and EXHIBIT_PAGE combination
mask = (
(result_df["FILE_NAME"] == file_name)
@@ -858,17 +869,14 @@ def fill_empty_dynamic(df):
# Get non-NA values
non_na_values = [
tuple(val) if isinstance(val, list) else val
for val in group[column]
val
for val in group[column].unique()
if not string_utils.is_empty(val)
]
non_na_values = pd.unique(non_na_values)
# If there's exactly one unique non-NA value, fill NA values with it
if len(non_na_values) == 1:
fill_value = non_na_values[0]
if isinstance(fill_value, tuple):
fill_value = list(fill_value)
# Only apply to rows with this FILE_NAME, EXHIBIT_PAGE, and LOB combination
mask = (
(result_df["FILE_NAME"] == file_name)
@@ -885,17 +893,14 @@ def fill_empty_dynamic(df):
for column in columns_to_fill:
# Get non-NA values
non_na_values = [
tuple(val) if isinstance(val, list) else val
for val in group[column]
val
for val in group[column].unique()
if not string_utils.is_empty(val)
]
non_na_values = pd.unique(non_na_values)
# If there's exactly one unique non-NA value, fill NA values with it
if len(non_na_values) == 1:
fill_value = non_na_values[0]
if isinstance(fill_value, tuple):
fill_value = list(fill_value)
# Only apply to rows with this FILE_NAME and EXHIBIT_PAGE combination
mask = (
(result_df["FILE_NAME"] == file_name)
src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py
+1 -1
View File
@@ -94,7 +94,7 @@ def process_single_field(
val = parsed.get(field.field_name)
if isinstance(val, list):
parsed[field.field_name] = val
parsed[field.field_name] = "|".join(map(str, val))
elif isinstance(val, str):
parsed[field.field_name] = val
field_value = parsed[field.field_name]
src/prompts/investment_prompts.json
+2 -2
View File
File diff suppressed because one or more lines are too long
src/prompts/prompt_templates.py
+115 -124
View File
@@ -3,14 +3,6 @@ from src.prompts.fieldset import FieldSet
from src.utils import string_utils
PIPE_FORMAT_INSTRUCTIONS = "Briefly explain your answer, then enclose your final answer in |pipes|. If the requested information doesn't apply to this context, return |N/A|. If the information should exist but cannot be clearly identified, return |UNKNOWN|."
JSON_LIST_FORMAT_INSTRUCTIONS = """Briefly explain your answer, then put the final answer in a properly formatted valid JSON list format. If the requested information doesn't apply to this context, return 'N/A'. If the information should exist but cannot be clearly identified, return 'UNKNOWN'. Formatting examples:
- Single string value: [\"value\"]
- Multiple string values: [\"value1\",\"value2\"]
- Date value: [\"YYYY-MM-DD\"]
- Not applicable: [\"N/A\"]
- Information unclear or missing: [\"UNKNOWN\"]
"""
JSON_DICT_FORMAT_INSTRUCTIONS = "Briefly explain your answer, then put the final answer in a properly formatted JSON dictionary. If the requested information doesn't apply to this context, return 'N/A'. If the information should exist but cannot be clearly identified, return 'UNKNOWN'."
def get_cacheable_instructions():
@@ -151,16 +143,15 @@ Here are the attributes to be included in the dictionary, and instructions on ho
{fields}
[GENERAL INSTRUCTIONS]
For each field, return ALL correct answers found. There may be more than one correct answer. Return all correct for each field value in a JSON LIST with all values.
For each field, return ALL correct answers found. There may be more than one correct answer. If there are multiple, return them in a pipe-separated list ("|").
If a list of valid values is provided, ONLY answer with the EXACT value from that list.
If there are no correct answers for a given field, write 'N/A' as the answer for that field.
Put each field and its corresponding JSON LIST answer in a properly formatted JSON dictionary.
Here is the text to analyze:
{context.replace('"', "'")}
{JSON_DICT_FORMAT_INSTRUCTIONS}.
Briefly explain your answer, then put the final answer in the properly-formatted JSON dictionary.
"""
@@ -172,6 +163,26 @@ def EXHIBIT_LEVEL_INSTRUCTION() -> str:
)
def DYNAMIC_PRIMARY_HEADER(context, field_name, field_prompt):
return f"""Extract attributes from the provided text, paying close attention to detail.
[ATTRIBUTE TO EXTRACT]
Here is the attribute, and instructions on how to correctly answer:
{field_name} : {field_prompt}
[GENERAL INSTRUCTIONS]
- Return all valid, explicitly stated values. If multiple values are found, list them separated by commas.
- If none of the valid values appear explicitly in the text, return N/A.
- Do not make assumptions. Only write an answer if it is clearly and explicitly stated.
Here is the text to analyze:
{context.replace('"', "'")}
Briefly explain your answer before putting the final answer in |pipes|.
"""
def DYNAMIC_PRIMARY_TEXT(context, field_name, field_prompt):
return f"""Extract attributes from the provided text, paying close attention to detail.
@@ -189,14 +200,14 @@ Here is the text to analyze:
{context.replace('"', "'")}
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer before putting the final answer in |pipes|.
"""
def DYNAMIC_PRIMARY_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Extract attribute values for dynamic fields from contract text. Use exact values and return in JSON list format."
"Extract attribute values for dynamic fields from contract text. Use exact values and return in pipes where instructed."
)
@@ -349,7 +360,7 @@ Here is the definition and valid values. Use this information to help you find t
{field_name} : {field_prompt}
[OUTPUT FORMAT]
Briefly explain your reasoning, then return a valid, properly formatted JSON dictionary where the key is the field name `{field_name}` and the value is always a list of strings. If there is one correct answer, return it as a single-element list. If there are multiple correct answers, return them as a comma-separated list within the array. If there is no correct answer, return "N/A". Follow `{JSON_DICT_FORMAT_INSTRUCTIONS}` exactly and do not include any extra keys or non-JSON text in the final output.
Briefly explain your answer, return a properly-formatted JSON object where the key is the field name `{field_name}` and the value is the correct answer. If there is more than one correct answer, return them in a comma-separated list. If there is no correct answer, return 'N/A'
"""
@@ -435,7 +446,7 @@ DECISION LOGIC:
- "All services in this exhibit" → applies to any service
- "Inpatient services only" → does NOT apply to outpatient services
If nothing applies → return ["N/A"]
If nothing applies → return N/A
3. COMBINE & FORMAT
@@ -445,19 +456,19 @@ DECISION LOGIC:
- Join multiple terms: "lesser of X, Y, or Z"
OUTPUT FORMAT:
["[SERVICE] paid at [combined statement]"]
|[SERVICE] paid at [combined statement]|
Examples:
- ["Lab Test paid at lesser of $100, billed charges, or 110% Medicare"]
- ["Radiology paid at $250 or charges"]
- ['N/A']
- |Lab Test paid at lesser of $100, billed charges, or 110% Medicare|
- |Radiology paid at $250 or charges|
- |N/A|
Example — Intra-exhibit statement references another exhibit → N/A:
- METHODOLOGY: "per visit basis at payment rates established in Attachment A"
- CROSS-EXHIBIT TEMPLATES: None
- EXHIBIT TEXT contains: "lesser of payment rates established in Attachment A or 100% of Medicare"
- Analysis: The only lesser-of statement found references "Attachment A" → IGNORE IT
- Result: ['N/A'] (no applicable statements remain)
- Result: |N/A| (no applicable statements remain)
RETURN N/A WHEN:
- No applicable lesser-of statements found
@@ -501,13 +512,13 @@ EXHIBIT TEXT:
{exhibit_text}
---
[OUTPUT FORMAT]
{JSON_LIST_FORMAT_INSTRUCTIONS}
Analyze the inputs above. Briefly explain your reasoning, then provide your final answer in pipes.
"""
def LESSER_OF_CHECK_INSTRUCTION() -> str:
return f"""Classify lesser-of payment statements from healthcare contracts.
return """Classify lesser-of payment statements from healthcare contracts.
**IMPORTANT: Check BOTH Service AND Reimbursement fields for exhibit references.**
@@ -552,10 +563,7 @@ Service: "MRI"
Reimbursement: "lesser of $1000 or Medicare"
→ STANDALONE
**CRITICAL: If you see ANY exhibit/attachment/schedule name in EITHER the Service OR Reimbursement field, it CANNOT be GLOBAL - it must be EXHIBIT_SPECIFIC (OTHER).**
Follow {JSON_DICT_FORMAT_INSTRUCTIONS} for output.
"""
**CRITICAL: If you see ANY exhibit/attachment/schedule name in EITHER the Service OR Reimbursement field, it CANNOT be GLOBAL - it must be EXHIBIT_SPECIFIC (OTHER).**"""
def LESSER_OF_CHECK(service_term: str, reimb_term: str, exhibit_title: str):
@@ -567,23 +575,23 @@ def LESSER_OF_CHECK(service_term: str, reimb_term: str, exhibit_title: str):
**Service:** {service_term}
**Reimbursement:** {reimb_term}
**Output JSON instructions:**
**Output JSON in |pipes|:**
{{
"service_term": "{service_term}",
"reimb_term": "{reimb_term}",
"scope": ["GLOBAL"] or ["EXHIBIT_SPECIFIC"] or ["STANDALONE"],
"target_exhibit": ["ALL"] or ["CURRENT"] or ["OTHER"] or [null],
"scope": "GLOBAL" | "EXHIBIT_SPECIFIC" | "STANDALONE",
"target_exhibit": "ALL" | "CURRENT" | "OTHER" | null,
"exhibit_reference": "<exhibit name if OTHER, else null>"
}}
{JSON_DICT_FORMAT_INSTRUCTIONS}."""
Briefly explain, then return JSON in |pipes|."""
def EXHIBIT_TITLE_MATCH_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Determine if two exhibit references refer to the same exhibit. Consider exact, partial, and descriptive matches.\n"
"Return YES or NO in JSON LIST format. "
"Return YES or NO in pipes."
)
@@ -605,9 +613,9 @@ Determine if these refer to the same exhibit. Consider:
- Different services: "Inpatient Services" does NOT match "Outpatient Services"
[OUTPUT FORMAT]
Return ONLY "YES" or "NO" in JSON LIST format.
Return ONLY "YES" or "NO" in |pipes|.
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your reasoning, then return your answer in |pipes|.
"""
@@ -738,7 +746,7 @@ Here is the term to analyze:
{term}
[OUTPUT FORMAT]
Follow {JSON_DICT_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer, then put your final answer in JSON dictionary format.
"""
@@ -868,7 +876,7 @@ Here is the Service and Methodology to analyze:
Service: {service.replace('"', "'")}
Methodology: {methodology.replace('"', "'")}
Follow {JSON_DICT_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer before putting the final answer in JSON dictionary format.
"""
@@ -893,7 +901,7 @@ Here is the Service Term to analyze:
{service.replace('"', "'")}
[OUTPUT FORMAT]
Explain your answer in 1-2 sentences. Write your final answer in JSON list format. Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output format instructions.
Explain your answer in 1-2 sentences. Write your final answer in JSON list format.
"""
@@ -911,7 +919,7 @@ def CODE_IMPLICIT_SPECIAL(service):
Here is the Service to analyze:
{service}
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer, but put your final answer in |pipes|
"""
@@ -956,7 +964,7 @@ If the term is a generic medical service, not a service at all, the name of a ho
Here is the service to analyze: {service}
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer, then put your final answer in |pipes|.
"""
@@ -1002,7 +1010,7 @@ Here is the Service Term to analyze:
{service.replace('"', "'")}
{reimb_term_section}{exhibit_context_section}
[OUTPUT FORMAT]
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer, then return your final answer in JSON list format.
"""
@@ -1011,21 +1019,6 @@ Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
#####################################################################################
def GLOBAL_LESSER_OF_STATEMENT_TEMPLATE(
context: str,
field_prompt: dict[str, str],
):
return f"""The following is a contract (or excerpt thereof).
Answer the following question: {field_prompt} with field name 'GLOBAL_LESSER_OF_STATEMENT'.
## START CONTRACT TEXT ##
{context}
## END CONTRACT TEXT ##
Briefly explain your answer, then put the final answer in a properly-formatted JSON dictionary, where the key is the field name and value is the answer. If there is no valid answer, return "N/A".
"""
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context: str, field_prompt: dict[str, str]):
return f"""The following is a contract (or excerpt thereof).
Answer the following question: {field_prompt}
@@ -1041,7 +1034,7 @@ Briefly explain your answer, then put the final answer in a properly-formatted J
def ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Answer a single, deterministic question about the contract text. Return result in JSON dictionary format as instructed.\n\n"
"Answer a single, deterministic question about the contract text. Return result in pipes as instructed."
)
@@ -1053,11 +1046,12 @@ IMPORTANT DISTINCTION:
- Providers: Physicians, physician groups, hospitals, clinics who DELIVER healthcare
- Payers: Insurance companies, health plans who PAY for healthcare services
CRITICAL: If a provider has multiple TINs or NPIs, include ALL of them in a JSON List in the same provider record. Do NOT create separate records for the same provider entity.
CRITICAL: If a provider has multiple TINs or NPIs, include ALL of them separated by pipes (|) in the same provider record. Do NOT create separate records for the same provider entity.
Examples:
- If you see "NPI: 1083087266 and 1396117412", return "NPI": ["1083087266", "1396117412"]
- If you see "Tax ID: 12-3456789, also 98-7654321", return "TIN": ["123456789", "987654321"]
- If you see "NPI: 1083087266 and 1396117412", return "NPI": "1083087266|1396117412"
- If you see "Tax ID: 12-3456789, also 98-7654321", return "TIN": "123456789|987654321"
Briefly explain your reasoning. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output.
[EXTRACTION AND FORMATTING INSTRUCTIONS]
@@ -1081,19 +1075,19 @@ Example of correct format:
FINAL PROVIDER INFO:
[
{
"TIN": ["123456789"],
"NPI": ["1234567890"],
"NAME": ["Main Provider Group"]
"TIN": "123456789",
"NPI": "1234567890",
"NAME": "Main Provider Group"
},
{
"TIN": ["987654321"],
"TIN": "987654321",
"NPI": null,
"NAME": ["Dr. John Smith"],
"NAME": "Dr. John Smith",
},
{
"TIN": null,
"NPI": ["1234567890"],
"NAME": ["Jane Doe, CRNA"]
"NPI": "1234567890",
"NAME": "Jane Doe, CRNA"
}
]
@@ -1126,9 +1120,9 @@ def FULL_CONTEXT_CLAIM_TYPES_ADDITIONAL_INSTRUCTION():
def FULL_CONTEXT_ADDITIONAL_INSTRUCTIONS(allow_na):
if allow_na:
return " Only provide an answer if that answer clearly applies to the ENTIRE contract. Otherwise, answer N/A. If there are multiple correct answers, return them in properly formatted JSON list."
return " Only provide an answer if that answer clearly applies to the ENTIRE contract. Otherwise, answer N/A. If there are multiple correct answers, return them in a pipe (|) separated list."
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 properly formatted JSON list."
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 FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION():
@@ -1140,12 +1134,12 @@ def CHECK_PROVIDER_NAME_MATCH_INSTRUCTION() -> str:
"[OBJECTIVE]\n"
"Determine if two provider names refer to the same healthcare organization.\n"
"Consider legal suffixes, D/B/A variations, acronyms, and common word variations.\n"
"Return Y or N in JSON LIST format."
"Return Y or N in pipes."
)
def CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name: str, hybrid_smart_chunking_provider_group_name: str | list
provider_name: str, hybrid_smart_chunking_provider_group_name: str
) -> str:
"""
Create prompt for LLM-based provider name matching.
@@ -1156,11 +1150,11 @@ def CHECK_PROVIDER_NAME_MATCH_PROMPT(
PROVIDER NAME A: {provider_name}
PROVIDER NAME B: {hybrid_smart_chunking_provider_group_name}
Note: PROVIDER NAME B is typically a single name but may contain multiple names as a JSON LIST. Determine if PROVIDER NAME A matches PROVIDER NAME B (or ANY name if multiple are present).
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).
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, check each name individually.
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").
@@ -1171,7 +1165,7 @@ CRITICAL - DO NOT MATCH:
- 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 properly formatted JSON list. Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output format instructions."""
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]):
@@ -1205,7 +1199,7 @@ The term length appears in two places but I'm selecting...
def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Answer multiple deterministic questions about contract text. Return results in JSON dictionary format as instructed.\n\n"
"Answer multiple deterministic questions about contract text. Follow JSON-only output rules in the prompt."
)
@@ -1238,7 +1232,7 @@ def EXHIBIT_HEADER_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Extract exhibit/attachment header identifiers from contract page excerpts.\n"
"Return full header text in JSON LIST format, or N/A if no header found."
"Return full header text in pipes, or N/A if no header found."
)
@@ -1269,14 +1263,14 @@ Rules for Inclusion:
* Do NOT include docusign IDs and other metadata
* DO NOT include unrelated text written in lowercase or title case.
If there is no Exhibit Header present, return ["N/A"]. Do not fill in values from the examples above, these are only examples. Exhibit header is only what is present in the context and only one header string value should be returned.
If there is no Exhibit Header present, return |N/A|. Do not fill in values from the examples above, these are only examples.
Here is the text to analyze:
{context.replace('"', "'")}
Before returning any output, ensure that all instructions for inclusion were followed.
- Output Format:
Follow {JSON_LIST_FORMAT_INSTRUCTIONS}.
Enclose your final answer in |pipes|, followed by a brief explanation.
"""
@@ -1284,7 +1278,7 @@ def EXHIBIT_LINKAGE_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Determine if two exhibit headers are meaningfully different from each other.\n"
'Return ["Y"] if different, ["N"] if same or continuation.'
"Return Y if different, N if same or continuation. Answer in pipes."
)
@@ -1300,7 +1294,7 @@ def EXHIBIT_LINKAGE(header1, header2):
* If the headers have the same numbers, but different descriptors, they are NOT meaningfully different. For example, "Attachment A" and "Attachment A: Professional" are NOT meaningfully different because they are both attachment A.
* If one header is a continuation of the other (e.g. "Continued", "CONT'D"), they are NOT meaningfully different.
* If Header 1 directly references Header 2 in a way that links the two, then they are NOT meaningfully different. This linkage also applies if Header 2 references Header 1.
* Return ["Y"] if the two Headers are meaningfully different. Return ["N"] if they are not.
* Return |Y| if the two Headers are meaningfully different. Return |N| if they are not.
[CONTEXT]
Header 1:
@@ -1309,8 +1303,7 @@ Header 1:
Header 2:
"{header2}"
[OUTPUT FORMAT]
Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output.
Briefly explain your answer, then enclose your final answer in |pipes|.
"""
@@ -1373,7 +1366,7 @@ Here are the SERVICE TERM and REIMBURSEMENT TERM to analyze:
SERVICE TERM: {service_term}
REIMBURSEMENT TERM: {reimb_term}
Briefly explain your reasoning, then return YES or NO in properly formatted JSON list."""
Briefly explain your reasoning, then return YES or NO in |pipes|."""
def VALIDATE_REIMBURSEMENTS_INSTRUCTION() -> str:
@@ -1399,28 +1392,29 @@ Rules:
4. Return N/A if: missing/ambiguous month or year components, or invalid dates. If just the day is missing, assume it's the first of the month.
5. For ambiguous input cases, assume American format (MM/DD/YYYY) and apply YYYY/MM/DD formatting
6. Any day values with spaces or leading zeros (e.g., '0 1' or '01') are interpreted as single-digit days
7. Return final answer in |pipes|
Examples:
Input: "7th day of February, 2002"
Output: ["2002/02/07"]
Output: |2002/02/07|
Input: "Sep 20, 2022"
Output: ["2022/09/20"]
Output: |2022/09/20|
Input: "3 years"
Output: ["3 years"]
Output: |3 years|
Input: "FEB 0 1 2020"
Output: ["2020/02/01"]
Output: |2020/02/01|
Input: "8.1.10" # (assumed MM.DD.YYYY)
Output: ["2010/08/01"]
Output: |2010/08/01|
Input: "March 2014" # (missing day, assume 1st)
Output: ["2014/03/01"]
Output: |2014/03/01|
Input: "Feb 26, 2_0_" # (incomplete year)
Output: ["N/A"]
Output: |N/A|
Please analyze this date:
@@ -1428,7 +1422,7 @@ Please analyze this date:
{context}
## END CONTEXT ##
{JSON_LIST_FORMAT_INSTRUCTIONS}
{PIPE_FORMAT_INSTRUCTIONS}
"""
@@ -1436,7 +1430,7 @@ def DATE_FIX_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Convert date strings to YYYY/MM/DD format. Handle durations, ambiguous dates, and American format assumptions.\n"
"Return result in JSON LIST format."
"Return result in pipes."
)
@@ -1465,14 +1459,14 @@ Reimbursement Term: {reimb_term.replace('"', "'")}
- Always return one value from the combined list of cases above; do not return 'N/A' or 'UNKNOWN'.
[OUTPUT FORMAT]
Briefly explain your answer (including why it was one of carveout cases or special cases), then enclose your final answer in properly formatted JSON list. Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output format instructions.
Briefly explain your answer (including why it was one of carveout cases or special cases), then enclose your final answer in |pipes|.
"""
def CARVEOUT_CHECK_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Determine if a carve-out applies for a service given a base reimbursement. Return result in JSON list format."
"Determine if a carve-out applies for a service given a base reimbursement. Return result in pipes as instructed."
)
@@ -1488,7 +1482,7 @@ If it is unclear, write 'N/A'.
Here is the Service:
{service_term}
Explain your answer in one or two sentences, but put your final answer in a properly formatted JSON list. Follow {JSON_LIST_FORMAT_INSTRUCTIONS} for output format instructions.
Explain your answer in one or two sentences, but put your final answer in |pipes|.
"""
@@ -1549,13 +1543,6 @@ Return the dates in the following JSON format:
Note: Use "N/A" for dates that cannot be determined or extracted from the input text."""
def SPLIT_REIMBURSEMENTS_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Split a compound reimbursement term into distinct components using only explicit contract text."
)
def LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values):
return f"""Analyze the exhibit from a Payer-Provider contract.
@@ -1581,27 +1568,12 @@ Here is the Exhibit to be analyzed:
{exhibit_text.replace('"', "'")}
[OUTPUT FORMAT - REQUIRED]
Follow these output rules strictly:
- {JSON_LIST_FORMAT_INSTRUCTIONS}
You MUST format your response with pipe delimiters. Example format:
"Based on the analysis... |Inclusive|" or "The relationship is... |Exclusive|"
Your final answer MUST be enclosed in pipe delimiters: |Inclusive| or |Exclusive| AND you must make sure that the pipe delimiters are only used around the terms Inclusive or Exclusive.
"""
def DERIVED_TERM_DATE_PROMPT(
effective_date: str = None, termination_information: str = None
) -> str:
"""
Generates a prompt to extract the termination date from a legal contract.
Args:
effective_date (str, optional): The effective date of the contract. Defaults to None.
termination_information (str, optional): Information regarding the termination of the contract. Defaults to None.
Returns:
str: A formatted string prompt requesting the termination date of the contract.
"""
return f"The following are excerpts from a legal contract related to effective and termination dates. Determine the initial termination date of the contract, excluding any renewals or extensions.\nIf termination information comes in as a duration, derive the termination date exclusive of the final day. For example, if effective date is 2023/02/01 and termination information is 'two years', termination date would be 2025/01/31 and not 2025/02/01. If effective date is 2023/02/01 and termination information is 'year-to-year', termination date would be 2024/01/31.\nIf the termination information comes in as a date, return that date. For example, if effective date is 2023/02/01 and termination information is 2024/07/01, return 2024/07/01.\nEffective Date: {effective_date}\nTermination Information: {termination_information}\nOnly provide the first termination date, ignoring any automatic or optional renewals. Provide the date in YYYY/MM/DD format only. \nFollow {JSON_LIST_FORMAT_INSTRUCTIONS} for the output format instructions."
def LOB_RELATIONSHIP_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
@@ -1609,15 +1581,34 @@ def LOB_RELATIONSHIP_INSTRUCTION() -> str:
"[OUTPUT FORMAT - CRITICAL]\n"
"You MUST format your response as follows:\n"
"1. Briefly explain your reasoning\n"
"2. Then follow {JSON_LIST_FORMAT_INSTRUCTIONS} for final answer format."
"2. Then put your final answer in pipe delimiters: |Inclusive| or |Exclusive|\n"
"Example: 'The contract shows... |Inclusive|'\n"
"Your response MUST include the pipe delimiters (|) around your final answer."
)
def DERIVED_TERM_DATE_PROMPT(
effective_date: str = None, termination_information: str = None
) -> str:
"""
Generates a prompt to extract the termination date from a legal contract.
Call DERIVED_TERM_DATE_INSTRUCTION() separately for the cached instruction.
Args:
effective_date (str, optional): The effective date of the contract. Defaults to None.
termination_information (str, optional): Information regarding the termination of the contract. Defaults to None.
Returns:
str: A formatted string prompt requesting the termination date of the contract.
"""
return f"The following are excerpts from a legal contract related to effective and termination dates. Determine the initial termination date of the contract, excluding any renewals or extensions.\nIf termination information comes in as a duration, derive the termination date exclusive of the final day. For example, if effective date is 2023/02/01 and termination information is 'two years', termination date would be 2025/01/31 and not 2025/02/01. If effective date is 2023/02/01 and termination information is 'year-to-year', termination date would be 2024/01/31.\nIf the termination information comes in as a date, return that date. For example, if effective date is 2023/02/01 and termination information is 2024/07/01, return 2024/07/01.\nEffective Date: {effective_date}\nTermination Information: {termination_information}\nOnly provide the first termination date, ignoring any automatic or optional renewals. Provide the date in YYYY/MM/DD format only. \n{PIPE_FORMAT_INSTRUCTIONS}"
def DERIVED_TERM_DATE_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Calculate contract termination date from effective date and term duration.\n"
"Return date in YYYY/MM/DD format in JSON LIST format."
"Return date in YYYY/MM/DD format in pipes."
)
@@ -1644,7 +1635,7 @@ Reimbursement: {reimbursement_dict["REIMB_TERM"]}
Return the integer of the {special_case_term} that is most associated with the REIMBURSEMENT TERM. Base your answer on context clues found in the text.
{JSON_LIST_FORMAT_INSTRUCTIONS}
{PIPE_FORMAT_INSTRUCTIONS}
"""
@@ -1652,5 +1643,5 @@ def SPECIAL_CASE_ASSIGNMENT_INSTRUCTION() -> str:
return (
"[OBJECTIVE]\n"
"Determine which special case term value is associated with a reimbursement term based on contract context.\n"
"Return the integer index of the matching term in JSON LIST format."
"Return the integer index of the matching term in pipes."
)
src/qc_qa/pipeline.py
+5 -34
View File
@@ -16,7 +16,7 @@ import os
import re
import warnings
from typing import Optional
import json
import pandas as pd
from src import config
@@ -234,25 +234,11 @@ def run_qc_qa_pipeline(
df[tin] = out.str[0]
def is_all_valid_tin(cell):
if cell is None:
return True
if isinstance(cell, (list, tuple)):
tokens = list(cell)
else:
if pd.isna(cell) or str(cell).strip() == "":
return True
try:
tokens = json.loads(cell)
if not isinstance(tokens, list):
return False
except Exception:
return False
tokens = [t for t in str(cell).split("|") if t.strip()]
if not tokens:
return True
for tok in tokens:
if not re.fullmatch(r"\d{9}", str(tok)):
if not re.fullmatch(r"\d{9}", tok):
return False
return True
@@ -269,21 +255,7 @@ def run_qc_qa_pipeline(
if npi in df.columns:
def clean_npi_cell(cell):
if cell is None:
return "[]"
if isinstance(cell, (list, tuple)):
tokens = list(cell)
else:
if pd.isna(cell) or str(cell).strip() == "":
return "[]"
try:
tokens = json.loads(cell)
if not isinstance(tokens, list):
return "[]"
except Exception:
return "[]"
tokens = re.split(r"[|,]", str(cell))
clean_npies = []
for tok in tokens:
tok = qc.safe_token_to_str(tok)
@@ -292,8 +264,7 @@ def run_qc_qa_pipeline(
digits = re.sub(r"\D", "", tok)
if len(digits) == 10:
clean_npies.append(digits)
return json.dumps(clean_npies)
return "|".join(clean_npies)
df[npi] = df[npi].apply(clean_npi_cell)
src/tests/test_code_funcs.py
+12 -12
View File
@@ -119,12 +119,12 @@ class TestCodeFuncs(unittest.TestCase):
self.assertIn("Drug B", result["PROCEDURE_CD_DESC"])
@patch("src.utils.llm_utils.invoke_claude")
@patch("src.utils.string_utils.universal_json_load")
@patch("src.utils.string_utils.extract_text_from_delimiters")
def test_code_implicit_special(self, mock_extract, mock_invoke_claude):
"""Test the code_implicit_special function for special case handling."""
# Setup mocks
mock_invoke_claude.return_value = "mock_response"
mock_extract.return_value = ["Drugs"]
mock_extract.return_value = "Drugs"
# Test drug category - returns a list
result = code_funcs.code_implicit_special("DRUG SERVICE", "test.pdf")
@@ -132,7 +132,7 @@ class TestCodeFuncs(unittest.TestCase):
self.assertEqual(result["PROCEDURE_CD_DESC"], "Drugs")
# Test no match
mock_extract.return_value = [""]
mock_extract.return_value = ""
result = code_funcs.code_implicit_special("OTHER SERVICE", "test.pdf")
self.assertEqual(result, {})
@@ -236,19 +236,19 @@ class TestCodeFuncs(unittest.TestCase):
self.assertIn("CODE_METHODOLOGY", result)
@patch("src.utils.llm_utils.invoke_claude")
@patch("src.utils.string_utils.universal_json_load")
@patch("src.utils.string_utils.extract_text_from_delimiters")
def test_code_last_check(self, mock_extract, mock_invoke_claude):
"""Test the code_last_check function for categorizing non-matched services."""
# Setup mocks
mock_invoke_claude.return_value = "mock_response"
# Test specific case
mock_extract.return_value = ["Specific"]
mock_extract.return_value = "Specific"
result = code_funcs.code_last_check("SPECIAL SERVICE", "test.pdf")
self.assertEqual(result, "Specific")
# Test generic case
mock_extract.return_value = ["Generic"]
mock_extract.return_value = "Generic"
result = code_funcs.code_last_check("GENERIC SERVICE", "test.pdf")
self.assertEqual(result, "Generic")
@@ -269,8 +269,8 @@ class TestCodeFuncs(unittest.TestCase):
self.constants.BILL_TYPE_REVERSE_MAPPING,
)
self.assertEqual(result["BILL_TYPE_CD"], ["11X"])
self.assertEqual(result["BILL_TYPE_CD_DESC"], ["Inpatient Hospital"])
self.assertEqual(result["BILL_TYPE_CD"], "11X")
self.assertEqual(result["BILL_TYPE_CD_DESC"], "Inpatient Hospital")
# Test with empty response
mock_json_load.return_value = []
@@ -403,13 +403,13 @@ class TestCodeFuncs(unittest.TestCase):
"""Test the fill_claim_type function for determining claim types."""
# Test with existing claim types
answer_dicts = [
{"AARETE_DERIVED_CLAIM_TYPE_CD": ["H"]},
{"AARETE_DERIVED_CLAIM_TYPE_CD": ["H"]},
{"AARETE_DERIVED_CLAIM_TYPE_CD": ["M"]},
{"AARETE_DERIVED_CLAIM_TYPE_CD": "H"},
{"AARETE_DERIVED_CLAIM_TYPE_CD": "H"},
{"AARETE_DERIVED_CLAIM_TYPE_CD": "M"},
{},
]
result = code_funcs.fill_claim_type(answer_dicts)
self.assertEqual(result[3]["AARETE_DERIVED_CLAIM_TYPE_CD"], ["H"])
self.assertEqual(result[3]["AARETE_DERIVED_CLAIM_TYPE_CD"], "H")
# Test with all empty claim types
answer_dicts = [{}, {}]
src/tests/test_crosswalk_utils.py
+1 -1
View File
@@ -140,7 +140,7 @@ def test_create_reverse_mapping():
builder.from_dict({"A": "1", "B": "1", "C": "2", "D": "2", "E": "3"})
reversed_mapping = builder.create_reverse_mapping()
assert reversed_mapping == {"1": ["A", "B"], "2": ["C", "D"], "3": ["E"]}
assert reversed_mapping == {"1": "A|B", "2": "C|D", "3": "E"}
# Defensive programming tests
src/tests/test_page_funcs.py
+5 -6
View File
@@ -299,9 +299,8 @@ class TestParseTableRows:
rows, num_columns = parse_table_rows(table_text, "1")
assert len(rows) == 3
assert num_columns == 2
assert rows[0] == ["Service", "Payment"]
assert rows[1] == ["Item 1", "$100"]
assert rows[2] == ["Item 2", "$200"]
assert "Service | Payment" in rows[0]
assert "Item 1 | $100" in rows[1]
def test_parse_unexpected_format_warning(self, caplog):
"""Test that unexpected format logs warning and returns empty."""
@@ -329,7 +328,7 @@ class TestParseTableRows:
# Should successfully parse the list structure
assert len(rows) == 1
assert num_columns == 2
assert rows[0] == ["Service", "Payment"]
assert "Service | Payment" in rows[0]
def test_parse_empty_table(self):
"""Test parsing empty table."""
@@ -344,8 +343,8 @@ class TestParseTableRows:
rows, num_columns = parse_table_rows(table_text, "1")
assert len(rows) == 2
assert num_columns == 2
assert rows[0] == ["Service", "Rate"]
assert rows[1] == ["Item 1", "50%"]
assert "Service | Rate" in rows[0]
assert "Item 1 | 50%" in rows[1]
def test_parse_table_many_columns(self):
"""Test parsing table with many columns."""
src/tests/test_postprocess.py
+57 -82
View File
@@ -38,7 +38,7 @@ class TestPostprocessFunctions(unittest.TestCase):
def test_remove_hyphens(self):
self.assertEqual(remove_hyphens("123-45-6789"), "123456789")
self.assertEqual(remove_hyphens("123-456"), "123456")
self.assertEqual(remove_hyphens(None), None)
self.assertEqual(remove_hyphens(None), "")
self.assertEqual(remove_hyphens(""), "")
def test_flatten_singleton_string_list(self):
@@ -297,23 +297,23 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 1: Basic GROUP removal
input_df1 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["123456789"]],
"PROV_GROUP_NPI": [["1234567890"]],
"PROV_GROUP_NAME_FULL": [["Main Hospital"]],
"PROV_OTHER_TIN": [["123456789", "987654321"]],
"PROV_OTHER_NPI": [["1234567890", "0987654321"]],
"PROV_OTHER_NAME_FULL": [["Main Hospital", "Other Clinic"]],
"PROV_GROUP_TIN": ["123456789"],
"PROV_GROUP_NPI": ["1234567890"],
"PROV_GROUP_NAME_FULL": ["Main Hospital"],
"PROV_OTHER_TIN": ["123456789|987654321"],
"PROV_OTHER_NPI": ["1234567890|0987654321"],
"PROV_OTHER_NAME_FULL": ["Main Hospital|Other Clinic"],
}
)
expected_df1 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["123456789"]],
"PROV_GROUP_NPI": [["1234567890"]],
"PROV_GROUP_NAME_FULL": [["Main Hospital"]],
"PROV_OTHER_TIN": [["987654321"]],
"PROV_OTHER_NPI": [["0987654321"]],
"PROV_OTHER_NAME_FULL": [["Other Clinic"]],
"PROV_GROUP_TIN": ["123456789"],
"PROV_GROUP_NPI": ["1234567890"],
"PROV_GROUP_NAME_FULL": ["Main Hospital"],
"PROV_OTHER_TIN": ["987654321"],
"PROV_OTHER_NPI": ["0987654321"],
"PROV_OTHER_NAME_FULL": ["Other Clinic"],
}
)
@@ -323,46 +323,27 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 2: Internal deduplication (your original example)
input_df2 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["061798267"]],
"PROV_GROUP_NPI": [["1111111111"]],
"PROV_GROUP_NAME_FULL": [["Group Practice"]],
"PROV_GROUP_TIN": ["061798267"],
"PROV_GROUP_NPI": ["1111111111"],
"PROV_GROUP_NAME_FULL": ["Group Practice"],
"PROV_OTHER_TIN": [
[
"061798267",
"061798267",
"UNKNOWN",
"UNKNOWN",
"061992277",
"061798267",
]
],
"PROV_OTHER_NPI": [
[
"1111111111",
"2222222222",
"2222222222",
"UNKNOWN",
]
"061798267|061798267|UNKNOWN|UNKNOWN|061992277|061798267"
],
"PROV_OTHER_NPI": ["1111111111|2222222222|2222222222|UNKNOWN"],
"PROV_OTHER_NAME_FULL": [
[
"Group Practice",
"Other Practice",
"Other Practice",
"UNKNOWN",
]
"Group Practice|Other Practice|Other Practice|UNKNOWN"
],
}
)
expected_df2 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["061798267"]],
"PROV_GROUP_NPI": [["1111111111"]],
"PROV_GROUP_NAME_FULL": [["Group Practice"]],
"PROV_OTHER_TIN": [["061992277"]],
"PROV_OTHER_NPI": [["2222222222"]],
"PROV_OTHER_NAME_FULL": [["Other Practice"]],
"PROV_GROUP_TIN": ["061798267"],
"PROV_GROUP_NPI": ["1111111111"],
"PROV_GROUP_NAME_FULL": ["Group Practice"],
"PROV_OTHER_TIN": ["061992277"],
"PROV_OTHER_NPI": ["2222222222"],
"PROV_OTHER_NAME_FULL": ["Other Practice"],
}
)
@@ -372,23 +353,23 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 3: Empty OTHER fields after deduplication
input_df3 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["123456789"]],
"PROV_GROUP_NPI": [["1234567890"]],
"PROV_GROUP_NAME_FULL": [["Main Hospital"]],
"PROV_OTHER_TIN": [["123456789", "123456789", "UNKNOWN"]],
"PROV_OTHER_NPI": [["1234567890", "UNKNOWN", "UNKNOWN"]],
"PROV_OTHER_NAME_FULL": [["Main Hospital", "UNKNOWN"]],
"PROV_GROUP_TIN": ["123456789"],
"PROV_GROUP_NPI": ["1234567890"],
"PROV_GROUP_NAME_FULL": ["Main Hospital"],
"PROV_OTHER_TIN": ["123456789|123456789|UNKNOWN"],
"PROV_OTHER_NPI": ["1234567890|UNKNOWN|UNKNOWN"],
"PROV_OTHER_NAME_FULL": ["Main Hospital|UNKNOWN"],
}
)
expected_df3 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["123456789"]],
"PROV_GROUP_NPI": [["1234567890"]],
"PROV_GROUP_NAME_FULL": [["Main Hospital"]],
"PROV_OTHER_TIN": [[]],
"PROV_OTHER_NPI": [[]],
"PROV_OTHER_NAME_FULL": [[]],
"PROV_GROUP_TIN": ["123456789"],
"PROV_GROUP_NPI": ["1234567890"],
"PROV_GROUP_NAME_FULL": ["Main Hospital"],
"PROV_OTHER_TIN": [""],
"PROV_OTHER_NPI": [""],
"PROV_OTHER_NAME_FULL": [""],
}
)
@@ -398,32 +379,26 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 4: Multiple rows
input_df4 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["111111111"], ["222222222"]],
"PROV_GROUP_NPI": [["1111111111"], ["2222222222"]],
"PROV_GROUP_NAME_FULL": [["Hospital A"], ["Hospital B"]],
"PROV_GROUP_TIN": ["111111111", "222222222"],
"PROV_GROUP_NPI": ["1111111111", "2222222222"],
"PROV_GROUP_NAME_FULL": ["Hospital A", "Hospital B"],
"PROV_OTHER_TIN": [
["111111111", "333333333", "444444444"],
["222222222", "444444444"],
],
"PROV_OTHER_NPI": [
["3333333333", "1111111111", "4444444444"],
["2222222222", "4444444444"],
],
"PROV_OTHER_NAME_FULL": [
["Clinic C", "Hospital A", "Clinic D"],
["Hospital B", "Clinic D"],
"111111111|333333333",
"444444444|222222222|444444444",
],
"PROV_OTHER_NPI": ["3333333333|1111111111", "4444444444|2222222222"],
"PROV_OTHER_NAME_FULL": ["Clinic C|Hospital A", "Clinic D|Hospital B"],
}
)
expected_df4 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["111111111"], ["222222222"]],
"PROV_GROUP_NPI": [["1111111111"], ["2222222222"]],
"PROV_GROUP_NAME_FULL": [["Hospital A"], ["Hospital B"]],
"PROV_OTHER_TIN": [["333333333", "444444444"], ["444444444"]],
"PROV_OTHER_NPI": [["3333333333", "4444444444"], ["4444444444"]],
"PROV_OTHER_NAME_FULL": [["Clinic C", "Clinic D"], ["Clinic D"]],
"PROV_GROUP_TIN": ["111111111", "222222222"],
"PROV_GROUP_NPI": ["1111111111", "2222222222"],
"PROV_GROUP_NAME_FULL": ["Hospital A", "Hospital B"],
"PROV_OTHER_TIN": ["333333333", "444444444"],
"PROV_OTHER_NPI": ["3333333333", "4444444444"],
"PROV_OTHER_NAME_FULL": ["Clinic C", "Clinic D"],
}
)
@@ -432,7 +407,7 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 5: Missing columns - should return unchanged
input_df5 = pd.DataFrame(
{"PROV_GROUP_TIN": [["123456789"]], "OTHER_COLUMN": [["value1"]]}
{"PROV_GROUP_TIN": ["123456789"], "OTHER_COLUMN": ["value1"]}
)
result_df5 = deduplicate_provider_columns(input_df5)
@@ -446,12 +421,12 @@ class TestPostprocessFunctions(unittest.TestCase):
# Test case 7: Empty OTHER fields (already empty strings)
input_df7 = pd.DataFrame(
{
"PROV_GROUP_TIN": [["123456789"]],
"PROV_GROUP_NPI": [["1234567890"]],
"PROV_GROUP_NAME_FULL": [["Main Hospital"]],
"PROV_OTHER_TIN": [[]],
"PROV_OTHER_NPI": [[]],
"PROV_OTHER_NAME_FULL": [[]],
"PROV_GROUP_TIN": ["123456789"],
"PROV_GROUP_NPI": ["1234567890"],
"PROV_GROUP_NAME_FULL": ["Main Hospital"],
"PROV_OTHER_TIN": [""],
"PROV_OTHER_NPI": [""],
"PROV_OTHER_NAME_FULL": [""],
}
)
src/tests/test_qc_qa_pipeline.py
+2 -2
View File
@@ -882,9 +882,9 @@ class TestQAQCUtils:
# Tests for format_or_preserve_tins
def test_format_or_preserve_tins_valid(self):
"""Test formatting valid TINs (format: XX-XXXXXXX)."""
result, is_valid = format_or_preserve_tins(["12-3456789"])
result, is_valid = format_or_preserve_tins("12-3456789")
assert is_valid is True
assert result == ["12-3456789"]
assert "12-3456789" in str(result)
def test_format_or_preserve_tins_list(self):
"""Test formatting list of TINs."""
src/tests/test_tin_npi_funcs.py
+23 -23
View File
@@ -45,27 +45,27 @@ class TestTinNpiFuncs:
def test_clean_provider_info(self):
providers = [
{
"TIN": ["12-345.6789"],
"NPI": ["12.34567890"],
"NAME": ["Test Provider"],
"TIN": "12-345.6789",
"NPI": "12.34567890",
"NAME": "Test Provider",
"IS_GROUP": "Y",
},
{
"TIN": [""],
"NPI": ["invalid"],
"NAME": [""],
"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[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"]
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(
@@ -74,24 +74,24 @@ class TestTinNpiFuncs:
"""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(names, provider_name, filename):
return "Group Provider" in names
def name_match_side_effect(provider_name, name, filename):
return name == "Group Provider"
mock_name_match_check.side_effect = name_match_side_effect
one_to_one_results = {
"PROVIDER_NAME": ["Group Provider"],
"PROVIDER_NAME": "Group Provider",
"PROV_INFO_JSON": json.dumps(
[
{
"TIN": ["123456789"],
"NPI": ["1234567890"],
"NAME": ["Group Provider"],
"TIN": "123456789",
"NPI": "1234567890",
"NAME": "Group Provider",
},
{
"TIN": ["987654321"],
"NPI": ["9876543210"],
"NAME": ["Other Provider"],
"TIN": "987654321",
"NPI": "9876543210",
"NAME": "Other Provider",
},
]
),
@@ -107,8 +107,8 @@ class TestTinNpiFuncs:
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 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"]
src/utils/qa_qc_utils.py
+26 -42
View File
@@ -1092,34 +1092,28 @@ def safe_token_to_str(tok: Any) -> str:
def format_or_preserve_tins(val: Any) -> Tuple[str, bool]:
"""
Format TINs to 9 digits, preserving JSON-list multiple values.
Returns tuple of (formatted_list, any_invalid).
Format TINs to 9 digits, preserving pipe-delimited multiple values.
Args:
val: TIN value (may be pipe-delimited)
Returns:
Tuple of (formatted_tins, has_invalid)
"""
if pd.isna(val) or val is None or str(val).strip() == "":
return ("", True)
# ---- critical fix: NEVER pd.isna() on list/array ----
if val is None:
return ("[]", True)
if isinstance(val, (list, tuple)):
parts = list(val)
else:
if pd.isna(val) or str(val).strip() == "":
return ("[]", True)
try:
parts = json.loads(val)
if not isinstance(parts, list):
parts = []
except Exception:
parts = []
parts = [
part.strip()
for part in str(val).split("|")
if part is not None and part.strip() != ""
]
out, any_invalid = [], False
for part in parts:
part = safe_token_to_str(part)
if not part:
continue
if part is None:
part = ""
elif len(part) == 9 and part.isdigit():
out.append(part)
elif QC_DIGITS_9_RE.fullmatch(part):
@@ -1130,34 +1124,24 @@ def format_or_preserve_tins(val: Any) -> Tuple[str, bool]:
out.append(part)
any_invalid = True
return (out, any_invalid)
return ("|".join([x if x is not None else "" for x in out]), any_invalid)
def is_all_valid_npi(cell: Any) -> bool:
"""
Check if all JSON-list tokens in cell are valid 10-digit NPIs.
Check if all pipe-delimited tokens in cell are valid 10-digit NPIs.
Args:
cell: Cell value to validate
Returns:
True if all tokens are valid NPIs, or if blank
"""
if cell is None:
return True
if isinstance(cell, (list, tuple)):
tokens = list(cell)
else:
if pd.isna(cell) or str(cell).strip() == "":
return True
try:
tokens = json.loads(cell)
if not isinstance(tokens, list):
return False
except Exception:
return False
tokens = [t for t in str(cell).split("|") if t.strip()]
if not tokens:
return True
for tok in tokens:
if not re.fullmatch(r"\d{10}", str(tok)):
if not re.fullmatch(r"\d{10}", tok):
return False
return True