From f736bde98707d24e372c5392a36f109abefb9edb Mon Sep 17 00:00:00 2001 From: Faizan Mohiuddin Date: Thu, 16 Oct 2025 20:32:00 +0000 Subject: [PATCH] Merged in feature/prompt-caching (pull request #737) Feature/prompt caching * caching implemented * caching * Merge branch 'main' into feature/prompt-caching * Merged main into feature/prompt-caching * Merged main into feature/prompt-caching * caching * fixed lob * Merged main into feature/prompt-caching Approved-by: Katon Minhas --- fieldExtraction/src/investment/main.py | 20 +- .../src/investment/one_to_one_funcs.py | 15 +- .../src/investment/prompt_calls.py | 106 +++++- .../src/investment/tin_npi_funcs.py | 4 +- .../src/prompts/prompt_templates.py | 308 ++++++++++++++--- fieldExtraction/src/utils/llm_utils.py | 316 ++++++++++++++++-- 6 files changed, 667 insertions(+), 102 deletions(-) diff --git a/fieldExtraction/src/investment/main.py b/fieldExtraction/src/investment/main.py index 9c8f791..02a2caa 100644 --- a/fieldExtraction/src/investment/main.py +++ b/fieldExtraction/src/investment/main.py @@ -10,7 +10,9 @@ import pandas as pd random.seed(42) import src.investment.file_processing as file_processing +import src.prompts.prompt_templates as prompt_templates import src.utils.io_utils as io_utils +import src.utils.llm_utils as llm_utils import src.utils.logging_utils as logging_utils from constants.constants import Constants from src import config @@ -76,7 +78,8 @@ def main(testing=False, test_params={}): # Set up per-file logging (creates separate log files for each document) logging_utils.setup_per_file_logging() - run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H:%M_{config.BATCH_ID}") + # Windows paths cannot contain ':'; use '-' in time component + run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H-%M_{config.BATCH_ID}") # Read Constants constants = Constants() @@ -95,6 +98,21 @@ def main(testing=False, test_params={}): logging.info(f"Total Input Files : {len(input_dict)}") + # Warm prompt caches before parallel processing + # This ensures the cache is registered before multiple threads try to use it + logging.info("Warming prompt caches before parallel processing...") + try: + cacheable_instructions = prompt_templates.get_cacheable_instructions(constants) + for cache_key, instruction in cacheable_instructions.items(): + llm_utils.warm_prompt_cache( + instruction=instruction, + cache_key=cache_key, + model_id="sonnet_latest" + ) + logging.info(f"Successfully warmed {len(cacheable_instructions)} prompt caches") + except Exception as e: + logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") + # Process files concurrently - each thread gets its own per-file logging context successful_results = [] error_results = [] diff --git a/fieldExtraction/src/investment/one_to_one_funcs.py b/fieldExtraction/src/investment/one_to_one_funcs.py index d69b79a..96e8fda 100644 --- a/fieldExtraction/src/investment/one_to_one_funcs.py +++ b/fieldExtraction/src/investment/one_to_one_funcs.py @@ -47,7 +47,14 @@ def extract_global_lesser_of(contract_text: str, filename: str) -> dict: f"Extracting global lesser of statement for {filename} with prompt: {prompt}" ) - response = llm_utils.invoke_claude(prompt, "sonnet_latest", filename) + response = llm_utils.invoke_claude( + prompt, + "sonnet_latest", + filename, + cache=True, + instruction=prompt_templates.ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION(), + usage_label="ONE_TO_ONE_SINGLE_FIELD", + ) logging.debug(f"Response for global lesser of statement extraction: {response}") return { @@ -596,7 +603,7 @@ def run_global_lesser_of_breakout( # Create a mock service term for the breakout mock_service = "Global Lesser Of (all services)" - prompt = METHODOLOGY_BREAKOUT( + instruction, prompt = METHODOLOGY_BREAKOUT( [mock_service, global_lesser_of_stmt], FieldSet( file_path=config.FIELD_JSON_PATH, field_type="methodology_breakout" @@ -606,7 +613,9 @@ def run_global_lesser_of_breakout( f"Running global lesser of breakout for {filename} with prompt: {prompt}" ) - llm_response = llm_utils.invoke_claude(prompt, "sonnet_latest", filename) + llm_response = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, cache=True, instruction=instruction + , usage_label="GLOBAL_LESSER_OF_BREAKOUT") try: breakout_results = string_utils.universal_json_load(llm_response) diff --git a/fieldExtraction/src/investment/prompt_calls.py b/fieldExtraction/src/investment/prompt_calls.py index a5e554d..1466316 100644 --- a/fieldExtraction/src/investment/prompt_calls.py +++ b/fieldExtraction/src/investment/prompt_calls.py @@ -56,7 +56,13 @@ def prompt_dynamic_primary( ): prompt = TEMPLATE(exhibit_text, field.field_name, field.get_prompt(constants)) logging.debug(f"Dynamic primary prompt for {filename}; {field}: {prompt}") - llm_answer_raw = llm_utils.invoke_claude(prompt, "sonnet_latest", filename) + llm_answer_raw = llm_utils.invoke_claude( + prompt, + "sonnet_latest", + filename, + cache=True, + instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(), + ) logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}") llm_answer_final = string_utils.extract_text_from_delimiters( llm_answer_raw, string_utils.Delimiter.PIPE @@ -76,12 +82,19 @@ def prompt_reimbursement_primary( ) -> list[dict]: field_prompts = reimbursement_level_fields.print_prompt_dict(constants) - prompt = prompt_templates.REIMBURSEMENT_PRIMARY(exhibit_text, field_prompts) + instruction, prompt = prompt_templates.REIMBURSEMENT_PRIMARY(exhibit_text, field_prompts) + logging.debug( f"""Running reimbursement primary prompt for {filename} with prompt: {prompt}""" ) llm_answer_raw = llm_utils.invoke_claude( - prompt, "sonnet_latest", filename, max_tokens=25000 + prompt, + "sonnet_latest", + filename, + max_tokens=25000, + cache=True, + instruction=instruction, + usage_label="REIMBURSEMENT_PRIMARY", ) logging.debug(f"""LLM raw output for {filename}: {llm_answer_raw}""") @@ -105,13 +118,19 @@ def prompt_special_case_primary( filename: str, ): field_prompts = special_case_primary_fields.print_prompt_dict(constants) - prompt = prompt_templates.SPECIAL_CASE_PRIMARY(exhibit_text, field_prompts) + instruction, prompt = prompt_templates.SPECIAL_CASE_PRIMARY(exhibit_text, field_prompts) logging.debug( f"""Running Special Case Primary prompt for {filename} with prompt: {prompt}""" ) llm_answer_raw = llm_utils.invoke_claude( - prompt, "sonnet_latest", filename, max_tokens=25000 + prompt, + "sonnet_latest", + filename, + max_tokens=25000, + cache=True, + instruction=instruction, + usage_label="SPECIAL_CASE_PRIMARY", ) logging.debug(f"""LLM raw output for {filename}: {llm_answer_raw}""") @@ -137,7 +156,7 @@ def prompt_methodology_breakout( constants: Constants, filename: str, ): - prompt = prompt_templates.METHODOLOGY_BREAKOUT( + instruction, prompt = prompt_templates.METHODOLOGY_BREAKOUT( [service, reimbursement_method], methodology_breakout_questions.print_prompt_dict(constants), ) @@ -146,6 +165,8 @@ def prompt_methodology_breakout( prompt, "sonnet_latest", filename, + cache=True, + instruction=instruction, ) logging.debug(f"LLM Response for {filename}: {llm_response}") try: @@ -224,6 +245,9 @@ def prompt_carveout_check( carveout_prompt, "sonnet_latest", filename, + cache=True, + instruction=prompt_templates.CARVEOUT_CHECK_INSTRUCTION(), + usage_label="CARVEOUT_CHECK", ) logging.debug(f"LLM raw output for carveout check in {filename}: {llm_answer_raw}") carveout_answer = string_utils.extract_text_from_delimiters( @@ -233,10 +257,36 @@ def prompt_carveout_check( def prompt_special_case_breakout(breakout_template, term_answer, filename): - prompt = breakout_template(term_answer) - logging.debug(f"Special Case Breakout Prompt for {filename} : {prompt}") - llm_answer_raw = llm_utils.invoke_claude(prompt, "sonnet_latest", filename) - logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}") + # Some breakout templates (e.g., OUTLIER_BREAKOUT) return (instruction, prompt) + # while others return just a prompt string. Handle both cases safely. + prompt_obj = breakout_template(term_answer) + instruction = None + if isinstance(prompt_obj, tuple) and len(prompt_obj) == 2: + instruction, prompt = prompt_obj + else: + prompt = prompt_obj + + # Ensure prompt is a string for Bedrock messages API + if not isinstance(prompt, str): + logging.warning( + f"Expected prompt string, got {type(prompt)} in prompt_special_case_breakout; coercing to str." + ) + prompt = str(prompt) + + # If we have an instruction, enable prompt caching for efficiency + if instruction: + llm_answer_raw = llm_utils.invoke_claude( + prompt, + "sonnet_latest", + filename, + cache=True, + instruction=instruction, + usage_label=(getattr(breakout_template, "__name__", "SPECIAL_CASE_BREAKOUT")), + ) + else: + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, usage_label="SPECIAL_CASE_BREAKOUT" + ) llm_answer_final = string_utils.universal_json_load(llm_answer_raw) return llm_answer_final @@ -254,7 +304,11 @@ def prompt_lob_relationship( prompt = prompt_templates.LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values) llm_answer_raw = llm_utils.invoke_claude( - prompt, model_id="sonnet_latest", filename=filename + prompt, + model_id="sonnet_latest", + filename=filename, + cache=True, + instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(), ) llm_answer_final = string_utils.extract_text_from_delimiters( llm_answer_raw, Delimiter.PIPE @@ -325,7 +379,12 @@ def prompt_full_context( ) try: claude_answer_raw = llm_utils.invoke_claude( - full_context_prompt, "sonnet_latest", filename, 8192 + full_context_prompt, + "sonnet_latest", + filename, + 8192, + cache=True, + instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(), ) full_context_answers_dict = string_utils.universal_json_load(claude_answer_raw) except Exception as e: @@ -358,7 +417,13 @@ def is_exhibit_lesser_of_redundant( prompt = prompt_templates.CHECK_REDUNDANT_LESSER(reimb_term, exhibit_lesser_of) - response = llm_utils.invoke_claude(prompt, "haiku_latest", filename=filename) + response = llm_utils.invoke_claude( + prompt, + "haiku_latest", + filename=filename, + cache=True, + instruction=prompt_templates.CHECK_REDUNDANT_LESSER_INSTRUCTION(), + ) logging.debug(f"LLM response for redundancy check in {filename}: {response}") return ( string_utils.extract_text_from_delimiters(response, Delimiter.PIPE) @@ -394,6 +459,8 @@ def validate_reimbursements_for_llm(answer_dict: dict, filename: str) -> bool: prompt, "sonnet_latest", filename, + cache=True, + instruction=prompt_templates.VALIDATE_REIMBURSEMENTS_INSTRUCTION(), ) logging.debug( f"LLM response for reimbursement validation in {filename}:\n{llm_response}" @@ -426,7 +493,12 @@ def split_compound_reimbursement_llm(answer_dict: dict, filename: str) -> list[d f"Invoking LLM for splitting reimbursements in {filename};\nPrompt:\n{prompt}" ) llm_response = llm_utils.invoke_claude( - prompt, "haiku_latest", filename, max_tokens=4096 + prompt, + "haiku_latest", + filename, + max_tokens=4096, + cache=True, + instruction=prompt_templates.SPLIT_REIMBURSEMENTS_INSTRUCTION(), ) logging.debug( f"LLM response for splitting reimbursements in {filename}:\n{llm_response}" @@ -477,7 +549,11 @@ def prompt_dynamic(text, field_prompts, filename): prompt = prompt_templates.EXHIBIT_LEVEL(text, field_prompts) logging.debug(f"Dynamic prompt for {filename}: {prompt}") llm_answer_raw = llm_utils.invoke_claude( - prompt, "sonnet_latest", filename + prompt, + "sonnet_latest", + filename, + cache=True, + instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(), ) # Returns dictionary of lists logging.debug(f"Dynamic answer for {filename}: {llm_answer_raw}") llm_answer_final = string_utils.universal_json_load(llm_answer_raw) diff --git a/fieldExtraction/src/investment/tin_npi_funcs.py b/fieldExtraction/src/investment/tin_npi_funcs.py index b19c5c4..a17274c 100644 --- a/fieldExtraction/src/investment/tin_npi_funcs.py +++ b/fieldExtraction/src/investment/tin_npi_funcs.py @@ -500,11 +500,11 @@ def get_provider_info(text_dict: dict, page_num: str, filename: str) -> list: provider_fields = FieldSet( file_path=config.FIELD_JSON_PATH, field_type="provider_info" ) - prompt = prompt_templates.TIN_NPI_TEMPLATE( + instruction, prompt = prompt_templates.TIN_NPI_TEMPLATE( chunk, provider_fields.print_prompt_dict() ) claude_answer_raw = llm_utils.invoke_claude( - prompt, "legacy_sonnet", filename, max_tokens=8192 + prompt, "legacy_sonnet", filename, max_tokens=8192, cache=True, instruction=instruction ) # Sometimes rosters can have 50+ provider logging.debug( diff --git a/fieldExtraction/src/prompts/prompt_templates.py b/fieldExtraction/src/prompts/prompt_templates.py index 293579d..2e3e68e 100644 --- a/fieldExtraction/src/prompts/prompt_templates.py +++ b/fieldExtraction/src/prompts/prompt_templates.py @@ -11,6 +11,115 @@ from src.utils.string_utils import extract_text_from_delimiters PIPE_FORMAT_INSTRUCTIONS = "Briefly explain your answer, then enclose your final answer in |pipes|. If the requested information doesn't apply to this context, return |N/A|. If the information should exist but cannot be clearly identified, return |UNKNOWN|." +def get_cacheable_instructions(constants=None): + """ + Returns a dictionary of all cacheable instructions for cache warming. + This should be called once before parallel processing starts. + + Args: + constants: Constants object (optional, needed for field prompts) + + Returns: + dict: Dictionary mapping cache keys to instruction text + """ + from src.prompts.fieldset import FieldSet + + cacheable_instructions = {} + + # REIMBURSEMENT_PRIMARY instruction + try: + reimbursement_fields = FieldSet( + config.FIELD_JSON_PATH, field_type="reimbursement_primary" + ) + if constants: + field_prompts = reimbursement_fields.print_prompt_dict(constants) + else: + field_prompts = reimbursement_fields.print_prompt_dict() + + instruction, _ = REIMBURSEMENT_PRIMARY("", field_prompts) + cacheable_instructions["REIMBURSEMENT_PRIMARY"] = instruction + except Exception as e: + pass # Skip if fields not available + + # SPECIAL_CASE_PRIMARY instruction + try: + special_case_fields = FieldSet( + config.FIELD_JSON_PATH, field_type="special_case_primary" + ) + if constants: + field_prompts = special_case_fields.print_prompt_dict(constants) + else: + field_prompts = special_case_fields.print_prompt_dict() + + instruction, _ = SPECIAL_CASE_PRIMARY("", field_prompts) + cacheable_instructions["SPECIAL_CASE_PRIMARY"] = instruction + except Exception as e: + pass + + # METHODOLOGY_BREAKOUT instruction + try: + methodology_fields = FieldSet( + config.FIELD_JSON_PATH, field_type="methodology_breakout" + ) + if constants: + field_prompts = methodology_fields.print_prompt_dict(constants) + else: + field_prompts = methodology_fields.print_prompt_dict() + + instruction, _ = METHODOLOGY_BREAKOUT("", field_prompts) + cacheable_instructions["METHODOLOGY_BREAKOUT"] = instruction + except Exception as e: + pass + + # OUTLIER_BREAKOUT instruction + try: + instruction, _ = OUTLIER_BREAKOUT("") + cacheable_instructions["OUTLIER_BREAKOUT"] = instruction + except Exception as e: + pass + + # TIN_NPI_TEMPLATE instruction + try: + provider_fields = FieldSet(config.FIELD_JSON_PATH, field_type="provider_info") + field_prompts = provider_fields.print_prompt_dict() + instruction, _ = TIN_NPI_TEMPLATE("", field_prompts) + cacheable_instructions["TIN_NPI_TEMPLATE"] = instruction + except Exception as e: + pass + + # ONE_TO_ONE templates (single and multi-field) + try: + cacheable_instructions["ONE_TO_ONE_SINGLE_FIELD"] = ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION() + cacheable_instructions["ONE_TO_ONE_MULTI_FIELD"] = ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() + except Exception: + pass + + # EXHIBIT_LEVEL + LESSER_OF + try: + cacheable_instructions["EXHIBIT_LEVEL"] = EXHIBIT_LEVEL_INSTRUCTION() + cacheable_instructions["EXHIBIT_LEVEL_LESSER_OF"] = EXHIBIT_LEVEL_LESSER_OF_INSTRUCTION() + except Exception: + pass + + # Validation / split / checks / relationships + try: + cacheable_instructions["VALIDATE_REIMBURSEMENTS"] = VALIDATE_REIMBURSEMENTS_INSTRUCTION() + cacheable_instructions["SPLIT_REIMBURSEMENTS"] = SPLIT_REIMBURSEMENTS_INSTRUCTION() + cacheable_instructions["CHECK_REDUNDANT_LESSER"] = CHECK_REDUNDANT_LESSER_INSTRUCTION() + cacheable_instructions["LOB_RELATIONSHIP"] = LOB_RELATIONSHIP_INSTRUCTION() + except Exception: + pass + + # Dynamic primary + carveout + try: + cacheable_instructions["DYNAMIC_PRIMARY"] = DYNAMIC_PRIMARY_INSTRUCTION() + cacheable_instructions["CARVEOUT_CHECK"] = CARVEOUT_CHECK_INSTRUCTION() + except Exception: + pass + + return cacheable_instructions + + ##################################################################################### ################################## PRIMARY PROMPTS ################################## ##################################################################################### @@ -34,8 +143,16 @@ Here is the text to analyze: {context.replace('"', "'")} -Briefly explain your answer, then put the final answer in the properly-formatted JSON dictionary. -""" + Briefly explain your answer, then put the final answer in the properly-formatted JSON dictionary. + """ + + +def EXHIBIT_LEVEL_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Extract structured attributes from contract text. Follow formatting and output rules in the prompt.\n" + "Be consistent and deterministic; avoid creative paraphrasing." + ) def EXHIBIT_LEVEL_LESSER_OF(context): @@ -74,8 +191,15 @@ If no exhibit-wide constraint exists, return 'N/A'. Here is the Exhibit to analyze: {context} -Briefly explain your answer, then put your final answer in |pipes| -""" + Briefly explain your answer, then put your final answer in |pipes| + """ + + +def EXHIBIT_LEVEL_LESSER_OF_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Identify presence of exhibit-level 'lesser of' or 'not to exceed' statements. Return answer in pipes." + ) def DYNAMIC_PRIMARY_HEADER(context, field_name, field_prompt): @@ -120,13 +244,12 @@ Briefly explain your answer before putting the final answer in |pipes|. def REIMBURSEMENT_PRIMARY(context, fields): - return f"""[OBJECTIVE] + instruction = f"""[OBJECTIVE] Extract reimbursement attributes from this payer-provider contract section. The output from this prompt will be used to populate downstream fields in the system. [A. OUTPUT FORMAT] -Return a JSON array of objects, each containing these attributes: -{fields} +Return a JSON array of objects with attributes specified in the prompt. If no reimbursement terms are found, return the text "NO_REIMBURSEMENT_TERMS_FOUND" instead of an empty JSON array. [B. KEY EXTRACTION RULES] @@ -261,23 +384,31 @@ Answer: "[{{'SERVICE_TERM': 'A6544', 'REIMB_TERM': 'lesser of $25.00 and Provide Note: Both services receive the same global cap treatment regardless of table position. -[G. ANALYSIS CONTEXT] -[Start Context] +[G. ANALYSIS CONTEXT]""" + + prompt = f""" +[FIELDS] +Return a JSON array of objects, each containing these attributes: +{fields} + +[START CONTEXT] {context.replace('"', "'")} -[End Context] +[END CONTEXT] Briefly talk through your reasoning, then return a properly formatted JSON array with all extracted reimbursement items. """ + return instruction, prompt + + def SPECIAL_CASE_PRIMARY(context, fields): - return f""" + instruction = f""" [OBJECTIVE] Extract relevant excerpts from this payer-provider contract section. The output will be used to populate downstream system fields. [OUTPUT FORMAT] -Return a JSON dictionary with the following attributes: -{fields} +Return a JSON dictionary with attributes specified in the prompt. [GUIDELINES] - For each attribute, extract all paragraphs or table entries that meet the specified criteria and return them as a list of strings. @@ -302,13 +433,22 @@ Should be combined into single entries preserving exact wording: - "Specialized Neurological Services Outlier Day Threshold - # of Days: 15; Specialized Neurological Services Days in Excess of Threshold - Per Diem: $6,369" - "Cardiac Services Outlier Day Threshold - # of Days: 12; Cardiac Services Days in Excess of Threshold - Per Diem: $6,368" -[ANALYSIS CONTEXT] +Keep all original terms exactly as they appear - do not modify the language. + +[ANALYSIS CONTEXT]""" + + prompt = f""" +[FIELDS] +Return a JSON dictionary with the following attributes: +{fields} + [Start Context] {context.replace('"', "'")} [End Context] -Return a properly formatted JSON dictionary containing lists of all extracted reimbursement items that match the specified criteria. -""" +Briefly talk through your reasoning, then return a properly formatted JSON dictionary containing lists of all extracted reimbursement items that match the specified criteria.""" + + return instruction, prompt ##################################################################################### @@ -325,13 +465,10 @@ def METHODOLOGY_BREAKOUT(term, questions): if isinstance(term, list): term = f"SERVICE: {term[0]}\nMETHODOLOGY: {term[1]}" - return f"""[OBJECTIVE] + instruction = f"""[OBJECTIVE] Analyze a given term from a Payer-Provider contract: -[FIELDS] -Populate a list of JSON dictionaries with the following fields: -{questions} -The output should always be a JSON list/array of dictionaries, even when there is only one methodology. + The output should always be a JSON list/array of dictionaries, even when there is only one methodology. For any fields that don't apply to a particular methodology, use "N/A" as the value. [KEY EXTRACTION RULES] @@ -355,13 +492,18 @@ For any fields that don't apply to a particular methodology, use "N/A" as the va In the case of 'RBRVS', 'RVU', or American Society of Anesthesiology ('ASA'), any dollar amount mentioned refers to a conversion factor (even if it is not explicitly labeled as such), and not to a flat fee rate. In such cases, the output should have 'REIMB_CONVERSION_FACTOR' field populated with the dollar amount, and 'REIMB_FEE_RATE' field should be blank. -[CONTEXT] -Here is the text to analyze and respond to: +[CONTEXT]""" + + prompt = f"""Here is the text to analyze and respond to: {term} [OUTPUT FORMAT] -Explain your reasoning as you work through the context. After your explanation, include the heading "FINAL REIMBURSEMENT METHODOLOGY:" followed by a properly formatted JSON list with your final output. -""" +Explain your reasoning as you work through the context. After your explanation, include the heading "FINAL REIMBURSEMENT METHODOLOGY:" followed by a properly formatted JSON list with your final output.""" + + # Include dynamic fields in the prompt (not instruction) for cache reuse + prompt += f"\n\n[FIELDS]\nPopulate a list of JSON dictionaries with the following fields:\n{questions}\n" + + return instruction, prompt def FEE_SCHEDULE_BREAKOUT(methodology, fee_schedule, questions): @@ -454,27 +596,30 @@ def OUTLIER_BREAKOUT(term): questions = FieldSet( config.FIELD_JSON_PATH, field_type="outlier_breakout" ).print_prompt_dict() - return f"""[OBJECTIVE] -Analyze Outlier Terms of a healthcare payer-provider contract to extract outlier payment information. Outlier payments are additional supplemental reimbursements triggered when claims exceed predetermined thresholds for cost, length of stay, or resource utilization. -[FIELDS] -{questions} - -[CONTEXT] -Here is the Outlier Term to analyze and respond to: -{term} + instruction = f"""[OBJECTIVE] +Analyze a healthcare payer-provider contract to extract outlier payment information. Outlier payments are additional supplemental reimbursements triggered when claims exceed predetermined thresholds for cost, length of stay, or resource utilization. **Extraction Guidelines:** - Extract exact field values (numbers, percentages, dollar amounts) as specified - Convert time units to days when requested (e.g., 'two weeks' = '14') - Include code types when extracting exclusions (e.g., 'CPT 12345') - For frequency fields, use exact contract language (e.g., 'per admission', 'per day') -- Base answers ONLY on information explicitly stated or directly inferable from the Outlier Term -- If no relevant information is found for a field, return 'N/A' +- Base answers ONLY on information explicitly stated or directly inferable from the contract text +""" + + prompt = f""" +[FIELDS] +{questions} + +[CONTEXT] +Here is the text to analyze and respond to: +{term} [OUTPUT FORMAT] -Briefly explain your answer, then provide the extracted outlier payment information in proper JSON format. For any information not found in the Term, return 'N/A' -""" +Briefly explain your answer, then provide the extracted outlier payment information in JSON format. For any information not found in the Term, return 'N/A'""" + + return instruction, prompt def FACILITY_ADJUSTMENT_BREAKOUT(term): questions = FieldSet( @@ -670,21 +815,25 @@ def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, question): {context} ## END CONTRACT TEXT ## -{PIPE_FORMAT_INSTRUCTIONS} -""" + {PIPE_FORMAT_INSTRUCTIONS} + """ + + +def ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Answer a single, deterministic question about the contract text. Return result in pipes as instructed." + ) def TIN_NPI_TEMPLATE(context, questions): - return f""" + instruction = f""" Analyze the following healthcare payer-provider contract text and extract provider entities - these are healthcare providers who deliver services, NOT payers/insurance companies who reimburse for services. IMPORTANT DISTINCTION: - Providers: Physicians, physician groups, hospitals, clinics who DELIVER healthcare - Payers: Insurance companies, health plans who PAY for healthcare services -For EACH provider entity found (group or individual practitioner), extract: -{questions} - CRITICAL: If a provider has multiple TINs or NPIs, include ALL of them separated by pipes (|) in the same provider record. Do NOT create separate records for the same provider entity. Examples: @@ -710,7 +859,7 @@ FINAL PROVIDER INFO: [ {{ "TIN": "123456789", - "NPI": "1234567890", + "NPI": "1234567890", "NAME": "Main Provider Group", "ON_SIGNATURE_PAGE": "Y", }}, @@ -722,10 +871,18 @@ FINAL PROVIDER INFO: }} ] +Contract text:""" + + prompt = f""" +[FIELDS TO EXTRACT] +{questions} + Contract text: -{context.replace('"', "'")} +{context.replace('"', "'" )} """ + return instruction, prompt + def GROUP_TIN_NPI_TEMPLATE(key_sections: str, provider_list: str) -> str: return f""" @@ -792,8 +949,15 @@ The term length appears in two places but I'm selecting... {{ "effective_date": "January 1, 2024", "term_length": "36 months" -}} -""" + }} + """ + + +def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Answer multiple deterministic questions about contract text. Follow JSON-only output rules in the prompt." + ) ##################################################################################### @@ -910,7 +1074,14 @@ Here is the Term to analyze: SERVICE: {service_term} REIMBURSEMENT TERM: {reimb_term} -Briefly explain your reasoning, then return YES or NO in |pipes|.""" + Briefly explain your reasoning, then return YES or NO in |pipes|.""" + + +def VALIDATE_REIMBURSEMENTS_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Validate whether a service-reimbursement pair constitutes a complete reimbursement methodology." + ) def SPLIT_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str: @@ -1077,8 +1248,15 @@ Specific reimbursement method details: {base_reimb_answer.replace('"', "'")} Determine which case description corresponds to the specific reimbursement method. It will always match one of the listed cases. If multiple cases apply, choose the one that describes the specific reimbursement method, or choose a special case if applicable. Always return a value from the list of cases above; do not return 'N/A' or 'UNKNOWN'. -{PIPE_FORMAT_INSTRUCTIONS} -""" + {PIPE_FORMAT_INSTRUCTIONS} + """ + + +def CHECK_REDUNDANT_LESSER_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Decide if two texts are semantically redundant with respect to 'lesser of' meaning. Return YES/NO in pipes." + ) def DUAL_LOB_CHECK(service_term): @@ -1151,7 +1329,28 @@ Return the dates in the following JSON format: "end_date": "YYYY/MM/DD" }} -Note: Use "N/A" for dates that cannot be determined or extracted from the input text.""" + Note: Use "N/A" for dates that cannot be determined or extracted from the input text.""" + + +def SPLIT_REIMBURSEMENTS_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Split a compound reimbursement term into distinct components using only explicit contract text." + ) + + +def DYNAMIC_PRIMARY_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Extract attribute values for dynamic fields from contract text. Use exact values and return in pipes where instructed." + ) + + +def CARVEOUT_CHECK_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Determine if a carve-out applies for a service given a base reimbursement. Return result in pipes as instructed." + ) def LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values): @@ -1200,6 +1399,13 @@ def DERIVED_TERM_DATE_PROMPT( return f"The following are excerpts from a legal contract related to effective and termination dates. Determine the initial termination date of the contract, excluding any renewals or extensions.\nIf termination information comes in as a duration, derive the termination date exclusive of the final day. For example, if effective date is 2023/02/01 and termination information is 'two years', termination date would be 2025/01/31 and not 2025/02/01. If effective date is 2023/02/01 and termination information is 'year-to-year', termination date would be 2024/01/31.\nIf the termination information comes in as a date, return that date. For example, if effective date is 2023/02/01 and termination information is 2024/07/01, return 2024/07/01.\nEffective Date: {effective_date}\nTermination Information: {termination_information}\nOnly provide the first termination date, ignoring any automatic or optional renewals. Provide the date in YYYY/MM/DD format only. \n{PIPE_FORMAT_INSTRUCTIONS}" +def LOB_RELATIONSHIP_INSTRUCTION() -> str: + return ( + "[OBJECTIVE]\n" + "Determine if LOB and Program relationship in the exhibit is Inclusive or Exclusive." + ) + + @cache def invoke_derived_term_date( effective_date: str = None, termination_information: str = None diff --git a/fieldExtraction/src/utils/llm_utils.py b/fieldExtraction/src/utils/llm_utils.py index d4a1814..56c6c08 100644 --- a/fieldExtraction/src/utils/llm_utils.py +++ b/fieldExtraction/src/utils/llm_utils.py @@ -1,9 +1,11 @@ import base64 import hashlib +import inspect import json import logging import math import random +import threading import time from collections import OrderedDict @@ -12,6 +14,23 @@ import src.utils.string_utils as string_utils from botocore.exceptions import ClientError, ReadTimeoutError from src import config +# Cache warming state +_cache_warmed = {} +_cache_warming_lock = threading.Lock() + + +def _supports_prompt_cache(model_id: str) -> bool: + """Return True if the given model supports prompt caching via system cache_control. + + Enable only for models that are known to support prompt caching (Claude 3.5 Sonnet v2+, 3.7 Sonnet, 4 Sonnet). + """ + supported_ids = { + getattr(config, "MODEL_ID_CLAUDE35_SONNET_V2", None), + getattr(config, "MODEL_ID_CLAUDE37_SONNET", None), + getattr(config, "MODEL_ID_CLAUDE4_SONNET", None), + } + return model_id in supported_ids + def update_stats(filename, tokens_sent, tokens_received, elapsed_time, model_type): # Update per-file statistics @@ -72,7 +91,13 @@ def get_cache_key(prompt, model_id): def invoke_claude( - prompt: str, model_id: str, filename: str, max_tokens: int = 4096 + prompt: str, + model_id: str, + filename: str, + max_tokens: int = 4096, + cache: bool = False, + instruction: str = None, + usage_label: str = None, ) -> str: """Invokes the Claude model with the given parameters. @@ -83,6 +108,8 @@ def invoke_claude( 'sonnet_latest', 'haiku_latest', etc.). filename (str): The name of the file being processed, for cost logging purposes. max_tokens (int, optional): The maximum number of tokens to generate. Defaults to 4096. + cache (bool, optional): Whether to enable prompt caching. Defaults to False. + instruction (str, optional): System instruction to use with caching. Defaults to None. Raises: ValueError: If the model_id is not supported after alias resolution @@ -95,7 +122,14 @@ def invoke_claude( - Includes automatic retry logic with exponential backoff for throttling - Cost logging is performed automatically for all successful requests - Model aliases are resolved via config.resolve_model_id() + - Prompt caching can be enabled to reduce costs for large repeated prompts """ + if usage_label is None: + try: + usage_label = inspect.stack()[1].function + except Exception: + usage_label = "UNLABELED" + cache_key = get_cache_key(prompt, model_id) # Check cache first - use get() method from LRUCache @@ -111,52 +145,110 @@ def invoke_claude( if resolved_model_id == config.MODEL_ID_CLAUDE2: response = local_claude_2(prompt, resolved_model_id, filename, max_tokens) input_cost_per_1k, output_cost_per_1k = 0.008, 0.024 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.0, 0.0 elif resolved_model_id == config.MODEL_ID_CLAUDE3_HAIKU: response = local_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.00025, 0.00125 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.0003, 0.00003 elif resolved_model_id == config.MODEL_ID_CLAUDE35_SONNET: response = local_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 elif resolved_model_id == config.MODEL_ID_CLAUDE37_SONNET: response = local_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 elif resolved_model_id == config.MODEL_ID_CLAUDE4_SONNET: response = local_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 else: raise ValueError(f"Unsupported model_id: {model_id}") elif config.RUN_MODE == "ec2": if resolved_model_id == config.MODEL_ID_CLAUDE2: response = ec2_claude2(prompt, resolved_model_id, filename, max_tokens) input_cost_per_1k, output_cost_per_1k = 0.008, 0.024 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.0, 0.0 elif resolved_model_id == config.MODEL_ID_CLAUDE3_HAIKU: response = ec2_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.00025, 0.00125 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.0003, 0.00003 elif resolved_model_id == config.MODEL_ID_CLAUDE35_SONNET: response = ec2_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 elif resolved_model_id == config.MODEL_ID_CLAUDE37_SONNET: response = ec2_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 elif resolved_model_id == config.MODEL_ID_CLAUDE4_SONNET: response = ec2_claude_3_and_up( - prompt, resolved_model_id, filename, max_tokens + prompt, + resolved_model_id, + filename, + max_tokens, + cache=cache, + instruction=instruction, + usage_label=usage_label, ) input_cost_per_1k, output_cost_per_1k = 0.003, 0.015 + cache_creation_cost_per_1k, cache_read_cost_per_1k = 0.00375, 0.0003 else: raise ValueError(f"Unsupported model_id: {model_id}") else: @@ -183,6 +275,7 @@ def get_cache_key_with_image_array(prompt, model_id, image_base64_list): def invoke_claude_with_image_array( prompt, model_id, filename, base64_images, max_tokens=4096 ): + usage_label = inspect.stack()[1].function cache_key = get_cache_key_with_image_array(prompt, model_id, base64_images) # Check cache first - use get() method from LRUCache if cache_key in claude_cache: @@ -196,7 +289,12 @@ def invoke_claude_with_image_array( if config.RUN_MODE == "local": if resolved_model_id == config.MODEL_ID_CLAUDE35_SONNET: response = local_claude_3_and_up_with_image_array( - prompt, resolved_model_id, filename, max_tokens, base64_images + prompt, + resolved_model_id, + filename, + max_tokens, + base64_images, + usage_label=usage_label, ) logging.info( f"Effective Date Response from local Claude 3 for {filename}: {response}" @@ -267,6 +365,9 @@ def local_claude_3_and_up( max_tokens: int, max_retries: int = 5, initial_delay: int = 1, + cache: bool = False, + instruction: str = None, + usage_label: str = None, ): """ Handles local invocation of Claude 3 and above models (3.x, 3.5, 4.x) via AWS Bedrock. @@ -281,6 +382,8 @@ def local_claude_3_and_up( max_tokens (int): Maximum number of tokens to generate in the response max_retries (int, optional): Maximum number of retry attempts for throttling errors. Defaults to 5. initial_delay (int, optional): Initial delay in seconds for exponential backoff. Defaults to 1. + cache (bool, optional): Whether to enable prompt caching. Defaults to False. + instruction (str, optional): System instruction to use with caching. Defaults to None. Returns: str: Generated text response from the Claude model @@ -294,20 +397,35 @@ def local_claude_3_and_up( - Supports answer tracking via config.ENABLE_ANSWER_TRACKING for caching/debugging - Temperature is set to 0.0 for deterministic responses - Automatically handles throttling with exponential backoff retry logic + - Prompt caching can be enabled to reduce costs for large repeated prompts """ if config.ENABLE_ANSWER_TRACKING and prompt in config.ANSWER_TRACKING_DICT.keys(): return config.ANSWER_TRACKING_DICT[prompt] - body = json.dumps( - { - "anthropic_version": "bedrock-2023-05-31", - "max_tokens": max_tokens, - "temperature": 0.0, - "messages": [ - {"role": "user", "content": [{"type": "text", "text": prompt}]} - ], - } - ) + # Build request body with caching support + request_body = { + "anthropic_version": "bedrock-2023-05-31", + "max_tokens": max_tokens, + "temperature": 0.0, + } + + # If instruction provided, prefer system placement. Only add cache_control when supported. + if instruction: + if cache and _supports_prompt_cache(model_id): + request_body["system"] = [ + { + "type": "text", + "text": instruction, + "cache_control": {"type": "ephemeral"}, + } + ] + else: + request_body["system"] = instruction + request_body["messages"] = [ + {"role": "user", "content": [{"type": "text", "text": prompt}]} + ] + + body = json.dumps(request_body) for attempt in range(max_retries): try: @@ -318,6 +436,12 @@ def local_claude_3_and_up( contentType="application/json", ) response_body = json.loads(response.get("body").read()) + # Print usage stats if available for debugging (cache creation/read tokens, etc.) + if "usage" in response_body: + label = usage_label or "UNLABELED" + print( + f"Usage stats [{label}] model={model_id} file={filename}: {response_body['usage']}" + ) response_text = response_body["content"][0]["text"] if config.ENABLE_ANSWER_TRACKING: @@ -350,6 +474,7 @@ def local_claude_3_and_up_with_image_array( max_retries=5, initial_delay=1, image_media_type="image/png", + usage_label: str = None, ): """ Processes a prompt along with an image in base64 format using the Claude 3 model via AWS Bedrock. @@ -408,6 +533,12 @@ def local_claude_3_and_up_with_image_array( contentType="application/json", ) response_body = json.loads(response.get("body").read()) + # Print usage stats if available for debugging (cache creation/read tokens, etc.) + if "usage" in response_body: + label = usage_label or "UNLABELED" + print( + f"Usage stats [{label}] model={model_id} file={filename}: {response_body['usage']}" + ) response_text = response_body["content"][0]["text"] if config.ENABLE_ANSWER_TRACKING: @@ -466,6 +597,9 @@ def ec2_claude_3_and_up( max_tokens: int, max_retries: int = 5, initial_delay: int = 1, + cache: bool = False, + instruction: str = None, + usage_label: str = None, ): """ Handles EC2 invocation of Claude 3 and above models. @@ -485,6 +619,9 @@ def ec2_claude_3_and_up( max_tokens (int): Maximum number of tokens to generate in the response max_retries (int, optional): Maximum retry attempts per runtime environment. Defaults to 5. initial_delay (int, optional): Initial delay in seconds for exponential backoff. Defaults to 1. + cache (bool, optional): Whether to enable prompt caching. Defaults to False. + instruction (str, optional): System instruction to use with caching. Defaults to None. + usage_label (str, optional): Label to surface in usage diagnostics. Defaults to None. Returns: str: Generated text response from the Claude model @@ -509,18 +646,34 @@ def ec2_claude_3_and_up( - When rotation is enabled, runtime gets full retry cycle before switching to next - Stats are only updated on successful responses - Temperature set to 0.0 for consistent, deterministic outputs + - Prompt caching can be enabled to reduce costs for large repeated prompts """ start_time = time.time() - body = json.dumps( - { - "anthropic_version": "bedrock-2023-05-31", - "max_tokens": max_tokens, - "temperature": 0.0, - "messages": [ - {"role": "user", "content": [{"type": "text", "text": prompt}]} - ], - } - ) + + # Build request body with caching support + request_body = { + "anthropic_version": "bedrock-2023-05-31", + "max_tokens": max_tokens, + "temperature": 0.0, + } + + # If instruction provided, prefer system placement. Only add cache_control when supported. + if instruction: + if cache and _supports_prompt_cache(model_id): + request_body["system"] = [ + { + "type": "text", + "text": instruction, + "cache_control": {"type": "ephemeral"}, + } + ] + else: + request_body["system"] = instruction + request_body["messages"] = [ + {"role": "user", "content": [{"type": "text", "text": prompt}]} + ] + + body = json.dumps(request_body) if config.ENABLE_RUNTIME_ROTATION: dev_runtime = config.assume_role_and_get_runtime( @@ -545,6 +698,12 @@ def ec2_claude_3_and_up( contentType="application/json", ) response_body = json.loads(response.get("body").read()) + # Print usage stats if available for debugging (cache creation/read tokens, etc.) + if "usage" in response_body: + label = usage_label or "UNLABELED" + print( + f"Usage stats [{label}] model={model_id} file={filename}: {response_body['usage']}" + ) response_text = response_body["content"][0]["text"] elapsed_time = time.time() - start_time @@ -608,6 +767,12 @@ def ec2_claude_3_and_up( contentType="application/json", ) response_body = json.loads(response.get("body").read()) + # Print usage stats if available for debugging (cache creation/read tokens, etc.) + if "usage" in response_body: + label = usage_label or "UNLABELED" + print( + f"Usage stats [{label}] model={model_id} file={filename}: {response_body['usage']}" + ) response_text = response_body["content"][0]["text"] elapsed_time = time.time() - start_time @@ -670,3 +835,94 @@ def get_image_array_answer(prompt, filename, base64_images): ) response_dict = json.loads(response_text) return response_dict + + +def warm_prompt_cache(instruction: str, cache_key: str, model_id: str = "sonnet_latest", max_tokens: int = 10): + """ + Warms the prompt cache by sending a minimal request with the instruction. + This should be called once before parallel processing to ensure the cache is registered. + + Args: + instruction (str): The instruction text to cache + cache_key (str): A unique identifier for this cache (e.g., "REIMBURSEMENT_PRIMARY") + model_id (str): The model ID to use. Defaults to "sonnet_latest" + max_tokens (int): Minimal tokens for cache warming. Defaults to 10 + + Returns: + bool: True if cache was warmed, False if it was already warm + """ + global _cache_warmed + + # Check if already warmed (thread-safe) + with _cache_warming_lock: + if _cache_warmed.get(cache_key, False): + logging.debug(f"Cache already warmed for {cache_key}") + return False + + # Mark as being warmed + _cache_warmed[cache_key] = True + + logging.info(f"Warming prompt cache for {cache_key}...") + + # Resolve model ID + resolved_model_id = config.resolve_model_id(model_id) + + # Create minimal prompt with the instruction to register the cache + minimal_prompt = "Respond with OK." + + try: + # Build request with caching enabled + request_body = { + "anthropic_version": "bedrock-2023-05-31", + "max_tokens": max_tokens, + "temperature": 0.0, + "system": [ + { + "type": "text", + "text": instruction, + "cache_control": {"type": "ephemeral"} + } + ], + "messages": [ + {"role": "user", "content": [{"type": "text", "text": minimal_prompt}]} + ] + } + + body = json.dumps(request_body) + + # Send the request to register the cache + if config.RUN_MODE == "local": + response = config.BEDROCK_RUNTIME.invoke_model( + body=body, + modelId=resolved_model_id, + accept="application/json", + contentType="application/json", + ) + elif config.RUN_MODE == "ec2": + response = config.BEDROCK_RUNTIME.invoke_model( + body=body, + modelId=resolved_model_id, + accept="application/json", + contentType="application/json", + ) + else: + logging.error("Invalid RUN_MODE for cache warming") + return False + + response_body = json.loads(response.get("body").read()) + logging.info(f"Successfully warmed cache for {cache_key}") + + # Log cache usage info if available + if "usage" in response_body: + usage = response_body["usage"] + if "cache_creation_input_tokens" in usage: + logging.info(f"Cache created with {usage.get('cache_creation_input_tokens', 0)} tokens for {cache_key}") + + return True + + except Exception as e: + logging.error(f"Error warming cache for {cache_key}: {str(e)}") + # Mark as not warmed on error so it can be retried + with _cache_warming_lock: + _cache_warmed[cache_key] = False + return False