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

Update code_funcs.py

Browse Source
This commit is contained in:
Katon Minhas
2026-02-02 21:47:30 -05:00
parent 309fa49ac6
commit 6c3f3f6f96
1 changed files with 58 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+58 -46
View File
@@ -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