From 6c3f3f6f96b598febe000f44bba4bdb31b6b3af6 Mon Sep 17 00:00:00 2001 From: Katon Minhas Date: Mon, 2 Feb 2026 21:47:30 -0500 Subject: [PATCH] Update code_funcs.py --- src/codes/code_funcs.py | 104 ++++++++++++++++++++++------------------ 1 file changed, 58 insertions(+), 46 deletions(-) diff --git a/src/codes/code_funcs.py b/src/codes/code_funcs.py index 6b94ac7..66dc380 100644 --- a/src/codes/code_funcs.py +++ b/src/codes/code_funcs.py @@ -77,13 +77,13 @@ def get_regex_answers(service, code_answer_dict): ) cpt_matches = re.findall(cpt_pattern, service) if len(cpt_matches) == 1: - code_answer_dict["PROCEDURE_CD"] = cpt_matches[0].replace("through", "-") + code_answer_dict["PROCEDURE_CD"] = [cpt_matches[0].replace("through", "-")] if "rev" in service.lower(): rev_pattern = r"\b(?:\d{3}|\d{2}X)(?:\s*(?:-|through)\s*(?:\d{3}|\d{2}X))?\b" rev_matches = re.findall(rev_pattern, service) if len(rev_matches) > 0: - code_answer_dict["REVENUE_CD"] = str(rev_matches) + code_answer_dict["REVENUE_CD"] = rev_matches return code_answer_dict @@ -106,15 +106,17 @@ def code_explicit(service, methodology, filename): """ # Prompt - field definitions are now included in CODE_EXPLICIT_INSTRUCTION() for caching - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.CODE_EXPLICIT(service, methodology), + prompt, _parser = prompt_templates.CODE_EXPLICIT(service, methodology) + + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, cache=True, instruction=prompt_templates.CODE_EXPLICIT_INSTRUCTION(), usage_label="CODE_EXPLICIT", ) - code_answer_dict = string_utils.universal_json_load(claude_answer_raw) + code_answer_dict = _parser(llm_answer_raw) return code_answer_dict @@ -139,19 +141,22 @@ def code_category(service, proc_category, hcpcs_level2_mapping, filename): for key, value in hcpcs_level2_mapping.items() if key.startswith(category) ] - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.CODE_CATEGORY(service, matching_values), + + prompt, _parser = prompt_templates.CODE_CATEGORY(service, matching_values) + + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, cache=True, instruction=prompt_templates.CODE_CATEGORY_INSTRUCTION(), ) - claude_answer_final = string_utils.universal_json_load( - claude_answer_raw + llm_answer_final = _parser( + llm_answer_raw ) # Returns list code_answer_dict = {"PROCEDURE_CD": [], "PROCEDURE_CD_DESC": []} - for answer in claude_answer_final: + for answer in llm_answer_final: if "0000" in answer: code_answer_dict["PROCEDURE_CD"].append(answer) code_answer_dict["PROCEDURE_CD_DESC"].append("") @@ -176,17 +181,16 @@ def code_implicit_special(service, filename): Returns: dict: A dictionary containing the implicit codes found for the service. If no codes are found, returns an empty dictionary. """ + prompt, _parser = prompt_templates.CODE_IMPLICIT_SPECIAL(service) - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.CODE_IMPLICIT_SPECIAL(service), + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, cache=True, instruction=prompt_templates.CODE_IMPLICIT_SPECIAL_INSTRUCTION(), ) - claude_answer_final = string_utils.extract_text_from_delimiters( - claude_answer_raw, Delimiter.PIPE - ) + llm_answer_final = _parser(llm_answer_raw) special_case_mapping = { "Drugs": ["J0000-J9999"], @@ -227,9 +231,9 @@ def code_implicit_special(service, filename): "ST": ["ST Codes TBD"], } code_answer_dict = {} - if claude_answer_final in special_case_mapping: - code_answer_dict["PROCEDURE_CD"] = special_case_mapping[claude_answer_final] - code_answer_dict["PROCEDURE_CD_DESC"] = claude_answer_final + if llm_answer_final[0] in special_case_mapping: + code_answer_dict["PROCEDURE_CD"] = special_case_mapping[llm_answer_final] + code_answer_dict["PROCEDURE_CD_DESC"] = llm_answer_final return code_answer_dict @@ -386,7 +390,7 @@ def code_implicit_rag(service, implicit_run_dict, filename, constants): if matching_codes: # Split pipe-delimited codes into individual codes for code in matching_codes: - proc_codes.extend(code.split("|")) + proc_codes.extend(code) proc_descs.append(description) if description in hcpcs_mapping.values(): matching_codes = [ @@ -395,7 +399,7 @@ def code_implicit_rag(service, implicit_run_dict, filename, constants): if matching_codes: # Split pipe-delimited codes into individual codes for code in matching_codes: - proc_codes.extend(code.split("|")) + proc_codes.extend(code) proc_descs.append(description) if description in rev_mapping.values(): matching_codes = [ @@ -404,7 +408,7 @@ def code_implicit_rag(service, implicit_run_dict, filename, constants): if matching_codes: # Split pipe-delimited codes into individual codes for code in matching_codes: - rev_codes.extend(code.split("|")) + rev_codes.extend(code) rev_descs.append(description) # Combine answers @@ -438,18 +442,21 @@ def code_last_check(service, filename): str: "Specific" if the service is a specific case that requires manual intervention, "Generic" if the service is a generic case that does not match any codes. """ + prompt, _parser = prompt_templates.CODE_LAST_CHECK(service) - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.CODE_LAST_CHECK(service), + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, cache=True, instruction=prompt_templates.CODE_LAST_CHECK_INSTRUCTION(), ) - claude_answer_final = string_utils.extract_text_from_delimiters( - claude_answer_raw, Delimiter.PIPE - ) - return claude_answer_final + try: + llm_answer_final = _parser(llm_answer_raw) + return llm_answer_final[0] + except: + return "Generic" + def fill_bill_type( @@ -477,47 +484,52 @@ def fill_bill_type( valid_bill_type = sorted(list(set(BILL_TYPE_MAPPING.values()))) + prompt, _parser = prompt_templates.FILL_BILL_TYPE( + service, valid_bill_type, reimb_term=reimb_term, exhibit_text=None + ) + + # First attempt: Service term + reimbursement term - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.FILL_BILL_TYPE( - service, valid_bill_type, reimb_term=reimb_term, exhibit_text=None - ), + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", "", cache=True, instruction=prompt_templates.FILL_BILL_TYPE_INSTRUCTION(), ) - claude_answer_final = string_utils.universal_json_load(claude_answer_raw) + llm_answer_final = string_utils.universal_json_load(llm_answer_raw) # Second attempt: Service term + reimbursement term + exhibit context (if first attempt failed and context available) - if not claude_answer_final and exhibit_text: - claude_answer_raw = llm_utils.invoke_claude( - prompt_templates.FILL_BILL_TYPE( - service, - valid_bill_type, - reimb_term=reimb_term, - exhibit_text=exhibit_text, - ), + if not llm_answer_final and exhibit_text: + prompt, _parser = prompt_templates.FILL_BILL_TYPE( + service, + valid_bill_type, + reimb_term=reimb_term, + exhibit_text=exhibit_text, + ) + + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", "", cache=True, instruction=prompt_templates.FILL_BILL_TYPE_INSTRUCTION(), ) - claude_answer_final = string_utils.universal_json_load(claude_answer_raw) + llm_answer_final = _parser(llm_answer_raw) # Keep legacy behavior: return answer_dict unchanged if no result found - if not claude_answer_final: + if not llm_answer_final: return answer_dict bill_codes, bill_descs = [], [] - for description in claude_answer_final: + for description in llm_answer_final: if description in BILL_TYPE_REVERSE_MAPPING: bill_codes.append(BILL_TYPE_REVERSE_MAPPING[description]) bill_descs.append(description) if bill_codes: - answer_dict["BILL_TYPE_CD"] = "|".join(bill_codes) - answer_dict["BILL_TYPE_CD_DESC"] = "|".join(bill_descs) + answer_dict["BILL_TYPE_CD"] = bill_codes + answer_dict["BILL_TYPE_CD_DESC"] = bill_descs return answer_dict @@ -569,7 +581,7 @@ def fill_grouper_cd_desc(answer_dict, constants: Constants): ) if grouper_descs: - answer_dict["GROUPER_CD_DESC"] = "|".join(list(set(grouper_descs))) + answer_dict["GROUPER_CD_DESC"] = list(set(grouper_descs)) return answer_dict