diff --git a/fieldExtraction/src/investment/code_funcs.py b/fieldExtraction/src/investment/code_funcs.py index f10728a..130ae98 100644 --- a/fieldExtraction/src/investment/code_funcs.py +++ b/fieldExtraction/src/investment/code_funcs.py @@ -9,7 +9,6 @@ import src.utils.embedding_utils as embedding_utils import pandas as pd import os import re -import string import ast import src.constants.investment_values as investment_values from crosswalk.crosswalk_utils import CrosswalkBuilder @@ -131,12 +130,9 @@ def find_containing_ranges(input_range: str, code_mapping: dict) -> list[dict]: if is_highest: highest_level_ranges.append(result['code_range']) - # if len(highest_level_ranges) == 0: - # highest_level_ranges = [f"{i}" for i in range(input_start, input_end + 1)] - return highest_level_ranges -def code_primary(answer_dicts, filename): +def code_primary(service, code_primary_questions, filename): """ Breaks down the direct code information from SERVICE_TERM and REIMB_TERM fields using an LLM prompt and adds the results to the provided answer dictionaries. @@ -148,22 +144,17 @@ def code_primary(answer_dicts, filename): Returns: list of dict: The input answer dictionaries with additional primary code breakout information. These are only the codes that are explicitly written in the Service/Methodology """ - code_primary_questions = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="code_primary_breakout").get_prompt_dict() - code_breakout_answers = [] - for answer_dict in answer_dicts: - service, methodology = answer_dict.get("SERVICE_TERM"), answer_dict.get("REIMB_TERM") - if service: - code_answer_dict = string_utils.universal_json_load( - llm_utils.invoke_claude( - investment_prompts.CODE_PRIMARY_BREAKOUT(service, methodology, code_primary_questions), + if not string_utils.is_empty(service): + claude_answer_raw = llm_utils.invoke_claude( + investment_prompts.CODE_PRIMARY_BREAKOUT(str(service), code_primary_questions), config.MODEL_ID_CLAUDE35_SONNET, filename ) - ) - else: - code_answer_dict = {} - code_breakout_answers.append({**answer_dict, **code_answer_dict}) - return code_breakout_answers + code_answer_dict = string_utils.universal_json_load(claude_answer_raw) + else: + code_answer_dict = {} + + return code_answer_dict def get_proc_crosswalk(): """ @@ -284,12 +275,12 @@ def get_best_match( return None, None, 0 # No valid match found -def code_indirect(answer_dicts: list[dict], all_dataset: dict) -> list[dict]: +def code_implicit(service, filename): """ Maps free-form service descriptions to the best-matching code and description. Caches results for efficiency and skips entries with existing codes or irrelevant text. - ONLY FOR PROC and DIAG CODES RN + ONLY FOR PROC CODES RN Args: answer_dicts (list): List of dictionaries containing 1:N answers @@ -298,41 +289,15 @@ def code_indirect(answer_dicts: list[dict], all_dataset: dict) -> list[dict]: Returns: list: Updated list of dictionaries with mapped CPT4 procedure codes and descriptions. """ - results = [] - already_seen = {} - for answer_dict in answer_dicts: - service = answer_dict.get("SERVICE_TERM", "") - proc_cd = answer_dict.get("CPT4_PROC_CD") - diag_cd = answer_dict.get("DIAG_CD") - - for code, code_value in zip(["proc_cd", "diag_cd"], [proc_cd, diag_cd]): - - if (not service # Service DNE - or (service in ["covered service", "covered services", "all covered services"]) # Is generic - or isinstance(code_value, list) # Is a list - or not string_utils.is_empty(code_value) # Is not empty - ): - pass # pass for now - we may need to add functionality here later - else: - if service in already_seen and code in already_seen[service]: - nearest_code = already_seen[service][code] - nearest_description = already_seen[service][code+'description'] - else: - nearest_code, nearest_description, score = get_best_match(code, all_dataset, service, model) - # print(service, " | ", description, " | ", code, " | ", score) - already_seen[service] = {} - already_seen[service][code] = nearest_code - already_seen[service][code+'description'] = nearest_description - - if code == 'proc_cd': - answer_dict["CPT4_PROC_CD"] = nearest_code - answer_dict["CPT4_PROC_CD_DESC"] = nearest_description - elif code == 'diag_cd': - answer_dict["DIAG_CD"] = nearest_code - answer_dict["DIAG_CD_DESC"] = nearest_description - - results.append(answer_dict) - return results + if not string_utils.is_empty(service): + claude_answer_raw = llm_utils.invoke_claude(investment_prompts.CODE_IMPLICIT(service), config.MODEL_ID_CLAUDE35_SONNET, filename) + code_answer_final = string_utils.universal_json_load(claude_answer_raw) + else: + code_answer_final = "" + + return code_answer_final + + def get_code_description(code, code_mapping): """ @@ -427,7 +392,6 @@ def get_mappings(): Returns: dict: A dictionary containing mappings for PROC, DIAG, Grouper, Rev, and Admit Type codes. """ - all_mappings = {} # PROC codes @@ -459,7 +423,7 @@ def get_mappings(): return all_mappings -def code_description_mapping(answer_dicts, all_mappings): +def code_description_mapping(answer_dict, all_mappings): """ Maps code descriptions to relevant fields in the answer dictionaries using provided mappings. @@ -470,30 +434,63 @@ def code_description_mapping(answer_dicts, all_mappings): Returns: list: Updated list of dictionaries with code descriptions added. """ - for answer_dict in answer_dicts: - proc_code = answer_dict.get("CPT4_PROC_CD") - proc_mod = answer_dict.get("CPT4_PROC_MOD") - diag_code = answer_dict.get("DIAG_CD") - rev_code = answer_dict.get("REVENUE_CD") - grouper_code = answer_dict.get("GROUPER_CD") - grouper_type = answer_dict.get("GROUPER_TYPE") - admit_type_code = answer_dict.get("CLAIM_ADMIT_TYPE_CD") + + proc_code = answer_dict.get("CPT4_PROC_CD") + proc_mod = answer_dict.get("CPT4_PROC_MOD") + diag_code = answer_dict.get("DIAG_CD") + rev_code = answer_dict.get("REVENUE_CD") + grouper_code = answer_dict.get("GROUPER_CD") + grouper_type = answer_dict.get("GROUPER_TYPE") + admit_type_code = answer_dict.get("CLAIM_ADMIT_TYPE_CD") + + answer_dict["CPT4_PROC_CD_DESC"] = get_code_description(proc_code, all_mappings['proc_mapping']) + answer_dict["CPT4_PROC_MOD_DESC"] = get_code_description(proc_mod, all_mappings['proc_mapping']) + answer_dict["DIAG_CD_DESC"] = get_code_description(diag_code, all_mappings['diag_mapping']) + answer_dict["REVENUE_CD_DESC"] = get_code_description(rev_code, all_mappings['rev_mapping']) + if grouper_type == 'MS-DRG': + answer_dict["GROUPER_CD_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping_ms']) + elif grouper_type == 'APR-DRG': + answer_dict["GROUPER_CD_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping_apr']) + elif string_utils.is_empty(grouper_code): + answer_dict["GROUPER_CD_DESC"] = "N/A" # No grouper code exists + else: + answer_dict["GROUPER_CD_DESC"] = "UNKNOWN" # Grouper code exists but type isn't recognized + answer_dict["AUTH_ADMIT_TYPE_DESC"] = get_code_description(admit_type_code, all_mappings['admit_mapping']) - answer_dict["CPT4_PROC_CD_DESC"] = get_code_description(proc_code, all_mappings['proc_mapping']) - answer_dict["CPT4_PROC_MOD_DESC"] = get_code_description(proc_mod, all_mappings['proc_mapping']) - answer_dict["DIAG_CD_DESC"] = get_code_description(diag_code, all_mappings['diag_mapping']) - answer_dict["REVENUE_CD_DESC"] = get_code_description(rev_code, all_mappings['rev_mapping']) - if grouper_type == 'MS-DRG': - answer_dict["GROUPER_CD_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping_ms']) - elif grouper_type == 'APR-DRG': - answer_dict["GROUPER_CD_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping_apr']) - elif string_utils.is_empty(grouper_code): - answer_dict["GROUPER_CD_DESC"] = "N/A" # No grouper code exists - else: - answer_dict["GROUPER_CD_DESC"] = "UNKNOWN" # Grouper code exists but type isn't recognized - answer_dict["AUTH_ADMIT_TYPE_DESC"] = get_code_description(admit_type_code, all_mappings['admit_mapping']) - - return answer_dicts + return answer_dict + + +def clean_service_term(service_term: str, + LOBS_TO_REMOVE, + PROGRAMS_TO_REMOVE, + NETWORKS_TO_REMOVE) -> str: + if string_utils.is_empty(service_term): + return str(service_term) + + service_term = service_term.upper() + + # replace whole words from LOBS_TO_REMOVE with "" + for lob in LOBS_TO_REMOVE: + service_term = re.sub(r'\b' + re.escape(lob.upper()) + r'\b', '', service_term) + + # replace whole words from PROGRAMS_TO_REMOVE with "" + for program in PROGRAMS_TO_REMOVE: + if program not in ["HOSPICE"]: + service_term = re.sub(r'\b' + re.escape(program.upper()) + r'\b', '', service_term) + + # replace whole words from NETWORKS_TO_REMOVE with "" + for network in NETWORKS_TO_REMOVE: + service_term = re.sub(r'\b' + re.escape(network.upper()) + r'\b', '', service_term) + + # replace specific phrases + for val in ["INPATIENT", "OUTPATIENT", + "PROFESSIONAL", "INSITUTIONAL", "FACILITY", "ANCILLARY", "HOSPITAL", "HEALTH"]: + service_term = re.sub(r'\b' + re.escape(val) + r'\b', '', service_term) + + # remove extra spaces + service_term = re.sub(r'\s+', ' ', service_term).strip() + + return service_term def get_code_breakout(answer_dicts: list[dict], filename: str, all_dataset: dict): """ @@ -508,13 +505,43 @@ def get_code_breakout(answer_dicts: list[dict], filename: str, all_dataset: dict dict: A dictionary with primary, indirect/derived, and mapped code descriptions. """ - # Code Primary (Pull codes directly from SERVICE_TERM and REIMB_TERM) - code_primary_answers = code_primary(answer_dicts, filename) + LOBS_TO_REMOVE = list(set(CrosswalkBuilder().from_json(path="crosswalk/mappings/crosswalk_lob.json").mapping.values())) + PROGRAMS_TO_REMOVE = list(set(CrosswalkBuilder().from_json(path="crosswalk/mappings/crosswalk_program.json").mapping.values())) + NETWORKS_TO_REMOVE = list(set(CrosswalkBuilder().from_json(path="crosswalk/mappings/crosswalk_network.json").mapping.values())) - # Only used for Implicit. Code Indirect (Derive codes indirectly from SERVICE_TERM and REIMB_TERM) - code_indirect_answers = code_indirect(code_primary_answers, all_dataset) + code_primary_questions = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="code_primary_breakout").get_prompt_dict() - # Code Descriptions (Pull standardized descriptions from codes) - code_description_answers = code_description_mapping(code_indirect_answers, all_dataset['codes_mappings']) - - return code_description_answers \ No newline at end of file + final_answer_dicts = [] + for answer_dict in answer_dicts: + service_term = answer_dict.get("SERVICE_TERM", "") + service_term_clean = clean_service_term(service_term, + LOBS_TO_REMOVE, + PROGRAMS_TO_REMOVE, + NETWORKS_TO_REMOVE) + + if service_term_clean in ['COVERED SERVICES', 'ALL COVERED SERVICES']: + final_answer_dicts.append(answer_dict) + continue + + # Get Explicit Codes + code_primary_answers = code_primary(service_term, code_primary_questions, filename) + answer_dict.update(code_primary_answers) + if all(string_utils.is_empty(value) for value in code_primary_answers.values()): + implicit_proc_cd = code_implicit(service_term_clean, filename) + answer_dict["CPT4_PROC_CD"] = implicit_proc_cd + + # Code Descriptions + try: + answer_dict = code_description_mapping(answer_dict, all_dataset['codes_mappings']) + except Exception as e: + print(f"Error in code_description_mapping: {e}") + answer_dict["CPT4_PROC_CD_DESC"] = "N/A" + answer_dict["CPT4_PROC_MOD_DESC"] = "N/A" + answer_dict["DIAG_CD_DESC"] = "N/A" + answer_dict["REVENUE_CD_DESC"] = "N/A" + answer_dict["GROUPER_CD_DESC"] = "N/A" + answer_dict["AUTH_ADMIT_TYPE_DESC"] = "N/A" + + final_answer_dicts.append(answer_dict) + + return final_answer_dicts \ No newline at end of file diff --git a/fieldExtraction/src/prompts/investment_prompts.py b/fieldExtraction/src/prompts/investment_prompts.py index 2ecdefb..d01a7e1 100644 --- a/fieldExtraction/src/prompts/investment_prompts.py +++ b/fieldExtraction/src/prompts/investment_prompts.py @@ -257,7 +257,7 @@ class FieldSet: ##################################################################################### def EXHIBIT_CHECK(page_start): - + return f"""Analyze the following contract excerpt and determine if it indicates the start of a section. Follow the following instructions: 1. Look specifically for the words 'Exhibit', 'Article', 'Amendment', 'Schedule', 'Attachment', or 'Addendum'. These words indicate the start of a new section and are usually followed by a number or letter (e.g., 'Attachment A'). @@ -504,23 +504,40 @@ Methodology: {methodology.replace('"', "'")} Write your JSON dictionary below: """ -def CODE_PRIMARY_BREAKOUT(service, methodology, questions): - return f"""Analyze a given medical service, found in a Payer-Provider contract: +def CODE_PRIMARY_BREAKOUT(service, questions): + return f"""Analyze a given medical service: Use the text in the Service to populate a JSON dictionary with the following fields: {questions}. -For each field, return a list of all codes seen for that field. If the contract gives a range of codes, without listing each code in the range individually, return the range in the format 'LowestCode-HighestCode'. -For the CPT4_PROC_CD and CPT4_PROC_MOD fields, ensure that each item in each list corresponds with each other. +For each field, return a list of all codes EXPLICITLY written for that field. The code itself must be written, not language that describes the code. + +If the contract gives a range of codes, without listing each code in the range individually, return the range in the format 'LowestCode-HighestCode'. If any of the codes are not found, do not write a list for that field. Simply populate the field with 'N/A'. -Here is the Service and Methodology to analyze: +Here is the Service to analyze: Service: {service.replace('"', "'")} -Methodology: {methodology.replace('"', "'")} -Write the JSON dictionary below: +Briefly explain your answer before putting the final answer in JSON dictionary format. +""" + +def CODE_IMPLICIT(service): + return f"""Analyze a given medical Service. + +What Procedure code or code range the Service does the Service describe? + +If the Service is broad enough to describe a range of codes, return the range in the format 'LowestCode-HighestCode'. The code descriptions for ranges may be very broad (for general services like 'Surgery', 'Radiology', etc.) or more specific (for services like 'Surgical Procedures on the Shoulder', 'Diagnostic Radiology Procedures of the Heart', etc.). +If the Service is specific, and therefore references a single code, return just that code. +If the Service is not specific enough to identify a code, return 'N/A'. + +Ensure that there is no meaningful difference between the description of the code you provide and the service you are analyzing. These codes will be used to price healthcare claims, so they must be accurate. Do not attempt to force a code to fit the service if it will result in different pricing. + +Here is the Service to analyze: +{service.replace('"', "'")} + +Explain your answer in 1-2 sentences. Then return your final answer in JSON list-of-string format. """ diff --git a/fieldExtraction/src/utils/string_utils.py b/fieldExtraction/src/utils/string_utils.py index 9311266..236184a 100644 --- a/fieldExtraction/src/utils/string_utils.py +++ b/fieldExtraction/src/utils/string_utils.py @@ -264,19 +264,34 @@ def universal_json_load(string_dict: str): the traceback as well as the offending string are printed. """ - match = re.search(r'\{.*\}|\[\s*?\{.*\}\s*?\]', string_dict, re.DOTALL) # Updated regex - if match: - matched_str = match.group() + # Try to match dictionary or list of dictionaries first + dict_match = re.search(r'\{.*\}|\[\s*?\{.*\}\s*?\]', string_dict, re.DOTALL) + if dict_match: + matched_str = dict_match.group() try: return json.loads(matched_str) except json.JSONDecodeError as e: print(f"Failed to parse JSON: {e}") - # print(f"Problematic JSON string: {matched_str}") print(f"Original string: {string_dict}") print(f"Traceback: {traceback.format_exc()}") raise ValueError(f"JSON parsing failed: {e.msg} at position {e.pos}") from e - else: - return {} + + # If no dict pattern found, try to match list of strings + list_match = re.search(r'\[(.*?)\]', string_dict, re.DOTALL) + if list_match: + matched_str = list_match.group() + try: + # Clean up the string list before parsing + cleaned_str = re.sub(r'(?