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

black format

Browse Source
This commit is contained in:
Venkat
2026-02-03 13:41:09 +00:00
parent e8e8f1ba5a
commit 2aa68f75b9
15 changed files with 253 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+7 -9
View File
@@ -141,9 +141,9 @@ def code_category(service, proc_category, hcpcs_level2_mapping, filename):
for key, value in hcpcs_level2_mapping.items()
if key.startswith(category)
]
prompt, _parser = prompt_templates.CODE_CATEGORY(service, matching_values)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
@@ -151,9 +151,7 @@ def code_category(service, proc_category, hcpcs_level2_mapping, filename):
cache=True,
instruction=prompt_templates.CODE_CATEGORY_INSTRUCTION(),
)
llm_answer_final = _parser(
llm_answer_raw
) # Returns list
llm_answer_final = _parser(llm_answer_raw) # Returns list
code_answer_dict = {"PROCEDURE_CD": [], "PROCEDURE_CD_DESC": []}
for answer in llm_answer_final:
@@ -355,8 +353,10 @@ def code_implicit_rag(service, implicit_run_dict, filename, constants):
)
)
prompt, _parser = prompt_templates.CODE_IMPLICIT(service, level_dict["match_list"])
prompt, _parser = prompt_templates.CODE_IMPLICIT(
service, level_dict["match_list"]
)
# Run Prompt
llm_answer_raw = llm_utils.invoke_claude(
prompt,
@@ -458,7 +458,6 @@ def code_last_check(service, filename):
return "Generic"
def fill_bill_type(
service,
answer_dict,
@@ -488,7 +487,6 @@ def fill_bill_type(
service, valid_bill_type, reimb_term=reimb_term, exhibit_text=None
)
# First attempt: Service term + reimbursement term
llm_answer_raw = llm_utils.invoke_claude(
prompt,
src/document_classification/main.py
+1 -1
View File
@@ -314,7 +314,7 @@ if __name__ == "__main__":
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
# Suppress AWS SDK logging
logging.getLogger("botocore").setLevel(logging.WARNING)
logging.getLogger("boto3").setLevel(logging.WARNING)
src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py
+28 -30
View File
@@ -5,7 +5,9 @@ import src.prompts.prompt_templates as prompt_templates
from src.constants.constants import Constants
from src.prompts.fieldset import Field, FieldSet
from src.utils import llm_utils, string_utils, timing_utils, rag_utils
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import (
format_chunks_for_llm,
)
import json
@@ -74,7 +76,9 @@ def prompt_exhibit_level_breakout(
def prompt_dynamic_primary(
exhibit_text: str, field: Field, constants: Constants, filename: str, TEMPLATE
):
prompt, _parser = TEMPLATE(exhibit_text, field.field_name, field.get_prompt(constants))
prompt, _parser = 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,
@@ -284,7 +288,9 @@ def prompt_lob_relationship(
f"LOB: {answer_dict.get("AARETE_DERIVED_LOB")}\nPROGRAM: {programs}"
)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(
exhibit_text, dynamic_primary_values
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
model_id="sonnet_latest",
@@ -292,9 +298,7 @@ def prompt_lob_relationship(
cache=True,
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
@@ -320,7 +324,7 @@ def prompt_special_case_assignment(
]
)
return answer_dict
prompt, _parser = prompt_templates.SPECIAL_CASE_ASSIGNMENT(
exhibit_text, reimbursement_dict, special_case_dicts, special_case_term
)
@@ -334,9 +338,7 @@ def prompt_special_case_assignment(
instruction=prompt_templates.SPECIAL_CASE_ASSIGNMENT_INSTRUCTION(),
usage_label="SPECIAL_CASE_ASSIGNMENT",
)
index_answer = _parser(
llm_answer_raw
)
index_answer = _parser(llm_answer_raw)
try:
if not string_utils.is_empty(index_answer):
return special_case_dicts[int(index_answer)]
@@ -405,7 +407,9 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
return False
service_term, reimb_term = answer_dict["SERVICE_TERM"], answer_dict["REIMB_TERM"]
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(service_term, reimb_term)
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(
service_term, reimb_term
)
logging.debug(f"Prompt for reimbursement validation in {filename}:\n{prompt}")
llm_response = llm_utils.invoke_claude(
@@ -464,9 +468,7 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
"""
prompt, _parser = prompt_templates.EXHIBIT_LINKAGE(page_header, previous_header)
claude_answer_raw = llm_utils.invoke_claude(prompt, "legacy_sonnet", filename)
exhibit_is_different = _parser(
claude_answer_raw
)
exhibit_is_different = _parser(claude_answer_raw)
return exhibit_is_different
@@ -494,9 +496,7 @@ def prompt_exhibit_header(page_content, filename):
instruction=prompt_templates.EXHIBIT_HEADER_INSTRUCTION(),
usage_label="EXHIBIT_HEADER",
)
llm_answer_extracted = _parser(
llm_answer_raw
)
llm_answer_extracted = _parser(llm_answer_raw)
return llm_answer_extracted
@@ -665,9 +665,7 @@ def prompt_lesser_of_distribution(
)
return reimb_term # Return original term unchanged
else:
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
@@ -702,7 +700,9 @@ def prompt_lesser_of_check(
f"LESSER_OF_CHECK input: service='{service_term[:50]}...', reimb_term='{reimb_term[:100]}...'"
)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(service_term, reimb_term, exhibit_title)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(
service_term, reimb_term, exhibit_title
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
@@ -714,9 +714,7 @@ def prompt_lesser_of_check(
try:
# Extract JSON from pipes
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
@@ -790,9 +788,7 @@ def prompt_exhibit_title_match(
llm_answer_raw = llm_utils.invoke_claude(prompt, "sonnet_latest", filename)
try:
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final.strip().upper() # Return "YES" or "NO" as string
except:
return "NO" # Default to "NO" string
@@ -811,9 +807,11 @@ def provider_name_match_check(
return False
# Create prompt for LLM
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = (
prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
)
)
try:
src/pipelines/clients/clover/prompts/prompt_calls.py
+32 -42
View File
@@ -5,7 +5,9 @@ import src.prompts.prompt_templates as prompt_templates
from src.constants.constants import Constants
from src.prompts.fieldset import Field, FieldSet
from src.utils import llm_utils, string_utils, timing_utils, rag_utils
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import (
format_chunks_for_llm,
)
import json
@@ -81,7 +83,9 @@ def prompt_exhibit_level_breakout(
def prompt_dynamic_primary(
exhibit_text: str, field: Field, constants: Constants, filename: str, TEMPLATE
):
prompt, _parser = TEMPLATE(exhibit_text, field.field_name, field.get_prompt(constants))
prompt, _parser = 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,
@@ -91,9 +95,7 @@ def prompt_dynamic_primary(
instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(),
)
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
@@ -245,9 +247,7 @@ def prompt_carveout_check(
usage_label="CARVEOUT_CHECK",
)
carveout_answer = _parser(
llm_answer_raw
)
carveout_answer = _parser(llm_answer_raw)
return carveout_answer
@@ -295,7 +295,9 @@ def prompt_lob_relationship(
f"LOB: {answer_dict.get("AARETE_DERIVED_LOB")}\nPROGRAM: {programs}"
)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(
exhibit_text, dynamic_primary_values
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
model_id="sonnet_latest",
@@ -303,9 +305,7 @@ def prompt_lob_relationship(
cache=True,
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
@@ -331,7 +331,7 @@ def prompt_special_case_assignment(
]
)
return answer_dict
prompt, _parser = prompt_templates.SPECIAL_CASE_ASSIGNMENT(
exhibit_text, reimbursement_dict, special_case_dicts, special_case_term
)
@@ -345,9 +345,7 @@ def prompt_special_case_assignment(
instruction=prompt_templates.SPECIAL_CASE_ASSIGNMENT_INSTRUCTION(),
usage_label="SPECIAL_CASE_ASSIGNMENT",
)
index_answer = _parser(
llm_answer_raw
)
index_answer = _parser(llm_answer_raw)
try:
if not string_utils.is_empty(index_answer):
return special_case_dicts[int(index_answer)]
@@ -435,7 +433,9 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
return False
service_term, reimb_term = answer_dict["SERVICE_TERM"], answer_dict["REIMB_TERM"]
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(service_term, reimb_term)
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(
service_term, reimb_term
)
logging.debug(f"Prompt for reimbursement validation in {filename}:\n{prompt}")
llm_response = llm_utils.invoke_claude(
@@ -448,11 +448,7 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
logging.debug(
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
)
final_answer = (
_parser(llm_response)
.strip()
.upper()
)
final_answer = _parser(llm_response).strip().upper()
return final_answer == "YES"
@@ -505,9 +501,7 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
instruction=prompt_templates.EXHIBIT_LINKAGE_INSTRUCTION(),
usage_label="EXHIBIT_LINKAGE",
)
exhibit_is_different = _parser(
claude_answer_raw
)
exhibit_is_different = _parser(claude_answer_raw)
return exhibit_is_different
@@ -535,9 +529,7 @@ def prompt_exhibit_header(page_content, filename):
instruction=prompt_templates.EXHIBIT_HEADER_INSTRUCTION(),
usage_label="EXHIBIT_HEADER",
)
llm_answer_extracted = _parser(
llm_answer_raw
)
llm_answer_extracted = _parser(llm_answer_raw)
return llm_answer_extracted
@@ -713,9 +705,7 @@ def prompt_lesser_of_distribution(
)
return reimb_term # Return original term unchanged
else:
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
@@ -750,7 +740,9 @@ def prompt_lesser_of_check(
f"LESSER_OF_CHECK input: service='{service_term[:50]}...', reimb_term='{reimb_term[:100]}...'"
)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(service_term, reimb_term, exhibit_title)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(
service_term, reimb_term, exhibit_title
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
@@ -761,9 +753,7 @@ def prompt_lesser_of_check(
)
try:
# Extract JSON from pipes
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
@@ -844,9 +834,7 @@ def prompt_exhibit_title_match(
)
try:
llm_answer_final = _parser(
llm_answer_raw
)
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final.strip().upper() # Return "YES" or "NO" as string
except:
return "NO" # Default to "NO" string
@@ -865,9 +853,11 @@ def provider_name_match_check(
return False
# Create prompt for LLM
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = (
prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
)
)
try:
@@ -905,4 +895,4 @@ def provider_name_match_check(
logging.error(
f"[is_provider_name_match_llm] Error calling LLM: {e}, defaulting to False"
)
return False
return False
src/pipelines/saas/prompts/prompt_calls.py
+24 -12
View File
@@ -5,7 +5,9 @@ import src.prompts.prompt_templates as prompt_templates
from src.constants.constants import Constants
from src.prompts.fieldset import Field, FieldSet
from src.utils import llm_utils, string_utils, timing_utils, rag_utils
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm
from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import (
format_chunks_for_llm,
)
import json
@@ -80,7 +82,9 @@ def prompt_exhibit_level_breakout(
def prompt_dynamic_primary(
exhibit_text: str, field: Field, constants: Constants, filename: str, TEMPLATE
):
prompt, _parser = TEMPLATE(exhibit_text, field.field_name, field.get_prompt(constants))
prompt, _parser = 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,
@@ -91,7 +95,7 @@ def prompt_dynamic_primary(
)
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
@@ -291,7 +295,9 @@ def prompt_lob_relationship(
f"LOB: {answer_dict.get("AARETE_DERIVED_LOB")}\nPROGRAM: {programs}"
)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values)
prompt, _parser = prompt_templates.LOB_RELATIONSHIP(
exhibit_text, dynamic_primary_values
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
model_id="sonnet_latest",
@@ -325,7 +331,7 @@ def prompt_special_case_assignment(
]
)
return answer_dict
prompt, _parser = prompt_templates.SPECIAL_CASE_ASSIGNMENT(
exhibit_text, reimbursement_dict, special_case_dicts, special_case_term
)
@@ -407,7 +413,9 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
return False
service_term, reimb_term = answer_dict["SERVICE_TERM"], answer_dict["REIMB_TERM"]
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(service_term, reimb_term)
prompt, _parser = prompt_templates.VALIDATE_REIMBURSEMENTS_PROMPT(
service_term, reimb_term
)
logging.debug(f"Prompt for reimbursement validation in {filename}:\n{prompt}")
llm_response = llm_utils.invoke_claude(
@@ -474,7 +482,7 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
usage_label="EXHIBIT_LINKAGE",
)
exhibit_is_different = _parser(claude_answer_raw)
return exhibit_is_different
@@ -713,7 +721,9 @@ def prompt_lesser_of_check(
f"LESSER_OF_CHECK input: service='{service_term}...', reimb_term='{reimb_term}...'"
)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(service_term, reimb_term, exhibit_title)
prompt, _parser = prompt_templates.LESSER_OF_CHECK(
service_term, reimb_term, exhibit_title
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
"sonnet_latest",
@@ -820,9 +830,11 @@ def provider_name_match_check(
return False
# Create prompt for LLM
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
CHECK_PROVIDER_NAME_MATCH_PROMPT, _parser = (
prompt_templates.CHECK_PROVIDER_NAME_MATCH_PROMPT(
provider_name=provider_name,
hybrid_smart_chunking_provider_group_name=hybrid_smart_chunking_provider_group_name,
)
)
try:
@@ -862,6 +874,7 @@ def provider_name_match_check(
)
return False
def prompt_split_reimb_dates(date_range: str, filename: str) -> dict:
prompt, _parser = prompt_templates.SPLIT_REIMB_DATES(date_range)
@@ -879,4 +892,3 @@ def prompt_split_reimb_dates(date_range: str, filename: str) -> dict:
# Extract the effective and termination dates from the LLM output
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
src/pipelines/saas/test.py
+4 -3
View File
@@ -1,4 +1,3 @@
from src.pipelines.shared.postprocessing import aarete_derived
from src.constants.constants import Constants
from src.utils import io_utils
@@ -9,10 +8,12 @@ from src.pipelines.shared.extraction.one_to_n_funcs import reimbursement_level
service_term = "Covered Services provided to KIDSfirst Members"
reimb_term = "One hundred percent (100%) of the prevailing yearly and current Medicaid fee schedule for the State of Texas or the Participating Ancillary Services Provider's usual and customary charge, whichever is less."
exhibit_title = 'Attachment B SECTION 1 - PCHP MEDICAID STAR HMO SERVICES'
exhibit_title = "Attachment B SECTION 1 - PCHP MEDICAID STAR HMO SERVICES"
filename = ""
lesser_of_answer_dict = prompt_calls.prompt_lesser_of_check(service_term, reimb_term, exhibit_title, filename)
lesser_of_answer_dict = prompt_calls.prompt_lesser_of_check(
service_term, reimb_term, exhibit_title, filename
)
print(lesser_of_answer_dict)
src/pipelines/shared/extraction/dynamic_funcs.py
+1 -2
View File
@@ -64,14 +64,13 @@ def dynamic_primary(
if has_medicare and has_medicaid:
if "duals" not in values_lower:
if exhibit_text_answer:
exhibit_text_answer.append("Duals")
exhibit_text_answer.append("Duals")
else:
exhibit_text_answer = ["Duals"]
logging.debug(
f"Added 'Duals' to LOB valid values because both Medicare and Medicaid were detected"
)
field.update_valid_values(exhibit_text_answer + ["N/A"])
dynamic_reimbursement_fields.add_field(field)
dynamic_primary_fields.remove_field(field.field_name)
src/pipelines/shared/extraction/one_to_n_funcs.py
+7 -3
View File
@@ -97,7 +97,7 @@ def reimbursement_level(
reimbursement_primary_answers = clean_reimbursement_primary(
reimbursement_primary_answers, constants, filename
)
return reimbursement_primary_answers
@@ -675,7 +675,9 @@ def split_reimb_dates(one_to_n_results: list, filename: str) -> list:
date_range = str(record["REIMB_DATES"]).strip()
try:
llm_answer_final = prompt_calls.prompt_split_reimb_dates(date_range, filename)
llm_answer_final = prompt_calls.prompt_split_reimb_dates(
date_range, filename
)
# Update the specific record with parsed dates
if llm_answer_final and isinstance(llm_answer_final, dict):
if "start_date" in llm_answer_final:
@@ -862,7 +864,9 @@ def lesser_of_distribution(
# STANDALONE: Complete rate - ADD to results
elif scope == "STANDALONE":
logging.debug(f"STANDALONE rate from page {page_num} - ADDED to results")
logging.debug(
f"STANDALONE rate from page {page_num} - ADDED to results"
)
final_reimbursement_level_answers.append(
answer_dict
) # ← Only this gets added
src/pipelines/shared/extraction/tin_npi_funcs.py
+6 -3
View File
@@ -203,6 +203,7 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
cleaned_providers.append(cleaned_provider)
return cleaned_providers
def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename):
"""
Merges provider_info into one_to_one_results.
@@ -223,8 +224,10 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
other_npis = []
other_names = []
provider_name = one_to_one_results.get("PROVIDER_NAME") # Returns List
provider_info = one_to_one_results["PROV_INFO_JSON"] # Returns string representation of list
provider_name = one_to_one_results.get("PROVIDER_NAME") # Returns List
provider_info = one_to_one_results[
"PROV_INFO_JSON"
] # Returns string representation of list
prov_info_json_list = (
json.loads(provider_info) if isinstance(provider_info, str) else provider_info
)
@@ -234,7 +237,7 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
for provider in prov_info_json_list:
name_matches = prompt_calls.provider_name_match_check(
provider_name, provider.get("NAME", "UNKNOWN"), filename
) # Returns Boolean
) # Returns Boolean
if name_matches:
found_match = True
src/pipelines/shared/postprocessing/aarete_derived.py
+14 -15
View File
@@ -1,4 +1,4 @@
import logging
import src.config as config
import src.utils.string_utils as string_utils
from src.constants.constants import Constants
@@ -33,7 +33,7 @@ def fill_na_mapping(answer_dicts):
# Get existing AARETE_DERIVED_LOB values (if any)
existing_lob_list = answer_dict.get("AARETE_DERIVED_LOB", "")
if not string_utils.is_empty(existing_lob_list):
for lob_val in existing_lob_list:
if "|" in existing_lob_list:
@@ -55,7 +55,8 @@ def fill_na_mapping(answer_dicts):
"src/constants/mappings/crosswalk_program_lob.json",
)
if (
not string_utils.is_empty(program_filled_value_list) and "N/A" not in program_filled_value_list
not string_utils.is_empty(program_filled_value_list)
and "N/A" not in program_filled_value_list
):
for program_lob_val in program_filled_value_list:
if "|" in program_lob_val:
@@ -64,10 +65,7 @@ def fill_na_mapping(answer_dicts):
for v in program_lob_val.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
program_lob_val.strip()
and program_lob_val.strip() != "N/A"
):
elif program_lob_val.strip() and program_lob_val.strip() != "N/A":
program_lob_values = {program_lob_val.strip()}
all_lob_values.update(program_lob_values)
@@ -91,16 +89,13 @@ def fill_na_mapping(answer_dicts):
for v in product_filled_value_list.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
product_lob_val.strip()
and product_lob_val.strip() != "N/A"
):
elif product_lob_val.strip() and product_lob_val.strip() != "N/A":
product_lob_values = {product_lob_val.strip()}
all_lob_values.update(product_lob_values)
# Set the merged, deduplicated value
if all_lob_values:
answer_dict["AARETE_DERIVED_LOB"] = "|".join(sorted(all_lob_values))
answer_dict["AARETE_DERIVED_LOB"] = sorted(all_lob_values)
# Set relationship flags if values came from PROGRAM or PRODUCT
# Only set if field is empty or "N/A" (preserve LLM-determined Inclusive/Exclusive)
if program_lob_values:
@@ -145,9 +140,9 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
to_field_value = answer_dict.get(to_field_name)
from_field_value = answer_dict.get(from_field_name)
# If from_field_value is populated and to_field_value is not populated, perform mapping
if not string_utils.is_empty(
from_field_value
) and (string_utils.is_empty(to_field_value) or "N/A" in to_field_value):
if not string_utils.is_empty(from_field_value) and (
string_utils.is_empty(to_field_value) or "N/A" in to_field_value
):
if isinstance(from_field_value, list):
from_field_value_list = from_field_value
else:
@@ -176,4 +171,8 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
)
)
answer_dict[to_field_name] = to_field_answer_list
# update from field value to list if not already
if not isinstance(from_field_value, list):
answer_dict[from_field_name] = from_field_value_list
return answer_dicts
src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py
+5 -3
View File
@@ -328,6 +328,7 @@ def extract_amendment_num_from_filename(answer_dict: dict, filename: str) -> dic
return answer_dict
def prompt_hsc_single_field(
field, retriever, reranker, rag_config, text_dict, constants, filename
):
@@ -365,7 +366,9 @@ def prompt_hsc_single_field(
llm_prompt = field.get_prompt_dict(
constants
) # Returns dict of {field_name : prompt}
prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, llm_prompt)
prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(
context, llm_prompt
)
logging.debug(f"Field: {field.field_name}")
logging.debug(f"Retrieval question: {retrieval_query}")
@@ -384,7 +387,7 @@ def prompt_hsc_single_field(
field_value = llm_answer_final.get(field.field_name)
if not isinstance(field_value, list):
field_value = [field_value]
if field_value[0] == "N/A":
field.field_type = "full_context"
if field.field_name == "PAYER_NAME":
@@ -412,4 +415,3 @@ def prompt_hsc_single_field(
f"Error on {field.field_name}: {type(e).__name__}: {str(e)}, marking as full_context"
)
return (field.field_name, "N/A", field)
src/prompts/prompt_templates.py
+114 -92
View File
@@ -138,9 +138,6 @@ def get_cacheable_instructions():
if config.FIELDS in ["all", "one_to_n"]:
try:
cacheable_instructions["DYNAMIC_PRIMARY"] = DYNAMIC_PRIMARY_INSTRUCTION()
cacheable_instructions["DYNAMIC_PRIMARY_TEXT"] = (
DYNAMIC_PRIMARY_TEXT_INSTRUCTION()
)
cacheable_instructions["CARVEOUT_CHECK"] = CARVEOUT_CHECK_INSTRUCTION()
cacheable_instructions["DYNAMIC_ASSIGNMENT"] = (
DYNAMIC_ASSIGNMENT_INSTRUCTION()
@@ -240,7 +237,7 @@ Briefly explain your answer, then put the final answer in the properly-formatted
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.
"""
@@ -251,7 +248,7 @@ Here are the attributes to be included in the dictionary, and instructions on ho
[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_dict_parser)
@@ -273,10 +270,12 @@ Briefly explain your answer before putting the final answer in a properly-format
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def DYNAMIC_PRIMARY(context, field_name, field_prompt) -> Tuple[str, Callable[[str], list]]:
def DYNAMIC_PRIMARY(
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.
"""
@@ -286,7 +285,7 @@ def DYNAMIC_PRIMARY(context, field_name, field_prompt) -> Tuple[str, Callable[[s
[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_list_parser)
@@ -369,7 +368,7 @@ def REIMB_DATES_ASSIGNMENT(
) -> 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.
"""
@@ -386,7 +385,7 @@ 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)
@@ -459,7 +458,7 @@ def DYNAMIC_ASSIGNMENT(
"""
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.
"""
@@ -477,7 +476,7 @@ 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)
@@ -485,7 +484,7 @@ 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.
"""
@@ -496,7 +495,7 @@ def REIMBURSEMENT_PRIMARY(context) -> Tuple[str, Callable[[str], list]]:
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)
@@ -618,7 +617,7 @@ def LESSER_OF_DISTRIBUTION(
"""
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.
"""
@@ -650,7 +649,7 @@ EXHIBIT TEXT:
Analyze the inputs above. Briefly explain your reasoning, then provide your final answer as a JSON list.
"""
return (prompt, _json_list_parser)
@@ -738,18 +737,20 @@ Briefly explain your reasoning, then return a properly-formatted JSON dictionary
{JSON_DICT_FORMAT_INSTRUCTIONS}"""
def LESSER_OF_CHECK(service_term: str, reimb_term: str, exhibit_title: str) -> Tuple[str, Callable[[str], dict]]:
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.
"""
prompt = f"""**Current Exhibit:** {exhibit_title}
**Service:** {service_term}
**Reimbursement:** {reimb_term}"""
return (prompt, _json_dict_parser)
@@ -774,17 +775,19 @@ Examples: ["YES"] or ["NO"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_TITLE_MATCH(current_exhibit_title: str, target_exhibit_reference: str) -> Tuple[str, Callable[[str], list]]:
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.
"""
prompt = f"""[CONTEXT]
Current Exhibit Title: "{current_exhibit_title}"
Target Exhibit Reference: "{target_exhibit_reference}" """
return (prompt, _json_list_parser)
@@ -793,14 +796,16 @@ Target Exhibit Reference: "{target_exhibit_reference}" """
#####################################################################################
def METHODOLOGY_BREAKOUT(service_term: str, reimb_term: str) -> Tuple[str, Callable[[str], list]]:
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.
"""
@@ -809,7 +814,7 @@ 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)
@@ -872,18 +877,20 @@ Populate a JSON dictionary with the following fields:
Briefly explain your reasoning, then write your JSON dictionary."""
def FEE_SCHEDULE_BREAKOUT(methodology, fee_schedule) -> Tuple[str, Callable[[str], dict]]:
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.
"""
prompt = f"""[CONTEXT]
Analyze and respond to the following text, specifically for {fee_schedule} Fee Schedule:
Methodology: {methodology.replace('"', "'")}"""
return (prompt, _json_dict_parser)
@@ -918,7 +925,7 @@ 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.
"""
@@ -926,14 +933,14 @@ def GROUPER_BREAKOUT(service, term) -> Tuple[str, Callable[[str], dict]]:
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) -> 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.
"""
@@ -951,7 +958,7 @@ 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)
@@ -975,7 +982,7 @@ def RATE_ESCALATOR_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1016,21 +1023,21 @@ 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.
"""
prompt = f"""[CONTEXT]
Here is the text to analyze and respond to:
{term}"""
return (prompt, _json_dict_parser)
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.
"""
@@ -1043,7 +1050,7 @@ def FACILITY_ADJUSTMENT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1056,7 +1063,7 @@ def SEQUESTRATION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1069,7 +1076,7 @@ def DISCOUNT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1082,7 +1089,7 @@ def PREMIUM_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1095,7 +1102,7 @@ def ADDITION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1143,7 +1150,7 @@ 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.
"""
@@ -1151,7 +1158,7 @@ def CODE_EXPLICIT(service, methodology) -> Tuple[str, Callable[[str], dict]]:
Here is the Service and Methodology to analyze:
Service: {service.replace('"', "'")}
Methodology: {methodology.replace('"', "'")}"""
return (prompt, _json_dict_parser)
@@ -1174,7 +1181,7 @@ Explain your answer in 1-2 sentences. Write your final answer in a properly-form
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.
"""
@@ -1185,7 +1192,7 @@ Here are the descriptions to choose from:
[CONTEXT]
Here is the Service Term to analyze:
{service.replace('"', "'")}"""
return (prompt, _json_list_parser)
@@ -1210,14 +1217,14 @@ Examples: ["Drugs"], ["Vaccines"], ["PT/OT/ST"], ["PT"], ["Surgery"], ["N/A"]
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)
@@ -1247,7 +1254,7 @@ Explain your answer, ensuring each point in the instructions is addressed. Then
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.
"""
@@ -1258,7 +1265,7 @@ Here are the descriptions to choose from:
[CONTEXT]
Here is the Service Term to analyze:
{service.replace('"', "'")}"""
return (prompt, _json_list_parser)
@@ -1283,13 +1290,13 @@ Examples: ["Specific"] or ["Generic"]
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)
@@ -1313,9 +1320,11 @@ 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) -> Tuple[str, Callable[[str], list]]:
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.
"""
@@ -1344,7 +1353,7 @@ Here are the descriptions to choose from:
[SERVICE TERM TO ANALYZE]
{service.replace('"', "'")}
{reimb_term_section}{exhibit_context_section}"""
return (prompt, _json_list_parser)
@@ -1353,10 +1362,12 @@ Here are the descriptions to choose from:
#####################################################################################
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context: str, field_prompt: dict[str, str]) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1369,7 +1380,7 @@ 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)
@@ -1437,11 +1448,13 @@ FINAL PROVIDER INFO:
Contract text:"""
def TIN_NPI_TEMPLATE(context, questions, payer_name) -> Tuple[str, Callable[[str], list]]:
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).
"""
@@ -1458,7 +1471,7 @@ def TIN_NPI_TEMPLATE(context, questions, payer_name) -> Tuple[str, Callable[[str
Contract text:
{context.replace('"', "'" )}
"""
return (prompt, _json_list_parser)
@@ -1527,7 +1540,7 @@ def CHECK_PROVIDER_NAME_MATCH_PROMPT(
) -> 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.
"""
@@ -1536,14 +1549,16 @@ 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]) -> Tuple[str, Callable[[str], dict]]:
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.
"""
@@ -1587,12 +1602,12 @@ def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str:
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)
@@ -1627,7 +1642,7 @@ def EXHIBIT_HEADER_INSTRUCTION() -> str:
EXHIBIT_HEADER_MARKERS = constants.EXHIBIT_HEADER_MARKERS
markers_text = "\n* ".join(EXHIBIT_HEADER_MARKERS)
return f"""[OBJECTIVE]
Analyze contract excerpts and extract the full exhibit/attachment header found.
Capture the complete hierarchy of document identifiers present in the text.
@@ -1665,14 +1680,14 @@ 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.
"""
prompt = f"""[CONTEXT]
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_list_parser)
@@ -1700,7 +1715,7 @@ Examples: ["Y"] or ["N"]
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.
"""
@@ -1710,7 +1725,7 @@ Header 1:
Header 2:
"{header2}" """
return (prompt, _json_list_parser)
@@ -1788,10 +1803,13 @@ 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) -> Tuple[str, Callable[[str], list]]:
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.
"""
@@ -1808,7 +1826,7 @@ REIMBURSEMENT TERM: {reimb_term}
If 1=yes, 2=yes, 3=no → return ["YES"]
Otherwise → return ["NO"]"""
return (prompt, _json_list_parser)
@@ -1859,14 +1877,14 @@ Output: ["N/A"]
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.
"""
prompt = f"""[CONTEXT]
Please analyze this date:
{context}"""
return (prompt, _json_list_parser)
@@ -1896,7 +1914,7 @@ def CARVEOUT_CHECK_INSTRUCTION() -> str:
special_case_text = "\n\n".join(
[f"{name} : {desc}" for name, desc in special_case_definitions.items()]
)
return f"""[OBJECTIVE]
Examine the Reimbursement Term and identify which case it falls under.
@@ -1919,11 +1937,13 @@ Example: ["OUTLIER_TERM"] or ["RATE_ESCALATOR"]
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def CARVEOUT_CHECK(service_term: str, reimb_term: str) -> Tuple[str, Callable[[str], list]]:
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.
"""
@@ -1931,14 +1951,14 @@ def CARVEOUT_CHECK(service_term: str, reimb_term: str) -> Tuple[str, Callable[[s
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() -> 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.
"""
@@ -1950,7 +1970,7 @@ def EFFECTIVE_DATE_FIX_PROMPT() -> Tuple[str, Callable[[str], dict]]:
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)
@@ -1995,13 +2015,13 @@ Note: Use "N/A" for dates that cannot be determined or extracted from the input
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)
@@ -2034,10 +2054,12 @@ Your response MUST include the JSON list with your final answer.
{JSON_LIST_FORMAT_INSTRUCTIONS}"""
def LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values) -> Tuple[str, Callable[[str], list]]:
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.
"""
@@ -2048,7 +2070,7 @@ It was previously identified that the Exhibit contains the following LOB and Pro
[CONTEXT]
Here is the Exhibit to be analyzed:
{exhibit_text.replace('"', "'")}"""
return (prompt, _json_list_parser)
@@ -2078,14 +2100,14 @@ def DERIVED_TERM_DATE_PROMPT(
) -> 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.
"""
prompt = f"""[CONTEXT]
Effective Date: {effective_date}
Termination Information: {termination_information}"""
return (prompt, _json_list_parser)
@@ -2119,7 +2141,7 @@ def SPECIAL_CASE_ASSIGNMENT(
) -> 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.
"""
@@ -2139,5 +2161,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)
return (prompt, _json_list_parser)
src/tests/test_json_parsers.py
+6 -6
View File
@@ -53,12 +53,12 @@ class TestParseJsonDict:
def test_dict_with_newlines(self):
"""Test parsing dict with newlines in the JSON."""
output = '''
output = """
{
"field": "value",
"count": 42
}
'''
"""
result = parse_json_dict(output)
assert result == {"field": "value", "count": 42}
@@ -181,13 +181,13 @@ class TestParseJsonList:
def test_list_with_newlines(self):
"""Test parsing list with newlines in the JSON."""
output = '''
output = """
[
"item1",
"item2",
"item3"
]
'''
"""
result = parse_json_list(output)
assert result == ["item1", "item2", "item3"]
@@ -317,11 +317,11 @@ class TestEdgeCases:
def test_json_in_markdown_code_block(self):
"""Test parsing JSON within markdown code blocks."""
output = '''
output = """
```json
{"field": "value"}
```
'''
"""
result = parse_json_dict(output)
assert result == {"field": "value"}
src/utils/json_utils.py
+3 -9
View File
@@ -43,9 +43,7 @@ def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
{'key': 'value'}
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
raise TypeError(f"Expected string input, got {type(raw_llm_output).__name__}")
found_dicts = []
i = 0
@@ -130,9 +128,7 @@ def parse_json_list(raw_llm_output: str) -> list[Any]:
['item1', 'item2']
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
raise TypeError(f"Expected string input, got {type(raw_llm_output).__name__}")
found_lists = []
i = 0
@@ -207,9 +203,7 @@ def parse_json_dict_or_list(raw_llm_output: str) -> dict[str, Any] | list[Any]:
TypeError: If raw_llm_output is not a string.
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
raise TypeError(f"Expected string input, got {type(raw_llm_output).__name__}")
found_objects = []
i = 0
src/utils/logging_utils.py
+1 -1
View File
@@ -218,7 +218,7 @@ def setup_per_file_logging() -> None:
]
for logger_name in noisy_loggers:
logging.getLogger(logger_name).setLevel(logging.WARNING)
# Specifically suppress bedrock-related loggers
logging.getLogger("botocore.handlers").setLevel(logging.WARNING)
logging.getLogger("botocore.endpoint").setLevel(logging.WARNING)