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

refactor: update all prompt templates to return (prompt, parser) tuples

Browse Source 
- Updated 24 prompt template functions to return (prompt_string, parser_function) tuples
- Added type hints: Tuple[str, Callable[[str], dict/list]]
- Changed return pattern from 'return f"""..."""' to 'prompt = f"""..."""' followed by 'return (prompt, parser)'
- Added parser functions: _json_dict_parser and _json_list_parser
- Updated docstrings to include Returns section explaining the tuple structure
- Functions updated include: LESSER_OF_CHECK, EXHIBIT_TITLE_MATCH, OUTLIER_BREAKOUT, CODE_EXPLICIT, CODE_CATEGORY, CODE_IMPLICIT_SPECIAL, CODE_IMPLICIT, CODE_LAST_CHECK, FILL_BILL_TYPE, ONE_TO_ONE_SINGLE_FIELD_TEMPLATE, TIN_NPI_TEMPLATE, CHECK_PROVIDER_NAME_MATCH_PROMPT, EXTRACT_AMENDMENT_NUM_FROM_FILENAME, EXHIBIT_HEADER, EXHIBIT_LINKAGE, DATE_FIX_PROMPT, EFFECTIVE_DATE_FIX_PROMPT, SPLIT_REIMB_DATES, DERIVED_TERM_DATE_PROMPT, and all breakout functions
- All changes verified with linter and Python compilation
- No syntax errors or linter issues
This commit is contained in:
ppanchigar
2026-02-02 17:12:10 -06:00
parent f633922381
commit 98554afc05
1 changed files with 334 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/prompts/prompt_templates.py
+334 -85
View File
@@ -1,9 +1,10 @@
from typing import Optional
from typing import Optional, Callable, Tuple, Any
import src.config as config
from src.prompts.fieldset import FieldSet
from src.constants.constants import Constants
from src.utils import string_utils
from src.utils import string_utils, json_utils
from src.constants.delimiters import Delimiter
# Legacy pipe format instructions (deprecated - use JSON formats instead)
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|."
@@ -31,6 +32,23 @@ JSON_LIST_OF_DICTS_FORMAT_INSTRUCTIONS = """Return your final answer as a valid
- Use "N/A" for fields within each dictionary where the information doesn't apply.
- Ensure the JSON is properly formatted with square brackets, curly braces, and double quotes."""
# Parser helper functions for prompt templates
def _json_dict_parser(raw_output: str) -> dict:
"""Parse JSON dictionary from LLM output."""
return json_utils.parse_json_dict(raw_output)
def _json_list_parser(raw_output: str) -> list:
"""Parse JSON list from LLM output."""
return json_utils.parse_json_list(raw_output)
def _pipe_delimiter_parser(raw_output: str) -> str:
"""Parse pipe-delimited output from LLM (legacy format)."""
return string_utils.extract_text_from_delimiters(raw_output, Delimiter.PIPE)
# Module-level cached Constants instance for instruction functions
# This is lazily initialized on first use and reused across all instruction calls
_cached_constants = None
@@ -228,17 +246,22 @@ Briefly explain your answer, then put the final answer in the properly-formatted
{JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_LEVEL(context, fields):
def EXHIBIT_LEVEL(context, fields) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for exhibit level extraction.
Call EXHIBIT_LEVEL_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[ATTRIBUTES]
prompt = f"""[ATTRIBUTES]
Here are the attributes to be included in the dictionary, and instructions on how to correctly answer:
{fields}
[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_dict_parser)
def DYNAMIC_PRIMARY_TEXT_INSTRUCTION() -> str:
@@ -259,16 +282,21 @@ Briefly explain your answer before putting the final answer in a properly-format
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def DYNAMIC_PRIMARY_TEXT(context, field_name, field_prompt):
def DYNAMIC_PRIMARY_TEXT(context, field_name, field_prompt) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for text-based dynamic primary extraction.
Call DYNAMIC_PRIMARY_TEXT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[ATTRIBUTE TO EXTRACT]
prompt = f"""[ATTRIBUTE TO EXTRACT]
{field_name} : {field_prompt}
[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_list_parser)
def DYNAMIC_PRIMARY_INSTRUCTION() -> str:
@@ -358,11 +386,14 @@ def REIMB_DATES_ASSIGNMENT(
field_prompt: str,
exhibit_text_simplified: str,
page_num: str,
):
) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for REIMB_DATES assignment.
Call REIMB_DATES_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[REIMB_DATES FIELD DEFINITION]
prompt = f"""[REIMB_DATES FIELD DEFINITION]
{field_prompt}
[CONTEXT]
@@ -375,6 +406,8 @@ Here is the section of the contract. Examine this section closely and identify w
This is the term you need to find the date range for. It appears on page {page_num} of the text.
Service Term: "{service_term}"
Reimbursement Term: "{reimb_term}" """
return (prompt, _json_dict_parser)
def DYNAMIC_ASSIGNMENT_INSTRUCTION() -> str:
@@ -442,12 +475,15 @@ def DYNAMIC_ASSIGNMENT(
field_prompt: str,
exhibit_text_simplified: str,
page_num: str,
):
) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt for dynamic assignment extraction.
Call DYNAMIC_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here is the section of the contract. Examine this section closely and pick the correct answer or answers for the given Reimbursement Term.
[START CONTEXT]
{exhibit_text_simplified}
@@ -461,20 +497,27 @@ Reimbursement Term: "{reimb_term}"
[{field_name} INFORMATION]
Here is the definition and valid values. Use this information to help you find the right answer.
{field_name} : {field_prompt}"""
return (prompt, _json_dict_parser)
def REIMBURSEMENT_PRIMARY(context):
def REIMBURSEMENT_PRIMARY(context) -> Tuple[str, Callable[[str], list]]:
"""
Returns prompt for reimbursement primary extraction.
Call REIMBURSEMENT_PRIMARY_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""
prompt = f"""
[START CONTEXT]
{context.replace('"', "'")}
[END CONTEXT]
Briefly talk through your reasoning. Then return a properly formatted JSON list of dictionaries with all extracted SERVICE_TERM AND REIMB_TERM values.
"""
return (prompt, _json_list_parser)
def REIMBURSEMENT_PRIMARY_INSTRUCTION() -> str:
@@ -591,10 +634,13 @@ def LESSER_OF_DISTRIBUTION(
page_num: str,
exhibit_text: str,
cross_exhibit_lesser_of: list[dict],
):
) -> Tuple[str, Callable[[str], list]]:
"""
Returns prompt for lesser-of distribution.
Call LESSER_OF_DISTRIBUTION_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
# Format cross-exhibit templates
if cross_exhibit_lesser_of:
@@ -611,7 +657,7 @@ CROSS-EXHIBIT TEMPLATES (pre-verified, include if applicable):
else:
cross_exhibit_section = "CROSS-EXHIBIT TEMPLATES: None"
return f"""INPUTS:
prompt = f"""INPUTS:
- SERVICE: {service_term}
- METHODOLOGY: {reimb_term}
- PAGE: {page_num}
@@ -624,6 +670,8 @@ EXHIBIT TEXT:
Analyze the inputs above. Briefly explain your reasoning, then provide your final answer as a JSON list.
"""
return (prompt, _json_list_parser)
def LESSER_OF_CHECK_INSTRUCTION() -> str:
@@ -709,14 +757,19 @@ Briefly explain your reasoning, then return a properly-formatted JSON dictionary
{JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS}"""
def LESSER_OF_CHECK(service_term: str, reimb_term: str, exhibit_title: str):
def LESSER_OF_CHECK(service_term: str, reimb_term: str, exhibit_title: str) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt for lesser-of check classification.
Call LESSER_OF_CHECK_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""**Current Exhibit:** {exhibit_title}
prompt = f"""**Current Exhibit:** {exhibit_title}
**Service:** {service_term}
**Reimbursement:** {reimb_term}"""
return (prompt, _json_dict_parser)
def EXHIBIT_TITLE_MATCH_INSTRUCTION() -> str:
@@ -740,13 +793,18 @@ Examples: ["YES"] or ["NO"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_TITLE_MATCH(current_exhibit_title: str, target_exhibit_reference: str):
def EXHIBIT_TITLE_MATCH(current_exhibit_title: str, target_exhibit_reference: str) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for exhibit title matching.
Call EXHIBIT_TITLE_MATCH_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Current Exhibit Title: "{current_exhibit_title}"
Target Exhibit Reference: "{target_exhibit_reference}" """
return (prompt, _json_list_parser)
#####################################################################################
@@ -754,19 +812,24 @@ Target Exhibit Reference: "{target_exhibit_reference}" """
#####################################################################################
def METHODOLOGY_BREAKOUT(service_term: str, reimb_term: str):
def METHODOLOGY_BREAKOUT(service_term: str, reimb_term: str) -> Tuple[str, Callable[[str], list]]:
"""
Returns ONLY dynamic content for methodology breakout extraction.
Call METHODOLOGY_BREAKOUT_INSTRUCTION() separately for the cached instruction.
All fields run by this prompt will end up becoming their own row, rather than being distributed across the rows of the exhibit.
Note: questions/fields are now included in METHODOLOGY_BREAKOUT_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""Here is the text to analyze and respond to:
prompt = f"""Here is the text to analyze and respond to:
Service Term: {service_term}
Reimbursement Term: {reimb_term}
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."""
return (prompt, _json_list_parser)
def METHODOLOGY_BREAKOUT_INSTRUCTION() -> str:
@@ -828,14 +891,19 @@ Populate a JSON dictionary with the following fields:
Briefly explain your reasoning, then write your JSON dictionary."""
def FEE_SCHEDULE_BREAKOUT(methodology, fee_schedule):
def FEE_SCHEDULE_BREAKOUT(methodology, fee_schedule) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for fee schedule breakout.
Call FEE_SCHEDULE_BREAKOUT_INSTRUCTION() separately for the cached instruction.
Note: Field definitions are now included in FEE_SCHEDULE_BREAKOUT_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Analyze and respond to the following text, specifically for {fee_schedule} Fee Schedule:
Methodology: {methodology.replace('"', "'")}"""
return (prompt, _json_dict_parser)
def GROUPER_BREAKOUT_INSTRUCTION() -> str:
@@ -865,22 +933,30 @@ Populate a JSON dictionary with the following fields:
Briefly explain your answer, then return a valid JSON dictionary using the exact field names from the questions as keys. For any information not found in the Term, return 'N/A'."""
def GROUPER_BREAKOUT(service, term):
def GROUPER_BREAKOUT(service, term) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for grouper breakout.
Call GROUPER_BREAKOUT_INSTRUCTION() separately for the cached instruction.
Note: Field definitions are now included in GROUPER_BREAKOUT_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here are the service and reimbursement terms to analyze:
SERVICE: {service}
REIMBURSEMENT TERM: {term}"""
return (prompt, _json_dict_parser)
def SPECIAL_CASE_BREAKOUT(term, questions):
def SPECIAL_CASE_BREAKOUT(term, questions) -> Tuple[str, Callable[[str], dict]]:
"""
Generic prompt for special case breakout fields when no additional descriptions are needed.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[OBJECTIVE]
prompt = f"""[OBJECTIVE]
Analyze a given term from a Payer-Provider contract:
[FIELDS]
@@ -894,9 +970,11 @@ Here is the term to analyze:
[OUTPUT FORMAT]
Briefly explain your answer, then put your final answer in JSON dictionary format.
"""
return (prompt, _json_dict_parser)
def RATE_ESCALATOR_BREAKOUT(term):
def RATE_ESCALATOR_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Creates a prompt to extract rate escalator field information from a rate escalator statement.
@@ -905,7 +983,7 @@ def RATE_ESCALATOR_BREAKOUT(term):
rate_escalator_questions: Dictionary or list of questions with instructions for each field
Returns:
Formatted prompt string for rate escalator field extraction
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="rate_escalator_breakout"
@@ -913,7 +991,13 @@ def RATE_ESCALATOR_BREAKOUT(term):
return SPECIAL_CASE_BREAKOUT(term, questions)
def TRIGGER_CAP_BREAKOUT(term):
def TRIGGER_CAP_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for trigger cap breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="trigger_cap_breakout"
).print_prompt_dict()
@@ -946,53 +1030,94 @@ Populate a JSON dictionary with the following fields:
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'."""
def OUTLIER_BREAKOUT(term):
def OUTLIER_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns ONLY dynamic content for outlier breakout extraction.
Call OUTLIER_BREAKOUT_INSTRUCTION() separately for the cached instruction.
Note: Field definitions are now included in OUTLIER_BREAKOUT_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here is the text to analyze and respond to:
{term}"""
return (prompt, _json_dict_parser)
def FACILITY_ADJUSTMENT_BREAKOUT(term):
def FACILITY_ADJUSTMENT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for facility adjustment breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="facility_adjustment_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def SEQUESTRATION_BREAKOUT(term):
def SEQUESTRATION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for sequestration breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="sequestration_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def DISCOUNT_BREAKOUT(term):
def DISCOUNT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for discount breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="discount_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def PREMIUM_BREAKOUT(term):
def PREMIUM_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for premium breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="premium_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def ADDITION_BREAKOUT(term):
def ADDITION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for addition breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="addition_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def STOP_LOSS_BREAKOUT(term):
def STOP_LOSS_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for stop loss breakout.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="stop_loss_breakout"
).print_prompt_dict()
@@ -1032,16 +1157,21 @@ Populate a JSON dictionary with the following fields:
Briefly explain your answer before putting the final answer in JSON dictionary format."""
def CODE_EXPLICIT(service, methodology):
def CODE_EXPLICIT(service, methodology) -> Tuple[str, Callable[[str], dict]]:
"""
Returns ONLY dynamic content for code explicit extraction.
Call CODE_EXPLICIT_INSTRUCTION() separately for the cached instruction.
Note: Field definitions are now included in CODE_EXPLICIT_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here is the Service and Methodology to analyze:
Service: {service.replace('"', "'")}
Methodology: {methodology.replace('"', "'")}"""
return (prompt, _json_dict_parser)
def CODE_CATEGORY_INSTRUCTION() -> str:
@@ -1061,15 +1191,21 @@ Explain your answer in 1-2 sentences. Write your final answer in a properly-form
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CODE_CATEGORY(service, choices):
"""Call CODE_CATEGORY_INSTRUCTION() separately for the cached instruction."""
return f"""[VALID DESCRIPTIONS]
def CODE_CATEGORY(service, choices) -> Tuple[str, Callable[[str], list]]:
"""Call CODE_CATEGORY_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
prompt = f"""[VALID DESCRIPTIONS]
Here are the descriptions to choose from:
{choices}
[CONTEXT]
Here is the Service Term to analyze:
{service.replace('"', "'")}"""
return (prompt, _json_list_parser)
def CODE_IMPLICIT_SPECIAL_INSTRUCTION() -> str:
@@ -1091,11 +1227,17 @@ Examples: ["Drugs"], ["Vaccines"], ["PT/OT/ST"], ["PT"], ["Surgery"], ["N/A"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CODE_IMPLICIT_SPECIAL(service):
"""Call CODE_IMPLICIT_SPECIAL_INSTRUCTION() separately for the cached instruction."""
return f"""[CONTEXT]
def CODE_IMPLICIT_SPECIAL(service) -> Tuple[str, Callable[[str], list]]:
"""Call CODE_IMPLICIT_SPECIAL_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
prompt = f"""[CONTEXT]
Here is the Service to analyze:
{service}"""
return (prompt, _json_list_parser)
def CODE_IMPLICIT_INSTRUCTION() -> str:
@@ -1122,15 +1264,21 @@ Explain your answer, ensuring each point in the instructions is addressed. Then
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CODE_IMPLICIT(service, choices):
"""Call CODE_IMPLICIT_INSTRUCTION() separately for the cached instruction."""
return f"""[VALID DESCRIPTIONS]
def CODE_IMPLICIT(service, choices) -> Tuple[str, Callable[[str], list]]:
"""Call CODE_IMPLICIT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
prompt = f"""[VALID DESCRIPTIONS]
Here are the descriptions to choose from:
{choices}
[CONTEXT]
Here is the Service Term to analyze:
{service.replace('"', "'")}"""
return (prompt, _json_list_parser)
def CODE_LAST_CHECK_INSTRUCTION() -> str:
@@ -1152,10 +1300,16 @@ Examples: ["Specific"] or ["Generic"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CODE_LAST_CHECK(service):
"""Call CODE_LAST_CHECK_INSTRUCTION() separately for the cached instruction."""
return f"""[SERVICE TO ANALYZE]
def CODE_LAST_CHECK(service) -> Tuple[str, Callable[[str], list]]:
"""Call CODE_LAST_CHECK_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
prompt = f"""[SERVICE TO ANALYZE]
{service}"""
return (prompt, _json_list_parser)
def FILL_BILL_TYPE_INSTRUCTION() -> str:
@@ -1178,8 +1332,12 @@ Briefly explain your answer, then return your final answer in a properly-formatt
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def FILL_BILL_TYPE(service, choices, reimb_term=None, exhibit_text=None):
"""Call FILL_BILL_TYPE_INSTRUCTION() separately for the cached instruction."""
def FILL_BILL_TYPE(service, choices, reimb_term=None, exhibit_text=None) -> Tuple[str, Callable[[str], list]]:
"""Call FILL_BILL_TYPE_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
# Build reimbursement term section conditionally
reimb_term_section = ""
if reimb_term and not string_utils.is_empty(reimb_term):
@@ -1198,13 +1356,15 @@ If the Service Term and Reimbursement Term together do not provide sufficient in
{exhibit_text.replace('"', "'")}
"""
return f"""[VALID DESCRIPTIONS]
prompt = f"""[VALID DESCRIPTIONS]
Here are the descriptions to choose from:
{choices}
[SERVICE TERM TO ANALYZE]
{service.replace('"', "'")}
{reimb_term_section}{exhibit_context_section}"""
return (prompt, _json_list_parser)
#####################################################################################
@@ -1212,8 +1372,14 @@ Here are the descriptions to choose from:
#####################################################################################
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context: str, field_prompt: dict[str, str]):
return f"""The following is a contract (or excerpt thereof).
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context: str, field_prompt: dict[str, str]) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for one-to-one single field extraction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = f"""The following is a contract (or excerpt thereof).
Answer the following question: {field_prompt}
## START CONTRACT TEXT ##
@@ -1222,6 +1388,8 @@ Answer the following question: {field_prompt}
Briefly explain your answer, then put the final answer in a properly-formatted JSON dictionary, where the key is the field name and value is the answer. If there is no valid answer, return "N/A".
"""
return (prompt, _json_dict_parser)
def ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION() -> str:
@@ -1288,17 +1456,20 @@ FINAL PROVIDER INFO:
Contract text:"""
def TIN_NPI_TEMPLATE(context, questions, payer_name):
def TIN_NPI_TEMPLATE(context, questions, payer_name) -> Tuple[str, Callable[[str], list]]:
"""
Returns prompt for TIN/NPI extraction.
Call TIN_NPI_TEMPLATE_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output (list of dicts).
"""
# Build payer-specific instruction for the prompt
payer_instruction = ""
if not string_utils.is_empty(payer_name):
payer_instruction = f'\n\nIMPORTANT: "{payer_name}" is the PAYER (insurance company) in this contract. Do NOT extract TIN or NPI information for "{payer_name}" - only extract information for healthcare providers who deliver services.'
return f"""[PAYER CONTEXT]{payer_instruction}
prompt = f"""[PAYER CONTEXT]{payer_instruction}
[FIELDS TO EXTRACT]
{questions}
@@ -1306,6 +1477,8 @@ def TIN_NPI_TEMPLATE(context, questions, payer_name):
Contract text:
{context.replace('"', "'" )}
"""
return (prompt, _json_list_parser)
def FULL_CONTEXT_CLAIM_TYPES_ADDITIONAL_INSTRUCTION():
@@ -1370,19 +1543,30 @@ Examples: ["Y"] or ["N"]
def CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name: str, hybrid_smart_chunking_provider_group_name: str
) -> str:
) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for provider name matching.
Call CHECK_PROVIDER_NAME_MATCH_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
PROVIDER NAME A: {provider_name}
PROVIDER NAME B: {hybrid_smart_chunking_provider_group_name}
Note: PROVIDER NAME B is typically a single name but may contain multiple names. Determine if PROVIDER NAME A matches PROVIDER NAME B (or ANY name if multiple are present)."""
return (prompt, _json_list_parser)
def ONE_TO_ONE_MULTI_FIELD_TEMPLATE(context, questions: dict[str, str]):
return f"""The following is a contract (or excerpt thereof). Answer the following questions: {questions}
def ONE_TO_ONE_MULTI_FIELD_TEMPLATE(context, questions: dict[str, str]) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for one-to-one multi-field extraction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = f"""The following is a contract (or excerpt thereof). Answer the following questions: {questions}
## START CONTRACT TEXT ##
{context.replace('"', "'")}
@@ -1397,6 +1581,8 @@ The JSON should:
- Contain only the final answers, not explanations
- NOT contain any curly brackets '{{' or '}}' inside.
- If curly bracket still exists inside JSON, replace them with '()' instead.
Example format:
I found the effective date in section 2.1 which states...
@@ -1407,6 +1593,7 @@ The term length appears in two places but I'm selecting...
"term_length": "36 months"
}}
"""
return (prompt, _json_dict_parser)
def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str:
@@ -1416,8 +1603,16 @@ def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str:
)
def EXTRACT_AMENDMENT_NUM_FROM_FILENAME(filename) -> str:
return f"""Filename: {filename}"""
def EXTRACT_AMENDMENT_NUM_FROM_FILENAME(filename) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for extracting amendment number from filename.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = f"""Filename: {filename}"""
return (prompt, _json_dict_parser)
def EXTRACT_AMENDMENT_NUM_FROM_FILENAME_INSTRUCTION() -> str:
@@ -1485,14 +1680,19 @@ Example: ["Exhibit A: Commercial Services"] or ["Attachment B - Provider Compens
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_HEADER(context):
def EXHIBIT_HEADER(context) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for exhibit header extraction.
Call EXHIBIT_HEADER_INSTRUCTION() separately for the cached instruction.
Note: Header markers are now included in EXHIBIT_HEADER_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_list_parser)
def EXHIBIT_LINKAGE_INSTRUCTION() -> str:
@@ -1516,16 +1716,21 @@ Examples: ["Y"] or ["N"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_LINKAGE(header1, header2):
def EXHIBIT_LINKAGE(header1, header2) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for exhibit linkage comparison.
Call EXHIBIT_LINKAGE_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Header 1:
"{header1}"
Header 2:
"{header2}" """
return (prompt, _json_list_parser)
#####################################################################################
@@ -1602,11 +1807,14 @@ Briefly explain your reasoning following the three conditions above, then return
Examples: ["YES"] or ["NO"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def VALIDATE_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str:
def VALIDATE_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for validation.
Call VALIDATE_REIMBURSEMENTS_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""## INPUT
prompt = f"""## INPUT
SERVICE TERM: {service_term}
REIMBURSEMENT TERM: {reimb_term}
@@ -1619,6 +1827,8 @@ REIMBURSEMENT TERM: {reimb_term}
If 1=yes, 2=yes, 3=no → return ["YES"]
Otherwise → return ["NO"]"""
return (prompt, _json_list_parser)
def DATE_FIX_INSTRUCTION() -> str:
@@ -1665,13 +1875,18 @@ Output: ["N/A"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def DATE_FIX_PROMPT(context: str) -> str:
def DATE_FIX_PROMPT(context: str) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for date fixing.
Call DATE_FIX_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Please analyze this date:
{context}"""
return (prompt, _json_list_parser)
def CARVEOUT_CHECK_INSTRUCTION() -> str:
@@ -1723,19 +1938,30 @@ Example: ["OUTLIER_TERM"] or ["RATE_ESCALATOR"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CARVEOUT_CHECK(service_term: str, reimb_term: str):
def CARVEOUT_CHECK(service_term: str, reimb_term: str) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for carveout check.
Call CARVEOUT_CHECK_INSTRUCTION() separately for the cached instruction.
Note: Case definitions are now included in CARVEOUT_CHECK_INSTRUCTION() for caching.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[REIMBURSEMENT TERM]
prompt = f"""[REIMBURSEMENT TERM]
Here is the Reimbursement Term to analyze:
Service: {service_term}
Reimbursement Term: {reimb_term.replace('"', "'")}"""
return (prompt, _json_list_parser)
def EFFECTIVE_DATE_FIX_PROMPT() -> str:
return """Extract the effective date of the contract or contract effective date or Effective Date of Agreement or Effective Date of Amendment or the amendment effective date from the given image data. ALSO LOOK FOR dates in phrases that semantically indicate an effective date (may or may not be explictly labelled as effective date), even if they do not exactly match the phrases above.\nExamples include but ARE NOT LIMITED TO:\nPhrases indicating when an contract or amendment begins or takes effect, Phrases specifying the start date of the contract or amendment, Phrases defining when the contract or amendment becomes operational, or Any phrase indicating contract or amendment creation or execution date.\nIt is possible that the contract will specify that the effective date is derived from elsewhere in the contract. Give the date in a JSON FORMAT with AARETE_DERIVED_EFFECTIVE_DATE as the key and effective date value in YYYY/MM/DD format as the value.
def EFFECTIVE_DATE_FIX_PROMPT() -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for effective date extraction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = """Extract the effective date of the contract or contract effective date or Effective Date of Agreement or Effective Date of Amendment or the amendment effective date from the given image data. ALSO LOOK FOR dates in phrases that semantically indicate an effective date (may or may not be explictly labelled as effective date), even if they do not exactly match the phrases above.\nExamples include but ARE NOT LIMITED TO:\nPhrases indicating when an contract or amendment begins or takes effect, Phrases specifying the start date of the contract or amendment, Phrases defining when the contract or amendment becomes operational, or Any phrase indicating contract or amendment creation or execution date.\nIt is possible that the contract will specify that the effective date is derived from elsewhere in the contract. Give the date in a JSON FORMAT with AARETE_DERIVED_EFFECTIVE_DATE as the key and effective date value in YYYY/MM/DD format as the value.
Rules:
1. Add 2000 to 2-digit years 0-49, 1900 to 50-99
2. Return N/A if: missing/ambiguous month or year components, or invalid dates. If just the day is missing, assume it's the first of the month.
@@ -1743,6 +1969,8 @@ def EFFECTIVE_DATE_FIX_PROMPT() -> str:
4. Any day values with spaces or leading zeros (e.g., '0 1' or '01') are interpreted as single-digit days
Return N/A only for date if NO valid effective date is found. give the answer in the format: {{\"AARETE_DERIVED_EFFECTIVE_DT\": \"YYYY/MM/DD\"}}. nothing else."""
return (prompt, _json_dict_parser)
def SPLIT_REIMB_DATES_INSTRUCTION() -> str:
@@ -1784,10 +2012,16 @@ Return the dates in the following JSON format:
Note: Use "N/A" for dates that cannot be determined or extracted from the input text."""
def SPLIT_REIMB_DATES(date_range: str) -> str:
"""Call SPLIT_REIMB_DATES_INSTRUCTION() separately for the cached instruction."""
return f"""[DATE RANGE TO EXTRACT]
def SPLIT_REIMB_DATES(date_range: str) -> Tuple[str, Callable[[str], dict]]:
"""Call SPLIT_REIMB_DATES_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
prompt = f"""[DATE RANGE TO EXTRACT]
{date_range}"""
return (prompt, _json_dict_parser)
def LOB_RELATIONSHIP_INSTRUCTION() -> str:
@@ -1819,17 +2053,22 @@ Your response MUST include the JSON list with your final answer.
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values):
def LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for LOB relationship analysis.
Call LOB_RELATIONSHIP_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[LOB AND PROGRAMS IDENTIFIED]
prompt = f"""[LOB AND PROGRAMS IDENTIFIED]
It was previously identified that the Exhibit contains the following LOB and Programs:
{dynamic_primary_values}
[CONTEXT]
Here is the Exhibit to be analyzed:
{exhibit_text.replace('"', "'")}"""
return (prompt, _json_list_parser)
def DERIVED_TERM_DATE_INSTRUCTION() -> str:
@@ -1855,13 +2094,18 @@ Example: ["2024/01/31"] or ["N/A"]
def DERIVED_TERM_DATE_PROMPT(
effective_date: str = None, termination_information: str = None
) -> str:
) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for derived term date calculation.
Call DERIVED_TERM_DATE_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Effective Date: {effective_date}
Termination Information: {termination_information}"""
return (prompt, _json_list_parser)
def SPECIAL_CASE_ASSIGNMENT_INSTRUCTION() -> str:
@@ -1891,15 +2135,18 @@ Example: ["0"] or ["2"] or ["5"]
def SPECIAL_CASE_ASSIGNMENT(
exhibit_text, reimbursement_dict, special_case_dicts, special_case_term
):
) -> Tuple[str, Callable[[str], list]]:
"""Returns ONLY dynamic content for special case assignment.
Call SPECIAL_CASE_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
special_case_list = [
f"{i} : {special_case_dicts[i][special_case_term]}"
for i in range(len(special_case_dicts))
]
return f"""[CONTEXT]
prompt = f"""[CONTEXT]
Here is the section of the contract:
{exhibit_text}
@@ -1911,3 +2158,5 @@ Reimbursement: {reimbursement_dict["REIMB_TERM"]}
{special_case_list}
Return the integer of the {special_case_term} that is most associated with the REIMBURSEMENT TERM."""
return (prompt, _json_list_parser)