Alex Galarce
6d6839dcd4
Hotfix/code description handling * Update date formatting in prompts to YYYY/MM/DD * fix tests for new date formatting * Enhance code parsing to support comma-separated values in get_code_description function * Merged main into hotfix/code-description-handling Approved-by: Katon Minhas
362 lines
17 KiB
Python
362 lines
17 KiB
Python
|
|
from src.prompts.investment_prompts import Field, FieldSet
|
|
import src.config as config
|
|
import src.utils.string_utils as string_utils
|
|
import src.utils.llm_utils as llm_utils
|
|
import src.prompts.investment_prompts as investment_prompts
|
|
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
|
|
|
|
import faiss
|
|
import numpy as np
|
|
from sentence_transformers import SentenceTransformer
|
|
|
|
# Load model
|
|
model = SentenceTransformer("all-roberta-large-v1")
|
|
def create_faiss_index(choices):
|
|
"""
|
|
Creates a FAISS index for a list of choices using cosine similarity (inner product).
|
|
|
|
Args:
|
|
choices (list): A list of text items to index.
|
|
|
|
Returns:
|
|
faiss.IndexFlatIP: A FAISS index for the given embeddings.
|
|
"""
|
|
embeddings = model.encode(choices, normalize_embeddings=True).astype("float32")
|
|
index = faiss.IndexFlatIP(embeddings.shape[1]) # Inner product for cosine similarity
|
|
index.add(embeddings)
|
|
return index
|
|
|
|
def get_matches_above_threshold(target, choices, index, threshold=0.7, top_k=5):
|
|
"""
|
|
Retrieves the top-k matches from a FAISS index with similarity scores above a given threshold.
|
|
|
|
Args:
|
|
target (str): The target text to compare against the choices.
|
|
choices (list): A list of choices to match the target with.
|
|
index (faiss.IndexFlatIP): The FAISS index for fast nearest-neighbor search.
|
|
threshold (float, optional): The minimum similarity score to consider a match. Defaults to 0.7.
|
|
top_k (int, optional): The number of top matches to retrieve. Defaults to 5.
|
|
|
|
Returns:
|
|
list: A list of tuples containing matched choice and score for each match above the threshold.
|
|
"""
|
|
target_vec = model.encode([target], normalize_embeddings=True).astype("float32")
|
|
similarity_scores, match_indices = index.search(target_vec, top_k)
|
|
return [
|
|
(choices[idx], score)
|
|
for idx, score in zip(match_indices[0], similarity_scores[0])
|
|
if score >= threshold
|
|
]
|
|
|
|
def crosswalk_levels():
|
|
"""
|
|
Loads and processes crosswalk mappings for different levels (e.g., CPT, HCPCS) from CSV files,
|
|
and returns a dictionary with clean-to-description and description-to-code mappings for each level.
|
|
|
|
Returns:
|
|
dict: A dictionary where each key is a level, and the value is a tuple containing:
|
|
- clean-to-description mapping (str -> str)
|
|
- description-to-code mapping (str -> str)
|
|
"""
|
|
# Get necessary mappings
|
|
crosswalk_dict = {}
|
|
for level in investment_values.levels:
|
|
mapping_df = pd.read_csv(f"crosswalk/mapping_csvs/proc_cd/{level}.csv", dtype=str)
|
|
level_crosswalk = CrosswalkBuilder().from_df(mapping_df, from_col="Code", to_col="Description")
|
|
clean_to_description = {get_clean_value(x) : x for x in level_crosswalk.mapping.values()}
|
|
description_to_code = level_crosswalk.create_reverse_mapping()
|
|
crosswalk_dict[level] = (clean_to_description, description_to_code)
|
|
|
|
return crosswalk_dict
|
|
|
|
def code_primary(answer_dicts, filename):
|
|
"""
|
|
Breaks down the direct code information from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD fields
|
|
using an LLM prompt and adds the results to the provided answer dictionaries.
|
|
|
|
Args:
|
|
answer_dicts (list of dict): A list of dictionaries containing service and reimbursement data.
|
|
filename (str): The name of the file being processed, used for LLM processing.
|
|
|
|
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("CONTRACT_SERVICE_CD_OR_DESC"), answer_dict.get("CONTRACT_REIMBURSEMENT_METHOD")
|
|
if service:
|
|
code_answer_dict = string_utils.universal_json_load(
|
|
llm_utils.invoke_claude(
|
|
investment_prompts.CODE_PRIMARY_BREAKOUT(service, methodology, 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
|
|
|
|
def get_proc_crosswalk():
|
|
"""
|
|
Loads and concatenates all CSV files in the 'proc_cd' directory, and creates a crosswalk mapping from the 'Code' to 'Description' columns.
|
|
|
|
Returns:
|
|
CrosswalkBuilder: A CrosswalkBuilder object containing the mapping from procedure codes to descriptions.
|
|
"""
|
|
all_mappings = []
|
|
mapping_dir = "crosswalk/mapping_csvs/proc_cd"
|
|
for filename in os.listdir(mapping_dir):
|
|
if filename.endswith(".csv"):
|
|
mapping_df = pd.read_csv(os.path.join(mapping_dir, filename))
|
|
all_mappings.append(mapping_df)
|
|
proc_df = pd.concat(all_mappings, ignore_index=True)
|
|
return CrosswalkBuilder().from_df(proc_df, from_col="Code", to_col="Description")
|
|
|
|
|
|
def get_clean_value(service_description):
|
|
"""
|
|
Cleans a service description by removing stop words, replacing certain terms, and standardizing acronyms.
|
|
|
|
Args:
|
|
service_description (str): The service description string to clean.
|
|
|
|
Returns:
|
|
str: The cleaned service description with stop words removed, specific terms replaced, and acronyms expanded.
|
|
"""
|
|
stop_words = {"the", "is", "in", "and", "to", "of", "a", "an", "that", "this", "it", "for", "on", "with", "as", "was", "were", "at", "by", "but", "be"} # Experiment with not removing 'and'
|
|
def remove_stop_words(text):
|
|
words = text.split()
|
|
return " ".join(word for word in words if word.lower() not in stop_words)
|
|
|
|
service_clean = service_description.lower()
|
|
# Replace "Services" and "Procedures"
|
|
service_clean = service_clean.replace("services", "").replace("procedures", "")
|
|
|
|
service_clean = service_clean.replace("DME", "Durable Medical Equipment") # Explore adding other acronyms
|
|
service_clean = remove_stop_words(service_clean)
|
|
|
|
# Explore lemmatization here
|
|
|
|
return service_clean
|
|
|
|
|
|
def get_best_match(crosswalk_dict, target, model, thresholds={
|
|
"cpt": 0.85, "cpt_level3": 0.8, "cpt_level2": 0.75, "cpt_level1": 0.7,
|
|
"hcpcs": 0.85, "hcpcs_level3": 0.8, "hcpcs_level2": 0.75, "hcpcs_level1": 0.7
|
|
}):
|
|
"""
|
|
Finds the best matching code from a crosswalk dictionary by comparing a target string to precomputed embeddings
|
|
using a model and similarity threshold values for different levels.
|
|
|
|
Args:
|
|
crosswalk_dict (dict): Dictionary containing the crosswalk data for different levels.
|
|
target (str): The target string to match.
|
|
model: The model used to encode the target string into an embedding.
|
|
thresholds (dict, optional): A dictionary of similarity thresholds for each level (default values provided).
|
|
|
|
Returns:
|
|
tuple: A tuple containing the best matching code, its description, and the similarity score.
|
|
If no match is found above the threshold, returns (None, None, 0).
|
|
"""
|
|
|
|
target_vec = model.encode([target], normalize_embeddings=True).astype("float32")
|
|
for level in investment_values.levels:
|
|
index, _, choices = embedding_utils.load_faiss_index(index_path=os.path.join("embeddings", level, "faiss_index.bin"),
|
|
embedding_path=os.path.join("embeddings", level, "embeddings.npy"),
|
|
choices_path=os.path.join("embeddings", level, "choices.pkl"),)
|
|
similarity_scores, match_indices = index.search(target_vec, 1)
|
|
best_match, best_score = choices[match_indices[0][0]], similarity_scores[0][0]
|
|
|
|
if best_score >= thresholds[level]:
|
|
best_description = crosswalk_dict[level][0][best_match]
|
|
best_code = crosswalk_dict[level][1][best_description]
|
|
return best_code, best_description, best_score
|
|
|
|
return None, None, 0 # No valid match found
|
|
|
|
def code_indirect(answer_dicts, crosswalk):
|
|
"""
|
|
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 CODES RN
|
|
|
|
Args:
|
|
answer_dicts (list): List of dictionaries containing 1:N answers
|
|
filename (str): Unused parameter (can be removed if unnecessary).
|
|
|
|
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("CONTRACT_SERVICE_CD_OR_DESC", "")
|
|
proc_code = answer_dict.get("CPT4_PROC_CD")
|
|
if (not service # Service DNE
|
|
or "covered service" in service.lower() # Is generic
|
|
or isinstance(proc_code, list) # Is a list
|
|
or not string_utils.is_empty(proc_code) # Is not empty
|
|
):
|
|
pass # pass for now - we may need to add functionality here later
|
|
else:
|
|
if service in already_seen:
|
|
code = already_seen[service]['code']
|
|
description = already_seen[service]['description']
|
|
else:
|
|
code, description, score = get_best_match(crosswalk, service, model)
|
|
# print(service, " | ", description, " | ", code, " | ", score)
|
|
|
|
answer_dict["CPT4_PROC_CD"] = code
|
|
answer_dict["CPT4_PROC_DESC"] = description
|
|
results.append(answer_dict)
|
|
return results
|
|
|
|
def get_code_description(code, code_mapping):
|
|
"""
|
|
Retrieves descriptions for a given code or list of codes using the provided code mapping.
|
|
If no description is found, attempts to resolve by modifying the code (e.g., prepending "0" or handling code patterns).
|
|
|
|
Args:
|
|
code (str, list): A single code or a list of codes to look up.
|
|
code_mapping (Crosswalk): The mapping object containing code-to-description mappings.
|
|
|
|
Returns:
|
|
str: A comma-separated string of descriptions for the provided codes.
|
|
"""
|
|
if isinstance(code, str):
|
|
if string_utils.is_empty(code):
|
|
return "N/A"
|
|
elif "[" in code and "]" in code:
|
|
code_list = ast.literal_eval(code)
|
|
elif "," in code:
|
|
code_list = code.split(",")
|
|
else:
|
|
code_list = [code]
|
|
elif isinstance(code, list):
|
|
code_list = code
|
|
else:
|
|
code_list = [code]
|
|
|
|
code_description = []
|
|
for code in code_list:
|
|
code_str = str(code).strip()
|
|
description = code_mapping.mapping.get(code_str, "mismap")
|
|
|
|
if string_utils.is_empty(description) or description == "mismap":
|
|
# If mismap, prepend "0" and try again (useful for rev codes)
|
|
code_str = "0" + code_str
|
|
description = code_mapping.mapping.get(code_str, "mismap")
|
|
|
|
# We need similar logic for Diag codes with X (e.g. 123X)
|
|
# Could either: a) convert to list ([1230, 1231, ...]), b) convert to range (1230-1239), c) leave as is (123X) but have the descriptions be a list ["desc1","desc2", etc]
|
|
# for i in range(10): code is '123' + i' map that code
|
|
|
|
if not string_utils.is_empty(description):
|
|
code_description.append(description)
|
|
|
|
return ",".join(code_description)
|
|
|
|
def get_mappings():
|
|
"""
|
|
Loads and processes multiple CSV and Excel files containing code mappings for various categories
|
|
(e.g., PROC, DIAG, Grouper, Rev, Admit Type), and returns a dictionary of crosswalks for each code type.
|
|
|
|
Returns:
|
|
dict: A dictionary containing mappings for PROC, DIAG, Grouper, Rev, and Admit Type codes.
|
|
"""
|
|
|
|
all_mappings = {}
|
|
|
|
# PROC codes
|
|
mapping_path = "crosswalk/mapping_csvs/"
|
|
all_proc_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "proc_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "proc_cd")) if file.endswith(".csv")])
|
|
proc_mapping = CrosswalkBuilder().from_df(all_proc_mappings, from_col="Code", to_col="Description")
|
|
all_mappings['proc_mapping'] = proc_mapping
|
|
|
|
# Diag codes
|
|
all_diag_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "diag_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "diag_cd")) if file.endswith(".csv")])
|
|
diag_mapping = CrosswalkBuilder().from_df(all_diag_mappings, from_col="Code", to_col="Description")
|
|
all_mappings['diag_mapping'] = diag_mapping
|
|
|
|
# Grouper codes
|
|
all_grouper_mappings = pd.concat([pd.read_csv(os.path.join(mapping_path, "grouper_cd", file), dtype=str) for file in os.listdir(os.path.join(mapping_path, "grouper_cd")) if file.endswith(".csv")])
|
|
grouper_mapping = CrosswalkBuilder().from_df(all_grouper_mappings, from_col="Code", to_col="Description")
|
|
grouper_version_mapping = CrosswalkBuilder().from_df(all_grouper_mappings, from_col="Code", to_col="Version")
|
|
all_mappings['grouper_mapping'] = grouper_mapping
|
|
all_mappings['grouper_version_mapping'] = grouper_version_mapping
|
|
|
|
# Rev codes
|
|
rev_mapping = CrosswalkBuilder().from_excel(path=os.path.join(mapping_path, "rev_cd", "rev_mapping.csv"), from_col="Code", to_col="Description")
|
|
all_mappings['rev_mapping'] = rev_mapping
|
|
|
|
# Admit Type codes
|
|
admit_mapping = CrosswalkBuilder().from_excel(path=os.path.join(mapping_path, "admit_type_cd", "admit_mapping.csv"), from_col="Code", to_col="Description")
|
|
all_mappings['admit_mapping'] = admit_mapping
|
|
|
|
return all_mappings
|
|
|
|
|
|
def code_description_mapping(answer_dicts, all_mappings):
|
|
"""
|
|
Maps code descriptions to relevant fields in the answer dictionaries using provided mappings.
|
|
|
|
Args:
|
|
answer_dicts (list): List of dictionaries containing codes to be mapped.
|
|
all_mappings (dict): Dictionary containing the mappings for different code types.
|
|
|
|
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("FACILITY_GROUPER_CD")
|
|
admit_type_code = answer_dict.get("CLAIM_ADMIT_TYPE_CD")
|
|
|
|
answer_dict["CPT4_PROC_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'])
|
|
answer_dict["FACILITY_GROUPER_DESC"] = get_code_description(grouper_code, all_mappings['grouper_mapping'])
|
|
answer_dict["GROUPER_TYPE"] = get_code_description(grouper_code, all_mappings['grouper_version_mapping'])
|
|
answer_dict["AUTH_ADMIT_TYPE_DESC"] = get_code_description(admit_type_code, all_mappings['admit_mapping'])
|
|
|
|
return answer_dicts
|
|
|
|
def get_code_breakout(answer_dicts: list[dict], filename: str, all_dataset: dict):
|
|
"""
|
|
Extracts and maps codes from contract data, including primary, indirect, and description mappings.
|
|
|
|
Args:
|
|
answer_dicts (list[dict]): List of dictionaries containing contract-related answers.
|
|
filename (str): The name of the file being processed.
|
|
all_dataset (dict): Dictionary containing crosswalk and mappings for code translation.
|
|
|
|
Returns:
|
|
dict: A dictionary with primary, indirect/derived, and mapped code descriptions.
|
|
"""
|
|
|
|
# Code Primary (Pull codes directly from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD)
|
|
code_primary_answers = code_primary(answer_dicts, filename)
|
|
|
|
# Only used for Implicit. Code Indirect (Derive codes indirectly from CONTRACT_SERVICE_CD_OR_DESC and CONTRACT_REIMBURSEMENT_METHOD)
|
|
code_indirect_answers = code_indirect(code_primary_answers, all_dataset['codes_crosswalk'])
|
|
|
|
# Code Descriptions (Pull standardized descriptions from codes)
|
|
code_description_answers = code_description_mapping(code_indirect_answers, all_dataset['codes_mappings'])
|
|
|
|
return code_description_answers |