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

Merged in DAIP2-2281-remove-implicit-logic-for-revenue-codes (pull request #932)

Browse Source 
REVENUE_CD implicit method removed

* REVENUE_CD implicit method removed

* Merged dev into DAIP2-2281-remove-implicit-logic-for-revenue-codes

* fix(constants): restore REV crosswalks for explicit revenue after implicit RAG removal

Keep REV_LEVEL1_MAPPING and REV_MAPPING for validate_explicit_codes and
fill_code_descriptions_from_mappings; revenue embeddings remain removed.


Approved-by: Praneel Panchigar
This commit is contained in:
Mayank Aamseek
2026-04-02 19:24:16 +00:00
committed by Praneel Panchigar
parent a9098ab2e8
commit e72e7667f7
9 changed files with 51 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+9 -45
View File
@@ -235,7 +235,7 @@ def code_implicit_special(service: str, filename: str) -> dict:
def get_embedding_levels(
level_suffix: int, implicit_run_dict: dict, constants: Constants
) -> tuple[list, dict, dict, dict]:
) -> tuple[list, dict, dict]:
"""
Retrieves the embedding levels and their corresponding mappings for a
given level suffix.
@@ -246,7 +246,7 @@ def get_embedding_levels(
constants: Application constants with mapping data.
Returns:
(embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping).
(embedding_levels, cpt_mapping, hcpcs_mapping).
Raises:
ValueError: If level_suffix is not 1 or 2.
@@ -254,20 +254,16 @@ def get_embedding_levels(
if level_suffix == 1:
cpt_mapping = constants.CPT_LEVEL1_MAPPING
hcpcs_mapping = constants.HCPCS_LEVEL1_MAPPING
rev_mapping = constants.REV_LEVEL1_MAPPING
embedding_levels = []
if implicit_run_dict["PROCEDURE_CD"]:
embedding_levels += [
f"cpt_level{level_suffix}",
f"hcpcs_level{level_suffix}",
]
if implicit_run_dict["REVENUE_CD"]:
embedding_levels += [f"rev_level{level_suffix}"]
elif level_suffix == 2:
cpt_mapping = constants.CPT_LEVEL2_MAPPING
hcpcs_mapping = constants.HCPCS_LEVEL2_MAPPING
rev_mapping = constants.REV_MAPPING
embedding_levels = []
if implicit_run_dict["PROCEDURE_CD"]:
@@ -275,13 +271,11 @@ def get_embedding_levels(
f"cpt_level{level_suffix}",
f"hcpcs_level{level_suffix}",
]
if implicit_run_dict["REVENUE_CD"]:
embedding_levels += [f"rev"]
else:
raise ValueError(f"Unsupported level_suffix: {level_suffix}")
return embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping
return embedding_levels, cpt_mapping, hcpcs_mapping
def get_match_list(
@@ -341,13 +335,13 @@ def code_implicit_rag(
Returns:
List of {"source": "Level 1"|"Level 2", "code_answer_dict": dict}.
Only includes entries that have PROCEDURE_CD or REVENUE_CD (no placeholder-only).
Only includes entries that have PROCEDURE_CD (no placeholder-only).
"""
# Get embeddings, mappings, and best matches for Levels 1 and 2.
level_dicts: list[dict] = []
for level_suffix in [1, 2]:
embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = (
get_embedding_levels(level_suffix, implicit_run_dict, constants)
embedding_levels, cpt_mapping, hcpcs_mapping = get_embedding_levels(
level_suffix, implicit_run_dict, constants
)
if not embedding_levels:
return []
@@ -362,7 +356,6 @@ def code_implicit_rag(
"highest_similarity": highest_similarity,
"cpt_mapping": cpt_mapping,
"hcpcs_mapping": hcpcs_mapping,
"rev_mapping": rev_mapping,
}
)
@@ -372,7 +365,6 @@ def code_implicit_rag(
for level_dict in level_dicts:
cpt_mapping = level_dict["cpt_mapping"]
hcpcs_mapping = level_dict["hcpcs_mapping"]
rev_mapping = level_dict["rev_mapping"]
prompt, _parser = prompt_templates.CODE_IMPLICIT(
service, level_dict["match_list"]
@@ -401,9 +393,7 @@ def code_implicit_rag(
code_answer_dict: dict = {}
proc_codes: list[str] = []
rev_codes: list[str] = []
proc_descs: list[str] = []
rev_descs: list[str] = []
for description in llm_answer_final:
if description in cpt_mapping.values():
matching_codes = [
@@ -419,18 +409,6 @@ def code_implicit_rag(
if matching_codes:
proc_codes.extend(matching_codes)
proc_descs.append(description)
if description in rev_mapping.values():
matching_codes = [
str(key) for key, val in rev_mapping.items() if val == description
]
if matching_codes:
# REV_LEVEL1_MAPPING keys are pipe-delimited (e.g. "0810|0811|0812").
for code in matching_codes:
if "|" in code:
rev_codes.extend(code.split("|"))
else:
rev_codes.append(code)
rev_descs.append(description)
if proc_codes:
code_answer_dict["PROCEDURE_CD"] = proc_codes
@@ -438,12 +416,6 @@ def code_implicit_rag(
code_answer_dict["CODE_METHODOLOGY"] = (
f"Implicit - Level {level_dict['level_suffix']}"
)
if rev_codes:
code_answer_dict["REVENUE_CD"] = rev_codes
code_answer_dict["REVENUE_CD_DESC"] = rev_descs
code_answer_dict["CODE_METHODOLOGY"] = (
f"Implicit - Level {level_dict['level_suffix']}"
)
if code_answer_dict:
source_label = f"Level {level_dict['level_suffix']}"
@@ -459,7 +431,7 @@ def _format_implicit_candidate_for_prompt(candidate: dict, index: int) -> str:
source = candidate.get("source", "Unknown")
code_dict = candidate.get("code_answer_dict", {})
parts = [f"Candidate {index} (Source: {source})"]
for key in ["PROCEDURE_CD", "PROCEDURE_CD_DESC", "REVENUE_CD", "REVENUE_CD_DESC"]:
for key in ["PROCEDURE_CD", "PROCEDURE_CD_DESC"]:
if key in code_dict and code_dict[key]:
val = code_dict[key]
if isinstance(val, list):
@@ -768,15 +740,8 @@ def get_implicit_runs(answer_dict: dict) -> dict:
claim_type = answer_dict.get("AARETE_DERIVED_CLAIM_TYPE_CD")
bill_type = answer_dict.get("BILL_TYPE_CD_DESC")
if claim_type == "H":
run_dict["REVENUE_CD"] = True
elif claim_type == "M":
if not string_utils.is_empty(bill_type):
run_dict["REVENUE_CD"] = True
else:
run_dict["REVENUE_CD"] = False
else:
run_dict["REVENUE_CD"] = False
# REVENUE_CD implicit extraction is disabled
run_dict["REVENUE_CD"] = False
# Run proc code implicit for all bill types except 2
if bill_type in ["Inpatient Hospital", "Skilled Nursing Facility"]:
@@ -1149,7 +1114,6 @@ def extract_codes_from_service(answer_dict: dict, constants: Constants) -> dict:
- NDC_CD: National Drug Codes (NDC)
Implicit Codes (Up to Level 2):
- PROCEDURE_CD: CPT, HCPCS
- REVENUE_CD: Revenue codes
"""
service, bill_type = answer_dict.get("SERVICE_TERM"), answer_dict.get(
src/constants/constants.py
+21 -23
View File
@@ -125,6 +125,26 @@ class Constants:
)
.mapping
)
# Revenue crosswalks (explicit extraction, validation, description fill).
# Implicit RAG for revenue is disabled; embeddings for rev are not loaded.
self.REV_LEVEL1_MAPPING = (
CrosswalkBuilder()
.from_excel(
"src/constants/mapping_csvs/rev_cd/rev_level1.csv",
"Code",
"Description",
)
.mapping
)
self.REV_MAPPING = (
CrosswalkBuilder()
.from_excel(
_path("constants/mapping_csvs/rev_cd/rev.csv"),
"Code",
"Description",
)
.mapping
)
# Full procedure code mappings (exact code to description) for validation and
# description fill. Used by code extraction validation and fill_code_descriptions.
self.CPT_FULL_MAPPING = (
@@ -146,22 +166,6 @@ class Constants:
.mapping
)
self.REV_LEVEL1_MAPPING = (
CrosswalkBuilder()
.from_excel(
"src/constants/mapping_csvs/rev_cd/rev_level1.csv",
"Code",
"Description",
)
.mapping
)
self.REV_MAPPING = (
CrosswalkBuilder()
.from_excel(
_path("constants/mapping_csvs/rev_cd/rev.csv"), "Code", "Description"
)
.mapping
)
self.GROUPER_APR_DRG_MAPPING = (
CrosswalkBuilder()
.from_excel(
@@ -203,9 +207,6 @@ class Constants:
self.HCPCS_LEVEL1_EMBEDDING = CodeEmbedding("hcpcs_level1")
self.HCPCS_LEVEL2_EMBEDDING = CodeEmbedding("hcpcs_level2")
self.REV_LEVEL1_EMBEDDING = CodeEmbedding("rev_level1")
self.REV_EMBEDDING = CodeEmbedding("rev")
#################################### Methodology Breakout ####################################
self.VALID_AARETE_DERIVED_FEE_SCHEDULE = (
ListBuilder()
@@ -312,15 +313,12 @@ class Constants:
Returns the CodeEmbedding attribute corresponding to the given level.
Args:
level (str): The embedding level identifier (e.g., "rev", "cpt_level1").
level (str): The embedding level identifier (e.g., "cpt_level1").
Returns:
CodeEmbedding: The matching embedding object, or None if not found.
"""
level_map = {
# Revenue code embeddings
"rev": self.REV_EMBEDDING,
"rev_level1": self.REV_LEVEL1_EMBEDDING,
# CPT code embeddings
"cpt_level1": self.CPT_LEVEL1_EMBEDDING,
"cpt_level2": self.CPT_LEVEL2_EMBEDDING,
src/embeddings/rev/choices.pkl
BIN
View File
Binary file not shown.
src/embeddings/rev/embeddings.npy
BIN
View File
Binary file not shown.
src/embeddings/rev/faiss_index.bin
BIN
View File
Binary file not shown.
src/embeddings/rev_level1/choices.pkl
BIN
View File
Binary file not shown.
src/embeddings/rev_level1/embeddings.npy
BIN
View File
Binary file not shown.
src/embeddings/rev_level1/faiss_index.bin
BIN
View File
Binary file not shown.
src/tests/test_code_funcs.py
+21 -25
View File
@@ -162,26 +162,24 @@ class TestCodeFuncs(unittest.TestCase):
def test_get_embedding_levels(self):
"""Test the get_embedding_levels function for determining embedding levels."""
# Test level 1 with both codes enabled
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = (
code_funcs.get_embedding_levels(1, implicit_run_dict, self.constants)
# Test level 1 with procedure code enabled
implicit_run_dict = {"PROCEDURE_CD": True}
embedding_levels, cpt_mapping, hcpcs_mapping = code_funcs.get_embedding_levels(
1, implicit_run_dict, self.constants
)
self.assertIn("cpt_level1", embedding_levels)
self.assertIn("hcpcs_level1", embedding_levels)
self.assertIn("rev_level1", embedding_levels)
self.assertEqual(cpt_mapping, self.constants.CPT_LEVEL1_MAPPING)
# Test level 2 with only procedure code enabled
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": False}
embedding_levels, cpt_mapping, hcpcs_mapping, rev_mapping = (
code_funcs.get_embedding_levels(2, implicit_run_dict, self.constants)
implicit_run_dict = {"PROCEDURE_CD": True}
embedding_levels, cpt_mapping, hcpcs_mapping = code_funcs.get_embedding_levels(
2, implicit_run_dict, self.constants
)
self.assertIn("cpt_level2", embedding_levels)
self.assertIn("hcpcs_level2", embedding_levels)
self.assertNotIn("rev", embedding_levels)
self.assertEqual(cpt_mapping, self.constants.CPT_LEVEL2_MAPPING)
def test_get_match_list(self):
@@ -228,7 +226,7 @@ class TestCodeFuncs(unittest.TestCase):
mock_json_load.return_value = ["Test Procedure"]
# Test with successful match
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
# Patch the get_embedding_levels function to return predictable values
with patch(
@@ -239,7 +237,6 @@ class TestCodeFuncs(unittest.TestCase):
["cpt_level1"],
{"12345": "Test Procedure"},
{},
{},
)
# Call the function
@@ -280,7 +277,7 @@ class TestCodeFuncs(unittest.TestCase):
mock_get_match_list.return_value = (["Test Procedure"], 0.9)
mock_invoke_claude.return_value = '["INVALID_SERVICE", "Test Procedure"]'
mock_json_load.return_value = ["INVALID_SERVICE", "Test Procedure"]
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
with patch(
"src.codes.code_funcs.get_embedding_levels"
@@ -289,7 +286,6 @@ class TestCodeFuncs(unittest.TestCase):
["cpt_level1"],
{"12345": "Test Procedure"},
{},
{},
)
result = code_funcs.code_implicit_rag(
"TEST SERVICE", implicit_run_dict, "test.pdf", self.constants
@@ -351,7 +347,7 @@ class TestCodeFuncs(unittest.TestCase):
mock_category.return_value = {}
mock_special.return_value = {}
mock_rag.return_value = [] # RAG returns list of candidates
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
result = code_funcs.build_implicit_candidates(
"SERVICE",
{},
@@ -374,7 +370,7 @@ class TestCodeFuncs(unittest.TestCase):
}
mock_special.return_value = {}
mock_rag.return_value = [] # RAG returns list of candidates
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
code_after_explicit = {"PROCEDURE_CD": ["Category: J"]}
result = code_funcs.build_implicit_candidates(
"J CODES",
@@ -400,7 +396,7 @@ class TestCodeFuncs(unittest.TestCase):
"PROCEDURE_CD_DESC": ["Drugs"],
}
mock_rag.return_value = [] # RAG returns list of candidates
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
result = code_funcs.build_implicit_candidates(
"DRUG SERVICE",
{},
@@ -437,7 +433,7 @@ class TestCodeFuncs(unittest.TestCase):
},
},
]
implicit_run_dict = {"PROCEDURE_CD": True, "REVENUE_CD": True}
implicit_run_dict = {"PROCEDURE_CD": True}
code_after_explicit = {"PROCEDURE_CD": ["Category: A"]}
result = code_funcs.build_implicit_candidates(
"SERVICE",
@@ -571,13 +567,13 @@ class TestCodeFuncs(unittest.TestCase):
"""Test the get_implicit_runs function for determining which code runs to execute."""
# Test hospital claim type
result = code_funcs.get_implicit_runs({"AARETE_DERIVED_CLAIM_TYPE_CD": "H"})
self.assertTrue(result["REVENUE_CD"])
self.assertFalse(result["REVENUE_CD"])
# Test medical claim type with bill type
result = code_funcs.get_implicit_runs(
{"AARETE_DERIVED_CLAIM_TYPE_CD": "M", "BILL_TYPE_CD_DESC": "Outpatient"}
)
self.assertTrue(result["REVENUE_CD"])
self.assertFalse(result["REVENUE_CD"])
self.assertTrue(result["PROCEDURE_CD"])
# Test inpatient bill type
@@ -587,7 +583,7 @@ class TestCodeFuncs(unittest.TestCase):
"BILL_TYPE_CD_DESC": "Inpatient Hospital",
}
)
self.assertTrue(result["REVENUE_CD"])
self.assertFalse(result["REVENUE_CD"])
self.assertFalse(result["PROCEDURE_CD"])
@patch("src.codes.code_funcs.clean_service")
@@ -616,7 +612,7 @@ class TestCodeFuncs(unittest.TestCase):
"SERVICE_TERM": "TEST SERVICE",
"BILL_TYPE_CD_DESC": "Outpatient",
}
mock_implicit_runs.return_value = {"PROCEDURE_CD": True, "REVENUE_CD": True}
mock_implicit_runs.return_value = {"PROCEDURE_CD": True}
# Test explicit code match - extract_codes_from_service normalizes to JSON list format
mock_explicit.return_value = {"PROCEDURE_CD": "12345"}
@@ -855,7 +851,7 @@ class TestCodeFuncs(unittest.TestCase):
with patch("src.codes.code_funcs.fill_bill_type", return_value={}):
with patch(
"src.codes.code_funcs.get_implicit_runs",
return_value={"PROCEDURE_CD": True, "REVENUE_CD": True},
return_value={"PROCEDURE_CD": True},
):
result = code_funcs.extract_codes_from_service(
{"SERVICE_TERM": "TEST"}, self.constants
@@ -876,7 +872,7 @@ class TestCodeFuncs(unittest.TestCase):
with patch("src.codes.code_funcs.fill_bill_type", return_value={}):
with patch(
"src.codes.code_funcs.get_implicit_runs",
return_value={"PROCEDURE_CD": True, "REVENUE_CD": True},
return_value={"PROCEDURE_CD": True},
):
result = code_funcs.extract_codes_from_service(
{"SERVICE_TERM": "Office visit"}, self.constants
@@ -916,7 +912,7 @@ class TestCodeFuncs(unittest.TestCase):
with patch("src.codes.code_funcs.fill_bill_type", return_value={}):
with patch(
"src.codes.code_funcs.get_implicit_runs",
return_value={"PROCEDURE_CD": True, "REVENUE_CD": True},
return_value={"PROCEDURE_CD": True},
):
result = code_funcs.extract_codes_from_service(
{"SERVICE_TERM": "J CODES"}, self.constants
@@ -956,7 +952,7 @@ class TestCodeFuncs(unittest.TestCase):
with patch("src.codes.code_funcs.fill_bill_type", return_value={}):
with patch(
"src.codes.code_funcs.get_implicit_runs",
return_value={"PROCEDURE_CD": True, "REVENUE_CD": True},
return_value={"PROCEDURE_CD": True},
):
result = code_funcs.extract_codes_from_service(
{"SERVICE_TERM": "J CODES"}, self.constants