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

Merged in feature/dynamic-codes-final (pull request #942)

Browse Source 
Feature/dynamic codes final

* Move Claim Type to dynamic code group

* Update CLaim Type and Bill TYpe prompts

* Prompt optimization

* Prompt tuning

* streamline

* Format

* Merged dev into feature/dynamic-codes

* test claim and bill type

* Merged dev into DAIP2-2186-service-claim-type-testing

* Merged dev into feature/dynamic-codes

* prompt updated for claim type

* Merge branch 'feature/dynamic-codes' into DAIP2-2186-service-claim-type-testing

* Merged dev into DAIP2-2186-service-claim-type-testing

* Remove print

* Switch to map

* Merged dev into feature/dynamic-codes-final

* prompt inconsistency fixed


Approved-by: Praneel Panchigar
Approved-by: Siddhant Medar
This commit is contained in:
Katon Minhas
2026-04-07 18:33:28 +00:00
parent 5743a4c1be
commit 491abd31c2
7 changed files with 178 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/constants/mappings/crosswalk_claim_type.json
+1 -3
View File
@@ -6,8 +6,6 @@
"mapping": {
"Professional": "M",
"Ancillary":"M",
"Facility":"H",
"Hospital":"H",
"Institutional":"H"
"Facility":"H"
}
}
src/document_classification/main.py
+1 -1
View File
@@ -108,7 +108,7 @@ def _load_lob_keywords_from_mappings():
if keyword and normalized_lob:
keyword_to_lob[keyword] = normalized_lob
logging.info(
logging.debug(
f"Loaded {len(keyword_to_lob)} LOB keywords from "
f"{len(_LOB_JSON_FILES)} mapping files"
)
src/pipelines/saas/file_processing.py
+7 -42
View File
@@ -89,7 +89,6 @@ def process_file(file_object, constants: Constants, run_timestamp):
(
one_to_n_results,
dynamic_one_to_one_fields,
first_reimbursement_page,
) = run_one_to_n_prompts(
pages_dict=pages_dict,
text_dict=text_dict,
@@ -107,18 +106,8 @@ def process_file(file_object, constants: Constants, run_timestamp):
one_to_n_results
)
logging.debug(f"{datetime_str()} One to N Complete - {filename}")
else:
first_reimbursement_page = "1"
logging.debug(
f"{datetime_str()} No Reimbursement Found, Skipping - {filename}"
)
final_results = one_to_n_results # Set as final results
else:
first_reimbursement_page = "1"
logging.debug(
f"{datetime_str()} Fields not configured for One to N, Skipping - {filename}"
)
# ONE TO ONE PROCESSING
if process_one_to_one:
@@ -129,9 +118,7 @@ def process_file(file_object, constants: Constants, run_timestamp):
filename,
contract_text,
text_dict,
top_sheet_dict,
dynamic_one_to_one_fields,
first_reimbursement_page,
constants,
specific_fields=specific_fields,
)
@@ -189,9 +176,7 @@ def run_one_to_one_prompts(
filename: str,
contract_text: str,
text_dict: dict[str, str],
top_sheet_dict,
dynamic_one_to_one_fields: FieldSet,
first_reimbursement_page: str,
constants: Constants,
specific_fields: Optional[list] = None,
):
@@ -282,12 +267,8 @@ def run_one_to_one_prompts(
def process_page_reimbursements(
exhibit: Exhibit,
page_num: str,
pages_dict: Optional[dict[str, "Page"]] = None,
text_dict: Optional[dict[str, str]] = None,
exhibit_page_nums: Optional[list[str]] = None,
constants: Optional[Constants] = None,
filename: Optional[str] = None,
specific_fields: Optional[list] = None,
relevant_chunks: Optional[list[ExhibitChunk]] = None,
):
"""
@@ -350,6 +331,11 @@ def process_page_reimbursements(
)
)
################################ Dynamic Code Assignment ################################
reimbursement_level_answers = one_to_n_funcs.dynamic_code_assignment(
reimbursement_level_answers, constants, filename
)
############################### Lesser of Distribution ###############################
# Note: We run lesser_of without dynamic fields in Step 1
reimbursement_level_answers = one_to_n_funcs.lesser_of_distribution(
@@ -420,9 +406,7 @@ def run_one_to_n_prompts(
)
# Process exhibits serially; within each exhibit, process pages in parallel
total_chunks_processed = 0
one_to_n_results = []
first_reimbursement_page = "1"
for exhibit in exhibits:
# Search for reimbursement-related chunks, then group by page
@@ -448,12 +432,8 @@ def run_one_to_n_prompts(
process_page_reimbursements,
exhibit,
page_num,
pages_dict,
text_dict,
exhibit.exhibit_page_nums,
constants,
filename,
specific_fields,
relevant_chunks=page_chunks_map.get(
page_num
), # None for pages without relevant chunks
@@ -461,8 +441,6 @@ def run_one_to_n_prompts(
for page_num in pages_to_process
}
total_pages_in_exhibit = len(pages_to_process)
completed_pages = 0
for future in concurrent.futures.as_completed(page_futures):
try:
page_num = page_futures[future]
@@ -470,15 +448,6 @@ def run_one_to_n_prompts(
exhibit.add_reimbursement_rows(
reimbursement_rows, special_case_rows
)
completed_pages += 1
total_chunks_processed += 1
if (
completed_pages % 20 == 0
or completed_pages == total_pages_in_exhibit
):
logging.debug(
f"{datetime_str()} Page progress for exhibit {exhibit.exhibit_page}: {completed_pages}/{total_pages_in_exhibit} - {filename}"
)
except Exception as e:
logging.error(f"Error processing page {page_num}: {str(e)}")
@@ -520,7 +489,7 @@ def run_one_to_n_prompts(
)
if skip_reimbursement_processing:
# For exhibit-level-only extraction (like claim_type), just create minimal rows
# For exhibit-level-only extraction, just create minimal rows
# with exhibit_level_answers to preserve the data
if exhibit.exhibit_level_answers:
minimal_row = exhibit.exhibit_level_answers.copy()
@@ -580,10 +549,6 @@ def run_one_to_n_prompts(
f"{len(combined_rows)} final row(s) after combine & clean - {filename}"
)
# Track first reimbursement page
if first_reimbursement_page == "1" and exhibit.has_reimbursements:
first_reimbursement_page = exhibit.exhibit_page
logging.debug(
f"{datetime_str()} STEP 3 COMPLETE: Combined {len(one_to_n_results)} total rows - {filename}"
)
@@ -596,4 +561,4 @@ def run_one_to_n_prompts(
################## CONVERT TO DF ##################
one_to_n_df = pd.DataFrame(one_to_n_results)
return one_to_n_df, dynamic_one_to_one_fields, first_reimbursement_page
return one_to_n_df, dynamic_one_to_one_fields
src/pipelines/saas/prompts/prompt_calls.py
+20 -6
View File
@@ -190,6 +190,26 @@ def prompt_reimbursement_primary(
raise
def prompt_dynamic_code_assignment(service_term: str, filename: str):
prompt, _parser = prompt_templates.DYNAMIC_CODE_ASSIGNMENT(service_term)
logging.debug(f"Dynamic Code Assignment Prompt for {filename}: {prompt}")
llm_response = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
filename,
cache=True,
instruction=prompt_templates.DYNAMIC_CODE_ASSIGNMENT_INSTRUCTION(),
usage_label="DYNAMIC_CODE_ASSIGNMENT",
)
logging.debug(f"LLM Response for {filename}: {llm_response}")
try:
dynamic_code_assignment_answers = _parser(llm_response)
except ValueError as e:
dynamic_code_assignment_answers = {}
return dynamic_code_assignment_answers
def prompt_methodology_breakout(
service_term: str,
reimb_term: str,
@@ -215,12 +235,6 @@ def prompt_methodology_breakout(
logging.debug(f"LLM Response for {filename}: {llm_response}")
try:
methodology_breakout_answers = _parser(llm_response)
# Normalize output - methodology_breakout returns list[dict] with methodology fields
# Use FieldSet to get field names from single source of truth
methodology_breakout_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="methodology_breakout"
)
methodology_field_names = methodology_breakout_fields.list_fields()
# Handle both list and single dict cases
if isinstance(methodology_breakout_answers, dict):
methodology_breakout_answers = [methodology_breakout_answers]
src/pipelines/shared/extraction/one_to_n_funcs.py
+51 -14
View File
@@ -12,6 +12,53 @@ from src.prompts.fieldset import FieldSet
from src.pipelines.shared.extraction.exhibit_funcs import Exhibit
def dynamic_code_assignment(
reimbursement_level_answers: list[dict], constants: Constants, filename: str
):
final_answers = []
# Parallelize dynamic code assignment processing
if reimbursement_level_answers:
with concurrent.futures.ThreadPoolExecutor(
max_workers=min(len(reimbursement_level_answers), 5)
) as executor:
futures = [
executor.submit(
dynamic_code_assignment_single_row, answer_dict, filename
)
for answer_dict in reimbursement_level_answers
]
for future in futures:
try:
final_answers += future.result()
except Exception as e:
logging.error(f"Error in dynamic code assignment: {str(e)}")
return final_answers
def dynamic_code_assignment_single_row(
answer_dict: dict,
filename: str,
) -> list[dict]:
service_term = answer_dict.get("SERVICE_TERM", "")
if not service_term:
return [answer_dict]
dynamic_code_assignment_answers = prompt_calls.prompt_dynamic_code_assignment(
service_term,
filename,
)
if not dynamic_code_assignment_answers:
return [answer_dict]
return [{**answer_dict, **dynamic_code_assignment_answers}]
def exhibit_level(
exhibit_text: str,
exhibit_header: str,
@@ -60,7 +107,10 @@ def exhibit_level(
"exhibit_level.prompt_exhibit_level", context=filename
):
exhibit_level_answers = prompt_calls.prompt_exhibit_level(
exhibit_text, exhibit_level_fields, constants, filename
exhibit_text,
exhibit_level_fields.combine(dynamic_code_fields),
constants,
filename,
)
with timing_utils.timed_block(
@@ -86,19 +136,6 @@ def exhibit_level(
)
)
# Dynamic Codes (skip if filtering to specific fields)
if dynamic_code_fields.contains_fields():
with timing_utils.timed_block("exhibit_level.dynamic_codes", context=filename):
exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic(
exhibit_text,
exhibit_header,
exhibit_level_answers,
constants,
filename,
dynamic_code_fields,
dynamic_reimbursement_fields,
)
# Dynamic Reimb Info (skip if filtering to specific fields)
if dynamic_reimb_info_fields.contains_fields():
with timing_utils.timed_block(
src/prompts/investment_prompts.json
+32 -46
View File
@@ -151,35 +151,6 @@
"crosswalk": "CROSSWALK_PROGRAM",
"base_field": "PROGRAM"
},
{
"field_name": "CLAIM_TYPE_CD",
"relationship": "one_to_n",
"field_type": "exhibit_level",
"prompt": "What is the Claim Type of the contract, exhibit, or services mentioned in the text? Choose ONLY from the following: {valid_values}.\n\nPRIORITY: First check the title or header for explicit claim type indicators before analyzing the body text.\n\nGuidelines:\n- Return 'Professional' if the title/header indicates physician, professional services, or provider group agreement, OR if the text refers to services rendered by individual healthcare providers or provider groups (e.g., physician services, office visits, outpatient procedures).\n- Return 'Institutional' if the title/header indicates hospital, facility, or institutional agreement, OR if the text refers to services provided by hospitals, facilities, or institutions (e.g., inpatient, outpatient facility, or surgery centers).\n- Return 'Ancillary' if the title/header indicates ancillary services, OR if the text refers to services such as Home Health, Durable Medical Equipment (DME), Laboratory, Radiology, Dialysis, Diagnostic Imaging, Genetic Testing, Blood Testing, Sleep Lab Services, Telemedicine, Behavioral or Mental Health, or other ancillary settings.\n- If multiple claim types are mentioned, return the one most closely associated with the title/header first, then the compensation or service description in the context.\n- If no claim type is mentioned or cannot be determined, return 'N/A'.\nDo not infer or guess.",
"valid_values": "VALID_CLAIM_TYPE",
"keywords": [
"AGREEMENT",
"AMENDMENT",
"ADDENDUM",
"PROVIDER",
"CONTRACT",
"MEMORANDUM",
"LETTER",
"REQUEST",
"EFFECTIVE"
],
"retrieval_question": "What is the title of this agreement? Look for the header, title, agreement name, or exhibit title. Common titles include Professional Services Agreement, Physician Agreement, Hospital Agreement, Institutional Provider Agreement, Facility Agreement, Ancillary Services Agreement, DME Provider Agreement, Home Health Agreement, Laboratory Services Agreement.",
"allow_na": true
},
{
"field_name": "AARETE_DERIVED_CLAIM_TYPE_CD",
"base_field": "CLAIM_TYPE_CD",
"relationship": "one_to_n",
"field_type": "crosswalk",
"valid_values": "TBD",
"prompt": "",
"crosswalk": "CROSSWALK_CLAIM_TYPE_CD"
},
{
"field_name": "CONTRACT_TITLE",
"relationship": "one_to_one",
@@ -380,6 +351,38 @@
"case_sensitive": "True",
"retrieval_question": "What is the length, duration, or period of the automatic renewal term, successive renewal term, or additional renewal period? How long is each renewal period or renewal term in years or months?"
},
{
"field_name": "CLAIM_TYPE_CD",
"relationship": "one_to_n",
"field_type": "dynamic_code",
"prompt": "Claim Type is a classification indicating whether rates apply to a Facility, Professional, or Ancillary provider. Choose ONLY from the following Valid Values: {valid_values}.\n\nSTRICT MATCHING RULES - DO NOT INFER OR ESTIMATE:\n- 'Facility': ONLY if the text explicitly contains one of these exact terms: 'Facility', 'Hospital', 'Institutional', 'Inpatient', 'Outpatient'\n- 'Professional': ONLY if the text explicitly contains one of these exact terms: 'Professional', 'Physician', 'Medical' \n- 'Ancillary': ONLY if the text explicitly contains the exact term: 'Ancillary' \n\nIf none of these specific terms appear in the text, the answer is N/A. Do not classify based on context, implication, or related terminology. The exact terms listed above MUST be present.",
"valid_values": "VALID_CLAIM_TYPE"
},
{
"field_name": "AARETE_DERIVED_CLAIM_TYPE_CD",
"base_field": "CLAIM_TYPE_CD",
"relationship": "one_to_n",
"field_type": "crosswalk",
"valid_values": "TBD",
"prompt": "",
"crosswalk": "CROSSWALK_CLAIM_TYPE_CD"
},
{
"field_name": "BILL_TYPE_CD",
"base_field": "BILL_TYPE_CD_DESC",
"relationship": "one_to_n",
"field_type": "dynamic_code",
"prompt": "Specifies the type of service, by location. Choose only from the following Valid Values: {valid_values}. If the text does not explicitly mention any of the valid values, return N/A.",
"crosswalk": "CROSSWALK_BILL_TYPE_CD",
"valid_values" : "VALID_BILL_TYPE"
},
{
"field_name": "BILL_TYPE_CD_DESC",
"base_field": "BILL_TYPE_CD",
"relationship": "one_to_n",
"field_type": "TBD",
"prompt": ""
},
{
"field_name": "PROV_SPECIALTY_CD",
"base_field": "PROV_SPECIALTY_CD_DESC",
@@ -414,23 +417,6 @@
"valid_values": "VALID_PLACE_OF_SERVICE",
"crosswalk": "CROSSWALK_PLACE_OF_SERVICE_CD"
},
{
"field_name": "BILL_TYPE_CD",
"base_field": "BILL_TYPE_CD_DESC",
"relationship": "one_to_n",
"field_type": "dynamic_code",
"prompt": "4-character alphanumeric codes. The first character is often a leading zero, which may or may not be included. The last character may either be a digit 0-9, or the letter X. If no codes are explicitly written in the text, return N/A.",
"crosswalk": "CROSSWALK_BILL_TYPE_CD"
},
{
"field_name": "BILL_TYPE_CD_DESC",
"base_field": "BILL_TYPE_CD",
"relationship": "one_to_n",
"field_type": "TBD",
"prompt": "",
"valid_values": "VALID_BILL_TYPE",
"crosswalk": "CROSSWALK_BILL_TYPE_CD"
},
{
"field_name": "PATIENT_AGE_RANGE",
"relationship": "one_to_n",
src/prompts/prompt_templates.py
+66 -6
View File
@@ -7,8 +7,7 @@ from src.utils import json_utils, string_utils
# JSON format instructions for standardized LLM outputs
JSON_DICT_FORMAT_INSTRUCTIONS = """Return your final answer as a valid JSON dictionary with the specified field names as keys.
- Use "N/A" for fields where the requested information doesn't apply to this context.
- Use "UNKNOWN" for fields where the information should exist but cannot be clearly identified.
- Use "N/A" for fields where the requested information doesn't apply to this context or cannot be clearly identified.
- Ensure the JSON is properly formatted with double quotes around keys and string values."""
JSON_LIST_FORMAT_INSTRUCTIONS = """Return your final answer as a valid JSON list (array).
@@ -18,8 +17,7 @@ JSON_LIST_FORMAT_INSTRUCTIONS = """Return your final answer as a valid JSON list
JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS = """Return your final answer as a valid JSON dictionary where values can be lists (arrays).
- Use lists for fields that may contain multiple values.
- Use "N/A" for fields where the requested information doesn't apply.
- Use "UNKNOWN" for fields where the information should exist but cannot be clearly identified.
- Use "N/A" for fields where the requested information doesn't apply or cannot be clearly identified
- Ensure the JSON is properly formatted with double quotes around keys and string values."""
JSON_LIST_OF_DICTS_FORMAT_INSTRUCTIONS = """Return your final answer as a valid JSON list (array) of dictionaries.
@@ -138,6 +136,15 @@ def get_cacheable_instructions():
except Exception:
pass
# DYNAMIC_CODE_ASSIGNMENT instruction
if config.FIELDS in ["all", "one_to_n"]:
try:
cacheable_instructions["DYNAMIC_CODE_ASSIGNMENT"] = (
DYNAMIC_CODE_ASSIGNMENT_INSTRUCTION()
)
except Exception:
pass
# Secondary breakout instructions
if config.FIELDS in ["all", "one_to_n"]:
try:
@@ -268,10 +275,18 @@ def EXHIBIT_LEVEL_INSTRUCTION() -> str:
Contains all formatting rules and general instructions.
"""
return f"""[OBJECTIVE]
Extract attributes from a section of a contract.
Extract Exhibit-Level attributes from a section of a contract.
[EXHIBIT LEVEL DETERMINATION]
- For each field, ONLY extract an answer if that answer applies to EVERY rate in the Exhibit. Some ways to determine if the answer applies to the entire Exhibit:
- The answer is clearly written in the Header or Title of the Exhibit.
- It is stated in a sentence or paragraph that the answer applies to all rates.
- The answer for each field must be clearly and explicitly written. Do not attempt to deduce the answer from other related values:
E.g. If the Exhibit only contains Services related to "Home", "Dialysis", and "Skilled Nursing Facility", then the Claim Type is not "Ancillary" even though those are all Ancillary Services.
In the example above, the Claim Type answer is not written explicitly, so the answer should be "N/A".
[GENERAL INSTRUCTIONS]
- For each field, return the correct answer. There will only be one correct answer.
- Return the correct answer. There will only be one correct answer.
- If a list of valid values is provided, ONLY answer with the EXACT value from that list.
- If there are no correct answers for a given field, write 'N/A' as the answer for that field.
- Be consistent and deterministic; avoid creative paraphrasing.
@@ -1012,6 +1027,51 @@ The output should always be a JSON list/array of dictionaries, even when there i
[CONTEXT]"""
def DYNAMIC_CODE_ASSIGNMENT_INSTRUCTION() -> str:
"""Static instruction for DYNAMIC_CODE_ASSIGNMENT prompt caching.
Contains extraction rules and dynamic_code field definitions with resolved valid_values.
Field definitions are loaded from investment_prompts.json.
"""
fields_text = _get_fields_text("dynamic_code")
return f"""[OBJECTIVE]
Analyze a Service Term from a Payer-Provider contract and classify it as one or more of the listed field values.
[INSTRUCTION]
- The answer for each field must be clearly and explicitly written. Do not attempt to deduce the answer from other related values.
- If there are multiple clear and obvious answers, use a list (array).
- If a field has valid values, choose ONLY from those valid values.
- Make appropriate assumptions only if the answer is obvious.
[FIELDS]
Populate a JSON dictionary with the following fields:
{fields_text}
[OUTPUT FORMAT]
Briefly explain your reasoning, then output a properly formatted JSON dictionary.
{JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS}
[CONTEXT]"""
def DYNAMIC_CODE_ASSIGNMENT(service_term: str) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for dynamic code assignment extraction.
Call DYNAMIC_CODE_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = f"""Here is the text to analyze and respond to:
Service Term: {service_term}
Explain your reasoning as you work through the context. After your explanation, include the heading "FINAL DYNAMIC CODE ASSIGNMENT:" followed by a properly formatted JSON dictionary with your final output."""
dynamic_code_fields = FieldSet(config.FIELD_JSON_PATH, field_type="dynamic_code")
field_names = [field.field_name for field in dynamic_code_fields.fields]
parser = _create_json_dict_parser(field_names=field_names)
return (prompt, parser)
def FEE_SCHEDULE_BREAKOUT_INSTRUCTION() -> str:
"""Static instruction for FEE_SCHEDULE_BREAKOUT prompt caching.
Contains objective, field definitions with resolved valid_values, and output format rules.