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

Merged in feature/prompt-caching (pull request #737)

Browse Source 
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
This commit is contained in:
Faizan Mohiuddin
2025-10-16 20:32:00 +00:00
committed by Katon Minhas
parent 64287a2c21
commit f736bde987
6 changed files with 667 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/investment/main.py
+19 -1
View File
@@ -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 = []
fieldExtraction/src/investment/one_to_one_funcs.py
+12 -3
View File
@@ -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)
fieldExtraction/src/investment/prompt_calls.py
+91 -15
View File
@@ -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)
fieldExtraction/src/investment/tin_npi_funcs.py
+2 -2
View File
@@ -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(
fieldExtraction/src/prompts/prompt_templates.py
+257 -51
View File
@@ -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
fieldExtraction/src/utils/llm_utils.py
+286 -30
View File
@@ -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