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

Merged in feature/implicit-proc-mapping (pull request #531)

Browse Source 
Feature/implicit proc mapping

* Testing scripts

* Refactor code primary

* implicit draft

* Merge branch 'main' into feature/implicit-proc-mapping

* Prompt update

* Prompt changes

* Experiments with preprocessing

* code updates

* Merged main into feature/implicit-proc-mapping

* Add condition to avoid Covered Services

* Add functions to main script

* remove explanations

* Error handling

* Genericize prompt

* Final prompt updates

* remove test file

* prep for merge

* Merged main into feature/implicit-proc-mapping

* Merged main into feature/implicit-proc-mapping


Approved-by: Alex Galarce
This commit is contained in:
Katon Minhas
2025-05-19 20:26:24 +00:00
parent e510e8f6c2
commit 597d28045b
3 changed files with 161 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/investment/code_funcs.py
+115 -88
View File
@@ -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
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
fieldExtraction/src/prompts/investment_prompts.py
+25 -8
View File
@@ -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.
"""
fieldExtraction/src/utils/string_utils.py
+21 -6
View File
@@ -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'(?<!\\)"', '"', matched_str) # Handle escaped quotes
cleaned_str = re.sub(r"'", '"', cleaned_str) # Replace single quotes with double quotes
return json.loads(cleaned_str)
except json.JSONDecodeError as e:
print(f"Failed to parse list: {e}")
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
return {}
reimbursement_strings = [ # These are for `method='keyword'`
"%",