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

Merged in feature/split-reimbursements (pull request #593)

Browse Source 
Feature/split reimbursements

* Update reimbursement test cases to reflect accurate reimbursement terms and improve deduplication logic

* move VALIDATE_REIMBURSEMENTS_PROMPT to investment_prompts.py

* black, isort formatting

* Remove IDENTIFY_REIMBURSEMENT_EXHIBITS_PROMPT function (unused)

* Add MODEL_ID_CLAUDE4_SONNET for new model integration (doesn't work right now)

* pipe delimited-output and parsing

* Merge remote-tracking branch 'origin/main' into feature/split-and-filter-reimbursements

* Merge remote-tracking branch 'origin/main' into feature/split-reimbursements

* Add functionality to split compound reimbursement terms into distinct methodologies

* splitting logic for reimbursement terms

* put in split compound reimbursement step

* Fix JSON output formatting in reimbursement terms example

* reimbursement splitting: move prompt to investment_prompts.py, move keywords to investment_values.py

* Merge remote-tracking branch 'origin/main' into feature/split-reimbursements

* Merge remote-tracking branch 'origin/main' into feature/split-reimbursements

* Change order (split -> filter over filter -> split), improve logging

* Refactor reimbursement filtering logic to use a two-stage approach with pattern-based negative filtering and LLM validation

* Replace clear reimbursement indicators with obvious non-reimbursement indicators for improved clarity in reimbursement filtering

* Simplify reimbursement filtering by removing pattern-based checks and relying solely on LLM validation

* Remove obvious non-reimbursement indicators from reimbursement filtering keywords

* Enhance reimbursement level test by adding LLM validation and updating mock return values for service terms

* Merge remote-tracking branch 'origin/main' into feature/split-reimbursements

* Fix failing test

* Merge remote-tracking branch 'origin/main' into feature/split-reimbursements

* update mock return values to fix failed tests


Approved-by: Siddhant Medar
This commit is contained in:
Alex Galarce
2025-07-02 17:33:18 +00:00
parent 5d65251851
commit 0230d07dda
5 changed files with 232 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/constants/investment_values.py
+32 -50
View File
@@ -149,53 +149,35 @@ VALID_SPECIAL = {
proc_levels = ["cpt", "hcpcs", "cpt_level3", "cpt_level2", "hcpcs_level2", "cpt_level1", "hcpcs_level1"]
diag_levels = ["diag"]
### Reimbursement Filtering Keywords
clear_reimbursement_indicators = [
"paid at",
"reimbursed at",
"will be paid",
"will be reimbursed at",
"%",
"$",
"conversion factor",
"lesser of",
"greater of",
"not to exceed",
"billed",
"allowable",
# Rate patterns
"reimbursement rate",
"payment rate",
"rate of $",
"hourly rate",
"daily rate",
"per diem rate",
"capitation rate",
"flat rate",
# Cost patterns
"cost sharing",
"cost plus",
"cost based",
"at cost",
"cost per",
# Fee schedule patterns
"% of fee schedule",
"fee schedule rate",
"according to fee schedule",
"based on fee schedule",
"per fee schedule",
# Medicare patterns
"% of medicare",
"medicare rate",
"medicare allowable",
"according to medicare",
"based on medicare",
# Contract patterns
"contract rate",
"contracted rate",
"contract amount",
"per contract",
"according to contract",
]
### Reimbursement splitting keywords
# Patterns that indicate compound reimbursements
compound_indicators = [
# Exception/fallback patterns
"for services where there is no",
"where no rate exists",
"if no rate is available",
"for which there is no",
"when no payment rate",
"where there is no payment rate",
# Alternative methodology patterns
"otherwise shall be",
"shall otherwise be",
"alternatively",
"in the alternative",
"or alternatively",
# Conditional patterns
"for certain services",
"for specific services",
"for other services",
"for all other",
"for any services",
# Exception clauses
"except for",
"with the exception of",
"excluding",
"however, for",
"notwithstanding"
]
fieldExtraction/src/investment/one_to_n_funcs.py
+136 -58
View File
@@ -39,7 +39,7 @@ def get_exhibit_level_answers(exhibit_chunk, filename):
return llm_answer_final
def get_reimbursement_primary(reimbursement_level_fields, exhibit_text, filename):
def get_reimbursement_primary(reimbursement_level_fields, exhibit_text, filename) -> list[dict]:
field_prompts = reimbursement_level_fields.print_prompt_dict()
prompt = investment_prompts.REIMBURSEMENT_LEVEL_PRIMARY(exhibit_text, field_prompts)
logging.debug(
@@ -481,27 +481,27 @@ def reimbursement_level(
) # Returns list of dicts
if reimbursement_primary_answers: # If any reimbursements found
# Deduplicate against previously seen (SERVICE_TERM, REIMB_TERM) pairs
# Step 1: Split compound reimbursements FIRST
split_answers = split_compound_reimbursements(reimbursement_primary_answers, filename)
# Step 2: Filter out any non-reimbursement content from split answers
filtered_answers = filter_services_without_reimbursements(
split_answers, filename
)
# Step 3: Deduplicate against previously seen (SERVICE_TERM, REIMB_TERM) pairs
deduplicated_answers = deduplicate_with_accumulator(
reimbursement_primary_answers, seen_pairs, exhibit_page, filename
filtered_answers, seen_pairs, exhibit_page, filename
)
if deduplicated_answers: # Only process if we have unique pairs
# Filter out services without actual reimbursement terms
filtered_answers = filter_services_without_reimbursements(
deduplicated_answers, filename
if deduplicated_answers: # Only process if we have valid reimbursement pairs
# Step 4: Get carveout and special case answers
# This will also run the methodology breakout prompt for non-carveout rows
carveout_answers = get_special_cases(deduplicated_answers, filename)
code_breakout_answers = code_funcs.get_code_breakout(
carveout_answers, filename, all_dataset
)
if filtered_answers: # Only continue if we have valid reimbursement pairs
carveout_answers = get_special_cases(filtered_answers, filename)
code_breakout_answers = code_funcs.get_code_breakout(
carveout_answers, filename, all_dataset
)
return code_breakout_answers
else:
logging.info(
f"No valid reimbursement pairs found in exhibit {exhibit_page} of {filename}."
)
return [] # Return empty list if no valid reimbursement pairs
return code_breakout_answers
else:
return [] # All pairs were duplicates, return empty list
@@ -554,56 +554,43 @@ def filter_services_without_reimbursements(
answers: list[dict], filename: str
) -> list[dict]:
"""
Filters out services that lack actual reimbursement methodologies.
Args:
answers (list[dict]): List of dictionaries containing reimbursement details.
filename (str): The name of the file being processed.
Filters out services that lack actual reimbursement methodologies.
Returns:
list[dict]: Filtered list of dictionaries containing only services with reimbursement methodologies.
Uses a single-step approach of LLM validation for remaining entries to determine if they contain actual reimbursement information.
Example:
>>> answers = [
{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee
schedule for year prior. Repairs for optical hardware are not covered."},
{"SERVICE_TERM": "Eyeglass Lenses", "REIMB_TERM": "Standard eyeglass lenses as defined by DMAP are CR39
lenses including scratch resistant coating, +/- 7D sphere up to -4D cylinder."}, # Note that there is no actual reimbursement information here
]
>>> filtered_answers = filter_services_without_reimbursements(answers, "example_file.txt")
>>> print(filtered_answers)
[{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee
schedule for year prior. Repairs for optical hardware are not covered."}]
Args:
answers (list[dict]): List of dictionaries containing reimbursement details.
filename (str): The name of the file being processed.
Returns:
list[dict]: Filtered list of dictionaries containing only services with reimbursement methodologies.
Example:
>>> answers = [
{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee schedule for year prior. Repairs for optical hardware are not covered."},
{"SERVICE_TERM": "Eyeglass Lenses", "REIMB_TERM": "Standard eyeglass lenses as defined by DMAP are CR39 lenses including scratch resistant coating, +/- 7D sphere up to -4D cylinder."}, # Note that there is no actual reimbursement information here
{"SERVICE_TERM": "Covered Services", "REIMB_TERM": "Notwithstanding the above, Provider agrees that in the event there is a retroactive change to rates, Health Plan shall have sole discretion to determine if claims will be reprocessed."} # Administrative content
]
>>> filtered_answers = filter_services_without_reimbursements(answers, "example_file.txt")
>>> print(filtered_answers)
[{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee schedule for year prior. Repairs for optical hardware are not covered."}]
"""
# Stage 1: Allow obvious reimbursement cases
clear_reimbursement = []
needs_llm_review = []
valid_reimbursements = []
for answer_dict in answers:
reimb_term = str(answer_dict.get("REIMB_TERM", "")).strip().lower()
if any(indicator in reimb_term for indicator in investment_values.clear_reimbursement_indicators):
clear_reimbursement.append(answer_dict)
if validate_reimbursements_for_llm(answer_dict, filename):
valid_reimbursements.append(answer_dict)
else:
needs_llm_review.append(answer_dict)
# Stage 2: LLM review for ambiguous cases
if needs_llm_review:
logging.debug(
f"Sending {len(needs_llm_review)} reimbursement terms for LLM validation in {filename}."
)
for answer_dict in needs_llm_review:
if validate_reimbursements_for_llm(answer_dict, filename):
clear_reimbursement.append(answer_dict)
else:
logging.debug(f"LLM filtered out: {answer_dict}")
filtered_count = len(answers) - len(clear_reimbursement)
logging.debug(f"LLM filtered out: {answer_dict}")
filtered_count = len(answers) - len(valid_reimbursements)
if filtered_count > 0:
logging.info(
f"Filtered {filtered_count} services without clear reimbursement methodologies in {filename}."
)
return clear_reimbursement
return valid_reimbursements
def validate_reimbursements_for_llm(answer_dict: dict, filename: str) -> bool:
@@ -638,3 +625,94 @@ def validate_reimbursements_for_llm(answer_dict: dict, filename: str) -> bool:
).strip().upper()
return final_answer == "YES"
def split_compound_reimbursements(answers: list[dict], filename: str) -> list[dict]:
"""
Splits reimbursement terms that contain multiple distinct methodologies.
Args:
answers (list[dict]): List of dictionaries containing reimbursement details.
filename (str): The name of the file being processed.
Returns:
list[dict]: List of dictionaries with split reimbursement terms.
"""
split_answers = []
for answer_dict in answers:
reimb_term = answer_dict.get("REIMB_TERM", "").lower().strip()
# Check if this looks like a compound reimbursement
has_compound = any(
indicator in reimb_term for indicator in investment_values.compound_indicators
)
if has_compound:
# Use LLM to split the compound reimbursement
split_result = split_compound_reimbursement_llm(
answer_dict, filename)
if split_result and len(split_result) > 1:
logging.info(f"Split compound reimbursement in {filename}: "
f"{answer_dict['SERVICE_TERM']} - {answer_dict['REIMB_TERM']}"
f" into {len(split_result)} parts.")
logging.info(f"Split results: {split_result}")
split_answers.extend(split_result)
else: # If LLM didn't split, keep the original
split_answers.append(answer_dict)
else: # If not compound, keep the original
split_answers.append(answer_dict)
return split_answers
def split_compound_reimbursement_llm(answer_dict: dict, filename: str) -> list[dict]:
"""
Uses LLM to split a compound reimbursement term into distinct methodologies.
Args:
answer_dict (dict): The service-reimbursement answer dictionary.
filename (str): The name of the file being processed.
Returns:
list[dict]: List of dictionaries with split reimbursement terms.
"""
service_term = answer_dict.get("SERVICE_TERM", "").strip()
reimb_term = answer_dict.get("REIMB_TERM", "").strip()
prompt = investment_prompts.SPLIT_REIMBURSEMENTS_PROMPT(
service_term, reimb_term)
try:
llm_response = llm_utils.invoke_claude(
prompt, config.MODEL_ID_CLAUDE3_HAIKU, filename, max_tokens=4096
)
split_data = string_utils.universal_json_load(llm_response)
if not split_data or len(split_data) == 1:
return [answer_dict] # No split needed, return original
# Create new answer dictionaries for each split
split_answers = []
for i, split_item in enumerate(split_data):
new_answer = answer_dict.copy()
# Update service and reimbursement terms
split_service = split_item.get("SERVICE_TERM", "").strip()
split_methodology = split_item.get("REIMB_TERM", "").strip()
new_answer['SERVICE_TERM'] = split_service
new_answer['REIMB_TERM'] = split_methodology
# Add a flag to indicate that this was split
new_answer["COMPOUND_SPLIT_IND"] = "Y"
new_answer["COMPOUND_SPLIT_SEQUENCE"] = str(i+1)
split_answers.append(new_answer)
return split_answers
except Exception as e:
logging.error(f"Error occurred while splitting reimbursement terms in {filename}: {e}")
return [answer_dict] # Return original if error occurs
fieldExtraction/src/prompts/investment_prompts.py
+43
View File
@@ -350,6 +350,49 @@ Examples:
Briefly explain your reasoning, then return YES or NO in |pipes|."""
def SPLIT_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str:
""" Prompt for splitting a service-reimbursement pair into its component parts.
Args:
service_term (str): The service term to split.
reimb_term (str): The reimbursement term to split.
Returns:
str: A prompt for the LLM to split the pair into its components.
"""
return f"""Analyze this reimbursement term to determine if it contains multiple distinct payment methodologies that should be separated.
SERVICE: {service_term}
REIMBURSEMENT: {reimb_term}
Look for patterns where:
1. There's a primary payment method AND a secondary/fallback method
2. Different conditions trigger different payment calculations
3. Exception clauses create alternative payment paths
4. "For services where..." or similar language introduces a different methodology
If you find multiple distinct methodologies, split them into separate entries. For each methodology, provide:
- The portion the service description it applies to
- The specific reimbursement methodology text
Return your answer as a JSON array. If no splitting is needed, return the original service and reimbursement as a single entry.
Example input:
SERVICE: Covered Services
REIMBURSEMENT: Covered Services shall be paid at an amount equivalent to eighty percent (80%) of the payment Hospital would otherwise have been entitled to had the Covered Services been billed directly under the prevailing local and geographically adjusted Medicare Fee-For-Service program allowable payment rates as of the date(s) of service. For certain Covered Services where there is no payment rate in the prevailing Medicare Fee-For-Service program as of the date(s) of service, payment for such Covered Services shall be at thirty percent (30%) of Hospital's billed charges.
Example output:
[
{{
"SERVICE_TERM": "Covered Services",
"REIMB_TERM": "Covered Services shall be paid at an amount equivalent to eighty percent (80%) of the payment Hospital would otherwise have been entitled to had the Covered Services been billed directly under the prevailing local and geographically adjusted Medicare Fee-For-Service program allowable payment rates as of the date(s) of service."
}},
{{
"SERVICE_TERM": "Covered Services",
"REIMB_TERM": "For certain Covered Services where there is no payment rate in the prevailing Medicare Fee-For-Service program as of the date(s) of service, payment for such Covered Services shall be at thirty percent (30%) of Hospital's billed charges."
}}
]
Feel free to justify your answer, but ensure the output is a valid JSON array with each entry containing SERVICE_TERM and REIMB_TERM keys.
"""
def EXHIBIT_LEVEL(context, fields):
return f"""Extract attributes from a section of a contract.
fieldExtraction/tests/dynamic_test.py
+6 -4
View File
@@ -132,12 +132,14 @@ class TestOneToN(unittest.TestCase):
@patch('src.investment.one_to_n_funcs.get_reimbursement_primary')
@patch('src.investment.one_to_n_funcs.get_special_cases')
@patch('src.investment.code_funcs.get_code_breakout')
def test_reimbursement_level(self, mock_code_breakout,
@patch('src.investment.one_to_n_funcs.validate_reimbursements_for_llm')
def test_reimbursement_level(self, mock_validate_llm, mock_code_breakout,
mock_special_cases, mock_reimbursement_primary):
# Setup
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1$"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1$", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1$", "CODE": "code1"}]
mock_reimbursement_primary.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1$"}]
mock_special_cases.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1$", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1$", "CODE": "code1"}]
mock_validate_llm.return_value = True # Mock all entries pass LLM validation
mock_fields = MagicMock()
mock_dataset = {}
fieldExtraction/tests/one_to_n_test.py
+15 -8
View File
@@ -94,12 +94,14 @@ class TestOneToN(unittest.TestCase):
@patch('src.investment.one_to_n_funcs.get_reimbursement_primary')
@patch('src.investment.one_to_n_funcs.get_special_cases')
@patch('src.investment.code_funcs.get_code_breakout')
def test_reimbursement_level(self, mock_code_breakout,
@patch('src.investment.one_to_n_funcs.validate_reimbursements_for_llm')
def test_reimbursement_level(self, mock_validate_llm, mock_code_breakout,
mock_special_cases, mock_reimbursement_primary):
# Setup
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1 paid at rate"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1 paid at rate", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1 paid at rate", "CODE": "code1"}]
mock_reimbursement_primary.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1 paid at rate"}]
mock_validate_llm.return_value = True
mock_special_cases.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1 paid at rate", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"SERVICE_TERM": "Test Service", "REIMB_TERM": "term1 paid at rate", "CODE": "code1"}]
mock_fields = MagicMock()
mock_dataset = {}
@@ -119,7 +121,8 @@ class TestOneToN(unittest.TestCase):
mock_code_breakout.assert_called_once()
@patch('src.investment.one_to_n_funcs.get_reimbursement_primary')
def test_reimbursement_level_deduplication(self, mock_reimbursement_primary):
@patch('src.investment.one_to_n_funcs.validate_reimbursements_for_llm')
def test_reimbursement_level_deduplication(self, mock_validate_llm, mock_reimbursement_primary):
"""Test that duplicate service-reimbursement pairs are properly filtered"""
# Setup - simulate getting duplicate pairs
mock_reimbursement_primary.return_value = [
@@ -127,6 +130,7 @@ class TestOneToN(unittest.TestCase):
{"SERVICE_TERM": "Service B", "REIMB_TERM": "$50 per visit"},
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"} # Duplicate
]
mock_validate_llm.return_value = True
mock_fields = MagicMock()
mock_dataset = {}
@@ -154,10 +158,12 @@ class TestOneToN(unittest.TestCase):
self.assertIn(("Service B", "$50 per visit"), seen_pairs)
@patch('src.investment.one_to_n_funcs.get_reimbursement_primary')
def test_reimbursement_level_cross_exhibit_deduplication(self, mock_reimbursement_primary):
@patch('src.investment.one_to_n_funcs.validate_reimbursements_for_llm')
def test_reimbursement_level_cross_exhibit_deduplication(self, mock_validate_llm, mock_reimbursement_primary):
"""Test that pairs seen in previous exhibits are filtered out"""
# Setup - pre-populate seen_pairs as if previous exhibit processed
seen_pairs = {("Service A", "paid at 80% of fee schedule"),}
mock_validate_llm.return_value = True
# Current exhibit has same pair plus new one
mock_reimbursement_primary.return_value = [
@@ -187,10 +193,11 @@ class TestOneToN(unittest.TestCase):
# Verify both pairs are now in seen_pairs
self.assertEqual(len(seen_pairs), 2)
def test_reimbursement_level_all_duplicates(self):
@patch('src.investment.one_to_n_funcs.validate_reimbursements_for_llm')
def test_reimbursement_level_all_duplicates(self, mock_validate_llm):
"""Test behavior when all pairs are duplicates"""
seen_pairs = {("Service A", "paid at 80% of fee schedule"), ("Service B", "$50 per visit")}
mock_validate_llm.return_value = True
with patch('src.investment.one_to_n_funcs.get_reimbursement_primary') as mock_primary:
mock_primary.return_value = [
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"},